V10.621 sync auto candidates to growth layer

services/external_market_offer_service.py

@@ -12,7 +12,7 @@ import logging
 import csv
 import io
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 
 from sqlalchemy import inspect, text
@@ -216,6 +216,18 @@ def _load_json_list(value: Any) -> list[Any]:
         return []
 
 
+def _load_json_dict(value: Any) -> dict[str, Any]:
+    if not value:
+        return {}
+    if isinstance(value, dict):
+        return value
+    try:
+        parsed = json.loads(value)
+        return parsed if isinstance(parsed, dict) else {}
+    except Exception:
+        return {}
+
+
 def _has_table(conn, table_name: str) -> bool:
     try:
         return inspect(conn).has_table(table_name)
@@ -661,6 +673,181 @@ def _upsert_external_offer(conn, payload: dict[str, Any]) -> None:
     conn.execute(text(sql), payload)
 
 
+def _targeted_candidate_auto_type(candidate: dict[str, Any]) -> str:
+    explicit_type = str(candidate.get("auto_compare_type") or "").strip()
+    if explicit_type:
+        return explicit_type
+    if candidate.get("can_auto_compare"):
+        return "total_price"
+    return "manual_review"
+
+
+def _targeted_candidate_to_external_offer(
+    candidate: dict[str, Any],
+    *,
+    observed_at: datetime,
+) -> tuple[dict[str, Any] | None, str]:
+    auto_type = _targeted_candidate_auto_type(candidate)
+    if auto_type not in {"total_price", "unit_price"}:
+        return None, "不是可自動使用的候選"
+
+    momo_sku = str(candidate.get("product_id") or candidate.get("goodsCode") or candidate.get("id") or "").strip()
+    pchome_product_id = str(candidate.get("target_pchome_product_id") or "").strip()
+    momo_price = _to_float(candidate.get("price"))
+    pchome_price = _to_float(candidate.get("target_pchome_price"))
+    if not momo_sku:
+        return None, "缺少 MOMO 商品 ID"
+    if not pchome_product_id:
+        return None, "缺少 PChome 商品 ID"
+    if not momo_price or momo_price <= 0:
+        return None, "缺少 MOMO 售價"
+
+    unit_price_comparison = (
+        candidate.get("target_unit_price_comparison")
+        if isinstance(candidate.get("target_unit_price_comparison"), dict)
+        else {}
+    )
+    is_unit_price = auto_type == "unit_price"
+    if is_unit_price and not unit_price_comparison.get("comparable"):
+        return None, "單位價證據不足"
+
+    match_score = _quality_score_from_match(candidate.get("target_match_score"))
+    if is_unit_price:
+        quality_score = max(match_score, 82.0)
+        price_basis = "unit_price"
+    else:
+        quality_score = match_score
+        price_basis = "total_price"
+        if quality_score < 76:
+            return None, "同款分數低於自動同步門檻"
+
+    title = str(candidate.get("name") or candidate.get("title") or momo_sku).strip()
+    notes = [
+        "由 PChome 商品自動反查 MOMO 候選同步",
+        "自動單位價比較" if is_unit_price else "可直接總價比價",
+    ]
+    raw_payload = {
+        "source": "pchome_targeted_momo_search",
+        "auto_compare_type": auto_type,
+        "price_basis": price_basis,
+        "pchome_public_price": pchome_price,
+        "pchome_public_name": candidate.get("target_pchome_name"),
+        "match_score": candidate.get("target_match_score"),
+        "match_reasons": candidate.get("target_match_reasons") or [],
+        "comparison_mode": candidate.get("target_comparison_mode"),
+        "hard_veto": bool(candidate.get("target_hard_veto")),
+        "target_gap_pct": candidate.get("target_gap_pct"),
+        "unit_price_comparison": unit_price_comparison,
+        "search_term": candidate.get("target_search_term"),
+        "tags": [
+            "identity_v2",
+            "source_targeted_momo_search",
+            f"price_basis_{price_basis}",
+            f"auto_compare_{auto_type}",
+        ],
+    }
+    return {
+        "source_code": "momo_reference",
+        "platform_code": "momo",
+        "source_product_id": momo_sku,
+        "source_offer_key": f"momo_reference:{momo_sku}:{pchome_product_id}:{price_basis}",
+        "title": title or momo_sku,
+        "brand": candidate.get("brand"),
+        "category_text": candidate.get("category") or candidate.get("category_text"),
+        "product_url": candidate.get("product_url") or candidate.get("url"),
+        "image_url": candidate.get("image_url"),
+        "price": momo_price,
+        "original_price": _to_float(candidate.get("original_price")),
+        "currency": "TWD",
+        "stock_status": None,
+        "sold_count": None,
+        "rating": None,
+        "review_count": None,
+        "observed_at": observed_at,
+        "expires_at": None,
+        "ingestion_method": "targeted_momo_search",
+        "connector_key": "pchome_targeted_momo_search",
+        "pchome_product_id": pchome_product_id,
+        "momo_sku": momo_sku,
+        "match_status": "verified",
+        "quality_score": round(quality_score, 2),
+        "data_quality_status": "verified",
+        "quality_notes_json": json.dumps(notes, ensure_ascii=False),
+        "raw_payload_json": json.dumps(raw_payload, ensure_ascii=False),
+    }, ""
+
+
+def sync_targeted_momo_candidates_to_external_offers(
+    engine,
+    candidates: list[dict[str, Any]],
+    *,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """把頁面自動找到的安全 MOMO 候選同步進 external_offers。
+
+    只接受 total_price 與 unit_price 自動候選；人工確認候選不寫入。
+    """
+    generated_at = datetime.now().isoformat(timespec="seconds")
+    candidates = list(candidates or [])
+    required_tables = {"external_market_sources", "external_offers"}
+
+    with engine.begin() as conn:
+        missing_tables = sorted(table for table in required_tables if not _has_table(conn, table))
+        if missing_tables:
+            return {
+                "success": False,
+                "status": "skipped",
+                "generated_at": generated_at,
+                "candidate_count": len(candidates),
+                "written_count": 0,
+                "dry_run": dry_run,
+                "message": "外部報價同步暫時無法執行，缺少必要資料表。",
+                "missing_tables": missing_tables,
+            }
+
+        _ensure_external_market_source_seeds(conn)
+        base_observed_at = datetime.now()
+        offers: list[dict[str, Any]] = []
+        skipped_reasons: dict[str, int] = {}
+        for index, candidate in enumerate(candidates):
+            offer, reason = _targeted_candidate_to_external_offer(
+                candidate,
+                observed_at=base_observed_at + timedelta(microseconds=index),
+            )
+            if offer:
+                offers.append(offer)
+            else:
+                skipped_reasons[reason] = skipped_reasons.get(reason, 0) + 1
+
+        if not dry_run:
+            for offer in offers:
+                _upsert_external_offer(conn, offer)
+
+    unit_count = sum(
+        1
+        for offer in offers
+        if _load_json_dict(offer.get("raw_payload_json")).get("price_basis") == "unit_price"
+    )
+    total_count = len(offers) - unit_count
+    return {
+        "success": True,
+        "status": "dry_run" if dry_run else "synced",
+        "generated_at": generated_at,
+        "candidate_count": len(candidates),
+        "written_count": 0 if dry_run else len(offers),
+        "dry_run": dry_run,
+        "source_code": "momo_reference",
+        "total_price_count": total_count,
+        "unit_price_count": unit_count,
+        "skipped_reasons": skipped_reasons,
+        "message": (
+            "已把自動比價候選同步到外部價格參考。"
+            if not dry_run
+            else "已完成自動比價候選同步預檢，尚未寫入資料。"
+        ),
+    }
+
+
 def sync_legacy_momo_reference_offers(engine, *, limit: int = 500, dry_run: bool = False) -> dict[str, Any]:
     """把既有已確認同款的比價快取自動同步到 external_offers。"""
     limit = max(1, min(int(limit or 500), 5000))

services/momo_crawler.py

@@ -706,6 +706,7 @@ def search_momo_products_for_pchome_products(
                     "product_id": product_id,
                     "target_pchome_product_id": pchome_id,
                     "target_pchome_name": pchome_name,
+                    "target_pchome_price": pchome_price,
                     "target_match_score": round(score, 3),
                     "target_search_term": term,
                     "target_match_reasons": list(getattr(diagnostics, "reasons", ()) or ()),

services/pchome_revenue_growth_service.py

@@ -202,6 +202,15 @@ def _match_score_from_quality(value: Any) -> float:
     return max(0, min(1, score))
 
 
+def _external_price_basis(raw_payload: dict[str, Any]) -> str:
+    price_basis = str(raw_payload.get("price_basis") or "").strip()
+    return price_basis if price_basis in {"total_price", "unit_price"} else "total_price"
+
+
+def _external_price_basis_label(price_basis: str) -> str:
+    return "單位價" if price_basis == "unit_price" else "商品總價"
+
+
 def _fetch_normalized_external_price_map(conn, pchome_product_ids: list[str]) -> dict[str, dict[str, Any]]:
     inspector = inspect(conn)
     if not inspector.has_table("external_offers"):
@@ -274,6 +283,8 @@ def _fetch_normalized_external_price_map(conn, pchome_product_ids: list[str]) ->
     result: dict[str, dict[str, Any]] = {}
     for row in rows:
         raw_payload = _json_dict(row.get("raw_payload_json"))
+        price_basis = _external_price_basis(raw_payload)
+        unit_price_comparison = _json_dict(raw_payload.get("unit_price_comparison"))
         key = str(row.get("pchome_product_id") or "").strip()
         if not key:
             continue
@@ -284,8 +295,16 @@ def _fetch_normalized_external_price_map(conn, pchome_product_ids: list[str]) ->
             "momo_name": row.get("momo_name"),
             "momo_price": row.get("momo_price"),
             "pchome_price": raw_payload.get("pchome_public_price"),
+            "price_basis": price_basis,
+            "price_basis_label": _external_price_basis_label(price_basis),
+            "unit_label": unit_price_comparison.get("unit_label"),
+            "momo_unit_price": unit_price_comparison.get("momo_unit_price"),
+            "pchome_unit_price": unit_price_comparison.get("competitor_unit_price"),
+            "unit_gap_pct": unit_price_comparison.get("unit_gap_pct"),
+            "momo_total_quantity": unit_price_comparison.get("momo_total_quantity"),
+            "pchome_total_quantity": unit_price_comparison.get("competitor_total_quantity"),
             "match_score": _match_score_from_quality(row.get("quality_score")),
-            "tags": ["external_offers", "verified"],
+            "tags": raw_payload.get("tags") or ["external_offers", "verified"],
             "crawled_at": row.get("observed_at"),
             "momo_price_at": row.get("observed_at"),
             "data_source": "external_offers",
@@ -428,9 +447,24 @@ def _score_opportunity(sales_row: dict[str, Any], external_row: dict[str, Any] |
     reason_lines = []
 
     if external_row:
+        price_basis = str(external_row.get("price_basis") or "total_price")
+        price_basis_label = external_row.get("price_basis_label") or _external_price_basis_label(price_basis)
         pchome_price = _to_float(external_row.get("pchome_price"))
         momo_price = _to_float(external_row.get("momo_price"))
-        gap_pct = ((momo_price - pchome_price) / pchome_price * 100) if pchome_price else None
+        pchome_compare_price = pchome_price
+        momo_compare_price = momo_price
+        if price_basis == "unit_price":
+            pchome_unit_price = _to_float(external_row.get("pchome_unit_price"))
+            momo_unit_price = _to_float(external_row.get("momo_unit_price"))
+            pchome_compare_price = pchome_unit_price
+            momo_compare_price = momo_unit_price
+            gap_pct = _to_float(external_row.get("unit_gap_pct"), default=None)
+            if gap_pct is None and pchome_compare_price and momo_compare_price:
+                gap_pct = (momo_compare_price - pchome_compare_price) / pchome_compare_price * 100
+        else:
+            pchome_unit_price = None
+            momo_unit_price = None
+            gap_pct = ((momo_price - pchome_price) / pchome_price * 100) if pchome_price else None
         tags = _load_json_tags(external_row.get("tags"))
         data_quality_score = 78 + min(12, _to_float(external_row.get("match_score")) * 12)
         external_payload = {
@@ -443,6 +477,13 @@ def _score_opportunity(sales_row: dict[str, Any], external_row: dict[str, Any] |
             "momo_name": external_row.get("momo_name"),
             "momo_price": round(momo_price, 2) if momo_price else None,
             "pchome_price": round(pchome_price, 2) if pchome_price else None,
+            "price_basis": price_basis,
+            "price_basis_label": price_basis_label,
+            "unit_label": external_row.get("unit_label") or "",
+            "momo_unit_price": round(momo_unit_price, 4) if momo_unit_price else None,
+            "pchome_unit_price": round(pchome_unit_price, 4) if pchome_unit_price else None,
+            "momo_total_quantity": external_row.get("momo_total_quantity"),
+            "pchome_total_quantity": external_row.get("pchome_total_quantity"),
             "gap_pct": round(gap_pct, 1) if gap_pct is not None else None,
             "match_score": round(_to_float(external_row.get("match_score")), 3),
             "tags": tags,
@@ -467,12 +508,13 @@ def _score_opportunity(sales_row: dict[str, Any], external_row: dict[str, Any] |
             action_message = "業績與外部價格暫無明顯異常，先保留在觀察清單。"
 
         if gap_pct is not None:
+            basis_prefix = "單位價" if price_basis == "unit_price" else "價格"
             if gap_pct > 0:
-                reason_lines.append(f"PChome 目前比 MOMO 低約 {abs(gap_pct):.1f}%。")
+                reason_lines.append(f"PChome {basis_prefix}目前比 MOMO 低約 {abs(gap_pct):.1f}%。")
             elif gap_pct < 0:
-                reason_lines.append(f"MOMO 目前比 PChome 低約 {abs(gap_pct):.1f}%。")
+                reason_lines.append(f"MOMO {basis_prefix}目前比 PChome 低約 {abs(gap_pct):.1f}%。")
             else:
-                reason_lines.append("PChome 與 MOMO 價格幾乎相同。")
+                reason_lines.append(f"PChome 與 MOMO {basis_prefix}幾乎相同。")
     else:
         data_quality_score -= 12
         reason_lines.append("尚未找到可確認的 MOMO 對照商品。")
@@ -523,7 +565,13 @@ def _score_opportunity(sales_row: dict[str, Any], external_row: dict[str, Any] |
         },
         "reason_lines": reason_lines[:4],
         "data_quality": {
-            "label": "資料可直接判斷" if external_row else "需要補資料",
+            "label": (
+                "資料可用單位價判斷"
+                if external_payload and external_payload.get("price_basis") == "unit_price"
+                else "資料可直接判斷"
+                if external_row
+                else "需要補資料"
+            ),
             "score": round(max(0, min(100, data_quality_score)), 1),
             "issues": issues,
         },