feat(growth): automate Yahoo same-item offer canary

scripts/ops/run_pchome_growth_yahoo_backfill.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Run or read back the bounded Yahoo public-offer automation lane."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+
+def main() -> int:
+    from sqlalchemy import create_engine
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--limit", type=int, default=8)
+    parser.add_argument("--receipt-root")
+    args = parser.parse_args()
+
+    from services.yahoo_shopping_public_offer_service import (
+        build_yahoo_shopping_runtime_readback,
+        run_pchome_growth_yahoo_backfill,
+    )
+
+    if not args.execute:
+        payload = build_yahoo_shopping_runtime_readback(
+            receipt_root=args.receipt_root,
+        )
+        print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
+        return 0 if payload.get("success") else 1
+
+    from config import DATABASE_PATH
+
+    engine = create_engine(DATABASE_PATH)
+    try:
+        payload = run_pchome_growth_yahoo_backfill(
+            engine,
+            limit=max(1, min(args.limit, 20)),
+            receipt_root=args.receipt_root,
+        )
+    finally:
+        engine.dispose()
+    print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
+    return 0 if payload.get("success") else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())