feat(pchome): close freshness and multi-market decision gaps

scripts/ops/report_pchome_mapping_backlog.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Read-only PChome mapping backlog reporter with production version guard."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.ops import check_production_version_truth as version_guard  # noqa: E402
+from services.pchome_mapping_backlog_service import summarize_pchome_mapping_backlog  # noqa: E402
+
+
+DEFAULT_API_URL = "https://mo.wooo.work/api/ai/pchome-growth/opportunities"
+DEFAULT_HEALTH_URL = "https://mo.wooo.work/health"
+
+
+def with_limit(url: str, limit: int) -> str:
+    """Add a clamped limit while preserving existing query parameters."""
+    parsed = urllib.parse.urlsplit(url)
+    query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
+    query = [(key, value) for key, value in query if key != "limit"]
+    query.append(("limit", str(max(5, min(int(limit or 20), 50)))))
+    return urllib.parse.urlunsplit((
+        parsed.scheme,
+        parsed.netloc,
+        parsed.path,
+        urllib.parse.urlencode(query),
+        parsed.fragment,
+    ))
+
+
+def fetch_json(url: str, timeout: float) -> dict[str, Any]:
+    request = urllib.request.Request(
+        url,
+        headers={"Accept": "application/json", "User-Agent": "ewoooc-pchome-backlog-report/1"},
+    )
+    with urllib.request.urlopen(request, timeout=timeout) as response:
+        payload = json.loads(response.read().decode("utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError("growth API payload must be a JSON object")
+    return payload
+
+
+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,
+    allow_local_version_drift: bool = False,
+) -> dict[str, Any]:
+    target_url = with_limit(api_url, limit)
+    guard_report = None
+    if not skip_version_truth:
+        try:
+            guard_report = version_guard.build_report(health_url, timeout)
+            guard_ok, guard_errors = version_guard.evaluate(
+                guard_report,
+                allow_local_version_drift=allow_local_version_drift,
+            )
+        except Exception as error:
+            return {
+                "policy": "read_only_production_pchome_mapping_backlog",
+                "result": "BLOCKED",
+                "api_url": target_url,
+                "errors": [f"production version truth unavailable: {error}"],
+            }
+        if not guard_ok:
+            return {
+                "policy": "read_only_production_pchome_mapping_backlog",
+                "result": "BLOCKED",
+                "api_url": target_url,
+                "version_truth": guard_report,
+                "errors": guard_errors,
+            }
+
+    try:
+        summary = summarize_payload(fetch_json(target_url, timeout))
+    except Exception as error:
+        return {
+            "policy": "read_only_production_pchome_mapping_backlog",
+            "result": "BLOCKED",
+            "api_url": target_url,
+            "version_truth": guard_report,
+            "errors": [f"growth API readback unavailable: {error}"],
+        }
+
+    return {
+        **summary,
+        "policy": "read_only_production_pchome_mapping_backlog",
+        "api_url": target_url,
+        "version_truth": guard_report,
+        "errors": [],
+    }
+
+
+def _render_text(payload: dict[str, Any]) -> str:
+    stats = payload.get("stats") or {}
+    backlog = payload.get("backlog") or {}
+    errors = payload.get("errors") or []
+    if errors:
+        return "\n".join([f"result={payload.get('result', 'BLOCKED')}", *errors])
+    return "\n".join([
+        f"result={payload.get('result', 'UNKNOWN')}",
+        f"mapping_rate={stats.get('mapping_rate', 0)}%",
+        f"needs_mapping={stats.get('needs_mapping_count', 0)}",
+        f"direct_mapping={backlog.get('direct_mapping_count', 0)}",
+        f"review_candidates={backlog.get('review_candidate_count', 0)}",
+    ])
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="輸出正式 PChome 商品對應 backlog。")
+    parser.add_argument("--api-url", default=DEFAULT_API_URL)
+    parser.add_argument("--health-url", default=DEFAULT_HEALTH_URL)
+    parser.add_argument("--limit", type=int, default=20)
+    parser.add_argument("--timeout", type=float, default=12.0)
+    parser.add_argument("--skip-version-truth", action="store_true")
+    parser.add_argument("--allow-local-version-drift", action="store_true")
+    parser.add_argument("--json", action="store_true")
+    args = parser.parse_args(argv)
+
+    payload = 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,
+        allow_local_version_drift=args.allow_local_version_drift,
+    )
+    if args.json:
+        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
+    else:
+        print(_render_text(payload))
+    return 1 if payload.get("result") == "BLOCKED" else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())