chore(dev): sync ewoooc development workbench state

scripts/ops/report_pchome_mapping_backlog.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+"""Read-only report for the production PChome growth mapping backlog."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any
+
+from scripts.ops import check_production_version_truth as version_guard
+from services.pchome_mapping_backlog_service import summarize_pchome_mapping_backlog
+
+
+DEFAULT_API_URL = "https://mo.wooo.work/api/ai/pchome-growth/opportunities"
+DEFAULT_LIMIT = 20
+
+
+def with_limit(url: str, limit: int) -> str:
+    parsed = urllib.parse.urlparse(url)
+    query = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
+    query["limit"] = [str(max(5, min(int(limit or DEFAULT_LIMIT), 50)))]
+    return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query, doseq=True)))
+
+
+def fetch_json(url: str, timeout: float) -> dict[str, Any]:
+    with urllib.request.urlopen(url, timeout=timeout) as response:
+        payload = response.read().decode("utf-8")
+    data = json.loads(payload)
+    if not isinstance(data, dict):
+        raise ValueError("API payload must be a JSON object")
+    return data
+
+
+def summarize_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    return summarize_pchome_mapping_backlog(payload)
+
+
+def build_report(
+    api_url: str,
+    limit: int,
+    timeout: float,
+    health_url: str,
+    skip_version_truth: bool,
+) -> dict[str, Any]:
+    version_report = None
+    version_errors: list[str] = []
+    if not skip_version_truth:
+        version_report = version_guard.build_report(health_url, timeout)
+        version_ok, version_errors = version_guard.evaluate(version_report, allow_local_version_drift=False)
+        if not version_ok:
+            return {
+                "policy": "read_only_production_pchome_mapping_backlog",
+                "result": "BLOCKED",
+                "api_url": with_limit(api_url, limit),
+                "version_truth": version_report,
+                "errors": version_errors,
+            }
+
+    final_url = with_limit(api_url, limit)
+    payload = fetch_json(final_url, timeout)
+    summary = summarize_payload(payload)
+    return {
+        **summary,
+        "policy": "read_only_production_pchome_mapping_backlog",
+        "api_url": final_url,
+        "version_truth": version_report,
+        "errors": version_errors,
+    }
+
+
+def format_text(report: dict[str, Any]) -> str:
+    stats = report.get("stats") or {}
+    backlog = report.get("backlog") or {}
+    version = ((report.get("version_truth") or {}).get("production") or {}).get("version")
+    lines = [
+        "pchome_mapping_backlog:",
+        f"- policy: {report.get('policy')}",
+        f"- result: {report.get('result')}",
+        f"- production_version: {version or 'not_checked'}",
+        f"- api_url: {report.get('api_url')}",
+        f"- generated_at: {report.get('generated_at')}",
+        f"- cache_state: {report.get('cache_state') or 'unknown'}",
+        f"- candidate_count: {stats.get('candidate_count')}",
+        f"- mapped_count: {stats.get('mapped_count')}",
+        f"- mapping_rate: {stats.get('mapping_rate')}%",
+        f"- needs_mapping_count: {stats.get('needs_mapping_count')}",
+        f"- review_candidate_count: {stats.get('review_candidate_count')}",
+        f"- overall_sales_7d: {stats.get('overall_sales_7d')}",
+        f"- action_counts: {json.dumps(stats.get('action_counts') or {}, ensure_ascii=False, sort_keys=True)}",
+        f"- direct_mapping_count: {backlog.get('direct_mapping_count')}",
+        f"- mapped_opportunity_count: {backlog.get('mapped_opportunity_count')}",
+    ]
+    for error in report.get("errors") or []:
+        lines.append(f"- blocker: {error}")
+
+    top_items = list((backlog.get("top_needs_mapping") or [])[:5])
+    if top_items:
+        lines.append("- top_needs_mapping:")
+        for item in top_items:
+            lines.append(
+                "  - "
+                f"{item.get('pchome_product_id')} | "
+                f"sales_7d={item.get('sales_7d')} | "
+                f"score={item.get('priority_score')} | "
+                f"action={item.get('action_label')}"
+            )
+    return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--api-url", default=DEFAULT_API_URL)
+    parser.add_argument("--health-url", default=version_guard.DEFAULT_HEALTH_URL)
+    parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT)
+    parser.add_argument("--timeout", type=float, default=12.0)
+    parser.add_argument("--json", action="store_true", help="Print machine-readable report")
+    parser.add_argument(
+        "--skip-version-truth",
+        action="store_true",
+        help="Skip production version guard; intended only for offline parser tests.",
+    )
+    args = parser.parse_args(argv)
+
+    try:
+        report = build_report(
+            api_url=args.api_url,
+            limit=args.limit,
+            timeout=args.timeout,
+            health_url=args.health_url,
+            skip_version_truth=args.skip_version_truth,
+        )
+    except (OSError, ValueError, urllib.error.URLError) as exc:
+        report = {
+            "policy": "read_only_production_pchome_mapping_backlog",
+            "result": "BLOCKED",
+            "api_url": with_limit(args.api_url, args.limit),
+            "errors": [str(exc)],
+        }
+        print(json.dumps(report, ensure_ascii=False, indent=2) if args.json else format_text(report))
+        return 2
+
+    if args.json:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+    else:
+        print(format_text(report))
+    return 0 if report.get("result") != "BLOCKED" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())