ai-contract-platform/backend/tools/verify_phase14_source_review_queue.py

from __future__ import annotations

import argparse
import csv
import gzip
import json
import os
import re
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


def load_csv(path: Path) -> list[dict[str, str]]:
    with path.open("r", encoding="utf-8-sig", newline="") as file_obj:
        return list(csv.DictReader(file_obj))


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def iter_prepared_rows(path: Path) -> list[dict[str, Any]]:
    opener = gzip.open if path.suffix == ".gz" else open
    with opener(path, "rt", encoding="utf-8") as file_obj:
        return [json.loads(line) for line in file_obj if line.strip()]


def login(api_base: str, admin_email: str, admin_password: str) -> str:
    request = urllib.request.Request(
        f"{api_base.rstrip('/')}/auth/login",
        method="POST",
        headers={"Content-Type": "application/json; charset=utf-8"},
        data=json.dumps({"account": admin_email, "password": admin_password}).encode("utf-8"),
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        payload = json.loads(response.read().decode("utf-8"))
    token = (((payload or {}).get("data") or {}).get("access_token") or "").strip()
    if not token:
        raise RuntimeError("Login response missing access_token")
    return token


def fetch_paged_items(api_base: str, token: str, path: str, params: dict[str, str]) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    page = 1
    while True:
        query = {**params, "page": str(page), "page_size": "500"}
        url = f"{api_base.rstrip('/')}/{path.lstrip('/')}?{urllib.parse.urlencode(query)}"
        request = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
        with urllib.request.urlopen(request, timeout=30) as response:
            payload = json.loads(response.read().decode("utf-8"))
        data = (payload or {}).get("data") or {}
        batch = list(data.get("items") or [])
        items.extend(batch)
        total = int(data.get("total") or len(items))
        if len(items) >= total or not batch:
            break
        page += 1
    return items


def split_remote_ids(value: str) -> list[str]:
    parts = re.split(r"[;,|]\s*", value.strip()) if value else []
    return [part for part in parts if part]


def normalize_source_id(value: str) -> str:
    text = value.strip()
    if text.startswith("FLK-"):
        return text[4:]
    return text


def classify(local_rows: list[dict[str, Any]], remote_rows: list[dict[str, Any]], checksum_reason: str) -> str:
    local_article_set = {str(row.get("article_no") or "").strip() for row in local_rows if str(row.get("article_no") or "").strip()}
    remote_article_set = {str(row.get("article_no") or "").strip() for row in remote_rows if str(row.get("article_no") or "").strip()}
    local_checksum_set = {str(row.get("text_checksum") or "").strip() for row in local_rows if str(row.get("text_checksum") or "").strip()}
    remote_checksum_set = {str(row.get("text_checksum") or "").strip() for row in remote_rows if str(row.get("text_checksum") or "").strip()}

    if checksum_reason:
        return "manual_review_required_checksum_diff"
    if local_article_set == remote_article_set and local_checksum_set == remote_checksum_set:
        return "verified_exact_equivalent"
    if local_article_set.issubset(remote_article_set) and local_checksum_set.issubset(remote_checksum_set):
        return "verified_remote_superset"
    if local_article_set & remote_article_set:
        return "manual_review_required_partial_overlap"
    return "manual_review_required_no_overlap"


def build_detail(local_rows: list[dict[str, Any]], remote_rows: list[dict[str, Any]]) -> dict[str, Any]:
    local_articles = sorted({str(row.get("article_no") or "").strip() for row in local_rows if str(row.get("article_no") or "").strip()})
    remote_articles = sorted({str(row.get("article_no") or "").strip() for row in remote_rows if str(row.get("article_no") or "").strip()})
    local_checksums = {str(row.get("text_checksum") or "").strip() for row in local_rows if str(row.get("text_checksum") or "").strip()}
    remote_checksums = {str(row.get("text_checksum") or "").strip() for row in remote_rows if str(row.get("text_checksum") or "").strip()}
    overlap_articles = sorted(set(local_articles) & set(remote_articles))
    overlap_checksums = sorted(local_checksums & remote_checksums)
    return {
        "local_article_count": len(local_articles),
        "remote_article_count": len(remote_articles),
        "article_overlap_count": len(overlap_articles),
        "checksum_overlap_count": len(overlap_checksums),
        "local_only_articles": [item for item in local_articles if item not in remote_articles][:10],
        "remote_only_articles": [item for item in remote_articles if item not in local_articles][:10],
        "sample_overlap_articles": overlap_articles[:10],
    }


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = [
        "priority",
        "source_id",
        "source_title",
        "remote_source_ids_same_title",
        "classification_result",
        "local_article_count",
        "remote_article_count",
        "article_overlap_count",
        "checksum_overlap_count",
        "checksum_mismatch_reason",
        "action_suggestion",
    ]
    with path.open("w", encoding="utf-8", newline="") as file_obj:
        writer = csv.DictWriter(file_obj, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow({key: row.get(key, "") for key in fieldnames})


def write_md(path: Path, summary: dict[str, Any], rows: list[dict[str, Any]]) -> None:
    lines = [
        "# Phase14 来源抽样核验报告",
        "",
        f"- 状态：`{summary['status']}`",
        f"- 生成时间：`{summary['generated_at']}`",
        f"- 队列条目：`{summary['queue_count']}`",
        f"- 自动核验通过：`{summary['verified_count']}`",
        f"- 仍需人工复核：`{summary['manual_review_count']}`",
        "",
        "## 分类汇总",
        "",
        "| 分类 | 数量 |",
        "|---|---:|",
    ]
    for key, value in summary["classification_counts"].items():
        lines.append(f"| {key} | {value} |")
    lines.extend(
        [
            "",
            "## 逐条结果",
            "",
            "| source_id | 标题 | 结果 | local | remote | overlap | 建议 |",
            "|---|---|---|---:|---:|---:|---|",
        ]
    )
    for row in rows:
        lines.append(
            f"| {row['source_id']} | {row['source_title']} | {row['classification_result']} | "
            f"{row['local_article_count']} | {row['remote_article_count']} | {row['article_overlap_count']} | "
            f"{row['action_suggestion']} |"
        )
    lines.extend(
        [
            "",
            "## 结论",
            "",
            "- `verified_exact_equivalent`：本地备份与远端同标题来源在条号和正文 checksum 上完全对齐，可继续视为无需重复导入。",
            "- `verified_remote_superset`：远端同标题来源覆盖了本地条号和 checksum，可继续视为无需重复导入，但远端可能还包含更多条文。",
            "- `manual_review_required_*`：应保留到抽样人工复核队列，优先看正文 checksum 漂移和条号差异。",
            "",
        ]
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines), encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description="Verify Phase14 source review queue against prepared JSONL and remote legal_articles.")
    parser.add_argument("--queue-csv", type=Path, required=True)
    parser.add_argument("--prepare-summary-json", type=Path, required=True)
    parser.add_argument("--api-base-url", required=True)
    parser.add_argument("--admin-email", default="admin@example.com")
    parser.add_argument("--admin-password")
    parser.add_argument("--output-dir", type=Path, required=True)
    args = parser.parse_args()

    admin_password = (
        args.admin_password
        or os.environ.get("AICP_ADMIN_PASSWORD")
        or os.environ.get("AI_CONTRACT_ADMIN_PASSWORD")
        or "123456"
    )
    queue_rows = load_csv(args.queue_csv)
    prepare_summary = load_json(args.prepare_summary_json)
    jsonl_path = Path(prepare_summary["jsonl_path"])
    prepared_rows = iter_prepared_rows(jsonl_path)
    prepared_by_source: dict[str, list[dict[str, Any]]] = {}
    for row in prepared_rows:
        source_id = str(row.get("source_id") or "").strip()
        if not source_id:
            continue
        prepared_by_source.setdefault(source_id, []).append(row)
        prepared_by_source.setdefault(normalize_source_id(source_id), []).append(row)

    token = login(args.api_base_url, args.admin_email, admin_password)
    result_rows: list[dict[str, Any]] = []
    classification_counts: dict[str, int] = {}

    for queue_row in queue_rows:
        remote_ids = split_remote_ids(queue_row.get("remote_source_ids_same_title", ""))
        remote_rows: list[dict[str, Any]] = []
        for remote_id in remote_ids:
            remote_rows.extend(
                fetch_paged_items(
                    args.api_base_url,
                    token,
                    "/legal-articles",
                    {"source_id": remote_id},
                )
            )
        local_rows = prepared_by_source.get(queue_row["source_id"], [])
        classification_result = classify(local_rows, remote_rows, queue_row.get("checksum_mismatch_reason", "").strip())
        classification_counts[classification_result] = classification_counts.get(classification_result, 0) + 1
        detail = build_detail(local_rows, remote_rows)
        result_rows.append(
            {
                **queue_row,
                "classification_result": classification_result,
                **detail,
            }
        )

    verified_count = sum(1 for row in result_rows if row["classification_result"].startswith("verified_"))
    manual_review_count = len(result_rows) - verified_count
    status = "PASS" if manual_review_count == 0 else "WARN"
    summary = {
        "status": status,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "queue_count": len(result_rows),
        "verified_count": verified_count,
        "manual_review_count": manual_review_count,
        "classification_counts": classification_counts,
    }

    write_json(args.output_dir / "phase14-source-review-verification.json", {"summary": summary, "rows": result_rows})
    write_csv(args.output_dir / "phase14-source-review-verification.csv", result_rows)
    write_md(args.output_dir / "phase14-source-review-verification.md", summary, result_rows)
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())