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())
|