docs: prepare Wiki-first migration (#43)

This commit is contained in:
QiuSW
2026-08-17 11:27:09 +08:00
parent cde2a84211
commit a4ac398ca0
17 changed files with 1457 additions and 103 deletions
+131
View File
@@ -0,0 +1,131 @@
"""把 Gitea Wiki 任务归档人工按需导出到 docs/task。"""
from __future__ import annotations
import argparse
import re
from pathlib import Path
from typing import Any
from new_task_archive import safe_title
from wiki_docs import (
DEFAULT_CONFIG,
ROOT,
WikiClient,
WikiDocsError,
dirty_paths,
load_config,
parse_mirror,
write_mirror,
)
TASK_PAGE_PATTERN = re.compile(r"^Task-(?P<number>\d+)-(?P<title>.+)$")
def task_revision(metadata: dict[str, Any], page_name: str) -> str:
last_commit = metadata.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:{page_name}")
return revision
def existing_task_mirrors(root: Path = ROOT) -> dict[str, Path]:
"""按镜像头匹配已有文件,兼容历史自定义文件名。"""
mirrors: dict[str, Path] = {}
task_dir = root / "docs" / "task"
if not task_dir.is_dir():
return mirrors
for path in task_dir.glob("*.md"):
try:
metadata, _ = parse_mirror(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, WikiDocsError) as exc:
raise WikiDocsError(f"已有任务镜像无效 {path.name}:{exc}") from exc
page_name = metadata.get("wiki_page", "")
if not TASK_PAGE_PATTERN.fullmatch(page_name):
raise WikiDocsError(f"已有任务镜像页面名无效 {path.name}:{page_name}")
if page_name in mirrors:
raise WikiDocsError(f"任务页面存在重复本地镜像:{page_name}")
mirrors[page_name] = path
return mirrors
def task_target(page_name: str, root: Path = ROOT) -> Path:
match = TASK_PAGE_PATTERN.fullmatch(page_name)
if match is None:
raise WikiDocsError(f"不是任务归档页面:{page_name}")
title = safe_title(match.group("title"))
if not title:
raise WikiDocsError(f"任务归档标题无效:{page_name}")
return root / "docs" / "task" / f"{match.group('number')}-{title}.md"
def export_task_archives(
client: WikiClient, *, export_all: bool = False, root: Path = ROOT
) -> list[str]:
"""增量或全量读取任务归档;绝不删除本地文件。"""
dirty = dirty_paths(["docs/task"], root)
if dirty:
raise WikiDocsError(
"本地任务镜像存在未提交改动,已停止以防覆盖:\n" + "\n".join(dirty)
)
existing = existing_task_mirrors(root)
pages = []
for metadata in client.list_pages():
title = metadata.get("title")
if isinstance(title, str) and TASK_PAGE_PATTERN.fullmatch(title):
pages.append((int(title.split("-", 2)[1]), title, metadata))
pages.sort(key=lambda item: (item[0], item[1]))
messages: list[str] = []
targets: set[Path] = set()
for _, page_name, metadata in pages:
target = existing.get(page_name, task_target(page_name, root))
if target in targets:
raise WikiDocsError(f"多个任务页面映射到同一本地路径:{target.name}")
targets.add(target)
revision = task_revision(metadata, page_name)
if not export_all and target.is_file():
local_metadata, _ = parse_mirror(target.read_text(encoding="utf-8"))
if (
local_metadata.get("wiki_page") == page_name
and local_metadata.get("wiki_revision") == revision
):
messages.append(f"跳过:{target.relative_to(root)} <- {page_name}@{revision[:12]}")
continue
page = client.get_page_from_metadata(metadata, page_name)
changed = write_mirror(target, page)
action = "已导出" if changed else "无变化"
messages.append(f"{action}:{target.relative_to(root)} <- {page_name}@{revision[:12]}")
return messages
def main() -> int:
parser = argparse.ArgumentParser(description="人工按需导出 Gitea Wiki 任务归档")
parser.add_argument(
"--all", action="store_true", help="全量读取全部线上任务归档;默认按 revision 增量"
)
parser.add_argument(
"--config", default=str(DEFAULT_CONFIG), help="核心 Wiki 映射配置"
)
args = parser.parse_args()
try:
config = load_config(Path(args.config).resolve())
messages = export_task_archives(
WikiClient(config), export_all=args.all
)
except WikiDocsError as exc:
print(f"错误:{exc}")
return 1
for message in messages:
print(message)
print("任务归档全量导出完成" if args.all else "任务归档增量导出完成")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+87
View File
@@ -0,0 +1,87 @@
"""只在 Gitea Wiki 创建任务归档;本地镜像由人工按需导出。"""
from __future__ import annotations
import argparse
import re
from datetime import date
from pathlib import Path
from wiki_docs import (
DEFAULT_CONFIG,
WikiClient,
WikiDocsError,
load_config,
)
def safe_title(title: str) -> str:
"""把标题转换为适合 Wiki 页面名和 Windows 文件名的短文本。"""
cleaned = re.sub(r'[<>:"/\\|?*]', "-", title.strip())
cleaned = re.sub(r"\s+", "-", cleaned)
cleaned = re.sub(r"-+", "-", cleaned)
return cleaned.strip(".-")
def build_archive(
template: str,
issue_number: str,
title: str,
page_name: str,
issue_url: str,
) -> str:
content = template.replace("<工单号>", issue_number, 1)
content = content.replace("<标题>", title.strip(), 1)
content = content.replace("YYYY-MM-DD", date.today().isoformat(), 1)
content = content.replace("<链接>", issue_url, 1)
return content.replace("<页面名>", page_name, 1)
def main() -> int:
parser = argparse.ArgumentParser(
description="在 Gitea Wiki 创建任务归档,不自动导出本地镜像"
)
parser.add_argument("issue_number", help="Gitea 工单号,例如 123")
parser.add_argument("title", help="简短任务标题")
parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="Wiki 映射配置")
args = parser.parse_args()
short_title = safe_title(args.title)
if not args.issue_number.isdigit():
print("错误:工单号必须是数字")
return 1
if not short_title:
print("错误:标题不能为空")
return 1
try:
config = load_config(Path(args.config).resolve())
page_name = f"Task-{args.issue_number}-{short_title}"
client = WikiClient(config)
if any(item.get("title") == page_name for item in client.list_pages()):
raise WikiDocsError(f"任务归档已经存在:{page_name}")
template = client.get_page("Task-Archive-Template").text
issue_url = (
f"{config.gitea_url}/{config.owner}/{config.repository}/issues/"
f"{args.issue_number}"
)
content = build_archive(
template, args.issue_number, args.title, page_name, issue_url
)
page = client.create_page(
page_name,
content,
f"docs: 创建任务 #{args.issue_number} 归档草稿",
)
except WikiDocsError as exc:
print(f"错误:{exc}")
return 1
print(f"已创建 Wiki:{page.html_url}")
print("未导出本地任务归档;需要时运行 export_task_archives.py")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+33
View File
@@ -0,0 +1,33 @@
"""从 Gitea Wiki 单向导出配置中的核心 docs 镜像。"""
from __future__ import annotations
import argparse
from pathlib import Path
from wiki_docs import DEFAULT_CONFIG, WikiClient, WikiDocsError, load_config, sync_all
def main() -> int:
parser = argparse.ArgumentParser(description="从 Gitea Wiki 单向同步核心 docs 镜像")
parser.add_argument(
"--check", action="store_true", help="只检查 Wiki 与镜像是否一致,不写文件"
)
parser.add_argument(
"--config", default=str(DEFAULT_CONFIG), help="Wiki 页面映射 JSON 文件"
)
args = parser.parse_args()
try:
config = load_config(Path(args.config).resolve())
messages = sync_all(config, WikiClient(config), check=args.check)
except WikiDocsError as exc:
print(f"错误:{exc}")
return 1
for message in messages:
print(message)
print("Wiki 镜像检查通过" if args.check else "Wiki 镜像同步完成")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+422
View File
@@ -0,0 +1,422 @@
"""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)