132 lines
4.7 KiB
Python
132 lines
4.7 KiB
Python
"""把 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())
|