799 lines
30 KiB
Python
799 lines
30 KiB
Python
"""Public Yahoo/PChome evidence adapter with no session or credential access."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import unicodedata
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Mapping
|
||
from urllib.parse import quote, urlparse
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
|
||
from services.public_source_http_service import read_bounded_response_text
|
||
|
||
|
||
POLICY = "controlled_yahoo_shopping_public_offer_v1"
|
||
ADAPTER_VERSION = "yahoo_shopping_public_jsonld_v1"
|
||
WORK_ITEM_ID = "GROWTH-P0-001-C"
|
||
SOURCE_CODE = "yahoo_shopping"
|
||
PLATFORM_CODE = "yahoo_shopping_tw"
|
||
SOURCE_LABEL = "Yahoo 購物"
|
||
INGESTION_METHOD = "public_product_jsonld"
|
||
CONNECTOR_KEY = "pchome_targeted_yahoo_public_search"
|
||
YAHOO_HOST = "tw.buy.yahoo.com"
|
||
PCHOME_API_HOST = "ecapi.pchome.com.tw"
|
||
YAHOO_SEARCH_URL = "https://tw.buy.yahoo.com/search/product?p={query}"
|
||
PCHOME_PUBLIC_PRODUCT_API = "https://ecapi.pchome.com.tw/ecshop/prodapi/v2/prod"
|
||
DEFAULT_RECEIPT_ROOT = os.getenv(
|
||
"PCHOME_GROWTH_YAHOO_RECEIPT_ROOT",
|
||
"/app/data/ai_automation/pchome_growth_yahoo_backfill_receipts"
|
||
if Path("/app/data").exists()
|
||
else "runtime_artifacts/pchome_growth_yahoo_backfill_receipts",
|
||
)
|
||
DEFAULT_USER_AGENT = "Mozilla/5.0 (compatible; EwoooC-PublicEvidence/1.0)"
|
||
MAX_RESPONSE_BYTES = 1_500_000
|
||
SELECTION_MARKERS = (
|
||
"任選",
|
||
"多款",
|
||
"可選",
|
||
"隨機",
|
||
"款式可選",
|
||
"色號可選",
|
||
"香味可選",
|
||
)
|
||
|
||
|
||
def _bounded_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 _bounded_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 _to_float(value: Any, default: float = 0.0) -> float:
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _canonical_hash(payload: Any) -> str:
|
||
encoded = json.dumps(
|
||
payload,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
default=str,
|
||
).encode("utf-8")
|
||
return hashlib.sha256(encoded).hexdigest()
|
||
|
||
|
||
def _compact_text(value: Any) -> str:
|
||
text = unicodedata.normalize("NFKC", str(value or ""))
|
||
return re.sub(r"\s+", "", text).lower()
|
||
|
||
|
||
def _plain_text(value: Any) -> str:
|
||
if value in (None, ""):
|
||
return ""
|
||
return BeautifulSoup(str(value), "html.parser").get_text(" ", strip=True)
|
||
|
||
|
||
def _is_public_url(url: str, *, host: str, path_prefix: str = "") -> bool:
|
||
parsed = urlparse(str(url or ""))
|
||
return bool(
|
||
parsed.scheme == "https"
|
||
and parsed.hostname == host
|
||
and (not path_prefix or parsed.path.startswith(path_prefix))
|
||
and not parsed.username
|
||
and not parsed.password
|
||
)
|
||
|
||
|
||
def _response_text(response: Any) -> str:
|
||
return read_bounded_response_text(response, max_bytes=MAX_RESPONSE_BYTES)
|
||
|
||
|
||
def _default_get(url: str, *, timeout: float) -> Any:
|
||
return requests.get(
|
||
url,
|
||
timeout=timeout,
|
||
allow_redirects=True,
|
||
stream=True,
|
||
headers={
|
||
"User-Agent": DEFAULT_USER_AGENT,
|
||
"Accept-Language": "zh-TW,zh;q=0.9",
|
||
"Accept": "text/html,application/xhtml+xml,application/json;q=0.9,*/*;q=0.8",
|
||
},
|
||
)
|
||
|
||
|
||
def _jsonld_nodes(value: Any):
|
||
if isinstance(value, dict):
|
||
yield value
|
||
graph = value.get("@graph")
|
||
if graph is not None:
|
||
yield from _jsonld_nodes(graph)
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
yield from _jsonld_nodes(item)
|
||
|
||
|
||
def _type_is(node: Mapping[str, Any], expected: str) -> bool:
|
||
value = node.get("@type") or node.get("type")
|
||
if isinstance(value, str):
|
||
return value.lower() == expected.lower()
|
||
if isinstance(value, list):
|
||
return any(str(item).lower() == expected.lower() for item in value)
|
||
return False
|
||
|
||
|
||
def _first_mapping(value: Any) -> dict[str, Any]:
|
||
if isinstance(value, dict):
|
||
return dict(value)
|
||
if isinstance(value, list):
|
||
for item in value:
|
||
if isinstance(item, dict):
|
||
return dict(item)
|
||
return {}
|
||
|
||
|
||
def _first_image(value: Any) -> str:
|
||
if isinstance(value, str):
|
||
return value.strip()
|
||
if isinstance(value, dict):
|
||
return str(value.get("url") or value.get("contentUrl") or "").strip()
|
||
if isinstance(value, list):
|
||
for item in value:
|
||
image = _first_image(item)
|
||
if image:
|
||
return image
|
||
return ""
|
||
|
||
|
||
def _clean_yahoo_product_name(value: Any) -> str:
|
||
text = str(value or "").strip()
|
||
return re.split(r"\s+\|\s+", text, maxsplit=1)[0].strip()
|
||
|
||
|
||
def _has_multi_variant_group_for_spec(title: str, target_spec: str) -> bool:
|
||
compact_spec = _compact_text(target_spec)
|
||
if not compact_spec:
|
||
return False
|
||
for group in re.findall(r"[（(]([^）)]+)[）)]", title or ""):
|
||
if compact_spec in _compact_text(group) and re.search(r"[/／|、]", group):
|
||
return True
|
||
return False
|
||
|
||
|
||
def parse_yahoo_search_html(html: str, *, search_url: str) -> dict[str, Any]:
|
||
"""Parse Yahoo public search state without executing page scripts."""
|
||
soup = BeautifulSoup(html or "", "html.parser")
|
||
script = soup.find("script", id="isoredux-data")
|
||
errors: list[str] = []
|
||
payload: dict[str, Any] = {}
|
||
if script is None:
|
||
errors.append("isoredux_data_missing")
|
||
else:
|
||
try:
|
||
parsed = json.loads(script.string or script.get_text() or "{}")
|
||
payload = parsed if isinstance(parsed, dict) else {}
|
||
except json.JSONDecodeError:
|
||
errors.append("isoredux_data_invalid")
|
||
|
||
search = payload.get("search") if isinstance(payload.get("search"), dict) else {}
|
||
ecsearch = (
|
||
search.get("ecsearch") if isinstance(search.get("ecsearch"), dict) else {}
|
||
)
|
||
hits = ecsearch.get("hits") if isinstance(ecsearch.get("hits"), list) else []
|
||
candidates: list[dict[str, Any]] = []
|
||
for item in hits:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
product_id = str(item.get("ec_productid") or "").strip()
|
||
title = str(item.get("ec_title") or "").strip()
|
||
product_url = str(item.get("ec_item_url") or "").strip()
|
||
price = _to_float(item.get("ec_price"))
|
||
if not product_id or not title or price <= 0:
|
||
continue
|
||
if not _is_public_url(product_url, host=YAHOO_HOST, path_prefix="/gdsale/"):
|
||
continue
|
||
candidates.append(
|
||
{
|
||
"product_id": product_id,
|
||
"name": title,
|
||
"search_price": price,
|
||
"original_price": _to_float(item.get("ec_listprice")),
|
||
"product_url": product_url,
|
||
"image_url": str(item.get("ec_image") or "").strip(),
|
||
"brand": str(item.get("ec_brand") or "").strip(),
|
||
"seller": str(item.get("ec_storename") or "").strip(),
|
||
"stock_status": {"1": "in_stock", "0": "out_of_stock"}.get(
|
||
str(item.get("ec_has_stock") or "")
|
||
),
|
||
"rating": _to_float(item.get("ec_quality_rating")),
|
||
"review_count": int(_to_float(item.get("ec_rating_count"))),
|
||
}
|
||
)
|
||
|
||
page_title = soup.title.get_text(strip=True) if soup.title else ""
|
||
blocked = any(
|
||
marker in page_title.lower()
|
||
for marker in ("access denied", "captcha", "robot check")
|
||
)
|
||
if blocked:
|
||
errors.append("blocked_page_detected")
|
||
return {
|
||
"success": not errors,
|
||
"policy": POLICY,
|
||
"adapter_version": ADAPTER_VERSION,
|
||
"source_method": "public_search_embedded_json",
|
||
"search_url": search_url,
|
||
"page_title": page_title,
|
||
"candidate_count": len(candidates),
|
||
"candidates": candidates,
|
||
"errors": errors,
|
||
"content_sha256": hashlib.sha256((html or "").encode("utf-8")).hexdigest(),
|
||
}
|
||
|
||
|
||
def parse_yahoo_product_jsonld(html: str, *, product_url: str) -> dict[str, Any]:
|
||
"""Parse and normalize one Yahoo public Product/Offer JSON-LD payload."""
|
||
soup = BeautifulSoup(html or "", "html.parser")
|
||
products: list[dict[str, Any]] = []
|
||
warnings: list[str] = []
|
||
for script in soup.find_all(
|
||
"script",
|
||
attrs={"type": re.compile(r"ld\+json", re.IGNORECASE)},
|
||
):
|
||
text_value = script.string or script.get_text() or ""
|
||
if not text_value.strip():
|
||
continue
|
||
try:
|
||
data = json.loads(text_value)
|
||
except json.JSONDecodeError:
|
||
warnings.append("invalid_jsonld_skipped")
|
||
continue
|
||
for node in _jsonld_nodes(data):
|
||
if _type_is(node, "Product"):
|
||
products.append(dict(node))
|
||
|
||
if not products:
|
||
return {
|
||
"success": False,
|
||
"policy": POLICY,
|
||
"adapter_version": ADAPTER_VERSION,
|
||
"product_url": product_url,
|
||
"errors": ["product_jsonld_missing"],
|
||
"warnings": warnings,
|
||
}
|
||
|
||
product = products[0]
|
||
offer = _first_mapping(product.get("offers"))
|
||
brand = _first_mapping(product.get("brand"))
|
||
seller = _first_mapping(offer.get("seller"))
|
||
rating = _first_mapping(product.get("aggregateRating"))
|
||
product_id = str(product.get("sku") or "").strip()
|
||
title = _clean_yahoo_product_name(product.get("name"))
|
||
price = _to_float(offer.get("price"))
|
||
currency = str(offer.get("priceCurrency") or "").strip().upper()
|
||
errors = []
|
||
if not product_id:
|
||
errors.append("product_id_missing")
|
||
if not title:
|
||
errors.append("product_title_missing")
|
||
if price <= 0:
|
||
errors.append("positive_price_missing")
|
||
if currency != "TWD":
|
||
errors.append("currency_not_twd")
|
||
if not _is_public_url(product_url, host=YAHOO_HOST, path_prefix="/gdsale/"):
|
||
errors.append("product_url_outside_allowlist")
|
||
|
||
normalized = {
|
||
"product_id": product_id,
|
||
"name": title,
|
||
"price": price,
|
||
"currency": currency,
|
||
"product_url": product_url,
|
||
"offer_url": str(offer.get("url") or "").strip(),
|
||
"image_url": _first_image(product.get("image")),
|
||
"brand": str(brand.get("name") or "").strip(),
|
||
"gtin13": str(product.get("gtin13") or "").strip(),
|
||
"seller": str(seller.get("name") or "").strip(),
|
||
"availability": str(offer.get("availability") or "").strip(),
|
||
"rating": _to_float(rating.get("ratingValue")),
|
||
"review_count": int(_to_float(rating.get("ratingCount"))),
|
||
}
|
||
return {
|
||
"success": not errors,
|
||
"policy": POLICY,
|
||
"adapter_version": ADAPTER_VERSION,
|
||
"source_method": "public_product_jsonld",
|
||
"product_url": product_url,
|
||
"product": normalized,
|
||
"errors": errors,
|
||
"warnings": warnings,
|
||
"content_sha256": hashlib.sha256((html or "").encode("utf-8")).hexdigest(),
|
||
"product_fingerprint_sha256": _canonical_hash(normalized),
|
||
}
|
||
|
||
|
||
def parse_pchome_public_product_jsonp(payload: str) -> dict[str, Any]:
|
||
"""Parse the public PChome JSONP wrapper without evaluating JavaScript."""
|
||
start = str(payload or "").find("jsonp(")
|
||
if start < 0:
|
||
return {}
|
||
start += len("jsonp(")
|
||
end = str(payload or "").find(");}catch", start)
|
||
if end < 0:
|
||
end = str(payload or "").find(");", start)
|
||
if end < 0:
|
||
return {}
|
||
try:
|
||
parsed = json.loads(str(payload or "")[start:end])
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
|
||
|
||
def _target_current_price(product: Mapping[str, Any], fallback: Any) -> tuple[float, str]:
|
||
prices = product.get("Price") if isinstance(product.get("Price"), dict) else {}
|
||
for key in ("P", "Low", "M"):
|
||
value = _to_float(prices.get(key))
|
||
if value > 0:
|
||
return value, key
|
||
return _to_float(fallback), "sales_report_fallback"
|
||
|
||
|
||
def enrich_targets_with_pchome_public_evidence(
|
||
targets: list[dict[str, Any]],
|
||
*,
|
||
request_get: Callable[..., Any] | None = None,
|
||
timeout: float | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Attach current public PChome name, price, and Spec to bounded targets."""
|
||
ids = [
|
||
str(item.get("product_id") or "").strip()
|
||
for item in targets or []
|
||
if str(item.get("product_id") or "").strip()
|
||
]
|
||
if not ids:
|
||
return {"success": True, "evidence_count": 0, "missing_count": 0}
|
||
timeout_seconds = timeout or _bounded_float_env(
|
||
"PCHOME_GROWTH_YAHOO_HTTP_TIMEOUT_SECONDS", 15, 3, 30
|
||
)
|
||
url = (
|
||
f"{PCHOME_PUBLIC_PRODUCT_API}?id={quote(','.join(ids), safe=',')}"
|
||
"&fields=Id,Name,Nick,Price,Spec,Qty&_callback=jsonp"
|
||
)
|
||
getter = request_get or _default_get
|
||
try:
|
||
response = getter(url, timeout=timeout_seconds)
|
||
final_url = str(getattr(response, "url", url) or url)
|
||
if not _is_public_url(final_url, host=PCHOME_API_HOST, path_prefix="/ecshop/"):
|
||
raise ValueError("pchome_public_api_redirect_outside_allowlist")
|
||
products = parse_pchome_public_product_jsonp(_response_text(response))
|
||
except Exception as exc:
|
||
return {
|
||
"success": False,
|
||
"evidence_count": 0,
|
||
"missing_count": len(ids),
|
||
"error_type": type(exc).__name__,
|
||
"error": str(exc)[:200],
|
||
}
|
||
|
||
evidence_count = 0
|
||
for target in targets:
|
||
product_id = str(target.get("product_id") or "").strip()
|
||
product = products.get(product_id)
|
||
if not isinstance(product, dict):
|
||
continue
|
||
public_name = str(product.get("Name") or target.get("name") or "").strip()
|
||
public_spec = _plain_text(product.get("Spec"))
|
||
public_price, public_price_field = _target_current_price(product, target.get("price"))
|
||
prices = product.get("Price") if isinstance(product.get("Price"), dict) else {}
|
||
original_name = str(target.get("name") or "").strip()
|
||
original_price = target.get("price")
|
||
target["name"] = (
|
||
f"{public_name} {public_spec}" if public_spec else public_name
|
||
).strip()
|
||
target["price"] = public_price
|
||
target["target_public_evidence"] = {
|
||
"source": "pchome_public_product_api",
|
||
"product_id": product_id,
|
||
"name": public_name,
|
||
"spec": public_spec,
|
||
"current_price": public_price,
|
||
"current_price_field": public_price_field,
|
||
"price_fields": {
|
||
key: _to_float(prices.get(key))
|
||
for key in ("P", "Low", "M")
|
||
if _to_float(prices.get(key)) > 0
|
||
},
|
||
"sales_realized_price": original_price,
|
||
"sales_report_name": original_name,
|
||
**({"quantity": int(_to_float(product.get("Qty")))} if product.get("Qty") not in (None, "") else {}),
|
||
}
|
||
evidence_count += 1
|
||
|
||
return {
|
||
"success": evidence_count == len(ids),
|
||
"evidence_count": evidence_count,
|
||
"missing_count": max(0, len(ids) - evidence_count),
|
||
"source_url": url,
|
||
"source_method": "pchome_public_product_api",
|
||
}
|
||
|
||
|
||
def _source_contract_guard(
|
||
*,
|
||
target: Mapping[str, Any],
|
||
search_candidate: Mapping[str, Any],
|
||
product_readback: Mapping[str, Any],
|
||
) -> dict[str, Any]:
|
||
detail = product_readback.get("product")
|
||
detail = detail if isinstance(detail, dict) else {}
|
||
reasons: list[str] = []
|
||
source_title = str(detail.get("name") or search_candidate.get("name") or "")
|
||
target_evidence = target.get("target_public_evidence")
|
||
target_evidence = target_evidence if isinstance(target_evidence, dict) else {}
|
||
target_spec = str(target_evidence.get("spec") or "").strip()
|
||
if not product_readback.get("success"):
|
||
reasons.append("product_jsonld_invalid")
|
||
if str(detail.get("product_id") or "") != str(
|
||
search_candidate.get("product_id") or ""
|
||
):
|
||
reasons.append("search_detail_product_id_mismatch")
|
||
if any(marker in source_title for marker in SELECTION_MARKERS):
|
||
reasons.append("source_selection_listing")
|
||
if _has_multi_variant_group_for_spec(source_title, target_spec):
|
||
reasons.append("source_multi_variant_listing")
|
||
if target_spec and _compact_text(target_spec) not in _compact_text(source_title):
|
||
reasons.append("target_spec_missing_from_source_title")
|
||
if not target_evidence.get("current_price") or target_evidence.get("current_price_field") == "sales_report_fallback":
|
||
reasons.append("pchome_current_price_missing")
|
||
if "quantity" in target_evidence and int(
|
||
_to_float(target_evidence.get("quantity"))
|
||
) <= 0:
|
||
reasons.append("pchome_stock_not_available")
|
||
source_stock_status = str(
|
||
detail.get("availability") or search_candidate.get("stock_status") or ""
|
||
).lower()
|
||
if source_stock_status and not source_stock_status.replace("_", "").endswith(
|
||
"instock"
|
||
):
|
||
reasons.append("source_stock_not_confirmed")
|
||
if str(detail.get("currency") or "") != "TWD":
|
||
reasons.append("source_currency_not_twd")
|
||
if _to_float(detail.get("price")) <= 0:
|
||
reasons.append("source_positive_price_missing")
|
||
if not _is_public_url(
|
||
str(detail.get("product_url") or ""),
|
||
host=YAHOO_HOST,
|
||
path_prefix="/gdsale/",
|
||
):
|
||
reasons.append("source_url_outside_allowlist")
|
||
return {
|
||
"policy": POLICY,
|
||
"version": ADAPTER_VERSION,
|
||
"passed": not reasons,
|
||
"reasons": reasons,
|
||
"public_source_boundary": True,
|
||
"rate_limit_contract": "bounded_targets_and_terms",
|
||
"provenance_contract": "search_html_hash_plus_product_jsonld_hash",
|
||
"target_spec_required_when_present": True,
|
||
"stock_gate": "reject_explicit_unavailable_evidence",
|
||
"selection_listing_rejected": True,
|
||
"direct_price_write_allowed": False,
|
||
"requires_independent_same_item_verifier": True,
|
||
"requires_activation_canary": True,
|
||
}
|
||
|
||
|
||
def search_yahoo_candidates_for_pchome_products(
|
||
targets: list[dict[str, Any]],
|
||
limit: int,
|
||
*,
|
||
request_get: Callable[..., Any] | None = None,
|
||
) -> tuple[bool, str, list[dict[str, Any]]]:
|
||
"""Discover and validate Yahoo candidates for revenue-weighted PChome targets."""
|
||
from services.marketplace_product_matcher import (
|
||
build_search_terms,
|
||
build_unit_price_comparison,
|
||
score_marketplace_match,
|
||
)
|
||
|
||
target_limit = max(1, min(int(limit or 12), 20))
|
||
term_limit = _bounded_int_env("PCHOME_GROWTH_YAHOO_MAX_TERMS_PER_TARGET", 1, 1, 3)
|
||
detail_limit = _bounded_int_env(
|
||
"PCHOME_GROWTH_YAHOO_DETAIL_CANDIDATES_PER_TARGET", 3, 1, 5
|
||
)
|
||
minimum_score = _bounded_float_env(
|
||
"PCHOME_GROWTH_YAHOO_MIN_SCORE", 0.55, 0.45, 0.85
|
||
)
|
||
timeout = _bounded_float_env("PCHOME_GROWTH_YAHOO_HTTP_TIMEOUT_SECONDS", 15, 3, 30)
|
||
raw_getter = request_get or _default_get
|
||
request_delay = _bounded_float_env(
|
||
"PCHOME_GROWTH_YAHOO_REQUEST_DELAY_SECONDS",
|
||
0.2,
|
||
0.0,
|
||
2.0,
|
||
)
|
||
last_request_at = 0.0
|
||
|
||
def getter(url: str, *, timeout: float):
|
||
nonlocal last_request_at
|
||
if request_get is None and request_delay > 0:
|
||
elapsed = time.monotonic() - last_request_at
|
||
if elapsed < request_delay:
|
||
time.sleep(request_delay - elapsed)
|
||
response = raw_getter(url, timeout=timeout)
|
||
last_request_at = time.monotonic()
|
||
return response
|
||
|
||
bounded_targets = list(targets or [])[:target_limit]
|
||
pchome_evidence = enrich_targets_with_pchome_public_evidence(
|
||
bounded_targets,
|
||
request_get=getter,
|
||
timeout=timeout,
|
||
)
|
||
if not pchome_evidence.get("success"):
|
||
return (
|
||
False,
|
||
"PChome 公開商品規格或目前售價未完整讀回，本輪已停止正式候選。",
|
||
[],
|
||
)
|
||
|
||
candidates: list[dict[str, Any]] = []
|
||
searched_target_count = 0
|
||
search_request_count = 0
|
||
product_request_count = 0
|
||
source_errors: list[str] = []
|
||
for target in bounded_targets:
|
||
target_id = str(target.get("product_id") or "").strip()
|
||
target_name = str(target.get("name") or "").strip()
|
||
target_price = _to_float(target.get("price"))
|
||
if not target_id or not target_name or target_price <= 0:
|
||
continue
|
||
searched_target_count += 1
|
||
original_name = str(
|
||
(target.get("target_public_evidence") or {}).get("name") or target_name
|
||
).strip()
|
||
terms = [original_name]
|
||
terms.extend(build_search_terms(original_name, max_terms=term_limit))
|
||
unique_terms: list[str] = []
|
||
for term in terms:
|
||
term = re.sub(r"\s+", " ", str(term or "").strip())
|
||
if term and term.lower() not in {item.lower() for item in unique_terms}:
|
||
unique_terms.append(term)
|
||
if len(unique_terms) >= term_limit:
|
||
break
|
||
|
||
discovered: dict[str, dict[str, Any]] = {}
|
||
for term in unique_terms:
|
||
search_url = YAHOO_SEARCH_URL.format(query=quote(term))
|
||
try:
|
||
response = getter(search_url, timeout=timeout)
|
||
search_request_count += 1
|
||
final_url = str(getattr(response, "url", search_url) or search_url)
|
||
if not _is_public_url(
|
||
final_url, host=YAHOO_HOST, path_prefix="/search/"
|
||
):
|
||
raise ValueError("yahoo_search_redirect_outside_allowlist")
|
||
search_readback = parse_yahoo_search_html(
|
||
_response_text(response),
|
||
search_url=final_url,
|
||
)
|
||
except Exception as exc:
|
||
source_errors.append(f"{target_id}:{type(exc).__name__}")
|
||
continue
|
||
for item in search_readback.get("candidates") or []:
|
||
product_id = str(item.get("product_id") or "")
|
||
if product_id and product_id not in discovered:
|
||
discovered[product_id] = {
|
||
**item,
|
||
"target_search_term": term,
|
||
"search_provenance": {
|
||
"search_url": final_url,
|
||
"content_sha256": search_readback.get("content_sha256"),
|
||
"source_method": search_readback.get("source_method"),
|
||
},
|
||
}
|
||
|
||
preliminaries: list[tuple[float, dict[str, Any]]] = []
|
||
for item in discovered.values():
|
||
diagnostics = score_marketplace_match(
|
||
str(item.get("name") or ""),
|
||
target_name,
|
||
momo_price=_to_float(item.get("search_price")),
|
||
competitor_price=target_price,
|
||
)
|
||
score = _to_float(getattr(diagnostics, "score", 0.0))
|
||
if score >= minimum_score:
|
||
preliminaries.append((score, item))
|
||
preliminaries.sort(key=lambda pair: pair[0], reverse=True)
|
||
|
||
for _, search_candidate in preliminaries[:detail_limit]:
|
||
product_url = str(search_candidate.get("product_url") or "")
|
||
try:
|
||
response = getter(product_url, timeout=timeout)
|
||
product_request_count += 1
|
||
final_url = str(getattr(response, "url", product_url) or product_url)
|
||
if not _is_public_url(
|
||
final_url, host=YAHOO_HOST, path_prefix="/gdsale/"
|
||
):
|
||
raise ValueError("yahoo_product_redirect_outside_allowlist")
|
||
detail_readback = parse_yahoo_product_jsonld(
|
||
_response_text(response),
|
||
product_url=final_url,
|
||
)
|
||
except Exception as exc:
|
||
source_errors.append(f"{target_id}:{type(exc).__name__}")
|
||
continue
|
||
detail = detail_readback.get("product")
|
||
detail = detail if isinstance(detail, dict) else {}
|
||
source_name = str(detail.get("name") or search_candidate.get("name") or "")
|
||
source_price = _to_float(detail.get("price"))
|
||
diagnostics = score_marketplace_match(
|
||
source_name,
|
||
target_name,
|
||
momo_price=source_price,
|
||
competitor_price=target_price,
|
||
)
|
||
score = _to_float(getattr(diagnostics, "score", 0.0))
|
||
if score < minimum_score:
|
||
continue
|
||
guard = _source_contract_guard(
|
||
target=target,
|
||
search_candidate=search_candidate,
|
||
product_readback=detail_readback,
|
||
)
|
||
comparison_mode = str(getattr(diagnostics, "comparison_mode", "") or "")
|
||
price_basis = str(getattr(diagnostics, "price_basis", "") or "")
|
||
alert_tier = str(getattr(diagnostics, "alert_tier", "") or "")
|
||
match_type = str(getattr(diagnostics, "match_type", "") or "")
|
||
hard_veto = bool(getattr(diagnostics, "hard_veto", True))
|
||
exact_ready = bool(
|
||
guard.get("passed")
|
||
and not hard_veto
|
||
and comparison_mode == "exact_identity"
|
||
and price_basis == "total_price"
|
||
and alert_tier == "price_alert_exact"
|
||
and match_type == "exact"
|
||
)
|
||
unit_comparison: dict[str, Any] = {}
|
||
unit_ready = False
|
||
if comparison_mode == "unit_comparable" or price_basis == "unit_price":
|
||
unit_comparison = build_unit_price_comparison(
|
||
source_name,
|
||
target_name,
|
||
momo_price=source_price,
|
||
competitor_price=target_price,
|
||
)
|
||
unit_ready = bool(
|
||
guard.get("passed") and unit_comparison.get("comparable")
|
||
)
|
||
auto_type = (
|
||
"total_price"
|
||
if exact_ready
|
||
else "unit_price"
|
||
if unit_ready
|
||
else "manual_review"
|
||
)
|
||
gap_pct = (
|
||
round((source_price - target_price) / target_price * 100, 2)
|
||
if source_price > 0 and target_price > 0
|
||
else None
|
||
)
|
||
source_provenance = {
|
||
**(search_candidate.get("search_provenance") or {}),
|
||
"product_url": final_url,
|
||
"product_content_sha256": detail_readback.get("content_sha256"),
|
||
"product_fingerprint_sha256": detail_readback.get(
|
||
"product_fingerprint_sha256"
|
||
),
|
||
"adapter_version": ADAPTER_VERSION,
|
||
"observed_at": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
candidates.append(
|
||
{
|
||
"product_id": detail.get("product_id")
|
||
or search_candidate.get("product_id"),
|
||
"name": source_name,
|
||
"title": source_name,
|
||
"price": source_price,
|
||
"original_price": search_candidate.get("original_price"),
|
||
"currency": detail.get("currency") or "TWD",
|
||
"product_url": final_url,
|
||
"image_url": detail.get("image_url")
|
||
or search_candidate.get("image_url"),
|
||
"brand": detail.get("brand") or search_candidate.get("brand"),
|
||
"gtin13": detail.get("gtin13"),
|
||
"seller": detail.get("seller") or search_candidate.get("seller"),
|
||
"stock_status": detail.get("availability")
|
||
or search_candidate.get("stock_status"),
|
||
"rating": detail.get("rating") or search_candidate.get("rating"),
|
||
"review_count": detail.get("review_count")
|
||
or search_candidate.get("review_count"),
|
||
"target_pchome_product_id": target_id,
|
||
"target_pchome_name": target_name,
|
||
"target_pchome_price": target_price,
|
||
"target_public_evidence": target.get("target_public_evidence")
|
||
or {},
|
||
"target_search_term": search_candidate.get("target_search_term"),
|
||
"target_match_score": round(score, 3),
|
||
"target_match_reasons": list(
|
||
getattr(diagnostics, "reasons", ()) or ()
|
||
),
|
||
"target_comparison_mode": comparison_mode,
|
||
"target_match_type": match_type,
|
||
"target_alert_tier": alert_tier,
|
||
"target_hard_veto": hard_veto,
|
||
"target_price_basis": "total_price" if exact_ready else price_basis,
|
||
"target_gap_pct": gap_pct,
|
||
"target_unit_price_comparison": unit_comparison,
|
||
"auto_compare_type": auto_type,
|
||
"can_auto_compare": auto_type in {"total_price", "unit_price"},
|
||
"source_strategy": CONNECTOR_KEY,
|
||
"source_contract_guard": guard,
|
||
"source_provenance": source_provenance,
|
||
}
|
||
)
|
||
|
||
exact_count = sum(
|
||
1 for item in candidates if item.get("auto_compare_type") == "total_price"
|
||
)
|
||
guarded_count = sum(
|
||
1
|
||
for item in candidates
|
||
if not bool((item.get("source_contract_guard") or {}).get("passed"))
|
||
)
|
||
success = bool(searched_target_count and (candidates or not source_errors))
|
||
message = (
|
||
f"已用 {searched_target_count} 筆高業績商品搜尋 Yahoo，"
|
||
f"取得 {len(candidates)} 筆商品頁候選，{exact_count} 筆通過來源前置檢查，"
|
||
f"{guarded_count} 筆因規格或選項風險阻擋。"
|
||
)
|
||
if source_errors and not candidates:
|
||
message = "Yahoo 公開來源本輪無可用候選，已安全停止且不寫正式價格。"
|
||
return success, message, candidates
|
||
|
||
|
||
__all__ = [
|
||
"ADAPTER_VERSION",
|
||
"CONNECTOR_KEY",
|
||
"DEFAULT_RECEIPT_ROOT",
|
||
"INGESTION_METHOD",
|
||
"PLATFORM_CODE",
|
||
"POLICY",
|
||
"SOURCE_CODE",
|
||
"SOURCE_LABEL",
|
||
"WORK_ITEM_ID",
|
||
"enrich_targets_with_pchome_public_evidence",
|
||
"parse_pchome_public_product_jsonp",
|
||
"parse_yahoo_product_jsonld",
|
||
"parse_yahoo_search_html",
|
||
"search_yahoo_candidates_for_pchome_products",
|
||
]
|