From ea1ba2c35ba1bc3e2401a995456dc99bcfc4ee25 Mon Sep 17 00:00:00 2001
From: yuqei <cyuqei@gmail.com>
Date: Fri, 22 May 2026 17:39:30 +0800
Subject: [PATCH] feat: add legal source diff review

---
 .../tests/test_legal_source_diff_report.py    | 128 ++++++
 backend/tools/legal_source_diff_report.py     | 405 ++++++++++++++++++
 ...µ·è�‰ä¸Žå®¡æŸ¥åŠ©æ‰‹-当å‰�工作进度ä¿�å­˜-2026-05-11.md |  32 ++
 scripts/run-legal-source-diff-report.ps1      | 243 +++++++++++
 4 files changed, 808 insertions(+)
 create mode 100644 backend/tests/test_legal_source_diff_report.py
 create mode 100644 backend/tools/legal_source_diff_report.py
 create mode 100644 scripts/run-legal-source-diff-report.ps1

diff --git a/backend/tests/test_legal_source_diff_report.py b/backend/tests/test_legal_source_diff_report.py
new file mode 100644
index 00000000..4a336edc
--- /dev/null
+++ b/backend/tests/test_legal_source_diff_report.py
@@ -0,0 +1,128 @@
+import io
+import json
+import tarfile
+import tempfile
+import unittest
+from datetime import datetime
+from pathlib import Path
+
+from backend.tools.legal_source_diff_report import build_diff_report
+
+
+def write_tar_with_metadata(archive_path: Path, entries: list[tuple[str, dict]]) -> None:
+    with tarfile.open(archive_path, "w:gz") as tar:
+        for folder_name, payload in entries:
+            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+            info = tarfile.TarInfo(name=f"cn-laws-download/output/{folder_name}/metadata.json")
+            info.size = len(data)
+            info.mtime = int(datetime.now().timestamp())
+            tar.addfile(info, io.BytesIO(data))
+
+            fulltext = b"sample fulltext"
+            fulltext_info = tarfile.TarInfo(name=f"cn-laws-download/output/{folder_name}/fulltext.txt")
+            fulltext_info.size = len(fulltext)
+            fulltext_info.mtime = int(datetime.now().timestamp())
+            tar.addfile(fulltext_info, io.BytesIO(fulltext))
+
+
+class LegalSourceDiffReportTestCase(unittest.TestCase):
+    def test_reports_missing_sources_and_same_title_coverage(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            archive_path = temp_path / "cn-laws-download.tar.gz"
+            output_dir = temp_path / "out"
+            remote_json = temp_path / "remote.json"
+            prepare_json = temp_path / "prepare.json"
+            import_json = temp_path / "import.json"
+
+            entries = [
+                (
+                    "src-a",
+                    {
+                        "bbbs": "SRC-A",
+                        "row": {
+                            "bbbs": "SRC-A",
+                            "title": "Alpha Regulation",
+                            "flxz": "地方法规",
+                            "zdjgName": "Authority A",
+                            "sxrq": "2026-01-01",
+                        },
+                    },
+                ),
+                (
+                    "src-b",
+                    {
+                        "bbbs": "SRC-B",
+                        "row": {
+                            "bbbs": "SRC-B",
+                            "title": "Beta Regulation",
+                            "flxz": "地方规章",
+                            "zdjgName": "Authority B",
+                            "sxrq": "2026-02-01",
+                        },
+                    },
+                ),
+                (
+                    "src-c",
+                    {
+                        "bbbs": "SRC-C",
+                        "row": {
+                            "bbbs": "SRC-C",
+                            "title": "Gamma Regulation",
+                            "flxz": "�法解释",
+                            "zdjgName": "Authority C",
+                            "sxrq": "2026-03-01",
+                        },
+                    },
+                ),
+            ]
+            write_tar_with_metadata(archive_path, entries)
+
+            remote_json.write_text(
+                json.dumps(
+                    {
+                        "tenant_id": "tenant-demo-001",
+                        "legal_article_count": 3,
+                        "remote_source_count": 2,
+                        "remote_title_count": 2,
+                        "sources": [
+                            {
+                                "source_id": "FLK-SRC-A",
+                                "source_title": "Alpha Regulation",
+                                "article_count": 10,
+                            },
+                            {
+                                "source_id": "FLK-SRC-Z",
+                                "source_title": "Gamma Regulation",
+                                "article_count": 5,
+                            },
+                        ],
+                    },
+                    ensure_ascii=False,
+                ),
+                encoding="utf-8",
+            )
+            prepare_json.write_text(json.dumps({"rows": 3}, ensure_ascii=False), encoding="utf-8")
+            import_json.write_text(json.dumps({"inserted": 3, "skipped_existing_duplicates": 0}, ensure_ascii=False), encoding="utf-8")
+
+            summary = build_diff_report(
+                laws_backup_dir=archive_path,
+                remote_source_json=remote_json,
+                prepare_summary_json=prepare_json,
+                import_summary_json=import_json,
+                output_dir=output_dir,
+                tenant_id="tenant-demo-001",
+            )
+
+            self.assertEqual(summary["status"], "WARN")
+            self.assertEqual(summary["counts"]["backup_sources"], 3)
+            self.assertEqual(summary["counts"]["missing_sources_in_remote"], 2)
+            self.assertEqual(summary["counts"]["covered_by_remote_same_title"], 1)
+            self.assertEqual(summary["counts"]["needs_incremental_review"], 1)
+            self.assertEqual(summary["counts"]["extra_remote_sources"], 1)
+            self.assertTrue((output_dir / "legal-source-diff-report.json").exists())
+            self.assertTrue((output_dir / "missing-sources.csv").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/backend/tools/legal_source_diff_report.py b/backend/tools/legal_source_diff_report.py
new file mode 100644
index 00000000..32eaa257
--- /dev/null
+++ b/backend/tools/legal_source_diff_report.py
@@ -0,0 +1,405 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import tarfile
+from collections import Counter, defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def read_json(path: Path | None) -> dict[str, Any]:
+    if not path or not path.exists():
+        return {}
+    return json.loads(path.read_text(encoding="utf-8-sig"))
+
+
+def latest_file(directory: Path, pattern: str) -> Path | None:
+    files = sorted(directory.glob(pattern), key=lambda item: item.stat().st_mtime, reverse=True)
+    return files[0] if files else None
+
+
+def find_laws_archive(laws_backup_dir: Path) -> Path:
+    if laws_backup_dir.is_file() and laws_backup_dir.name.endswith(".tar.gz"):
+        return laws_backup_dir
+    archive = latest_file(laws_backup_dir, "*.tar.gz")
+    if archive is None:
+        raise FileNotFoundError(f"cn-laws-download archive not found under {laws_backup_dir}")
+    return archive
+
+
+def normalize_text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def normalize_source_id(value: Any) -> str:
+    source_id = normalize_text(value)
+    if source_id.upper().startswith("FLK-"):
+        return source_id[4:]
+    return source_id
+
+
+def metadata_parent(member_name: str) -> str:
+    return str(Path(member_name).parent).replace("\\", "/")
+
+
+def pick_source_id(payload: dict[str, Any], member_name: str) -> str:
+    row = payload.get("row") or {}
+    detail_data = ((payload.get("detail") or {}).get("data") or {})
+    source_id = normalize_text(payload.get("bbbs") or row.get("bbbs") or detail_data.get("bbbs"))
+    if source_id:
+        return source_id
+    parts = Path(member_name).parts
+    if len(parts) >= 3 and parts[-1] == "metadata.json":
+        return parts[-2]
+    return ""
+
+
+def scan_backup_sources(laws_backup_dir: Path) -> dict[str, Any]:
+    archive = find_laws_archive(laws_backup_dir)
+    sources_by_id: dict[str, dict[str, Any]] = {}
+    duplicate_ids: list[dict[str, Any]] = []
+    fulltext_parents: set[str] = set()
+
+    with tarfile.open(archive, "r:gz") as tar:
+        for member in tar:
+            if member.isfile() and member.name.endswith("/fulltext.txt"):
+                fulltext_parents.add(metadata_parent(member.name))
+                continue
+            if not (member.isfile() and member.name.endswith("/metadata.json")):
+                continue
+            file_obj = tar.extractfile(member)
+            if file_obj is None:
+                continue
+            try:
+                payload = json.loads(file_obj.read().decode("utf-8"))
+            except Exception as exc:  # noqa: BLE001
+                duplicate_ids.append({"source_id": "", "metadata_path": member.name, "error": str(exc)})
+                continue
+
+            row = payload.get("row") or {}
+            detail_data = ((payload.get("detail") or {}).get("data") or {})
+            source_id = pick_source_id(payload, member.name)
+            title = normalize_text(row.get("title") or detail_data.get("title"))
+            source = {
+                "source_id": source_id,
+                "canonical_source_id": normalize_source_id(source_id),
+                "source_title": title,
+                "source_type": normalize_text(row.get("flxz") or detail_data.get("flxz") or "未知"),
+                "authority": normalize_text(row.get("zdjgName") or detail_data.get("zdjgName")),
+                "effective_date": normalize_text(row.get("sxrq") or detail_data.get("sxrq")),
+                "publish_date": normalize_text(row.get("gbrq") or detail_data.get("gbrq")),
+                "validity_status": normalize_text(row.get("sxx") or detail_data.get("sxx")),
+                "source_page": payload.get("source_page"),
+                "metadata_path": member.name,
+                "source_directory": metadata_parent(member.name),
+                "fulltext_length": int(payload.get("fulltext_length") or 0),
+                "has_reader_file": bool(payload.get("reader_file")),
+                "has_ofd_path": bool(payload.get("ofd_path")),
+                "saved_at": normalize_text(payload.get("saved_at")),
+            }
+            if not source_id:
+                duplicate_ids.append(source | {"error": "missing_source_id"})
+                continue
+            if source_id in sources_by_id:
+                duplicate_ids.append(source | {"error": "duplicate_source_id"})
+                continue
+            sources_by_id[source_id] = source
+
+    for source in sources_by_id.values():
+        source["has_fulltext"] = source["source_directory"] in fulltext_parents
+
+    return {
+        "archive": str(archive),
+        "sources": sources_by_id,
+        "duplicate_or_invalid_metadata": duplicate_ids,
+        "fulltext_documents": len(fulltext_parents),
+    }
+
+
+def build_title_index(sources: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
+    title_index: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for source in sources:
+        title = normalize_text(source.get("source_title"))
+        if title:
+            title_index[title].append(source)
+    return title_index
+
+
+def classify_missing_source(source: dict[str, Any], remote_title_index: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
+    title = normalize_text(source.get("source_title"))
+    title_matches = remote_title_index.get(title, [])
+    if title_matches:
+        remote_article_count = sum(int(item.get("article_count") or 0) for item in title_matches)
+        remote_source_ids = [item.get("source_id") for item in title_matches]
+        return {
+            **source,
+            "classification": "covered_by_remote_same_title",
+            "remote_source_ids_same_title": ";".join(remote_source_ids),
+            "remote_same_title_count": len(title_matches),
+            "remote_same_title_article_count": remote_article_count,
+            "action_suggestion": "暂���导入；�标题内容已在�行库中覆盖，�续�需抽样核验。",
+        }
+    return {
+        **source,
+        "classification": "needs_incremental_review",
+        "remote_source_ids_same_title": "",
+        "remote_same_title_count": 0,
+        "remote_same_title_article_count": 0,
+        "action_suggestion": "需�进入增�补抓/补导入�核队列。",
+    }
+
+
+def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
+    with path.open("w", encoding="utf-8-sig", newline="") as file_obj:
+        writer = csv.DictWriter(file_obj, fieldnames=fieldnames, extrasaction="ignore")
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def markdown_table(rows: list[list[Any]]) -> str:
+    if not rows:
+        return ""
+    lines = [
+        "| " + " | ".join(str(item) for item in rows[0]) + " |",
+        "| " + " | ".join("---" for _ in rows[0]) + " |",
+    ]
+    for row in rows[1:]:
+        lines.append("| " + " | ".join(str(item).replace("\n", " ") for item in row) + " |")
+    return "\n".join(lines)
+
+
+def build_diff_report(
+    *,
+    laws_backup_dir: Path,
+    remote_source_json: Path,
+    prepare_summary_json: Path | None,
+    import_summary_json: Path | None,
+    output_dir: Path,
+    tenant_id: str,
+) -> dict[str, Any]:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    backup_scan = scan_backup_sources(laws_backup_dir)
+    remote_payload = read_json(remote_source_json)
+    prepare_summary = read_json(prepare_summary_json)
+    import_summary = read_json(import_summary_json)
+
+    backup_sources_by_id = backup_scan["sources"]
+    remote_sources = remote_payload.get("sources") or []
+    remote_sources_by_id = {
+        normalize_source_id(item.get("source_id")): item
+        for item in remote_sources
+        if normalize_source_id(item.get("source_id"))
+    }
+    backup_sources_by_canonical_id = {
+        normalize_source_id(source.get("source_id")): source
+        for source in backup_sources_by_id.values()
+        if normalize_source_id(source.get("source_id"))
+    }
+    remote_title_index = build_title_index(remote_sources)
+
+    backup_ids = set(backup_sources_by_canonical_id)
+    remote_ids = set(remote_sources_by_id)
+    missing_sources = [
+        classify_missing_source(backup_sources_by_canonical_id[source_id], remote_title_index)
+        for source_id in sorted(backup_ids - remote_ids)
+    ]
+    extra_remote_sources = [
+        remote_sources_by_id[source_id]
+        for source_id in sorted(remote_ids - backup_ids)
+    ]
+
+    classification_counts = Counter(item["classification"] for item in missing_sources)
+    source_type_counts = Counter(normalize_text(item.get("source_type") or "未知") for item in missing_sources)
+    remote_duplicate_titles = [
+        {
+            "source_title": title,
+            "remote_source_count": len(items),
+            "remote_source_ids": ";".join(str(item.get("source_id")) for item in items),
+            "remote_article_count": sum(int(item.get("article_count") or 0) for item in items),
+        }
+        for title, items in remote_title_index.items()
+        if len(items) > 1
+    ]
+    backup_title_index = build_title_index(list(backup_sources_by_id.values()))
+    backup_duplicate_titles = [
+        {
+            "source_title": title,
+            "backup_source_count": len(items),
+            "backup_source_ids": ";".join(str(item.get("source_id")) for item in items),
+        }
+        for title, items in backup_title_index.items()
+        if len(items) > 1
+    ]
+
+    needs_incremental_review = [
+        item for item in missing_sources if item["classification"] == "needs_incremental_review"
+    ]
+    covered_by_title = [
+        item for item in missing_sources if item["classification"] == "covered_by_remote_same_title"
+    ]
+
+    status = "PASS" if not needs_incremental_review else "WARN"
+    summary = {
+        "status": status,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "tenant_id": tenant_id,
+        "laws_backup_dir": str(laws_backup_dir),
+        "archive": backup_scan["archive"],
+        "remote_source_json": str(remote_source_json),
+        "prepare_summary_json": str(prepare_summary_json) if prepare_summary_json else None,
+        "import_summary_json": str(import_summary_json) if import_summary_json else None,
+        "counts": {
+            "backup_sources": len(backup_sources_by_id),
+            "backup_fulltext_documents": backup_scan["fulltext_documents"],
+            "remote_sources": len(remote_sources_by_id),
+            "remote_legal_articles": int(remote_payload.get("legal_article_count") or 0),
+            "missing_sources_in_remote": len(missing_sources),
+            "covered_by_remote_same_title": len(covered_by_title),
+            "needs_incremental_review": len(needs_incremental_review),
+            "extra_remote_sources": len(extra_remote_sources),
+            "backup_duplicate_title_groups": len(backup_duplicate_titles),
+            "remote_duplicate_title_groups": len(remote_duplicate_titles),
+            "invalid_or_duplicate_metadata": len(backup_scan["duplicate_or_invalid_metadata"]),
+            "prepared_article_rows": int(prepare_summary.get("rows") or 0),
+            "import_inserted": int(import_summary.get("inserted") or 0),
+            "import_skipped_existing_duplicates": int(import_summary.get("skipped_existing_duplicates") or 0),
+        },
+        "classification_counts": dict(classification_counts),
+        "missing_source_type_counts": dict(source_type_counts.most_common()),
+        "missing_sources": missing_sources,
+        "extra_remote_sources": extra_remote_sources,
+        "backup_duplicate_titles": backup_duplicate_titles[:200],
+        "remote_duplicate_titles": remote_duplicate_titles[:200],
+        "invalid_or_duplicate_metadata": backup_scan["duplicate_or_invalid_metadata"][:200],
+        "conclusion": (
+            "本轮没有�现必须立�补导入的��；缺失 source_id ��被远程�标题��覆盖。"
+            if not needs_incremental_review
+            else "存在远程�标题无法覆盖的��，需�进入增�补抓/补导入�核队列。"
+        ),
+    }
+
+    json_path = output_dir / "legal-source-diff-report.json"
+    md_path = output_dir / "legal-source-diff-report.md"
+    missing_csv = output_dir / "missing-sources.csv"
+    extra_csv = output_dir / "extra-remote-sources.csv"
+    covered_csv = output_dir / "covered-by-remote-title.csv"
+
+    json_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
+    common_fields = [
+        "source_id",
+        "canonical_source_id",
+        "source_title",
+        "source_type",
+        "authority",
+        "effective_date",
+        "publish_date",
+        "classification",
+        "remote_source_ids_same_title",
+        "remote_same_title_count",
+        "remote_same_title_article_count",
+        "action_suggestion",
+        "metadata_path",
+    ]
+    write_csv(missing_csv, missing_sources, common_fields)
+    write_csv(covered_csv, covered_by_title, common_fields)
+    write_csv(
+        extra_csv,
+        extra_remote_sources,
+        ["source_id", "source_title", "source_type", "authority", "effective_date", "article_count"],
+    )
+    write_markdown_report(md_path, summary, missing_sources)
+    return summary
+
+
+def write_markdown_report(path: Path, summary: dict[str, Any], missing_sources: list[dict[str, Any]]) -> None:
+    counts = summary["counts"]
+    sample_rows = [["source_id", "标题", "分类", "建议"]]
+    for item in missing_sources[:20]:
+        sample_rows.append(
+            [
+                item.get("source_id"),
+                item.get("source_title"),
+                item.get("classification"),
+                item.get("action_suggestion"),
+            ]
+        )
+
+    lines = [
+        "# 法律库��差异�核报告",
+        "",
+        f"- 状�：`{summary['status']}`",
+        f"- 生�时间：{summary['generated_at']}",
+        f"- 租户：`{summary['tenant_id']}`",
+        f"- 备份包：`{summary['archive']}`",
+        "- 说明：对比时已规范化远程 `source_id`，自动剥离 `FLK-` �缀���差异判断。",
+        "",
+        "## 结论",
+        "",
+        f"- {summary['conclusion']}",
+        f"- 备份包��数：`{counts['backup_sources']}`；远程�行库��数：`{counts['remote_sources']}`。",
+        f"- 远程 `legal_articles` �文数：`{counts['remote_legal_articles']}`。",
+        f"- 备份包有但远程 source_id 没有的��：`{counts['missing_sources_in_remote']}`。",
+        f"- 其中�标题已覆盖：`{counts['covered_by_remote_same_title']}`；需�增��核：`{counts['needs_incremental_review']}`。",
+        f"- 远程�外��：`{counts['extra_remote_sources']}`。",
+        "",
+        "## 关键统计",
+        "",
+        markdown_table(
+            [
+                ["指标", "数�"],
+                ["备份��", counts["backup_sources"]],
+                ["备份 fulltext 文档", counts["backup_fulltext_documents"]],
+                ["远程��", counts["remote_sources"]],
+                ["远程 legal_articles", counts["remote_legal_articles"]],
+                ["缺失 source_id", counts["missing_sources_in_remote"]],
+                ["�标题覆盖", counts["covered_by_remote_same_title"]],
+                ["需增��核", counts["needs_incremental_review"]],
+                ["远程�外��", counts["extra_remote_sources"]],
+                ["备份��标题组", counts["backup_duplicate_title_groups"]],
+                ["远程��标题组", counts["remote_duplicate_title_groups"]],
+            ]
+        ),
+        "",
+        "## 缺失��样例",
+        "",
+        markdown_table(sample_rows) if len(sample_rows) > 1 else "无。",
+        "",
+        "## 输出文件",
+        "",
+        "- `legal-source-diff-report.json`",
+        "- `missing-sources.csv`",
+        "- `covered-by-remote-title.csv`",
+        "- `extra-remote-sources.csv`",
+        "",
+    ]
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Compare cn-laws-download source metadata with remote legal_articles sources.")
+    parser.add_argument("--laws-backup-dir", type=Path, required=True)
+    parser.add_argument("--remote-source-json", type=Path, required=True)
+    parser.add_argument("--prepare-summary-json", type=Path)
+    parser.add_argument("--import-summary-json", type=Path)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--tenant-id", default="tenant-demo-001")
+    args = parser.parse_args()
+
+    summary = build_diff_report(
+        laws_backup_dir=args.laws_backup_dir,
+        remote_source_json=args.remote_source_json,
+        prepare_summary_json=args.prepare_summary_json,
+        import_summary_json=args.import_summary_json,
+        output_dir=args.output_dir,
+        tenant_id=args.tenant_id,
+    )
+    marker = "PASS" if summary["status"] == "PASS" else "WARN"
+    print(f"LEGAL_SOURCE_DIFF_REPORT_{marker} {args.output_dir}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/project-docs/AI��起�与审查助手-当�工作进度�存-2026-05-11.md b/project-docs/AI��起�与审查助手-当�工作进度�存-2026-05-11.md
index c09f4b26..2895ed10 100644
--- a/project-docs/AI��起�与审查助手-当�工作进度�存-2026-05-11.md
+++ b/project-docs/AI��起�与审查助手-当�工作进度�存-2026-05-11.md
@@ -4663,3 +4663,35 @@
   - Phase13 模��审查规则�标准�款已完�正��行表�收�版本化�治�资产绑定��布和远程真相��核。
   - 本轮没有 required 阻塞项。
   - �续若继续补资产能力，下一步应转�“法律库增�抓�/��差异�核�或“全�法规分批��化任务�，而�是继续补 Phase13 模�/规则/�款�行表。
+
+## 2026-05-22 法律库��差异�核闭环
+
+- 本轮目标：
+  - ��盲目抓�，先把备份包 `cn-laws-download_20260519_155247` 与远程 `legal_articles` ���级差异�核。
+  - �点确认：哪些��真的缺失，哪些�是 `source_id` �缀��但内容已覆盖，哪些需��续增�补抓。
+- 新增工具：
+  - `backend/tools/legal_source_diff_report.py`
+  - `backend/tests/test_legal_source_diff_report.py`
+  - `scripts/run-legal-source-diff-report.ps1`
+- �核规则：
+  - 远程�� `source_id` 会先�规范化，自动剥离 `FLK-` �缀��对比。
+  - 如果�标题内容已�在�行库中覆盖，则标记为“暂���导入�。
+  - �有确实没有�标题覆盖的���进入�续增�补抓。
+- 本地验�：
+  - `python -m compileall backend\tools\legal_source_diff_report.py backend\tests\test_legal_source_diff_report.py`
+  - `python -m unittest backend.tests.test_legal_source_diff_report`
+  - 结果：`Ran 1 test in 0.022s`，`OK`。
+- 远程�核：
+  - 命令：`$env:SSHPASS='***'; powershell -NoProfile -ExecutionPolicy Bypass -File .\scripts\run-legal-source-diff-report.ps1 -UseSshpass`
+  - 结果：`LEGAL_SOURCE_DIFF_REPORT_PASS D:\ai-contract-platform\project-docs\qa-artifacts\legal-source-diff-report\2026-05-22T17-37-17`
+  - 报告结论：本轮没有�现必须立�补导入的��；缺失 `source_id` ��被远程�标题��覆盖。
+  - 关键统计：
+    - 备份�� `17235`
+    - 远程�� `17219`
+    - 缺失 `source_id` `16`
+    - �标题覆盖 `16`
+    - 需增��核 `0`
+    - 远程�外�� `0`
+- 当�判断：
+  - 法律库这批备份包�需���全�导入。
+  - 下一步若继续法律库方�，优先�“这 16 个�标题覆盖��的抽样核验�，或者转去“法律库增�抓�/全�法规分批��化�。
diff --git a/scripts/run-legal-source-diff-report.ps1 b/scripts/run-legal-source-diff-report.ps1
new file mode 100644
index 00000000..4e9a7d06
--- /dev/null
+++ b/scripts/run-legal-source-diff-report.ps1
@@ -0,0 +1,243 @@
+param(
+  [string]$LawsBackupDir = "E:\DB_Backups\192.168.3.90_20260519_114618\file_backups\cn-laws-download_20260519_155247",
+  [string]$ArtifactRoot = "",
+  [string]$RemoteHost = "192.168.3.90",
+  [string]$RemoteUser = "root",
+  [string]$BackendContainer = "ai-contract-backend-restore",
+  [string]$SshKeyPath = "$HOME\.ssh\id_rsa",
+  [switch]$UseSshpass,
+  [string]$TenantId = "tenant-demo-001",
+  [string]$PrepareSummaryJson = "",
+  [string]$ImportSummaryJson = ""
+)
+
+$ErrorActionPreference = "Stop"
+
+function Resolve-SshpassPath {
+  $command = Get-Command sshpass.exe -ErrorAction SilentlyContinue
+  if ($null -ne $command) {
+    return $command.Source
+  }
+
+  $candidates = @(
+    (Join-Path $env:LOCALAPPDATA "Microsoft\WinGet\Packages\xhcoding.sshpass-win32_Microsoft.Winget.Source_8wekyb3d8bbwe\sshpass.exe")
+  )
+  foreach ($candidate in $candidates) {
+    if (Test-Path $candidate) {
+      return $candidate
+    }
+  }
+
+  return $null
+}
+
+function Invoke-LoggedCommand {
+  param(
+    [string]$Name,
+    [string]$FilePath,
+    [string[]]$Arguments,
+    [string]$LogDir
+  )
+
+  $logPath = Join-Path $LogDir "$Name.log"
+  $previousErrorActionPreference = $ErrorActionPreference
+  $ErrorActionPreference = "Continue"
+  try {
+    $output = & $FilePath @Arguments 2>&1
+    $exitCode = if ($LASTEXITCODE -is [int]) { $LASTEXITCODE } else { 0 }
+  } finally {
+    $ErrorActionPreference = $previousErrorActionPreference
+  }
+
+  $outputText = ($output -join [Environment]::NewLine)
+  $outputText | Set-Content -Encoding UTF8 -Path $logPath
+  if ($exitCode -ne 0) {
+    throw "$Name failed with exit code $exitCode. See $logPath"
+  }
+  return $outputText
+}
+
+function Get-JsonFromOutput {
+  param([string]$Text)
+
+  $start = $Text.IndexOf("{")
+  if ($start -lt 0) {
+    throw "Remote output did not contain a JSON object."
+  }
+  return $Text.Substring($start)
+}
+
+$repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path
+if (-not (Test-Path "$HOME\.Codex\skills\gstack\bin")) {
+  throw "GSTACK_MISSING: gstack is required for AI-assisted work in this repo."
+}
+
+if ([string]::IsNullOrWhiteSpace($ArtifactRoot)) {
+  $ArtifactRoot = Join-Path $repoRoot "project-docs\qa-artifacts\legal-source-diff-report"
+}
+$timestamp = Get-Date -Format "yyyy-MM-ddTHH-mm-ss"
+$artifactDir = Join-Path $ArtifactRoot $timestamp
+New-Item -ItemType Directory -Force -Path $artifactDir | Out-Null
+
+if ([string]::IsNullOrWhiteSpace($PrepareSummaryJson)) {
+  $PrepareSummaryJson = Join-Path $repoRoot "project-docs\qa-artifacts\cn-laws-download-import\2026-05-20T09-53-00\cn_laws_prepare_summary_20260520.json"
+}
+if ([string]::IsNullOrWhiteSpace($ImportSummaryJson)) {
+  $ImportSummaryJson = Join-Path $repoRoot "project-docs\qa-artifacts\cn-laws-download-import\2026-05-20T09-53-00\cn_laws_import_summary_20260520.json"
+}
+
+foreach ($requiredPath in @($LawsBackupDir, $PrepareSummaryJson, $ImportSummaryJson)) {
+  if (-not (Test-Path $requiredPath)) {
+    throw "Required input path does not exist: $requiredPath"
+  }
+}
+
+$authMode = ""
+$sshpassPath = $null
+if ($UseSshpass -or -not [string]::IsNullOrWhiteSpace($env:SSHPASS)) {
+  $sshpassPath = Resolve-SshpassPath
+  if ([string]::IsNullOrWhiteSpace($sshpassPath)) {
+    throw "sshpass.exe not found. Install sshpass or use SSH key auth."
+  }
+  if ([string]::IsNullOrWhiteSpace($env:SSHPASS)) {
+    throw "SSHPASS environment variable is empty."
+  }
+  $authMode = "sshpass"
+} elseif (Test-Path $SshKeyPath) {
+  $authMode = "key"
+} else {
+  throw "SSH key not found and SSHPASS is not configured."
+}
+
+$remoteScript = @'
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timezone
+from sqlalchemy import create_engine, text
+
+
+tenant_id = "__TENANT_ID__"
+database_url = os.environ.get("DATABASE_URL")
+if not database_url:
+    raise RuntimeError("DATABASE_URL is required in backend container")
+
+engine = create_engine(database_url)
+with engine.connect() as conn:
+    summary = dict(
+        conn.execute(
+            text(
+                """
+                select
+                  count(*) as legal_article_count,
+                  count(distinct source_id) as remote_source_count,
+                  count(distinct source_title) as remote_title_count
+                from legal_articles
+                where tenant_id = :tenant_id
+                """
+            ),
+            {"tenant_id": tenant_id},
+        ).first()._mapping
+    )
+    sources = [
+        dict(row._mapping)
+        for row in conn.execute(
+            text(
+                """
+                select
+                  source_id,
+                  min(source_title) as source_title,
+                  min(source_type) as source_type,
+                  min(authority) as authority,
+                  min(effective_date) as effective_date,
+                  min(validity_status) as validity_status,
+                  min(region) as region,
+                  min(official_url) as official_url,
+                  count(*) as article_count,
+                  count(distinct text_checksum) as text_checksum_count,
+                  min(created_at) as first_created_at,
+                  max(updated_at) as last_updated_at
+                from legal_articles
+                where tenant_id = :tenant_id
+                group by source_id
+                order by source_id
+                """
+            ),
+            {"tenant_id": tenant_id},
+        )
+    ]
+
+payload = {
+    "generated_at": datetime.now(timezone.utc).isoformat(),
+    "tenant_id": tenant_id,
+    "legal_article_count": int(summary["legal_article_count"] or 0),
+    "remote_source_count": int(summary["remote_source_count"] or 0),
+    "remote_title_count": int(summary["remote_title_count"] or 0),
+    "sources": sources,
+}
+print(json.dumps(payload, ensure_ascii=False, default=str, indent=2))
+'@
+$remoteScript = $remoteScript.Replace("__TENANT_ID__", $TenantId)
+
+$localRemoteScript = Join-Path $artifactDir "remote-legal-source-snapshot.py"
+$remoteSnapshotPath = Join-Path $artifactDir "remote-legal-source-snapshot.json"
+$remoteScript | Set-Content -Encoding UTF8 -Path $localRemoteScript
+
+$target = "${RemoteUser}@${RemoteHost}"
+$remoteTempScript = "/tmp/legal_source_diff_remote_snapshot.py"
+
+if ($authMode -eq "sshpass") {
+  Invoke-LoggedCommand -Name "scp-remote-legal-source-script" -FilePath $sshpassPath -Arguments @(
+    "-e",
+    "scp",
+    "-o", "StrictHostKeyChecking=accept-new",
+    "-o", "ConnectTimeout=10",
+    $localRemoteScript,
+    "${target}:$remoteTempScript"
+  ) -LogDir $artifactDir | Out-Null
+  $remoteOutput = Invoke-LoggedCommand -Name "remote-legal-source-snapshot" -FilePath $sshpassPath -Arguments @(
+    "-e",
+    "ssh",
+    "-o", "StrictHostKeyChecking=accept-new",
+    "-o", "ConnectTimeout=10",
+    $target,
+    "docker cp $remoteTempScript ${BackendContainer}:$remoteTempScript && docker exec $BackendContainer sh -lc 'cd /app && python $remoteTempScript'"
+  ) -LogDir $artifactDir
+} else {
+  Invoke-LoggedCommand -Name "scp-remote-legal-source-script" -FilePath "scp" -Arguments @(
+    "-i", $SshKeyPath,
+    "-o", "BatchMode=yes",
+    "-o", "StrictHostKeyChecking=accept-new",
+    "-o", "ConnectTimeout=10",
+    $localRemoteScript,
+    "${target}:$remoteTempScript"
+  ) -LogDir $artifactDir | Out-Null
+  $remoteOutput = Invoke-LoggedCommand -Name "remote-legal-source-snapshot" -FilePath "ssh" -Arguments @(
+    "-i", $SshKeyPath,
+    "-o", "BatchMode=yes",
+    "-o", "StrictHostKeyChecking=accept-new",
+    "-o", "ConnectTimeout=10",
+    $target,
+    "docker cp $remoteTempScript ${BackendContainer}:$remoteTempScript && docker exec $BackendContainer sh -lc 'cd /app && python $remoteTempScript'"
+  ) -LogDir $artifactDir
+}
+
+Get-JsonFromOutput $remoteOutput | Set-Content -Encoding UTF8 -Path $remoteSnapshotPath
+
+$toolPath = Join-Path $repoRoot "backend\tools\legal_source_diff_report.py"
+$toolOutput = & python $toolPath `
+  --laws-backup-dir $LawsBackupDir `
+  --remote-source-json $remoteSnapshotPath `
+  --prepare-summary-json $PrepareSummaryJson `
+  --import-summary-json $ImportSummaryJson `
+  --tenant-id $TenantId `
+  --output-dir $artifactDir
+
+$toolExitCode = if ($LASTEXITCODE -is [int]) { $LASTEXITCODE } else { 0 }
+($toolOutput -join [Environment]::NewLine) | Set-Content -Encoding UTF8 -Path (Join-Path $artifactDir "legal-source-diff-report-tool.log")
+if ($toolExitCode -ne 0) {
+  throw "legal_source_diff_report.py failed with exit code $toolExitCode."
+}
+
+Write-Output ($toolOutput -join [Environment]::NewLine)