"""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 = "" MIRROR_END = "" HEADER_PATTERN = re.compile( rf"\A{re.escape(MIRROR_START)}\n(?P.*?)\n" rf"{re.escape(MIRROR_END)}\n\n(?P.*)\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 _metadata_revision(metadata: dict[str, Any]) -> str | None: last_commit = metadata.get("last_commit") revision = last_commit.get("sha") if isinstance(last_commit, dict) else None return revision if isinstance(revision, str) and revision else None def _find_page_metadata( pages: list[dict[str, Any]], page_name: str ) -> dict[str, Any]: metadata = next( ( item for item in 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 metadata def _metadata_identity( config: Config, metadata: dict[str, Any], fallback_title: str ) -> tuple[str, str] | None: title = metadata.get("title") sub_url = metadata.get("sub_url") if not isinstance(title, str) or not title: title = fallback_title if not isinstance(sub_url, str) or not sub_url: return None url = ( f"{config.gitea_url}/{quote(config.owner, safe='')}/" f"{quote(config.repository, safe='')}/wiki/{quote(sub_url, safe='%')}" ) return title, url def _local_revision( path: Path, *, expected_title: str, expected_url: str ) -> str | None: if not path.is_file(): return None try: metadata, _body = parse_mirror(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, WikiDocsError): return None if ( metadata.get("wiki_page") != expected_title or metadata.get("wiki_url") != expected_url or not metadata.get("synchronized_at") ): return None revision = metadata.get("wiki_revision") return revision if revision else None def sync_all( config: Config, client: WikiClient, *, check: bool = False, deep_check: bool = False, ) -> list[str]: """检查或写入所有显式映射；绝不处理映射外的文件。""" if not check: dirty = dirty_mirror_paths(config) if dirty: details = "\n".join(dirty) raise WikiDocsError( "已映射的本地镜像存在未提交改动，已停止以防覆盖：\n" + details ) pages = client.list_pages() messages: list[str] = [] for mapping in config.mappings: metadata = _find_page_metadata(pages, mapping.page) target = ROOT / PurePosixPath(mapping.path) remote_revision = _metadata_revision(metadata) identity = _metadata_identity(config, metadata, mapping.page) local_revision = ( _local_revision( target, expected_title=identity[0], expected_url=identity[1] ) if identity is not None else None ) if not deep_check and remote_revision and local_revision == remote_revision: action = "一致" if check else "无变化" messages.append( f"{action}：{mapping.path} <- {mapping.page}@{remote_revision[:12]}" ) continue page = client.get_page_from_metadata(metadata, mapping.page) 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)