feat: add legal source diff review

This commit is contained in:
2026-05-22 17:39:30 +08:00
parent ed2ce8a40d
commit ea1ba2c35b
4 changed files with 808 additions and 0 deletions
@@ -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()
+405
View File
@@ -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())
@@ -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 个同标题覆盖来源的抽样核验”,或者转去“法律库增量抓取/全量法规分批向量化”。
+243
View File
@@ -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)