OPC/lexgo
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf: 优化 Wiki 镜像增量同步 (#29)

Browse Source
This commit is contained in:
ila
2026-09-02 09:38:09 +08:00
parent 4bbacf4d7f
commit cbdd27d861
5 changed files with 297 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dev_scripts/harness.py
+42 -19
View File
@@ -620,32 +620,47 @@ def run_check(args: argparse.Namespace) -> int:
def run_sync(args: argparse.Namespace) -> int:
"""--verify 依次执行导出、结构检查和一致性校验，替代原来的三条命令。"""
"""同步核心镜像；--verify 复用本轮完整读取完成初始化验证。"""
if args.verify:
steps = (
("同步", lambda: run_sync(
argparse.Namespace(check=False, verify=False, config=args.config))),
("结构检查", lambda: run_check(argparse.Namespace(strict=True))),
("一致性校验", lambda: run_sync(
argparse.Namespace(check=True, verify=False, config=args.config))),
)
for name, step in steps:
code = step()
if code != 0:
print(f"错误：{name}未通过，已停止")
return code
try:
config = load_config(Path(args.config).resolve())
messages = sync_all(
config, WikiClient(config), check=False, deep_check=True
)
except WikiDocsError as exc:
print(f"错误：{exc}")
return 1
for message in messages:
print(message)
code = run_check(argparse.Namespace(strict=True))
if code != 0:
print("错误：结构检查未通过，已停止")
return code
print("Wiki 镜像初始化验证通过")
return 0
try:
config = load_config(Path(args.config).resolve())
messages = sync_all(config, WikiClient(config), check=args.check)
deep_check = getattr(args, "deep_check", False)
messages = sync_all(
config,
WikiClient(config),
check=args.check or deep_check,
deep_check=deep_check,
)
except WikiDocsError as exc:
print(f"错误：{exc}")
return 1
for message in messages:
print(message)
print("Wiki 镜像检查通过" if args.check else "Wiki 镜像同步完成")
print(
"Wiki 镜像深度检查通过"
if deep_check
else "Wiki 镜像检查通过"
if args.check
else "Wiki 镜像同步完成"
)
return 0
@@ -710,13 +725,21 @@ def main() -> int:
p_check.set_defaults(func=run_check)
p_sync = sub.add_parser("sync", help="从 Gitea Wiki 单向同步核心 docs 镜像")
p_sync.add_argument(
"--check", action="store_true", help="只检查 Wiki 与镜像是否一致，不写文件"
sync_mode = p_sync.add_mutually_exclusive_group()
sync_mode.add_argument(
"--check",
action="store_true",
help="按 revision 快速检查 Wiki 与镜像，不写文件",
)
p_sync.add_argument(
sync_mode.add_argument(
"--deep-check",
action="store_true",
help="下载全部 Wiki 正文并逐页检查镜像，不写文件",
)
sync_mode.add_argument(
"--verify",
action="store_true",
help="依次执行导出、check --strict 和一致性校验",
help="完整读取并导出 Wiki，再执行 check --strict 初始化验证",
)
p_sync.add_argument(
"--config", default=str(DEFAULT_CONFIG), help="Wiki 页面映射 JSON 文件"
dev_scripts/wiki_docs.py
+85 -2
View File
@@ -381,7 +381,72 @@ def check_mirror(mapping: Mapping, page: WikiPage, path: Path) -> list[str]:
return errors
def sync_all(config: Config, client: WikiClient, *, check: bool = False) -> list[str]:
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:
@@ -392,10 +457,28 @@ def sync_all(config: Config, client: WikiClient, *, check: bool = False) -> list
"已映射的本地镜像存在未提交改动，已停止以防覆盖：\n" + details
)
pages = client.list_pages()
messages: list[str] = []
for mapping in config.mappings:
page = client.get_page(mapping.page)
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: