dev_scripts/wiki_docs.py

"""Gitea Wiki 到本地 docs 镜像的共享实现。"""

from __future__ import annotations

import base64
import json
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = ROOT / "wiki-docs.json"
MIRROR_START = "<!-- gitea-wiki-mirror:start -->"
MIRROR_END = "<!-- gitea-wiki-mirror:end -->"
HEADER_PATTERN = re.compile(
    rf"\A{re.escape(MIRROR_START)}\n(?P<metadata>.*?)\n"
    rf"{re.escape(MIRROR_END)}\n\n(?P<body>.*)\Z",
    re.DOTALL,
)


class WikiDocsError(RuntimeError):
    """可供命令行直接展示的 Wiki 文档错误。"""


@dataclass(frozen=True)
class Mapping:
    page: str
    path: str


@dataclass(frozen=True)
class Config:
    path: Path
    gitea_url: str
    owner: str
    repository: str
    mappings: tuple[Mapping, ...]


@dataclass(frozen=True)
class WikiPage:
    title: str
    sub_url: str
    text: str
    revision: str
    html_url: str


def _required_string(data: dict[str, Any], key: str) -> str:
    value = data.get(key)
    if not isinstance(value, str) or not value.strip():
        raise WikiDocsError(f"配置字段 {key!r} 必须是非空字符串")
    return value.strip()


def validate_mappings(raw_mappings: Any) -> tuple[Mapping, ...]:
    """校验显式页面映射，确保只会写入 docs 下的 Markdown。"""

    if not isinstance(raw_mappings, list) or not raw_mappings:
        raise WikiDocsError("配置字段 'mappings' 必须是非空数组")

    mappings: list[Mapping] = []
    pages: set[str] = set()
    paths: set[str] = set()
    for index, item in enumerate(raw_mappings, start=1):
        if not isinstance(item, dict):
            raise WikiDocsError(f"第 {index} 个映射必须是对象")
        page = _required_string(item, "page")
        path = _required_string(item, "path").replace("\\", "/")
        pure_path = PurePosixPath(path)
        if (
            pure_path.is_absolute()
            or ".." in pure_path.parts
            or not pure_path.parts
            or pure_path.parts[0] != "docs"
            or pure_path.suffix.lower() != ".md"
        ):
            raise WikiDocsError(f"镜像路径必须是 docs/ 下的 Markdown：{path}")
        if page in pages:
            raise WikiDocsError(f"Wiki 页面重复映射：{page}")
        if path in paths:
            raise WikiDocsError(f"本地路径重复映射：{path}")
        pages.add(page)
        paths.add(path)
        mappings.append(Mapping(page=page, path=path))
    return tuple(mappings)


def load_config(path: Path = DEFAULT_CONFIG) -> Config:
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise WikiDocsError(f"无法读取 Wiki 映射配置 {path}: {exc}") from exc
    if not isinstance(raw, dict) or raw.get("schema_version") != 1:
        raise WikiDocsError("wiki-docs.json 的 schema_version 必须为 1")
    configured_url = _required_string(raw, "gitea_url")
    gitea_url = os.environ.get("GITEA_URL", configured_url).rstrip("/")
    if gitea_url.endswith("/api/v1"):
        gitea_url = gitea_url[: -len("/api/v1")]
    return Config(
        path=path,
        gitea_url=gitea_url,
        owner=_required_string(raw, "owner"),
        repository=_required_string(raw, "repository"),
        mappings=validate_mappings(raw.get("mappings")),
    )


class WikiClient:
    """只使用标准库访问 Gitea Wiki API。"""

    def __init__(self, config: Config, token: str | None = None) -> None:
        self.config = config
        self.token = token if token is not None else os.environ.get("GITEA_TOKEN")

    def _request(
        self,
        method: str,
        api_path: str,
        *,
        payload: dict[str, Any] | None = None,
        query: dict[str, Any] | None = None,
    ) -> Any:
        url = f"{self.config.gitea_url}/api/v1{api_path}"
        if query:
            url = f"{url}?{urlencode(query)}"
        headers = {"Accept": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        data = None
        if payload is not None:
            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            headers["Content-Type"] = "application/json"
        request = Request(url, data=data, headers=headers, method=method)
        try:
            with urlopen(request, timeout=30) as response:
                body = response.read()
        except HTTPError as exc:
            if self.token and method == "GET" and exc.code in {401, 403, 404}:
                # 公共仓库可能可匿名读取，而当前 shell 中的通用令牌属于
                # 另一个实例或已失效。只对只读请求安全降级为匿名访问。
                anonymous_headers = {"Accept": "application/json"}
                anonymous_request = Request(
                    url, data=data, headers=anonymous_headers, method=method
                )
                try:
                    with urlopen(anonymous_request, timeout=30) as response:
                        body = response.read()
                except HTTPError as anonymous_exc:
                    detail = anonymous_exc.read().decode("utf-8", errors="replace")
                    raise WikiDocsError(
                        f"Gitea API {method} {api_path} 返回 "
                        f"{anonymous_exc.code}: {detail}"
                    ) from anonymous_exc
                except URLError as anonymous_exc:
                    raise WikiDocsError(
                        f"无法连接 Gitea：{anonymous_exc.reason}"
                    ) from anonymous_exc
            else:
                detail = exc.read().decode("utf-8", errors="replace")
                raise WikiDocsError(
                    f"Gitea API {method} {api_path} 返回 {exc.code}: {detail}"
                ) from exc
        except URLError as exc:
            raise WikiDocsError(f"无法连接 Gitea：{exc.reason}") from exc
        if not body:
            return None
        try:
            return json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise WikiDocsError("Gitea API 返回了无效的 UTF-8 JSON") from exc

    def list_pages(self) -> list[dict[str, Any]]:
        pages: list[dict[str, Any]] = []
        page_number = 1
        while True:
            batch = self._request(
                "GET",
                f"/repos/{quote(self.config.owner, safe='')}/"
                f"{quote(self.config.repository, safe='')}/wiki/pages",
                query={"page": page_number, "limit": 50},
            )
            if not isinstance(batch, list):
                raise WikiDocsError("Gitea Wiki 页面列表格式无效")
            pages.extend(item for item in batch if isinstance(item, dict))
            if len(batch) < 50:
                return pages
            page_number += 1

    def get_page(self, page_name: str) -> WikiPage:
        metadata = next(
            (
                item
                for item in self.list_pages()
                if item.get("title") == page_name or item.get("sub_url") == page_name
            ),
            None,
        )
        if metadata is None:
            raise WikiDocsError(
                f"Wiki 页面不存在：{page_name}；不会自动删除或重命名本地镜像"
            )
        return self.get_page_from_metadata(metadata, page_name)

    def get_page_from_metadata(
        self, metadata: dict[str, Any], page_name: str | None = None
    ) -> WikiPage:
        """使用页面列表元数据读取正文，避免重复获取完整页面列表。"""

        resolved_name = page_name or _required_string(metadata, "title")
        sub_url = _required_string(metadata, "sub_url")
        page = self._request(
            "GET",
            f"/repos/{quote(self.config.owner, safe='')}/"
            f"{quote(self.config.repository, safe='')}/wiki/page/"
            f"{quote(sub_url, safe='%')}",
        )
        if not isinstance(page, dict):
            raise WikiDocsError(f"Wiki 页面响应格式无效：{resolved_name}")
        encoded_content = page.get("content_base64")
        if not isinstance(encoded_content, str):
            raise WikiDocsError(f"Wiki 页面没有 content_base64：{resolved_name}")
        try:
            text = base64.b64decode(encoded_content, validate=True).decode("utf-8")
        except (ValueError, UnicodeDecodeError) as exc:
            raise WikiDocsError(
                f"Wiki 页面不是有效的 UTF-8 Markdown：{resolved_name}"
            ) from exc
        last_commit = page.get("last_commit")
        revision = last_commit.get("sha") if isinstance(last_commit, dict) else None
        if not isinstance(revision, str) or not revision:
            raise WikiDocsError(f"Wiki 页面缺少 revision：{resolved_name}")
        title = page.get("title")
        resolved_title = title if isinstance(title, str) and title else resolved_name
        html_url = (
            f"{self.config.gitea_url}/{quote(self.config.owner, safe='')}/"
            f"{quote(self.config.repository, safe='')}/wiki/{quote(sub_url, safe='%')}"
        )
        return WikiPage(
            title=resolved_title,
            sub_url=sub_url,
            text=normalize_body(text),
            revision=revision,
            html_url=html_url,
        )

    def create_page(self, title: str, content: str, message: str) -> WikiPage:
        if not self.token:
            raise WikiDocsError("创建 Wiki 页面需要通过 GITEA_TOKEN 提供写入令牌")
        encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
        self._request(
            "POST",
            f"/repos/{quote(self.config.owner, safe='')}/"
            f"{quote(self.config.repository, safe='')}/wiki/new",
            payload={"title": title, "content_base64": encoded, "message": message},
        )
        return self.get_page(title)


def normalize_body(text: str) -> str:
    return text.replace("\r\n", "\n").replace("\r", "\n").rstrip() + "\n"


def parse_mirror(text: str) -> tuple[dict[str, str], str]:
    match = HEADER_PATTERN.match(text.replace("\r\n", "\n").replace("\r", "\n"))
    if match is None:
        raise WikiDocsError("缺少或损坏 gitea-wiki-mirror 元数据头")
    metadata: dict[str, str] = {}
    for line in match.group("metadata").splitlines():
        key, separator, value = line.partition(": ")
        if not separator or not key or not value:
            raise WikiDocsError(f"无效的镜像元数据行：{line}")
        metadata[key] = value
    return metadata, normalize_body(match.group("body"))


def render_mirror(page: WikiPage, existing: str | None = None) -> str:
    synchronized_at: str | None = None
    if existing is not None:
        try:
            metadata, body = parse_mirror(existing)
        except WikiDocsError:
            pass
        else:
            if metadata.get("wiki_revision") == page.revision and body == page.text:
                synchronized_at = metadata.get("synchronized_at")
    if not synchronized_at:
        synchronized_at = datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
            "+00:00", "Z"
        )
    header = "\n".join(
        (
            MIRROR_START,
            "generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件)",
            f"wiki_page: {page.title}",
            f"wiki_url: {page.html_url}",
            f"wiki_revision: {page.revision}",
            f"synchronized_at: {synchronized_at}",
            MIRROR_END,
        )
    )
    return f"{header}\n\n{page.text}"


def dirty_mirror_paths(config: Config, root: Path = ROOT) -> list[str]:
    return dirty_paths([mapping.path for mapping in config.mappings], root)


def dirty_paths(paths: list[str], root: Path = ROOT) -> list[str]:
    """返回指定路径中已有、修改或未跟踪的工作区条目。"""

    if not paths:
        return []
    result = subprocess.run(
        ["git", "status", "--porcelain", "--", *paths],
        cwd=root,
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return [line for line in result.stdout.splitlines() if line.strip()]


def _write_atomic(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    handle, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
    )
    try:
        with os.fdopen(handle, "w", encoding="utf-8", newline="\n") as stream:
            stream.write(content)
        os.replace(temporary_name, path)
    except BaseException:
        Path(temporary_name).unlink(missing_ok=True)
        raise


def write_mirror(path: Path, page: WikiPage) -> bool:
    """写入一份 Wiki 镜像；内容无变化时返回 False。"""

    existing = path.read_text(encoding="utf-8") if path.is_file() else None
    rendered = render_mirror(page, existing)
    if existing == rendered:
        return False
    _write_atomic(path, rendered)
    return True


def check_mirror(mapping: Mapping, page: WikiPage, path: Path) -> list[str]:
    if not path.is_file():
        return [f"缺少镜像：{mapping.path}"]
    try:
        metadata, body = parse_mirror(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, WikiDocsError) as exc:
        return [f"镜像无效 {mapping.path}: {exc}"]
    expected = {
        "wiki_page": page.title,
        "wiki_url": page.html_url,
        "wiki_revision": page.revision,
    }
    errors = [
        f"{mapping.path} 的 {key} 不一致"
        for key, value in expected.items()
        if metadata.get(key) != value
    ]
    if not metadata.get("synchronized_at"):
        errors.append(f"{mapping.path} 缺少 synchronized_at")
    if body != page.text:
        errors.append(f"{mapping.path} 的正文与 Wiki 不一致")
    return errors


def sync_all(config: Config, client: WikiClient, *, check: bool = False) -> list[str]:
    """检查或写入所有显式映射；绝不处理映射外的文件。"""

    if not check:
        dirty = dirty_mirror_paths(config)
        if dirty:
            details = "\n".join(dirty)
            raise WikiDocsError(
                "已映射的本地镜像存在未提交改动，已停止以防覆盖：\n" + details
            )

    messages: list[str] = []
    for mapping in config.mappings:
        page = client.get_page(mapping.page)
        target = ROOT / PurePosixPath(mapping.path)
        if check:
            errors = check_mirror(mapping, page, target)
            if errors:
                raise WikiDocsError("\n".join(errors))
            messages.append(f"一致：{mapping.path} <- {page.title}@{page.revision[:12]}")
            continue
        existing = target.read_text(encoding="utf-8") if target.is_file() else None
        rendered = render_mirror(page, existing)
        if existing != rendered:
            _write_atomic(target, rendered)
            messages.append(f"已更新：{mapping.path} <- {page.title}@{page.revision[:12]}")
        else:
            messages.append(f"无变化：{mapping.path} <- {page.title}@{page.revision[:12]}")
    return messages


def append_mapping(config: Config, mapping: Mapping) -> None:
    raw = json.loads(config.path.read_text(encoding="utf-8"))
    mappings = validate_mappings(raw.get("mappings"))
    if any(item.page == mapping.page or item.path == mapping.path for item in mappings):
        raise WikiDocsError(f"页面或路径已经登记：{mapping.page} -> {mapping.path}")
    raw["mappings"].append({"page": mapping.page, "path": mapping.path})
    rendered = json.dumps(raw, ensure_ascii=False, indent=2) + "\n"
    _write_atomic(config.path, rendered)