From cbdd27d8618efdd744c4fa90f0f26e72fd9ebec3 Mon Sep 17 00:00:00 2001 From: ila Date: Wed, 2 Sep 2026 09:38:09 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E4=BC=98=E5=8C=96=20Wiki=20=E9=95=9C?= =?UTF-8?q?=E5=83=8F=E5=A2=9E=E9=87=8F=E5=90=8C=E6=AD=A5=20(#29)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 5 +- dev_scripts/harness.py | 61 ++++++++++----- dev_scripts/wiki_docs.py | 87 ++++++++++++++++++++- docs/01-workflow.md | 9 ++- tests/test_wiki_docs.py | 161 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 297 insertions(+), 26 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 1b4d176..3f3629d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,8 @@ | 检查模板结构 | `python dev_scripts/harness.py check --strict` | | 运行单元测试 | `python -m unittest discover -s tests -v` | | 导出核心 Wiki 镜像 | `python dev_scripts/harness.py sync` | -| 检查核心 Wiki 镜像 | `python dev_scripts/harness.py sync --check` | +| 快速检查核心 Wiki 镜像 | `python dev_scripts/harness.py sync --check` | +| 深度检查核心 Wiki 镜像 | `python dev_scripts/harness.py sync --deep-check` | | 创建可选任务快照 | `python dev_scripts/harness.py archive 123 "修复登录超时"` | | 增量导出已有快照 | `python dev_scripts/harness.py export` | | 全量导出已有快照 | `python dev_scripts/harness.py export --all` | @@ -102,7 +103,7 @@ Gitea 不可用时，输出完整工单草稿并说明阻塞。未经用户明 - `建工单并做`：依次建单和执行，`建工单，做`、`建工单,做` 含义相同；停在“待验收”。 - `继续工单 #N`：优先核对当前状态、最新评论、Git 和必要 Wiki 证据，从首个未完成步骤继续；关键前提未变化时不重复读取和分析仍有效的内容。 - `检查工单 #N`：只读核对范围、验收、测试和证据并输出报告；不自动修复。 -- `同步文档`：读取 Wiki、导出核心 `docs/` 并检查一致性，不处理任务归档；不修改 Wiki、不自动提交。 +- `同步文档`：按 revision 增量读取 Wiki、导出核心 `docs/` 并快速检查一致性，不处理任务归档；不修改 Wiki、不自动提交。只有明确要求深度检查时才逐页下载正文。 - `导出原型 #N`：人工触发导出指定工单已确认的原型版本，按工单和版本写入 `prototypes/`；不扩展范围、不自动提交。 - `导出全部原型`：人工触发导出当前项目已明确范围内的全部已确认原型；不自动提交。 - `导出任务归档`：人工触发 `python dev_scripts/harness.py export`，只导出新增或 revision 已变化的任务归档；不删除本地文件、不自动提交。 diff --git a/dev_scripts/harness.py b/dev_scripts/harness.py index a8d5fe3..cba590a 100644 --- a/dev_scripts/harness.py +++ b/dev_scripts/harness.py @@ -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 文件" diff --git a/dev_scripts/wiki_docs.py b/dev_scripts/wiki_docs.py index 6b1008c..814cf61 100644 --- a/dev_scripts/wiki_docs.py +++ b/dev_scripts/wiki_docs.py @@ -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: diff --git a/docs/01-workflow.md b/docs/01-workflow.md index 9449279..2b80144 100644 --- a/docs/01-workflow.md +++ b/docs/01-workflow.md @@ -2,8 +2,8 @@ generated: true (请先修改 Gitea Wiki，禁止直接编辑本文件) wiki_page: Development-Workflow wiki_url: https://git.ilapage.cn/OPC/dev_harness/wiki/Development-Workflow.- -wiki_revision: 4e30058e244453bf5da5cf2b5a0aa59176649127 -synchronized_at: 2026-08-26T08:01:22Z +wiki_revision: fb52a69363b53f3bce6d5f2bdc016aead0c19f90 +synchronized_at: 2026-09-02T01:36:08Z # 开发工作流 @@ -193,7 +193,10 @@ python dev_scripts/harness.py export --all # 全量导出已有归档 - 所有同步和导出只实现 Wiki → `docs/`，不提供反向同步。 - 镜像头必须记录页面名、页面地址、revision 和同步时间。 - 已跟踪镜像存在未提交改动时，同步必须停止；确认改动来源后再处理。 -- 核心同步的 `--check` 只检查核心镜像，不要求线上任务归档全部存在于本地。 +- `sync` 和 `sync --check` 先读取一次页面列表，并用远端 revision 与本地镜像头比较；revision 未变化时不下载正文，只有页面新增、变化、元数据缺失或本地镜像元数据无效时才读取正文。 +- `sync --check` 是核心镜像的日常快速检查，不要求线上任务归档全部存在于本地；它验证 revision 和镜像元数据，不替代正文审计。 +- `sync --deep-check` 显式下载全部映射页面正文并逐页比较，用于疑似镜像损坏、同步算法审计或人工要求的完整核对。 +- `sync --verify` 用于新项目 Wiki 初始化门禁，完整读取并写入核心镜像后执行严格结构检查；同一次运行复用读取结果，不重复下载正文。 - 已经导出的任务镜像仍必须具有来源页面、revision 和同步时间，并通过 Harness 格式检查。 - 页面删除和重命名不会自动传播，必须先更新工单并人工确认映射变化。 - Wiki 更新成功而导出失败时，在工单记录部分完成状态，不得把任务标为完成。 diff --git a/tests/test_wiki_docs.py b/tests/test_wiki_docs.py index bc6b58e..6bb9bb6 100644 --- a/tests/test_wiki_docs.py +++ b/tests/test_wiki_docs.py @@ -5,6 +5,7 @@ import os import sys import tempfile import unittest +from argparse import Namespace from pathlib import Path from unittest.mock import Mock, patch @@ -17,6 +18,7 @@ from harness import ( # noqa: E402 existing_task_mirrors, export_task_archives, main as harness_main, + run_sync, safe_title, task_target, ) @@ -122,6 +124,165 @@ class MirrorTests(unittest.TestCase): sync_all(config, client) client.get_page.assert_not_called() + @patch("wiki_docs.dirty_mirror_paths", return_value=[]) + def test_incremental_sync_lists_once_and_skips_unchanged_body(self, _dirty) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "docs" / "README.md" + target.parent.mkdir(parents=True) + target.write_text(render_mirror(self.page), encoding="utf-8") + config = Config( + path=root / "wiki-docs.json", + gitea_url="http://gitea.example", + owner="o", + repository="r", + mappings=(Mapping("Home", "docs/README.md"),), + ) + client = Mock() + client.list_pages.return_value = [ + { + "title": "Home", + "sub_url": "Home", + "last_commit": {"sha": self.page.revision}, + } + ] + with patch("wiki_docs.ROOT", root): + messages = sync_all(config, client) + self.assertTrue(messages[0].startswith("无变化：")) + client.list_pages.assert_called_once_with() + client.get_page_from_metadata.assert_not_called() + + @patch("wiki_docs.dirty_mirror_paths", return_value=[]) + def test_incremental_sync_fetches_changed_page(self, _dirty) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "docs" / "README.md" + target.parent.mkdir(parents=True) + target.write_text(render_mirror(self.page), encoding="utf-8") + changed = WikiPage( + title="Home", + sub_url="Home", + text="# 新首页\n", + revision="b" * 40, + html_url=self.page.html_url, + ) + metadata = { + "title": "Home", + "sub_url": "Home", + "last_commit": {"sha": changed.revision}, + } + config = Config( + path=root / "wiki-docs.json", + gitea_url="http://gitea.example", + owner="o", + repository="r", + mappings=(Mapping("Home", "docs/README.md"),), + ) + client = Mock() + client.list_pages.return_value = [metadata] + client.get_page_from_metadata.return_value = changed + with patch("wiki_docs.ROOT", root): + messages = sync_all(config, client) + _metadata, body = parse_mirror(target.read_text(encoding="utf-8")) + self.assertTrue(messages[0].startswith("已更新：")) + self.assertEqual(body, changed.text) + client.get_page_from_metadata.assert_called_once_with(metadata, "Home") + + @patch("wiki_docs.dirty_mirror_paths", return_value=[]) + def test_sync_fetches_when_remote_revision_is_missing(self, _dirty) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "docs" / "README.md" + target.parent.mkdir(parents=True) + target.write_text(render_mirror(self.page), encoding="utf-8") + metadata = {"title": "Home", "sub_url": "Home"} + config = Config( + path=root / "wiki-docs.json", + gitea_url="http://gitea.example", + owner="o", + repository="r", + mappings=(Mapping("Home", "docs/README.md"),), + ) + client = Mock() + client.list_pages.return_value = [metadata] + client.get_page_from_metadata.return_value = self.page + with patch("wiki_docs.ROOT", root): + sync_all(config, client) + client.get_page_from_metadata.assert_called_once_with(metadata, "Home") + + @patch("wiki_docs.dirty_mirror_paths", return_value=[]) + def test_sync_fetches_when_local_metadata_is_invalid(self, _dirty) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "docs" / "README.md" + target.parent.mkdir(parents=True) + invalid = render_mirror(self.page).replace( + "wiki_page: Home", "wiki_page: Wrong" + ) + target.write_text(invalid, encoding="utf-8") + metadata = { + "title": "Home", + "sub_url": "Home", + "last_commit": {"sha": self.page.revision}, + } + config = Config( + path=root / "wiki-docs.json", + gitea_url="http://gitea.example", + owner="o", + repository="r", + mappings=(Mapping("Home", "docs/README.md"),), + ) + client = Mock() + client.list_pages.return_value = [metadata] + client.get_page_from_metadata.return_value = self.page + with patch("wiki_docs.ROOT", root): + sync_all(config, client) + client.get_page_from_metadata.assert_called_once_with(metadata, "Home") + + def test_deep_check_fetches_unchanged_page_body(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "docs" / "README.md" + target.parent.mkdir(parents=True) + target.write_text(render_mirror(self.page), encoding="utf-8") + metadata = { + "title": "Home", + "sub_url": "Home", + "last_commit": {"sha": self.page.revision}, + } + config = Config( + path=root / "wiki-docs.json", + gitea_url="http://gitea.example", + owner="o", + repository="r", + mappings=(Mapping("Home", "docs/README.md"),), + ) + client = Mock() + client.list_pages.return_value = [metadata] + client.get_page_from_metadata.return_value = self.page + with patch("wiki_docs.ROOT", root): + sync_all(config, client, check=True, deep_check=True) + client.list_pages.assert_called_once_with() + client.get_page_from_metadata.assert_called_once_with(metadata, "Home") + + @patch("harness.run_check", return_value=0) + @patch("harness.sync_all", return_value=["已更新：docs/README.md"]) + @patch("harness.WikiClient") + @patch("harness.load_config") + def test_verify_reuses_one_deep_sync( + self, load, client_class, sync, check + ) -> None: + config = Mock() + load.return_value = config + result = run_sync( + Namespace(verify=True, check=False, deep_check=False, config="config.json") + ) + self.assertEqual(result, 0) + sync.assert_called_once_with( + config, client_class.return_value, check=False, deep_check=True + ) + check.assert_called_once() + class WikiClientTests(unittest.TestCase): def test_encoded_unicode_sub_url_is_not_double_encoded(self) -> None: