221 lines
8.2 KiB
Python
221 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""PChome 高業績商品主動反查 MOMO 候選，寫入外部報價層。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any, Callable
|
||
|
||
|
||
def _int_env(name: str, default: int, *, minimum: int, maximum: int) -> int:
|
||
try:
|
||
value = int(os.getenv(name, str(default)))
|
||
except (TypeError, ValueError):
|
||
value = default
|
||
return max(minimum, min(value, maximum))
|
||
|
||
|
||
def _float_env(name: str, default: float, *, minimum: float, maximum: float) -> float:
|
||
try:
|
||
value = float(os.getenv(name, str(default)))
|
||
except (TypeError, ValueError):
|
||
value = default
|
||
return max(minimum, min(value, maximum))
|
||
|
||
|
||
def candidate_auto_compare_type(candidate: dict[str, Any]) -> str:
|
||
auto_type = str(candidate.get("auto_compare_type") or "").strip()
|
||
if auto_type in {"total_price", "unit_price"}:
|
||
return auto_type
|
||
if candidate.get("can_auto_compare") is True:
|
||
return "total_price"
|
||
return "manual_review"
|
||
|
||
|
||
def build_momo_backfill_targets(payload: dict[str, Any], limit: int) -> list[dict[str, Any]]:
|
||
opportunities = list((payload or {}).get("opportunities") or [])
|
||
targets: list[dict[str, Any]] = []
|
||
for item in opportunities:
|
||
action = item.get("recommended_action") or {}
|
||
if item.get("external_price"):
|
||
continue
|
||
if action.get("code") != "map_external_product":
|
||
continue
|
||
product_id = str(item.get("pchome_product_id") or "").strip()
|
||
product_name = str(item.get("product_name") or "").strip()
|
||
if not product_id or not product_name:
|
||
continue
|
||
targets.append({
|
||
"product_id": product_id,
|
||
"name": product_name,
|
||
"price": item.get("pchome_price"),
|
||
"sales_7d": item.get("sales_7d"),
|
||
"priority_score": item.get("priority_score"),
|
||
})
|
||
if len(targets) >= limit:
|
||
break
|
||
return targets
|
||
|
||
|
||
def _default_build_payload(engine, limit: int) -> dict[str, Any]:
|
||
from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
|
||
|
||
return build_pchome_growth_opportunities(engine, limit=limit)
|
||
|
||
|
||
def _default_search_candidates(targets: list[dict[str, Any]], limit: int):
|
||
from services.momo_crawler import search_momo_products_for_pchome_products
|
||
|
||
return search_momo_products_for_pchome_products(
|
||
targets,
|
||
max_products=limit,
|
||
limit_per_product=_int_env(
|
||
"PCHOME_GROWTH_MOMO_BACKFILL_LIMIT_PER_TERM",
|
||
8,
|
||
minimum=3,
|
||
maximum=12,
|
||
),
|
||
max_terms_per_product=_int_env(
|
||
"PCHOME_GROWTH_MOMO_BACKFILL_MAX_TERMS",
|
||
8,
|
||
minimum=3,
|
||
maximum=10,
|
||
),
|
||
min_score=_float_env(
|
||
"PCHOME_GROWTH_MOMO_BACKFILL_MIN_SCORE",
|
||
0.45,
|
||
minimum=0.35,
|
||
maximum=0.8,
|
||
),
|
||
)
|
||
|
||
|
||
def _default_sync_candidates(engine, candidates: list[dict[str, Any]]) -> dict[str, Any]:
|
||
from services.external_market_offer_service import sync_targeted_momo_candidates_to_external_offers
|
||
|
||
return sync_targeted_momo_candidates_to_external_offers(engine, candidates, dry_run=False)
|
||
|
||
|
||
def _default_sync_review_candidates(engine, candidates: list[dict[str, Any]]) -> dict[str, Any]:
|
||
from services.external_market_offer_service import sync_targeted_momo_review_candidates_to_external_offers
|
||
|
||
return sync_targeted_momo_review_candidates_to_external_offers(engine, candidates, dry_run=False)
|
||
|
||
|
||
def run_pchome_growth_momo_backfill(
|
||
engine,
|
||
*,
|
||
limit: int = 12,
|
||
build_payload_func: Callable[[Any, int], dict[str, Any]] | None = None,
|
||
search_func: Callable[[list[dict[str, Any]], int], tuple[bool, str, list[dict[str, Any]]]] | None = None,
|
||
sync_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None,
|
||
sync_review_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""補高業績 PChome 商品的 MOMO 對應。
|
||
|
||
不呼叫 LLM，只搜尋 MOMO 候選，並只把可自動判斷的 total_price / unit_price
|
||
寫入 external_offers；需人工確認的候選會以 needs_review 保存，不進價格判斷。
|
||
"""
|
||
limit = max(1, min(int(limit or 12), 20))
|
||
build_payload = build_payload_func or _default_build_payload
|
||
search_candidates = search_func or _default_search_candidates
|
||
sync_candidates = sync_func or _default_sync_candidates
|
||
sync_review_candidates = sync_review_func or _default_sync_review_candidates
|
||
|
||
before_payload = build_payload(engine, max(limit, 16))
|
||
targets = build_momo_backfill_targets(before_payload, limit)
|
||
if not targets:
|
||
return {
|
||
"success": True,
|
||
"message": "目前高業績清單沒有需要補 MOMO 對應的商品。",
|
||
"data": {
|
||
"scanned_products": 0,
|
||
"target_count": 0,
|
||
"candidate_count": 0,
|
||
"exact_compare_count": 0,
|
||
"unit_compare_count": 0,
|
||
"auto_compare_count": 0,
|
||
"review_count": 0,
|
||
"external_offer_sync": {
|
||
"success": True,
|
||
"status": "not_needed",
|
||
"written_count": 0,
|
||
"message": "沒有需要同步的自動候選。",
|
||
},
|
||
"review_candidate_sync": {
|
||
"success": True,
|
||
"status": "not_needed",
|
||
"written_count": 0,
|
||
"message": "沒有需要保存的待確認候選。",
|
||
},
|
||
"before_stats": before_payload.get("stats") or {},
|
||
"after_stats": before_payload.get("stats") or {},
|
||
"targets": [],
|
||
"review_candidates": [],
|
||
},
|
||
}
|
||
|
||
search_success, search_message, candidates = search_candidates(targets, limit)
|
||
candidates = list(candidates or [])
|
||
exact_candidates = [
|
||
item for item in candidates
|
||
if candidate_auto_compare_type(item) == "total_price"
|
||
]
|
||
unit_candidates = [
|
||
item for item in candidates
|
||
if candidate_auto_compare_type(item) == "unit_price"
|
||
]
|
||
review_candidates = [
|
||
item for item in candidates
|
||
if candidate_auto_compare_type(item) not in {"total_price", "unit_price"}
|
||
]
|
||
auto_candidates = [*exact_candidates, *unit_candidates]
|
||
external_offer_sync = {
|
||
"success": True,
|
||
"status": "not_found",
|
||
"written_count": 0,
|
||
"message": "已搜尋 MOMO，但尚未找到可自動寫入的同款或單位價候選。",
|
||
}
|
||
if auto_candidates:
|
||
external_offer_sync = sync_candidates(engine, auto_candidates)
|
||
review_candidate_sync = {
|
||
"success": True,
|
||
"status": "not_found",
|
||
"written_count": 0,
|
||
"message": "沒有需要保存的待確認候選。",
|
||
}
|
||
if review_candidates:
|
||
review_candidate_sync = sync_review_candidates(engine, review_candidates)
|
||
|
||
after_payload = build_payload(engine, max(limit, 16))
|
||
written_count = int(external_offer_sync.get("written_count") or 0)
|
||
message = (
|
||
f"已掃描 {len(targets)} 個高業績商品，找到 {len(candidates)} 筆 MOMO 候選，"
|
||
f"自動寫入 {written_count} 筆。"
|
||
)
|
||
if not search_success and not candidates:
|
||
message = search_message or "已搜尋 MOMO，但沒有找到可用候選。"
|
||
|
||
return {
|
||
"success": True,
|
||
"message": message,
|
||
"data": {
|
||
"search_success": bool(search_success),
|
||
"search_message": search_message,
|
||
"scanned_products": len(targets),
|
||
"target_count": len(targets),
|
||
"candidate_count": len(candidates),
|
||
"exact_compare_count": len(exact_candidates),
|
||
"unit_compare_count": len(unit_candidates),
|
||
"auto_compare_count": len(auto_candidates),
|
||
"review_count": len(review_candidates),
|
||
"external_offer_sync": external_offer_sync,
|
||
"review_candidate_sync": review_candidate_sync,
|
||
"before_stats": before_payload.get("stats") or {},
|
||
"after_stats": after_payload.get("stats") or {},
|
||
"targets": targets[:8],
|
||
"review_candidates": review_candidates[:8],
|
||
},
|
||
}
|