457 lines
14 KiB
Python
457 lines
14 KiB
Python
"""Read-only product-page and package evidence parsing."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import unicodedata
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
|
||
from services.pchome_mapping_backlog.contracts import (
|
||
_ai_exception_compatibility_fields,
|
||
)
|
||
from services.pchome_mapping_backlog.policies import (
|
||
PCHOME_FETCH_ALLOWED_DOMAIN,
|
||
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
|
||
)
|
||
|
||
|
||
_MEASURE_UNIT_ALIASES = {
|
||
"ml": "ml",
|
||
"m l": "ml",
|
||
"毫升": "ml",
|
||
"l": "l",
|
||
"公升": "l",
|
||
"g": "g",
|
||
"公克": "g",
|
||
"克": "g",
|
||
"mg": "mg",
|
||
"毫克": "mg",
|
||
"kg": "kg",
|
||
"公斤": "kg",
|
||
"oz": "oz",
|
||
"floz": "floz",
|
||
"fl oz": "floz",
|
||
"fl.oz": "floz",
|
||
}
|
||
_MEASURE_RE = re.compile(
|
||
r"(?P<value>\d+(?:\.\d+)?)\s*(?P<unit>fl\.?\s*oz|floz|ml|m\s*l|毫升|公升|l|mg|毫克|kg|公斤|g|公克|克|oz)",
|
||
re.IGNORECASE,
|
||
)
|
||
_COUNT_UNIT_ALIASES = {
|
||
"入": "ct",
|
||
"瓶": "ct",
|
||
"支": "ct",
|
||
"條": "ct",
|
||
"盒": "ct",
|
||
"包": "ct",
|
||
"袋": "ct",
|
||
"片": "ct",
|
||
"顆": "ct",
|
||
"粒": "ct",
|
||
"錠": "ct",
|
||
"枚": "ct",
|
||
"件": "ct",
|
||
"罐": "ct",
|
||
"蕊": "ct",
|
||
"張": "ct",
|
||
"抽": "ct",
|
||
"組": "ct",
|
||
"pcs": "ct",
|
||
"pc": "ct",
|
||
"ct": "ct",
|
||
}
|
||
_COUNT_UNIT_PATTERN = "|".join(sorted(map(re.escape, _COUNT_UNIT_ALIASES), key=len, reverse=True))
|
||
_COUNT_RE = re.compile(rf"(?P<count>\d+)\s*(?P<unit>{_COUNT_UNIT_PATTERN})", re.IGNORECASE)
|
||
_CHINESE_COUNT_RE = re.compile(rf"(?P<count>[一二兩三四五六七八九十])\s*(?P<unit>{_COUNT_UNIT_PATTERN})")
|
||
_MULTIPLIER_RE = re.compile(r"(?:x|X)\s*(?P<count>\d+)")
|
||
_CHINESE_DIGITS = {
|
||
"一": 1,
|
||
"二": 2,
|
||
"兩": 2,
|
||
"三": 3,
|
||
"四": 4,
|
||
"五": 5,
|
||
"六": 6,
|
||
"七": 7,
|
||
"八": 8,
|
||
"九": 9,
|
||
"十": 10,
|
||
}
|
||
_VARIANT_KEYWORDS = (
|
||
"任選",
|
||
"多款",
|
||
"色號",
|
||
"色選",
|
||
"顏色",
|
||
"款式",
|
||
"香味",
|
||
"香調",
|
||
"口味",
|
||
"尺寸",
|
||
"規格可選",
|
||
)
|
||
_BUNDLE_KEYWORDS = (
|
||
"套組",
|
||
"組合",
|
||
"超值組",
|
||
"買一送一",
|
||
"贈",
|
||
"加贈",
|
||
"禮盒",
|
||
"福袋",
|
||
)
|
||
_EXPIRY_KEYWORDS = ("即期", "效期", "有效期限")
|
||
_SAMPLE_KEYWORDS = ("試用", "小樣", "體驗", "旅行組")
|
||
_UNIT_BASE_MEASURE = {
|
||
"ml": {"value": 100, "unit": "ml"},
|
||
"l": {"value": 1, "unit": "l"},
|
||
"g": {"value": 100, "unit": "g"},
|
||
"mg": {"value": 100, "unit": "mg"},
|
||
"kg": {"value": 1, "unit": "kg"},
|
||
"oz": {"value": 1, "unit": "oz"},
|
||
"floz": {"value": 1, "unit": "floz"},
|
||
"ct": {"value": 1, "unit": "ct"},
|
||
}
|
||
|
||
|
||
def _to_float(value: Any) -> float:
|
||
try:
|
||
return float(value or 0)
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
|
||
|
||
def _action_code(item: dict[str, Any]) -> str:
|
||
action = item.get("recommended_action") or {}
|
||
return str(action.get("code") or "")
|
||
|
||
|
||
def _action_label(item: dict[str, Any]) -> str:
|
||
action = item.get("recommended_action") or {}
|
||
return str(action.get("label") or _action_code(item) or "unknown")
|
||
|
||
|
||
def _first_present(*values: Any) -> Any:
|
||
for value in values:
|
||
if value not in (None, ""):
|
||
return value
|
||
return None
|
||
|
||
|
||
def _pchome_product_url(product_id: str) -> str | None:
|
||
if not product_id:
|
||
return None
|
||
return f"https://24h.pchome.com.tw/prod/{product_id}"
|
||
|
||
|
||
def _normalize_package_text(value: str) -> str:
|
||
normalized = unicodedata.normalize("NFKC", value or "")
|
||
normalized = normalized.replace("×", "x").replace("＊", "x").replace("*", "x")
|
||
return re.sub(r"\s+", " ", normalized).strip().lower()
|
||
|
||
|
||
def _canonical_measure_unit(unit: str) -> str:
|
||
compact = re.sub(r"\s+", " ", unit or "").strip().lower()
|
||
return _MEASURE_UNIT_ALIASES.get(compact, compact)
|
||
|
||
|
||
def _round_quantity(value: float) -> int | float:
|
||
return int(value) if float(value).is_integer() else round(value, 3)
|
||
|
||
|
||
def _risk_signals(normalized_name: str) -> list[str]:
|
||
signals = []
|
||
if any(keyword in normalized_name for keyword in _VARIANT_KEYWORDS):
|
||
signals.append("variant_selection")
|
||
if any(keyword in normalized_name for keyword in _BUNDLE_KEYWORDS):
|
||
signals.append("bundle_or_promo")
|
||
if any(keyword in normalized_name for keyword in _EXPIRY_KEYWORDS):
|
||
signals.append("freshness_or_expiry")
|
||
if any(keyword in normalized_name for keyword in _SAMPLE_KEYWORDS):
|
||
signals.append("sample_or_travel_size")
|
||
return signals
|
||
|
||
|
||
def _dedupe_quantity_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
seen = set()
|
||
deduped = []
|
||
for row in rows:
|
||
key = (row.get("value"), row.get("unit"), row.get("raw"))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
deduped.append(row)
|
||
return deduped
|
||
|
||
|
||
def _jsonld_nodes(value: Any):
|
||
if isinstance(value, dict):
|
||
yield value
|
||
for child in value.values():
|
||
yield from _jsonld_nodes(child)
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
yield from _jsonld_nodes(item)
|
||
|
||
|
||
def _jsonld_type_includes(node: dict[str, Any], expected_type: str) -> bool:
|
||
node_type = node.get("@type") or node.get("type")
|
||
if isinstance(node_type, str):
|
||
return node_type.lower() == expected_type.lower()
|
||
if isinstance(node_type, list):
|
||
return any(str(item).lower() == expected_type.lower() for item in node_type)
|
||
return False
|
||
|
||
|
||
def _first_image_url(image_value: Any) -> str | None:
|
||
if isinstance(image_value, str) and image_value.strip():
|
||
return image_value.strip()
|
||
if isinstance(image_value, dict):
|
||
return _first_present(image_value.get("url"), image_value.get("contentUrl"))
|
||
if isinstance(image_value, list):
|
||
for item in image_value:
|
||
image = _first_image_url(item)
|
||
if image:
|
||
return image
|
||
return None
|
||
|
||
|
||
def _normalize_schema_availability(value: Any) -> str | None:
|
||
if value in (None, ""):
|
||
return None
|
||
text = str(value).strip()
|
||
lowered = text.lower()
|
||
compact = re.sub(r"[\s_-]+", "", lowered)
|
||
if "instock" in compact:
|
||
return "in_stock"
|
||
if "outofstock" in compact or "soldout" in compact:
|
||
return "out_of_stock"
|
||
if "preorder" in compact:
|
||
return "preorder"
|
||
if "backorder" in compact:
|
||
return "backorder"
|
||
if "discontinued" in compact:
|
||
return "discontinued"
|
||
return "unknown"
|
||
|
||
|
||
def parse_pchome_product_page_evidence_html(html: str, product_url: str | None = None) -> dict[str, Any]:
|
||
"""Parse product-page evidence from HTML fixture text without fetching or writing."""
|
||
soup = BeautifulSoup(html or "", "html.parser")
|
||
warnings = []
|
||
image_url = None
|
||
availability = None
|
||
availability_raw = None
|
||
jsonld_product_found = False
|
||
jsonld_offer_found = False
|
||
fallbacks_used = []
|
||
|
||
for script in soup.find_all("script", attrs={"type": re.compile("ld\\+json", re.IGNORECASE)}):
|
||
text = script.string or script.get_text() or ""
|
||
if not text.strip():
|
||
continue
|
||
try:
|
||
data = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
warnings.append("invalid_jsonld_skipped")
|
||
continue
|
||
for node in _jsonld_nodes(data):
|
||
if _jsonld_type_includes(node, "Product"):
|
||
jsonld_product_found = True
|
||
image_url = image_url or _first_image_url(node.get("image"))
|
||
if _jsonld_type_includes(node, "Offer"):
|
||
jsonld_offer_found = True
|
||
availability_raw = availability_raw or node.get("availability")
|
||
availability = availability or _normalize_schema_availability(availability_raw)
|
||
|
||
if not image_url:
|
||
og_image = soup.find("meta", property="og:image")
|
||
if og_image and og_image.get("content"):
|
||
image_url = str(og_image.get("content")).strip()
|
||
fallbacks_used.append("og:image")
|
||
|
||
if not availability:
|
||
product_availability = soup.find("meta", attrs={"property": "product:availability"})
|
||
if product_availability and product_availability.get("content"):
|
||
availability_raw = str(product_availability.get("content")).strip()
|
||
availability = _normalize_schema_availability(availability_raw)
|
||
fallbacks_used.append("product:availability")
|
||
|
||
return {
|
||
"policy": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
|
||
"source": "html_fixture",
|
||
"product_url": product_url,
|
||
"image_url": image_url,
|
||
"availability": availability,
|
||
"availability_raw": availability_raw,
|
||
"jsonld_product_found": jsonld_product_found,
|
||
"jsonld_offer_found": jsonld_offer_found,
|
||
"fallbacks_used": fallbacks_used,
|
||
"parser_warnings": warnings,
|
||
"safety": {
|
||
"fetches_external_sites": False,
|
||
"writes_database": False,
|
||
"executes_search": False,
|
||
"dispatches_telegram": False,
|
||
"llm_calls": False,
|
||
},
|
||
}
|
||
|
||
|
||
def _is_allowed_pchome_product_url(product_url: str | None) -> bool:
|
||
if not product_url:
|
||
return False
|
||
parsed = urlparse(product_url)
|
||
return (
|
||
parsed.scheme in {"http", "https"}
|
||
and parsed.netloc == PCHOME_FETCH_ALLOWED_DOMAIN
|
||
and parsed.path.startswith("/prod/")
|
||
)
|
||
|
||
|
||
def _response_content_bytes(response: Any, max_html_bytes: int) -> bytes:
|
||
content = getattr(response, "content", None)
|
||
if content is None:
|
||
content = str(getattr(response, "text", "") or "").encode("utf-8")
|
||
if len(content) > max_html_bytes:
|
||
raise ValueError("html_size_cap_exceeded")
|
||
return bytes(content)
|
||
|
||
|
||
def _fetch_product_page_html(
|
||
product_url: str,
|
||
*,
|
||
timeout_seconds: int,
|
||
max_html_bytes: int,
|
||
http_get: Any = None,
|
||
) -> tuple[str, dict[str, Any]]:
|
||
getter = http_get or requests.get
|
||
response = getter(
|
||
product_url,
|
||
timeout=timeout_seconds,
|
||
headers={
|
||
"User-Agent": "MOMO-Pro-Evidence-Gate/1.0 (+read-only; no-write)",
|
||
"Accept": "text/html,application/xhtml+xml",
|
||
},
|
||
)
|
||
status_code = int(getattr(response, "status_code", 0) or 0)
|
||
if status_code >= 400:
|
||
raise ValueError(f"http_status_{status_code}")
|
||
content = _response_content_bytes(response, max_html_bytes=max_html_bytes)
|
||
encoding = getattr(response, "encoding", None) or "utf-8"
|
||
return content.decode(encoding, errors="replace"), {
|
||
"http_status": status_code,
|
||
"content_bytes": len(content),
|
||
}
|
||
|
||
|
||
def parse_unit_package_basis(product_name: str) -> dict[str, Any]:
|
||
"""Parse unit/package evidence from a product title without fetching or writing."""
|
||
normalized_name = _normalize_package_text(product_name)
|
||
quantities = []
|
||
for match in _MEASURE_RE.finditer(normalized_name):
|
||
value = float(match.group("value"))
|
||
unit = _canonical_measure_unit(match.group("unit"))
|
||
quantities.append(
|
||
{
|
||
"value": _round_quantity(value),
|
||
"unit": unit,
|
||
"raw": match.group(0).strip(),
|
||
}
|
||
)
|
||
|
||
counts = []
|
||
for match in _COUNT_RE.finditer(normalized_name):
|
||
count = int(match.group("count"))
|
||
counts.append({"count": count, "unit": match.group("unit"), "canonical_unit": "ct", "raw": match.group(0)})
|
||
for match in _CHINESE_COUNT_RE.finditer(normalized_name):
|
||
count = _CHINESE_DIGITS.get(match.group("count"))
|
||
if count:
|
||
counts.append({"count": count, "unit": match.group("unit"), "canonical_unit": "ct", "raw": match.group(0)})
|
||
counts = _dedupe_quantity_rows(counts)
|
||
|
||
multipliers = [int(match.group("count")) for match in _MULTIPLIER_RE.finditer(normalized_name)]
|
||
for row in counts:
|
||
if row["count"] > 1 and row["count"] not in multipliers:
|
||
multipliers.append(row["count"])
|
||
|
||
risk_signals = _risk_signals(normalized_name)
|
||
primary_quantity = quantities[0] if quantities else None
|
||
primary_count = counts[0] if counts else None
|
||
unit_label = primary_quantity["unit"] if primary_quantity else ("ct" if primary_count else None)
|
||
multiplier_product = 1
|
||
for multiplier in multipliers:
|
||
multiplier_product *= max(multiplier, 1)
|
||
|
||
estimated_total_quantity = None
|
||
if primary_quantity:
|
||
estimated_total_quantity = float(primary_quantity["value"]) * multiplier_product
|
||
elif primary_count:
|
||
estimated_total_quantity = float(primary_count["count"])
|
||
|
||
if primary_quantity and risk_signals:
|
||
package_basis = "variant_sensitive_quantity_candidate"
|
||
elif primary_quantity and multiplier_product > 1:
|
||
package_basis = "multi_pack_quantity_candidate"
|
||
elif primary_quantity:
|
||
package_basis = "single_unit_quantity_candidate"
|
||
elif primary_count:
|
||
package_basis = "count_package_candidate"
|
||
elif risk_signals:
|
||
package_basis = "catalog_or_variant_review"
|
||
else:
|
||
package_basis = "insufficient"
|
||
|
||
has_basis = package_basis != "insufficient"
|
||
confidence = 0.0
|
||
if primary_quantity and not risk_signals:
|
||
confidence = 0.86 if multiplier_product == 1 else 0.78
|
||
elif primary_quantity:
|
||
confidence = 0.62
|
||
elif primary_count and not risk_signals:
|
||
confidence = 0.68
|
||
elif has_basis:
|
||
confidence = 0.36
|
||
|
||
unit_pricing_measure = None
|
||
unit_pricing_base_measure = None
|
||
if estimated_total_quantity is not None and unit_label:
|
||
unit_pricing_measure = {
|
||
"value": _round_quantity(estimated_total_quantity),
|
||
"unit": unit_label,
|
||
}
|
||
unit_pricing_base_measure = _UNIT_BASE_MEASURE.get(unit_label)
|
||
|
||
ai_exception_required = bool(risk_signals) or not has_basis
|
||
|
||
return {
|
||
"source": "deterministic_product_title_parser",
|
||
"mode": "local_parse_only",
|
||
"product_name": product_name or "",
|
||
"package_basis": package_basis,
|
||
"quantities": quantities,
|
||
"counts": counts,
|
||
"multipliers": multipliers,
|
||
"estimated_total_quantity": _round_quantity(estimated_total_quantity) if estimated_total_quantity is not None else None,
|
||
"unit_label": unit_label,
|
||
"unit_pricing_measure": unit_pricing_measure,
|
||
"unit_pricing_base_measure": unit_pricing_base_measure,
|
||
"risk_signals": risk_signals,
|
||
"parser_confidence": confidence,
|
||
**_ai_exception_compatibility_fields(ai_exception_required),
|
||
"writes_database": False,
|
||
"fetches_external_sites": False,
|
||
"llm_calls": False,
|
||
}
|
||
|
||
|
||
__all__ = (
|
||
"parse_pchome_product_page_evidence_html",
|
||
"parse_unit_package_basis",
|
||
)
|