"""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", ]