"""
submit-pull-request Skill 的 Git、GitHub 和路径安全共享逻辑。
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import secrets
import subprocess
import sys
import tempfile
import time
import urllib.parse
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any, Iterator, Optional


def _find_repo_root() -> Path:
    """
    从当前工作目录或 Skill 路径定位 MoviePilot 源码根目录。
    """
    script_path = Path(__file__).resolve()
    candidates = []
    configured_root = os.environ.get("MOVIEPILOT_ROOT", "").strip()
    if configured_root:
        candidates.append(Path(configured_root).expanduser())
    candidates.extend([Path.cwd().resolve(), *Path.cwd().resolve().parents])
    candidates.extend([script_path.parent, *script_path.parents])
    for candidate in candidates:
        if (candidate / "app" / "runtime" / "config.py").is_file():
            return candidate
    raise RuntimeError(
        "无法定位 MoviePilot 程序目录；请在程序目录执行，或设置 MOVIEPILOT_ROOT="
        "<程序目录> 后重试。"
    )


SCRIPT_ROOT = _find_repo_root()
if str(SCRIPT_ROOT) not in sys.path:
    sys.path.insert(0, str(SCRIPT_ROOT))

# 脚本先补充本地模块或项目根目录，再导入对应依赖。
from app.adapters.network.http import RequestUtils  # noqa: E402
# 脚本先补充本地模块或项目根目录，再导入对应依赖。
from app.runtime.config import settings  # noqa: E402
# 脚本先补充本地模块或项目根目录，再导入对应依赖。
from app.runtime.settings import get_runtime_setting  # noqa: E402

GITHUB_API_BASE = "https://api.github.com"
GITHUB_API_VERSION = "2022-11-28"
DEFAULT_TIMEOUT_SECONDS = 30
MAX_FILE_BYTES = 10 * 1024 * 1024
MAX_CHANGED_FILES = 200

_REPO_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
_BRANCH_PATTERN = re.compile(r"^[A-Za-z0-9._/-]+$")
_TOKEN_PATTERN = re.compile(r"(?i)(bearer|token)\s+[^\s,;]+")
_SSH_REPO_PATTERN = re.compile(r"^(?:[^@]+@)?([^:]+):([^/]+/[^/]+?)(?:\.git)?$")

_REJECTED_PATH_PARTS = {
    ".git",
    ".env",
    "config",
    "data",
    "cache",
    "logs",
    "tmp",
    "__pycache__",
    ".pytest_cache",
    ".mypy_cache",
    ".ruff_cache",
    "node_modules",
}
_REJECTED_SUFFIXES = {
    ".db",
    ".sqlite",
    ".sqlite3",
    ".log",
    ".key",
    ".pem",
    ".p12",
    ".pfx",
    ".secret",
}


class PullRequestError(RuntimeError):
    """
    PR 发布流程遇到的可向 Agent 返回的稳定错误。
    """

    def __init__(self, reason: str, message: str, *, status: int = 0) -> None:
        """
        保存机器可判断的原因、用户可读消息和可选 HTTP 状态。
        """
        super().__init__(message)
        self.reason = reason
        self.message = message
        self.status = status


class GitHubApiError(PullRequestError):
    """
    GitHub API 返回非成功状态或无法解析响应。
    """


@dataclass(frozen=True)
class Change:
    """
    表示一个待提交文件的最终内容和 Git 工作树摘要。

    :param path: 目标仓库内的相对路径
    :param source_path: 本地待提交内容的绝对路径
    :param operation: 文件操作类型：create 创建、update 更新、delete 删除
    :param before_sha256: 本地克隆仓库 HEAD 中文件内容的摘要；新建或无法读取时为 None
    :param after_sha256: 本地最终内容摘要；删除时为 None
    :param size_bytes: 本地最终内容的字节数；删除时为 0
    :param mode: Git 文件模式，普通文件为 100644，可执行文件为 100755；删除项使用 100644
    """

    path: str
    source_path: Path
    operation: str
    before_sha256: Optional[str]
    after_sha256: Optional[str]
    size_bytes: int
    mode: str


def result_payload(**payload: Any) -> str:
    """
    将脚本结果序列化为稳定 JSON，供 Agent 解析。
    """
    return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)


def redact_text(value: Any) -> str:
    """
    去除错误信息中可能出现的 Authorization 凭据。
    """
    text = str(value or "")
    return _TOKEN_PATTERN.sub(r"\1 <REDACTED>", text)


def normalize_repo(repo: str) -> str:
    """
    规范化并校验 GitHub owner/repo 标识。
    """
    value = (repo or "").strip().removesuffix(".git")
    if value.startswith(("http://", "https://")):
        parsed = urllib.parse.urlparse(value)
        if parsed.netloc.lower() not in {"github.com", "www.github.com"}:
            raise ValueError("目标仓库必须位于 GitHub")
        parts = [part for part in parsed.path.strip("/").split("/") if part]
        if len(parts) >= 2:
            value = f"{parts[0]}/{parts[1]}"
    if not _REPO_PATTERN.fullmatch(value):
        raise ValueError("目标仓库必须是 owner/repo 或 GitHub URL")
    return value


def normalize_branch(branch: str) -> str:
    """
    校验分支名，拒绝 ref 前缀、空段和路径穿越。
    """
    value = (branch or "").strip()
    if value.startswith("refs/") or not _BRANCH_PATTERN.fullmatch(value):
        raise ValueError("分支名包含不支持的字符")
    if value.startswith("/") or value.endswith("/") or ".." in value or "//" in value:
        raise ValueError("分支名不能包含空段或连续点")
    return value


def normalize_relative_path(path: str) -> str:
    """
    将文件路径规范化为仓库内 POSIX 相对路径。
    """
    raw_value = str(path or "").replace("\\", "/")
    if not raw_value or raw_value.startswith("/") or re.match(r"^[A-Za-z]:/", raw_value):
        raise ValueError(f"文件路径不在源目录内: {path}")
    value = raw_value.strip("/")
    pure_path = PurePosixPath(value)
    if not value or value == "." or pure_path.is_absolute() or ".." in pure_path.parts:
        raise ValueError(f"文件路径不在源目录内: {path}")
    return pure_path.as_posix()


def is_rejected_path(path: str) -> bool:
    """
    判断路径是否指向运行时状态、缓存或明显敏感文件。
    """
    normalized = normalize_relative_path(path)
    parts = set(PurePosixPath(normalized).parts)
    if parts & _REJECTED_PATH_PARTS:
        return True
    return PurePosixPath(normalized).suffix.lower() in _REJECTED_SUFFIXES


def sha256_bytes(content: bytes) -> str:
    """
    返回原始字节的 SHA-256 摘要。
    """
    return hashlib.sha256(content).hexdigest()


def sha256_file(path: Path) -> str:
    """
    按块计算文件摘要，避免把大文件一次性载入内存。
    """
    digest = hashlib.sha256()
    with path.open("rb") as file_handle:
        for chunk in iter(lambda: file_handle.read(64 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def runtime_dir() -> Path:
    """
    返回普通 Agent 用户可读写的 Skill 工作目录，不把草稿写入源码仓库。
    """
    directory = Path(get_runtime_setting("CONFIG_PATH")) / "agent" / "submit-pull-request"
    directory.mkdir(parents=True, exist_ok=True)
    return directory


def write_json_file(path: Path, payload: dict[str, Any]) -> None:
    """
    以原子替换方式写入 Skill 运行状态 JSON。
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{secrets.token_hex(4)}.tmp")
    try:
        temporary.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def read_json_file(path: Path, *, reason: str = "invalid_state") -> dict[str, Any]:
    """
    读取并校验 Skill 运行状态 JSON。
    """
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise PullRequestError(reason, f"无法读取运行状态文件: {path}") from error
    if not isinstance(payload, dict):
        raise PullRequestError(reason, f"运行状态文件不是 JSON 对象: {path}")
    return payload


def _run_git(
    root: Path,
    *arguments: str,
    env: Optional[dict[str, str]] = None,
) -> subprocess.CompletedProcess[bytes]:
    """
    运行不经过 shell 的 Git 命令，避免用户输入触发命令注入。
    """
    try:
        return subprocess.run(
            ["git", *arguments],
            cwd=root,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
    except OSError as error:
        raise PullRequestError("git_unavailable", "当前环境没有可用的 Git 命令。") from error


def _git_output(root: Path, *arguments: str, env: Optional[dict[str, str]] = None) -> bytes:
    """
    运行 Git 并在失败时返回不含凭据的稳定错误。
    """
    result = _run_git(root, *arguments, env=env)
    if result.returncode != 0:
        detail = redact_text(result.stderr.decode("utf-8", errors="replace"))[:500]
        raise PullRequestError(
            "git_command_failed",
            f"Git 命令失败: {arguments[0]}{(': ' + detail) if detail else ''}",
        )
    return result.stdout


def run_git(
    root: Path,
    *arguments: str,
    env: Optional[dict[str, str]] = None,
) -> subprocess.CompletedProcess[bytes]:
    """
    运行 Git 并返回完整结果，供克隆、提交和推送脚本复用。
    """
    return _run_git(root, *arguments, env=env)


def git_output(root: Path, *arguments: str, env: Optional[dict[str, str]] = None) -> bytes:
    """
    运行 Git 并要求命令成功，供需要稳定失败原因的脚本复用。
    """
    return _git_output(root, *arguments, env=env)


def find_git_root(source_root: Path) -> Optional[Path]:
    """
    返回源目录对应的 Git 根目录；没有仓库时返回 None。
    """
    result = _run_git(source_root, "rev-parse", "--show-toplevel")
    if result.returncode != 0:
        return None
    try:
        root = Path(result.stdout.decode("utf-8").strip()).resolve()
    except OSError:
        return None
    return root if root.is_dir() else None


def require_git_root(source_root: Path) -> Path:
    """
    要求源目录本身是克隆的 Git 根目录。
    """
    source_root = source_root.expanduser().resolve()
    if not source_root.is_dir():
        raise PullRequestError("invalid_source_root", f"源目录不存在或不是目录: {source_root}")
    git_root = find_git_root(source_root)
    if git_root is None:
        raise PullRequestError("git_required", "PR 提交必须基于先前 git clone 的仓库目录。")
    if git_root != source_root:
        raise PullRequestError(
            "source_root_must_be_git_root",
            f"请把 clone 的仓库根目录作为 source_root: {git_root}",
        )
    return git_root


def _git_sha256(root: Path, path: str) -> Optional[str]:
    """
    读取 HEAD 中的文件内容并计算摘要，用于记录变更前版本。
    """
    result = _run_git(root, "show", f"HEAD:{path}")
    if result.returncode != 0:
        return None
    return sha256_bytes(result.stdout)


def _file_mode(path: Path) -> str:
    """
    返回普通文件或可执行文件对应的 Git 文件模式。
    """
    return "100755" if path.is_file() and path.stat().st_mode & 0o111 else "100644"


def _git_changes(root: Path) -> list[dict[str, Any]]:
    """
    收集 HEAD 到当前工作树的变更，包括未跟踪文件。
    """
    result = _run_git(root, "diff", "--name-status", "--no-renames", "-z", "HEAD", "--")
    if result.returncode != 0:
        raise PullRequestError("git_status_failed", "无法读取当前 Git 工作树变更。")

    changes: dict[str, str] = {}
    tokens = result.stdout.decode("utf-8", errors="surrogateescape").split("\0")
    index = 0
    while index + 1 < len(tokens):
        status, path = tokens[index], tokens[index + 1]
        index += 2
        if status and path:
            changes[normalize_relative_path(path)] = status[0]

    untracked = _run_git(root, "ls-files", "--others", "--exclude-standard", "-z")
    if untracked.returncode != 0:
        raise PullRequestError("git_status_failed", "无法读取未跟踪文件列表。")
    for path in untracked.stdout.decode("utf-8", errors="surrogateescape").split("\0"):
        if path:
            changes[normalize_relative_path(path)] = "A"

    result_changes = []
    for path, status in sorted(changes.items()):
        if is_rejected_path(path):
            raise PullRequestError("sensitive_path", f"拒绝提交运行时或敏感路径: {path}")
        local_path = root / path
        if local_path.is_symlink():
            raise PullRequestError("unsupported_change", f"拒绝提交符号链接文件: {path}")
        source_path = local_path.resolve(strict=False)
        if not _is_within(source_path, root):
            raise PullRequestError("path_outside_root", f"文件路径超出源目录: {path}")
        exists = source_path.is_file()
        if status == "D" or not exists:
            if status != "D":
                raise PullRequestError("unsupported_change", f"变更不是普通文件: {path}")
            operation = "delete"
        else:
            operation = "create" if status == "A" else "update"
        result_changes.append(
            {
                "path": path,
                "source_path": str(source_path),
                "operation": operation,
                "before_sha256": None if operation == "create" else _git_sha256(root, path),
                "after_sha256": sha256_file(source_path) if exists else None,
                "size_bytes": source_path.stat().st_size if exists else 0,
                "mode": _file_mode(source_path),
            }
        )
    return result_changes


def _is_within(path: Path, root: Path) -> bool:
    """
    判断解析后的路径是否仍位于指定根目录。
    """
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False


def git_current_branch(root: Path) -> str:
    """
    读取当前克隆的分支名，拒绝 detached HEAD。
    """
    branch = _git_output(root, "symbolic-ref", "--quiet", "--short", "HEAD").decode().strip()
    if not branch:
        raise PullRequestError("detached_head", "当前 clone 处于 detached HEAD，不能安全提交 PR。")
    try:
        return normalize_branch(branch)
    except ValueError as error:
        raise PullRequestError("invalid_branch", str(error)) from error


def git_remote_url(root: Path, remote: str = "origin") -> str:
    """
    读取指定 Git 远端地址。
    """
    return _git_output(root, "remote", "get-url", remote).decode().strip()


def remote_repo_from_url(remote_url: str) -> str:
    """
    从 HTTPS、SSH 或 scp 风格远端地址提取 GitHub 仓库名。
    """
    value = (remote_url or "").strip()
    if value.startswith(("http://", "https://", "ssh://")):
        parsed = urllib.parse.urlparse(value)
        if parsed.hostname and parsed.hostname.lower() not in {"github.com", "www.github.com"}:
            raise PullRequestError("unsupported_remote", "Git remote 必须指向 GitHub。")
        return normalize_repo(parsed.path.strip("/"))
    match = _SSH_REPO_PATTERN.fullmatch(value)
    if not match or match.group(1).lower() not in {"github.com", "www.github.com"}:
        raise PullRequestError("unsupported_remote", "无法确认 Git remote 的 GitHub 仓库。")
    return normalize_repo(match.group(2))


def collect_changes(source_root: Path) -> tuple[str, list[dict[str, Any]]]:
    """
    只从先前克隆的 Git 工作树收集代码变更。
    """
    git_root = require_git_root(source_root)
    changes = _git_changes(git_root)
    if not changes:
        raise PullRequestError("no_changes", "没有检测到可提交的 Git 工作树变更。")
    if len(changes) > MAX_CHANGED_FILES:
        raise PullRequestError("too_many_files", f"单次 PR 最多允许 {MAX_CHANGED_FILES} 个文件。")
    return "git", changes


def change_fingerprint(changes: list[dict[str, Any]]) -> str:
    """
    根据路径、操作、最终摘要和文件模式生成幂等变更指纹。
    """
    stable = [
        {
            "path": item["path"],
            "operation": item["operation"],
            "after_sha256": item.get("after_sha256"),
            "mode": item.get("mode", "100644"),
        }
        for item in sorted(changes, key=lambda value: value["path"])
    ]
    return hashlib.sha256(
        json.dumps(stable, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def build_branch_name(target_repo: str, base: str) -> str:
    """
    生成含时间与随机摘要的 Agent 工作分支名，降低与已有分支重名的概率。
    """
    seed = f"{target_repo}:{base}:{time.time_ns()}:{secrets.token_hex(8)}".encode()
    return normalize_branch(f"agent/moviepilot/{hashlib.sha256(seed).hexdigest()[:16]}")


def load_github_headers(repo: str) -> tuple[dict[str, str], Any]:
    """
    读取仓库专属或全局 GitHub 请求头及代理配置。

    返回的请求头可能包含认证令牌，只供内部请求使用，不得直接写入日志或公开结果。
    """
    headers: dict[str, str] = {}
    proxy: Any = None
    try:
        headers = dict(settings.REPO_GITHUB_HEADERS(repo=repo))
        proxy = settings.PROXY
    except Exception:
        proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
        proxy = {"http": proxy_url, "https": proxy_url} if proxy_url else None
    has_authorization = any(key.lower() == "authorization" and value for key, value in headers.items())
    if not has_authorization:
        token = (
            os.environ.get("MOVIEPILOT_GITHUB_TOKEN")
            or os.environ.get("GITHUB_TOKEN")
            or os.environ.get("GH_TOKEN")
            or ""
        ).strip()
        if token:
            headers["Authorization"] = f"Bearer {token}"
    if proxy is None:
        proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
        proxy = {"http": proxy_url, "https": proxy_url} if proxy_url else None
    headers.setdefault("Accept", "application/vnd.github+json")
    headers.setdefault("User-Agent", "MoviePilot-Agent-PR")
    headers.setdefault("X-GitHub-Api-Version", GITHUB_API_VERSION)
    return headers, proxy


def has_github_auth(headers: dict[str, str]) -> bool:
    """
    判断请求头是否包含认证信息，不解析或输出 Token。
    """
    return any(key.lower() == "authorization" and value for key, value in headers.items())


def github_token_from_headers(headers: dict[str, str]) -> str:
    """
    从内部 GitHub Authorization 请求头提取 Git HTTPS 所需的令牌。
    """
    authorization = next(
        (value for key, value in headers.items() if key.lower() == "authorization"),
        "",
    )
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() not in {"bearer", "token"} or not token.strip():
        raise PullRequestError("unsupported_token", "当前 GitHub Token 格式不能用于 Git HTTPS 推送。")
    return token.strip()


@contextmanager
def git_auth_environment(headers: dict[str, str]) -> Iterator[dict[str, str]]:
    """
    通过临时 askpass 和进程环境向 Git 提供令牌，避免令牌出现在命令行。
    """
    token = github_token_from_headers(headers)
    with tempfile.TemporaryDirectory(prefix="git-askpass-", dir=runtime_dir()) as directory:
        askpass = Path(directory) / "askpass.sh"
        askpass.write_text(
            "#!/bin/sh\n"
            'case "${1:-}" in\n'
            "  *[Uu]sername*) printf '%s\\n' 'x-access-token' ;;\n"
            "  *) printf '%s\\n' \"${MOVIEPILOT_GIT_TOKEN}\" ;;\n"
            "esac\n",
            encoding="utf-8",
        )
        askpass.chmod(0o700)
        environment = os.environ.copy()
        environment.update(
            {
                "GIT_ASKPASS": str(askpass),
                "GIT_TERMINAL_PROMPT": "0",
                "MOVIEPILOT_GIT_TOKEN": token,
            }
        )
        yield environment


def github_clone_url(repo: str) -> str:
    """
    返回不嵌入令牌的 GitHub HTTPS 克隆 URL。
    """
    return f"https://github.com/{normalize_repo(repo)}.git"


def _quote_path(value: str) -> str:
    """
    编码 GitHub API 路径中的单个或多个层级。
    """
    return "/".join(urllib.parse.quote(part, safe="") for part in value.split("/"))


class GitHubClient:
    """
    通过 MoviePilot RequestUtils 调用 GitHub REST API。
    """

    def __init__(
        self,
        *,
        headers: dict[str, str],
        proxies: Any = None,
        timeout: int = DEFAULT_TIMEOUT_SECONDS,
    ) -> None:
        """
        保存 HTTP 配置；Token 仅保留在请求头内。
        """
        self.headers = headers
        self.proxies = proxies
        self.timeout = timeout

    def request(self, method: str, path: str, payload: Optional[dict[str, Any]] = None) -> Any:
        """
        发起一次 JSON 请求并将 GitHub 错误转换为稳定异常。
        """
        body = None
        headers = dict(self.headers)
        if payload is not None:
            body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        response = None
        try:
            response = RequestUtils(
                proxies=self.proxies,
                headers=headers,
                timeout=self.timeout,
            ).request(
                method=method.lower(),
                url=f"{GITHUB_API_BASE}{path}",
                data=body,
                raise_exception=False,
            )
            if response is None:
                raise GitHubApiError("github_unavailable", "GitHub API 未返回响应。")
            raw = response.content or b""
            if response.status_code >= 400:
                message = raw.decode("utf-8", errors="replace")[:500]
                raise GitHubApiError(
                    "github_api_error",
                    f"GitHub API 返回 HTTP {response.status_code}: {redact_text(message)}",
                    status=response.status_code,
                )
            if not raw:
                return None
            try:
                return json.loads(raw.decode("utf-8"))
            except json.JSONDecodeError as error:
                raise GitHubApiError("invalid_github_response", "GitHub API 响应不是合法 JSON。") from error
        except GitHubApiError:
            raise
        except Exception as error:
            raise GitHubApiError("network_error", f"访问 GitHub API 失败: {redact_text(error)}") from error
        finally:
            if response is not None:
                response.close()

    def get_user(self) -> dict[str, Any]:
        """
        读取当前 Token 对应的 GitHub 用户。
        """
        data = self.request("GET", "/user")
        if not isinstance(data, dict) or not data.get("login"):
            raise GitHubApiError("invalid_github_user", "无法确定 GitHub Token 对应的用户。")
        return data

    def get_repo(self, repo: str) -> dict[str, Any]:
        """
        读取仓库详情。
        """
        data = self.request("GET", f"/repos/{normalize_repo(repo)}")
        if not isinstance(data, dict):
            raise GitHubApiError("invalid_github_repo", f"GitHub 仓库响应无效: {repo}")
        return data

    def get_ref(self, repo: str, branch: str) -> dict[str, Any]:
        """
        读取分支引用。
        """
        branch = normalize_branch(branch)
        data = self.request(
            "GET",
            f"/repos/{normalize_repo(repo)}/git/ref/heads/{_quote_path(branch)}",
        )
        if not isinstance(data, dict) or not isinstance(data.get("object"), dict):
            raise GitHubApiError("invalid_github_ref", f"GitHub 分支响应无效: {repo}:{branch}")
        return data

    def create_fork(self, repo: str) -> dict[str, Any]:
        """
        请求为当前认证用户创建 Fork。
        """
        data = self.request("POST", f"/repos/{normalize_repo(repo)}/forks", {"default_branch_only": True})
        if not isinstance(data, dict):
            raise GitHubApiError("invalid_fork_response", "GitHub Fork 响应无效。")
        return data

    def list_open_pulls(self, repo: str, *, head: str, base: str) -> list[dict[str, Any]]:
        """
        列出指定 head/base 的开放 PR，用于幂等重试。
        """
        query = urllib.parse.urlencode({"state": "open", "head": head, "base": base, "per_page": "20"})
        data = self.request("GET", f"/repos/{normalize_repo(repo)}/pulls?{query}")
        return data if isinstance(data, list) else []

    def create_pull(self, repo: str, payload: dict[str, Any]) -> dict[str, Any]:
        """
        创建上游 Pull Request。
        """
        data = self.request("POST", f"/repos/{normalize_repo(repo)}/pulls", payload)
        if not isinstance(data, dict) or not data.get("html_url"):
            raise GitHubApiError("invalid_pull_response", "GitHub PR 响应缺少 URL。")
        return data


def ensure_fork(client: GitHubClient, upstream_repo: str, login: str) -> tuple[str, bool]:
    """
    复用当前用户对目标仓库的已有 Fork，否则创建并等待可读。
    """
    upstream = normalize_repo(upstream_repo)
    owner, name = upstream.split("/", 1)
    if owner.lower() == login.lower():
        repository = client.get_repo(upstream)
        permissions = repository.get("permissions") if isinstance(repository, dict) else None
        if isinstance(permissions, dict) and permissions.get("push") is False:
            raise PullRequestError("fork_permission", f"当前 Token 没有向目标仓库推送的权限: {upstream}")
        return upstream, False
    fork_repo = normalize_repo(f"{login}/{name}")
    try:
        fork_data = client.get_repo(fork_repo)
    except GitHubApiError as error:
        if error.status != 404:
            raise
        client.create_fork(upstream)
        deadline = time.monotonic() + 60
        fork_data = None
        while time.monotonic() < deadline:
            try:
                fork_data = client.get_repo(fork_repo)
                break
            except GitHubApiError as retry_error:
                if retry_error.status != 404:
                    raise
                time.sleep(1)
        if fork_data is None:
            raise PullRequestError("fork_timeout", f"Fork {fork_repo} 创建后未在规定时间内可用。")
        created = True
    else:
        created = False

    parent = fork_data.get("parent") if isinstance(fork_data, dict) else None
    if not isinstance(fork_data, dict) or fork_data.get("fork") is not True or not isinstance(parent, dict):
        raise PullRequestError("invalid_fork", f"{fork_repo} 存在，但不是目标仓库的 Fork。")
    try:
        parent_full_name = normalize_repo(str(parent.get("full_name") or ""))
    except ValueError as error:
        raise PullRequestError("invalid_fork", f"无法确认 {fork_repo} 的上游仓库。") from error
    if parent_full_name.lower() != upstream.lower():
        raise PullRequestError("wrong_fork_parent", f"{fork_repo} 的上游不是 {upstream}。")
    return fork_repo, created


def pull_head(target_repo: str, fork_repo: str, login: str, branch: str) -> str:
    """
    返回 GitHub 创建 PR 所需的 head 表达式。
    """
    if normalize_repo(target_repo).lower() == normalize_repo(fork_repo).lower():
        return normalize_branch(branch)
    return f"{login}:{normalize_branch(branch)}"


def changes_from_payload(payload: dict[str, Any]) -> list[Change]:
    """
    读取并验证预览中的变更，确认 Git 工作树仍与预览一致。
    """
    source_root = Path(str(payload.get("source_root") or "")).expanduser().resolve()
    changes_file = Path(str(payload.get("changes_file") or "")).expanduser().resolve()
    require_git_root(source_root)
    if not changes_file.is_file():
        raise PullRequestError("missing_changes", "PR 变更清单不存在，请重新准备预览。")
    try:
        data = json.loads(changes_file.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise PullRequestError("invalid_changes", "PR 变更清单无法读取。") from error
    raw_changes = data.get("changes") if isinstance(data, dict) else None
    if not isinstance(raw_changes, list) or not raw_changes:
        raise PullRequestError("no_changes", "PR 变更清单为空。")

    result = []
    for raw in raw_changes:
        if not isinstance(raw, dict):
            raise PullRequestError("invalid_changes", "PR 变更清单包含无效条目。")
        path = normalize_relative_path(str(raw.get("path") or ""))
        local_path = source_root / path
        if local_path.is_symlink():
            raise PullRequestError("unsupported_change", f"拒绝提交符号链接文件: {path}")
        source_path = local_path.resolve(strict=False)
        if not _is_within(source_path, source_root):
            raise PullRequestError("path_outside_root", f"文件路径超出源目录: {path}")
        if is_rejected_path(path):
            raise PullRequestError("sensitive_path", f"拒绝提交运行时或敏感路径: {path}")
        operation = str(raw.get("operation") or "")
        if operation not in {"create", "update", "delete"}:
            raise PullRequestError("invalid_changes", f"文件操作无效: {path}")
        exists = source_path.is_file()
        after_sha256 = sha256_file(source_path) if exists else None
        if after_sha256 != raw.get("after_sha256"):
            raise PullRequestError("changes_modified", f"预览后文件发生变化: {path}")
        if operation != "delete" and not exists:
            raise PullRequestError("missing_file", f"待提交文件不存在: {path}")
        if operation == "delete" and exists:
            raise PullRequestError("delete_changed", f"待删除文件重新出现: {path}")
        size_bytes = source_path.stat().st_size if exists else 0
        if size_bytes > MAX_FILE_BYTES:
            raise PullRequestError("file_too_large", f"文件超过 {MAX_FILE_BYTES // 1024 // 1024} MiB: {path}")
        mode = _file_mode(source_path)
        if mode != raw.get("mode", "100644"):
            raise PullRequestError("changes_modified", f"预览后文件权限发生变化: {path}")
        result.append(
            Change(
                path=path,
                source_path=source_path,
                operation=operation,
                before_sha256=raw.get("before_sha256"),
                after_sha256=after_sha256,
                size_bytes=size_bytes,
                mode=mode,
            )
        )

    _, current_changes = collect_changes(source_root)
    expected_fingerprint = data.get("fingerprint") if isinstance(data, dict) else None
    current_fingerprint = change_fingerprint(current_changes)
    if expected_fingerprint != current_fingerprint:
        raise PullRequestError("changes_modified", "Git 工作树与已确认的 PR 预览不一致。")
    return result


__all__ = [
    "Change",
    "GitHubApiError",
    "GitHubClient",
    "MAX_CHANGED_FILES",
    "PullRequestError",
    "build_branch_name",
    "change_fingerprint",
    "changes_from_payload",
    "collect_changes",
    "ensure_fork",
    "find_git_root",
    "git_output",
    "git_auth_environment",
    "git_current_branch",
    "git_remote_url",
    "github_clone_url",
    "github_token_from_headers",
    "has_github_auth",
    "is_rejected_path",
    "load_github_headers",
    "normalize_branch",
    "normalize_repo",
    "pull_head",
    "read_json_file",
    "remote_repo_from_url",
    "require_git_root",
    "result_payload",
    "run_git",
    "runtime_dir",
    "write_json_file",
]