wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(pchome): extract backlog policy and evidence modules

Browse Source
This commit is contained in:
ogt
2026-07-11 11:08:38 +08:00
parent c7906d5b80
commit ff47ddf02f
11 changed files with 969 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.py
View File

@@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
SYSTEM_VERSION = "V10.779"
SYSTEM_VERSION = "V10.780"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示

2
docs/AI_INTELLIGENCE_MODULE_SOT.md
View File

@@ -2,7 +2,7 @@
> **最後更新**: 2026-07-11 (台北時間)
> **狀態**: 🟠 Partial。四 Agent、PixelRAG、MCP/RAG registry、PChome controlled-preview families 與多項 smoke/readback 已建立；但 access control、database identity/RBAC、full asset reconciliation、software supply chain、same-run controlled apply、restore drill、internal RAG canary 與 MCP/RAG runtime closure 尚未全部完成。任何局部 receipt 不得代表整體閉環完成。
> **適用版本**: V10.779
> **適用版本**: V10.780
---

2
docs/memory/code_modularization_inventory_20260430.md
View File

@@ -129,7 +129,7 @@
| 行數 | 檔案 | 分類 | 拆分方向 |
|---:|---|---|---|
| 44888 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service | 立即停止新增 closeout package；拆 policy、executor、verifier、receipt、reporter |
| 44261 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | 96 個 policy、AI exception contract、商品頁與單位 evidence parser 已移到 `services/pchome_mapping_backlog/`；下一步依序拆 executor、verifier、receipt、reporter |
| 14289 | `services/ai_automation_smoke_service.py` | P0 smoke 巨型 service | 拆 family registry、collectors、health projection、metrics adapter |
| 9383 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint、拆分進行中 | scheduler hook 已移到 `services/openclaw_bot/scheduled_jobs.py`；下一步拆 report、command |
| 7641 | `routes/ai_routes.py` | P0 巨型 Blueprint | PChome growth、AI automation、recommendation route extension 分離 |

1
services/codebase_modularization_performance_service.py
View File

@@ -19,6 +19,7 @@ IN_PROGRESS_PYTHON_PATHS = {
"app.py",
"routes/openclaw_bot_routes.py",
"run_scheduler.py",
"services/pchome_mapping_backlog_service.py",
}
IGNORED_PARTS = {
".claude",

1
services/pchome_mapping_backlog/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Cohesive PChome mapping backlog components."""

66
services/pchome_mapping_backlog/contracts.py Normal file
View File

@@ -0,0 +1,66 @@
"""AI-first compatibility contracts for PChome mapping receipts."""
from __future__ import annotations
from typing import Any
from services.ai_exception_contract import (
AI_EXCEPTION_MODE_KEY,
AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
AI_EXCEPTION_REQUIRED_COUNT_KEY,
AI_EXCEPTION_REQUIRED_KEY,
LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY,
LEGACY_HUMAN_REVIEW_REQUIRED_KEY,
LEGACY_HUMAN_REVIEW_REQUIRED_LEGACY_KEY,
LEGACY_PRIMARY_FLOW_COUNT_KEY,
LEGACY_REVIEW_MODE_EXCEPTION_ONLY,
LEGACY_REVIEW_MODE_KEY,
LEGACY_REVIEW_REQUIRED_COUNT_KEY,
LEGACY_REVIEW_REQUIRED_KEY,
PRIMARY_HUMAN_GATE_COUNT_KEY,
)
def _ai_exception_compatibility_fields(ai_exception_required: bool) -> dict[str, Any]:
"""Return the AI-first exception contract with legacy readback aliases."""
return {
LEGACY_HUMAN_REVIEW_REQUIRED_KEY: False,
LEGACY_HUMAN_REVIEW_REQUIRED_LEGACY_KEY: bool(ai_exception_required),
AI_EXCEPTION_REQUIRED_KEY: bool(ai_exception_required),
PRIMARY_HUMAN_GATE_COUNT_KEY: 0,
AI_EXCEPTION_MODE_KEY: AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
}
def _legacy_review_compatibility_fields(ai_exception_required: bool = False) -> dict[str, Any]:
"""Keep old review-mode keys false while exposing the AI exception state."""
return {
LEGACY_REVIEW_REQUIRED_KEY: False,
LEGACY_REVIEW_MODE_KEY: LEGACY_REVIEW_MODE_EXCEPTION_ONLY,
AI_EXCEPTION_REQUIRED_KEY: bool(ai_exception_required),
AI_EXCEPTION_MODE_KEY: AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
}
def _evidence_requires_ai_exception(evidence: dict[str, Any]) -> bool:
return bool(
evidence.get(AI_EXCEPTION_REQUIRED_KEY)
or evidence.get(LEGACY_HUMAN_REVIEW_REQUIRED_LEGACY_KEY)
or evidence.get(LEGACY_HUMAN_REVIEW_REQUIRED_KEY)
)
def _summary_exception_count(summary: dict[str, Any]) -> int:
return int(
summary.get(AI_EXCEPTION_REQUIRED_COUNT_KEY)
or summary.get(LEGACY_REVIEW_REQUIRED_COUNT_KEY)
or 0
)
__all__ = (
"_ai_exception_compatibility_fields",
"_evidence_requires_ai_exception",
"_legacy_review_compatibility_fields",
"_summary_exception_count",
)

456
services/pchome_mapping_backlog/evidence.py Normal file
View File

@@ -0,0 +1,456 @@
"""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",
)

280
services/pchome_mapping_backlog/policies.py Normal file
View File

@@ -0,0 +1,280 @@
"""Policy registry for the PChome mapping backlog automation lanes."""
from __future__ import annotations
BACKLOG_POLICY = "read_only_pchome_growth_mapping_backlog"
OPERATOR_PREVIEW_POLICY = "read_only_pchome_growth_mapping_operator_preview"
DIRECT_MAPPING_AUTO_SEARCH_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_auto_search_package"
)
DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_decision_package"
)
DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout"
)
DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_decision_package"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_auto_resolution"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_resolution_closeout"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_INPUT_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_input"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREVIEW_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preview"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_MATERIALIZATION_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_materialization"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREFLIGHT_VERIFIER_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preflight_verifier"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_executor"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_recovery"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence"
)
AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
AI_AUTOMATION_SURFACE_SUMMARY_POLICY = (
"read_only_pchome_growth_ai_automation_surface_summary"
)
EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY = "read_only_pchome_product_page_evidence_parser"
EVIDENCE_FETCH_GATE_POLICY = "controlled_read_only_pchome_product_page_evidence_fetch_gate"
EVIDENCE_MERGE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_merge_preview"
AUTO_POLICY_RECEIPT_GATE_POLICY = "read_only_pchome_growth_auto_policy_receipt_gate"
AUTO_POLICY_PERSISTENCE_GATE_POLICY = "read_only_pchome_growth_auto_policy_persistence_gate"
AUTO_POLICY_SCHEMA_MIGRATION_PREVIEW_POLICY = "read_only_pchome_growth_auto_policy_schema_migration_preview"
AUTO_POLICY_MIGRATION_FILE_PREVIEW_POLICY = "read_only_pchome_growth_auto_policy_migration_file_preview"
AUTO_POLICY_APPLY_READINESS_CLOSEOUT_POLICY = "read_only_pchome_growth_auto_policy_apply_readiness_closeout"
AUTO_POLICY_MIGRATION_FILE_GENERATION_REQUEST_POLICY = (
"read_only_pchome_growth_auto_policy_migration_file_generation_request"
)
AUTO_POLICY_MIGRATION_APPLY_GATE_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_migration_apply_gate_preview"
)
AUTO_POLICY_DB_APPLY_REQUEST_GATE_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_request_gate_preview"
)
AUTO_POLICY_DB_APPLY_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_package"
)
AUTO_POLICY_DB_APPLY_VERIFIER_ARTIFACT_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_verifier_artifact_preview"
)
AUTO_POLICY_DB_APPLY_FINAL_HANDOFF_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_final_handoff_package"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_shell_preview"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_shell_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_INTAKE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_request_intake"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_request_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_LANE_GUARD_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_decision_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_decision_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_ISSUER_GATE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_issuer_gate"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_decision_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_decision_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_GUARD_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_issuer_guard"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_issuer_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_execution_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_EVIDENCE_INTAKE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_evidence_intake"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DETACHED_VERIFICATION_EVIDENCE_VALIDATION_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_detached_verification_evidence_validation"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_VERIFIER_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_verifier_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_evidence_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_evidence_execution_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_APPLY_FINAL_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_apply_final_preflight"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_package"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_READINESS_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_readiness"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PLAN_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_plan_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_COMMAND_ARTIFACT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_command_artifact_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_EXECUTION_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_execution_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_POST_RECEIPT_PARSER_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_post_receipt_parser_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_APPLY_ENFORCEMENT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_apply_enforcement_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_EXECUTOR_GUARD_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_final_executor_guard_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PRE_APPLY_REPLAY_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_pre_apply_replay_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_APPLY_EXECUTOR_READINESS_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_apply_executor_readiness_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_INVOCATION_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_invocation_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_INVOCATION_PACKAGE_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_write_invocation_package_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PREFLIGHT_GUARD_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_preflight_guard_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_INVOCATION_BOUNDARY_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_invocation_boundary_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_EXECUTION_RECEIPT_HANDOFF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_execution_receipt_handoff_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_NO_RUNNER_EXECUTION_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_final_no_runner_execution_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_CONTROLLED_EXECUTOR_QUARANTINE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_controlled_executor_quarantine_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_ENVELOPE_FREEZE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_envelope_freeze_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FROZEN_ENVELOPE_VERIFIER_HANDOFF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_frozen_envelope_verifier_handoff_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_INVOCATION_LOCK_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_invocation_lock_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_NO_EXECUTION_RECEIPT_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_no_execution_receipt_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_RECEIPT_PERSISTENCE_GUARD_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_receipt_persistence_guard_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_PERSISTENCE_STORAGE_BOUNDARY_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_receipt_persistence_storage_boundary_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_STORAGE_BOUNDARY_NO_WRITE_LEDGER_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_storage_boundary_no_write_ledger_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_LEDGER_RETENTION_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_write_ledger_retention_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RETENTION_BOUNDARY_NO_WRITE_ARCHIVE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_retention_boundary_no_write_archive_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_ARCHIVE_RETENTION_SEALED_HANDOFF_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_archive_retention_sealed_handoff_proof_closeout"
)
PCHOME_FETCH_ALLOWED_DOMAIN = "24h.pchome.com.tw"
PCHOME_FETCH_MAX_BATCH_SIZE = 12
PCHOME_FETCH_DEFAULT_TIMEOUT_SECONDS = 5
PCHOME_FETCH_MAX_HTML_BYTES = 512_000
EXTERNAL_BENCHMARK_REFERENCES = [
{
"source": "Google Merchant Center product data specification",
"url": "https://support.google.com/merchants/answer/7052112",
"applies_to": "accurate_product_feed_matching",
},
{
"source": "Google Search Product structured data",
"url": "https://developers.google.com/search/docs/appearance/structured-data/product",
"applies_to": "rich_product_information_and_offer_visibility",
},
{
"source": "Google Merchant listing structured data",
"url": "https://developers.google.com/search/docs/appearance/structured-data/merchant-listing",
"applies_to": "product_offer_price_currency_availability",
},
{
"source": "Baymard ecommerce product and search UX benchmark",
"url": "https://baymard.com/research/product-page",
"applies_to": "operator_search_and_product_detail_quality",
},
]
__all__ = tuple(name for name in globals() if name.isupper())

857
services/pchome_mapping_backlog_service.py
View File

@@ -2,16 +2,13 @@
from __future__ import annotations
import json
import hashlib
import json
import re
import unicodedata
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
import requests
from bs4 import BeautifulSoup
from services.ai_exception_contract import (
AI_EXCEPTION_MODE_KEY,
@@ -30,747 +27,123 @@ from services.ai_exception_contract import (
)
BACKLOG_POLICY = "read_only_pchome_growth_mapping_backlog"
OPERATOR_PREVIEW_POLICY = "read_only_pchome_growth_mapping_operator_preview"
DIRECT_MAPPING_AUTO_SEARCH_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_auto_search_package"
from services.pchome_mapping_backlog.policies import (
BACKLOG_POLICY,
OPERATOR_PREVIEW_POLICY,
DIRECT_MAPPING_AUTO_SEARCH_PACKAGE_POLICY,
DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY,
DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY,
DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY,
DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_INPUT_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREVIEW_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_MATERIALIZATION_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREFLIGHT_VERIFIER_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY,
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY,
AI_AUTOMATION_READINESS_POLICY,
AI_AUTOMATION_SURFACE_SUMMARY_POLICY,
EVIDENCE_ENRICHMENT_PREVIEW_POLICY,
EVIDENCE_SOURCE_PREVIEW_POLICY,
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
EVIDENCE_FETCH_GATE_POLICY,
EVIDENCE_MERGE_PREVIEW_POLICY,
AUTO_POLICY_RECEIPT_GATE_POLICY,
AUTO_POLICY_PERSISTENCE_GATE_POLICY,
AUTO_POLICY_SCHEMA_MIGRATION_PREVIEW_POLICY,
AUTO_POLICY_MIGRATION_FILE_PREVIEW_POLICY,
AUTO_POLICY_APPLY_READINESS_CLOSEOUT_POLICY,
AUTO_POLICY_MIGRATION_FILE_GENERATION_REQUEST_POLICY,
AUTO_POLICY_MIGRATION_APPLY_GATE_PREVIEW_POLICY,
AUTO_POLICY_DB_APPLY_REQUEST_GATE_PREVIEW_POLICY,
AUTO_POLICY_DB_APPLY_EXECUTION_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_PACKAGE_POLICY,
AUTO_POLICY_DB_APPLY_VERIFIER_ARTIFACT_PREVIEW_POLICY,
AUTO_POLICY_DB_APPLY_FINAL_HANDOFF_PACKAGE_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_PREVIEW_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_INTAKE_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_LANE_GUARD_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_ISSUER_GATE_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_GUARD_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_EVIDENCE_INTAKE_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DETACHED_VERIFICATION_EVIDENCE_VALIDATION_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_VERIFIER_RECEIPT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_APPLY_FINAL_PREFLIGHT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_READINESS_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PLAN_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_COMMAND_ARTIFACT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_EXECUTION_RECEIPT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_POST_RECEIPT_PARSER_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_APPLY_ENFORCEMENT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_EXECUTOR_GUARD_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PRE_APPLY_REPLAY_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_APPLY_EXECUTOR_READINESS_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_INVOCATION_RECEIPT_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_INVOCATION_PACKAGE_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PREFLIGHT_GUARD_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_INVOCATION_BOUNDARY_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_EXECUTION_RECEIPT_HANDOFF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_NO_RUNNER_EXECUTION_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_CONTROLLED_EXECUTOR_QUARANTINE_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_ENVELOPE_FREEZE_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FROZEN_ENVELOPE_VERIFIER_HANDOFF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_INVOCATION_LOCK_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_NO_EXECUTION_RECEIPT_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_RECEIPT_PERSISTENCE_GUARD_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_PERSISTENCE_STORAGE_BOUNDARY_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_STORAGE_BOUNDARY_NO_WRITE_LEDGER_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_LEDGER_RETENTION_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RETENTION_BOUNDARY_NO_WRITE_ARCHIVE_PROOF_CLOSEOUT_POLICY,
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_ARCHIVE_RETENTION_SEALED_HANDOFF_PROOF_CLOSEOUT_POLICY,
PCHOME_FETCH_ALLOWED_DOMAIN,
PCHOME_FETCH_MAX_BATCH_SIZE,
PCHOME_FETCH_DEFAULT_TIMEOUT_SECONDS,
PCHOME_FETCH_MAX_HTML_BYTES,
EXTERNAL_BENCHMARK_REFERENCES,
)
from services.pchome_mapping_backlog.contracts import (
_ai_exception_compatibility_fields,
_evidence_requires_ai_exception,
_legacy_review_compatibility_fields,
_summary_exception_count,
)
from services.pchome_mapping_backlog.evidence import (
_action_code,
_action_label,
_fetch_product_page_html,
_first_present,
_is_allowed_pchome_product_url,
_pchome_product_url,
_to_float,
parse_pchome_product_page_evidence_html,
parse_unit_package_basis,
)
DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_decision_package"
)
DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout"
)
DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_decision_package"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_auto_resolution"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_resolution_closeout"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_INPUT_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_input"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREVIEW_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preview"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_MATERIALIZATION_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_materialization"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREFLIGHT_VERIFIER_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preflight_verifier"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_executor"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_POLICY = (
"ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_recovery"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention"
)
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence"
)
AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
AI_AUTOMATION_SURFACE_SUMMARY_POLICY = (
"read_only_pchome_growth_ai_automation_surface_summary"
)
EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY = "read_only_pchome_product_page_evidence_parser"
EVIDENCE_FETCH_GATE_POLICY = "controlled_read_only_pchome_product_page_evidence_fetch_gate"
EVIDENCE_MERGE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_merge_preview"
AUTO_POLICY_RECEIPT_GATE_POLICY = "read_only_pchome_growth_auto_policy_receipt_gate"
AUTO_POLICY_PERSISTENCE_GATE_POLICY = "read_only_pchome_growth_auto_policy_persistence_gate"
AUTO_POLICY_SCHEMA_MIGRATION_PREVIEW_POLICY = "read_only_pchome_growth_auto_policy_schema_migration_preview"
AUTO_POLICY_MIGRATION_FILE_PREVIEW_POLICY = "read_only_pchome_growth_auto_policy_migration_file_preview"
AUTO_POLICY_APPLY_READINESS_CLOSEOUT_POLICY = "read_only_pchome_growth_auto_policy_apply_readiness_closeout"
AUTO_POLICY_MIGRATION_FILE_GENERATION_REQUEST_POLICY = (
"read_only_pchome_growth_auto_policy_migration_file_generation_request"
)
AUTO_POLICY_MIGRATION_APPLY_GATE_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_migration_apply_gate_preview"
)
AUTO_POLICY_DB_APPLY_REQUEST_GATE_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_request_gate_preview"
)
AUTO_POLICY_DB_APPLY_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_package"
)
AUTO_POLICY_DB_APPLY_VERIFIER_ARTIFACT_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_verifier_artifact_preview"
)
AUTO_POLICY_DB_APPLY_FINAL_HANDOFF_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_final_handoff_package"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_PREVIEW_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_shell_preview"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_SHELL_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_shell_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_INTAKE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_request_intake"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_request_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_LANE_GUARD_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_decision_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DECISION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_decision_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_ISSUER_GATE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_issuer_gate"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_decision_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_DECISION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_decision_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_GUARD_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_issuer_guard"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_ISSUER_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_issuer_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNING_EXECUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signing_execution_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_SIGNED_RECEIPT_EVIDENCE_INTAKE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_signed_receipt_evidence_intake"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_DETACHED_VERIFICATION_EVIDENCE_VALIDATION_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_detached_verification_evidence_validation"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_VERIFIER_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_verifier_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_evidence_execution_preflight"
)
AUTO_POLICY_DB_APPLY_AUTHORIZATION_EVIDENCE_EXECUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_authorization_evidence_execution_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_APPLY_FINAL_PREFLIGHT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_apply_final_preflight"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_package"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_READINESS_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_readiness"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PLAN_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_plan_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_COMMAND_ARTIFACT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_command_artifact_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_EXECUTION_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_execution_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_POST_RECEIPT_PARSER_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_post_receipt_parser_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_APPLY_ENFORCEMENT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_apply_enforcement_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_EXECUTOR_GUARD_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_final_executor_guard_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PRE_APPLY_REPLAY_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_pre_apply_replay_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_APPLY_EXECUTOR_READINESS_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_apply_executor_readiness_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_INVOCATION_RECEIPT_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_invocation_receipt_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_INVOCATION_PACKAGE_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_write_invocation_package_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_PREFLIGHT_GUARD_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_preflight_guard_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RUNNER_INVOCATION_BOUNDARY_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_runner_invocation_boundary_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_EXECUTION_RECEIPT_HANDOFF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_execution_receipt_handoff_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FINAL_NO_RUNNER_EXECUTION_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_final_no_runner_execution_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_CONTROLLED_EXECUTOR_QUARANTINE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_controlled_executor_quarantine_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_EXECUTION_ENVELOPE_FREEZE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_execution_envelope_freeze_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_FROZEN_ENVELOPE_VERIFIER_HANDOFF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_frozen_envelope_verifier_handoff_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_INVOCATION_LOCK_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_invocation_lock_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_NO_EXECUTION_RECEIPT_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_no_execution_receipt_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_VERIFIER_RECEIPT_PERSISTENCE_GUARD_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_verifier_receipt_persistence_guard_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RECEIPT_PERSISTENCE_STORAGE_BOUNDARY_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_receipt_persistence_storage_boundary_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_STORAGE_BOUNDARY_NO_WRITE_LEDGER_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_storage_boundary_no_write_ledger_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_NO_WRITE_LEDGER_RETENTION_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_no_write_ledger_retention_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_RETENTION_BOUNDARY_NO_WRITE_ARCHIVE_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_retention_boundary_no_write_archive_proof_closeout"
)
AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_ARCHIVE_RETENTION_SEALED_HANDOFF_PROOF_CLOSEOUT_POLICY = (
"read_only_pchome_growth_auto_policy_db_apply_controlled_dry_run_archive_retention_sealed_handoff_proof_closeout"
)
PCHOME_FETCH_ALLOWED_DOMAIN = "24h.pchome.com.tw"
PCHOME_FETCH_MAX_BATCH_SIZE = 12
PCHOME_FETCH_DEFAULT_TIMEOUT_SECONDS = 5
PCHOME_FETCH_MAX_HTML_BYTES = 512_000
EXTERNAL_BENCHMARK_REFERENCES = [
{
"source": "Google Merchant Center product data specification",
"url": "https://support.google.com/merchants/answer/7052112",
"applies_to": "accurate_product_feed_matching",
},
{
"source": "Google Search Product structured data",
"url": "https://developers.google.com/search/docs/appearance/structured-data/product",
"applies_to": "rich_product_information_and_offer_visibility",
},
{
"source": "Google Merchant listing structured data",
"url": "https://developers.google.com/search/docs/appearance/structured-data/merchant-listing",
"applies_to": "product_offer_price_currency_availability",
},
{
"source": "Baymard ecommerce product and search UX benchmark",
"url": "https://baymard.com/research/product-page",
"applies_to": "operator_search_and_product_detail_quality",
},
]
def _ai_exception_compatibility_fields(ai_exception_required: bool) -> dict[str, Any]:
"""Return the AI-first exception contract with legacy readback aliases."""
return {
LEGACY_HUMAN_REVIEW_REQUIRED_KEY: False,
LEGACY_HUMAN_REVIEW_REQUIRED_LEGACY_KEY: bool(ai_exception_required),
AI_EXCEPTION_REQUIRED_KEY: bool(ai_exception_required),
PRIMARY_HUMAN_GATE_COUNT_KEY: 0,
AI_EXCEPTION_MODE_KEY: AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
}
def _legacy_review_compatibility_fields(ai_exception_required: bool = False) -> dict[str, Any]:
"""Keep old review-mode keys false while exposing the AI exception state."""
return {
LEGACY_REVIEW_REQUIRED_KEY: False,
LEGACY_REVIEW_MODE_KEY: LEGACY_REVIEW_MODE_EXCEPTION_ONLY,
AI_EXCEPTION_REQUIRED_KEY: bool(ai_exception_required),
AI_EXCEPTION_MODE_KEY: AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
}
def _evidence_requires_ai_exception(evidence: dict[str, Any]) -> bool:
return bool(
evidence.get(AI_EXCEPTION_REQUIRED_KEY)
or evidence.get(LEGACY_HUMAN_REVIEW_REQUIRED_LEGACY_KEY)
or evidence.get(LEGACY_HUMAN_REVIEW_REQUIRED_KEY)
)
def _summary_exception_count(summary: dict[str, Any]) -> int:
return int(
summary.get(AI_EXCEPTION_REQUIRED_COUNT_KEY)
or summary.get(LEGACY_REVIEW_REQUIRED_COUNT_KEY)
or 0
)
_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,
}
def _evidence_completeness(item: dict[str, Any], review_candidate: dict[str, Any], external_price: dict[str, Any]) -> dict[str, Any]:
product_id = str(item.get("pchome_product_id") or "").strip()

4
tests/test_codebase_modularization_performance_service.py
View File

@@ -28,8 +28,8 @@ def test_inventory_generates_complete_prioritized_work_items():
assert all(item["work_item_id"] for item in result["python_work_items"])
by_path = {item["path"]: item for item in result["python_work_items"]}
assert by_path["routes/openclaw_bot_routes.py"]["status"] == "in_progress"
assert by_path["services/pchome_mapping_backlog_service.py"]["status"] == "not_started"
assert summary["work_item_status_counts"]["in_progress"] == 3
assert by_path["services/pchome_mapping_backlog_service.py"]["status"] == "in_progress"
assert summary["work_item_status_counts"]["in_progress"] == 4
assert sum(summary["work_item_status_counts"].values()) == (
summary["large_python_module_count"] + summary["large_frontend_asset_count"]
)

45
tests/test_pchome_mapping_backlog_modularization.py Normal file
View File

@@ -0,0 +1,45 @@
from pathlib import Path
from services import pchome_mapping_backlog_service as compatibility_facade
from services.pchome_mapping_backlog import contracts, evidence, policies
ROOT = Path(__file__).resolve().parents[1]
def test_policy_registry_is_extracted_without_changing_legacy_exports():
assert len(policies.__all__) == 96
assert compatibility_facade.BACKLOG_POLICY == policies.BACKLOG_POLICY
assert (
compatibility_facade.AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY
== policies.AUTO_POLICY_DB_APPLY_CONTROLLED_DRY_RUN_PACKAGE_POLICY
)
assert compatibility_facade.PCHOME_FETCH_ALLOWED_DOMAIN == "24h.pchome.com.tw"
def test_evidence_and_contract_functions_keep_legacy_import_identity():
assert (
compatibility_facade.parse_pchome_product_page_evidence_html
is evidence.parse_pchome_product_page_evidence_html
)
assert compatibility_facade.parse_unit_package_basis is evidence.parse_unit_package_basis
assert (
compatibility_facade._ai_exception_compatibility_fields
is contracts._ai_exception_compatibility_fields
)
def test_extracted_unit_parser_preserves_ai_exception_and_no_write_contract():
parsed = compatibility_facade.parse_unit_package_basis("理膚寶水 B5 40ml x2 超值組")
assert parsed["estimated_total_quantity"] == 80
assert parsed["ai_exception_required"] is True
assert parsed["primary_human_gate_count"] == 0
assert parsed["writes_database"] is False
assert parsed["fetches_external_sites"] is False
def test_legacy_facade_is_smaller_after_first_p0_extraction():
facade = ROOT / "services/pchome_mapping_backlog_service.py"
assert sum(1 for _ in facade.open(encoding="utf-8")) < 44_300