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

refactor(pchome): extract evidence enrichment

Browse Source
This commit is contained in:
ogt
2026-07-11 15:42:00 +08:00
parent 99bb867e05
commit b120230cae
6 changed files with 445 additions and 315 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.784"
SYSTEM_VERSION = "V10.785"
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.784
> **適用版本**: V10.785
---

2
docs/memory/code_modularization_inventory_20260430.md
View File

@@ -129,7 +129,7 @@
| 行數 | 檔案 | 分類 | 拆分方向 |
|---:|---|---|---|
| 43183 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier、receipt replay 與 reporter 已移到 `services/pchome_mapping_backlog/`；下一步拆 evidence enrichment orchestration |
| 42881 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier、receipt replay、reporter 與 evidence enrichment/source preview 已移到 `services/pchome_mapping_backlog/`；下一步拆 fetch/merge orchestration |
| 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 分離 |

388
services/pchome_mapping_backlog/enrichment.py Normal file
View File

@@ -0,0 +1,388 @@
"""Read-only evidence enrichment and source-wiring previews."""
from __future__ import annotations
from typing import Any
from services.ai_exception_contract import LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY
from services.pchome_mapping_backlog.contracts import (
_ai_exception_compatibility_fields,
_evidence_requires_ai_exception,
)
from services.pchome_mapping_backlog.policies import (
EVIDENCE_ENRICHMENT_PREVIEW_POLICY,
EVIDENCE_SOURCE_PREVIEW_POLICY,
PCHOME_FETCH_MAX_BATCH_SIZE,
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
)
from services.pchome_mapping_backlog.reporter import (
build_pchome_mapping_operator_preview,
)
def _field_enrichment_sources(field: str) -> list[dict[str, Any]]:
source_map = {
"image": [
{
"source": "PChome product page structured data",
"mode": "future_read_only_fetch",
"writes_database": False,
"expected_output": "primary product image URL",
},
{
"source": "existing marketplace catalog payload",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "cached image_url",
},
],
"availability": [
{
"source": "PChome product page offer availability",
"mode": "future_read_only_fetch",
"writes_database": False,
"expected_output": "in_stock / out_of_stock / unknown",
},
{
"source": "merchant listing structured data",
"mode": "future_read_only_parse",
"writes_database": False,
"expected_output": "schema.org Offer availability",
},
],
"unit_price_or_package_basis": [
{
"source": "deterministic product title parser",
"mode": "local_parse_preview",
"writes_database": False,
"expected_output": "size, count, unit label, package basis",
},
{
"source": "external_price.price_basis",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "total_price / unit_price",
},
],
"price": [
{
"source": "growth opportunity payload",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "PChome listed price",
},
],
}
return source_map.get(field, [])
def _source_plan_for_field(field: str, missing_count: int) -> dict[str, Any]:
plans = {
"image": {
"payload_keys_checked": ["image_url", "image", "product_image_url"],
"preferred_current_source": "existing marketplace catalog payload",
"future_read_only_fetch_gate": {
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"product_url_required": True,
"parse_targets": [
"schema.org Product.image",
"og:image",
"primary product image",
],
"check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
"availability": {
"payload_keys_checked": [
"availability",
"stock_status",
"is_available",
],
"preferred_current_source": "existing marketplace catalog payload",
"future_read_only_fetch_gate": {
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"product_url_required": True,
"parse_targets": [
"schema.org Offer.availability",
"merchant listing offer availability",
],
"check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
"price": {
"payload_keys_checked": [
"pchome_price",
"external_price.pchome_price",
"review_candidate.pchome_price",
],
"preferred_current_source": "growth opportunity payload",
"future_read_only_fetch_gate": None,
"payload_mapping_probe": {
"goal": "Confirm whether missing price is a source payload gap or summary field mapping gap.",
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
}
plan = dict(plans.get(field, {}))
if not plan:
return {}
plan["field"] = field
plan["status"] = (
"missing_in_current_payload"
if missing_count
else "covered_by_current_payload"
)
plan["missing_count"] = missing_count
return plan
def _build_fetch_gate_candidates(
tasks: list[dict[str, Any]],
) -> list[dict[str, Any]]:
candidates = []
for task in tasks:
missing_fields = set(task.get("missing_fields") or [])
fetch_fields = [
field
for field in ("image", "availability")
if field in missing_fields
]
if not fetch_fields:
continue
candidates.append(
{
"pchome_product_id": task.get("pchome_product_id") or "",
"product_name": task.get("product_name") or "",
"product_url": task.get("product_url"),
"fields": fetch_fields,
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"executes_fetch_in_preview": False,
"writes_database": False,
}
)
return candidates[:PCHOME_FETCH_MAX_BATCH_SIZE]
def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]:
evidence = target.get("evidence_completeness") or {}
missing_fields = list(evidence.get("missing_fields") or [])
blocking_missing_fields = list(evidence.get("blocking_missing_fields") or [])
enrichment_steps = [
{
"field": field,
"blocking": field in blocking_missing_fields,
"sources": _field_enrichment_sources(field),
}
for field in missing_fields
]
ai_exception_required = _evidence_requires_ai_exception(evidence)
return {
"lane": lane,
"pchome_product_id": target.get("pchome_product_id") or "",
"product_name": target.get("product_name") or "",
"product_url": target.get("product_url") or evidence.get("product_url"),
"sales_7d": target.get("sales_7d"),
"priority_score": target.get("priority_score"),
"action_code": target.get("action_code"),
"evidence_score": evidence.get("score"),
"present_fields": list(evidence.get("present_fields") or []),
"missing_fields": missing_fields,
"blocking_missing_fields": blocking_missing_fields,
"auto_accept_ready": bool(evidence.get("auto_accept_ready")),
**_ai_exception_compatibility_fields(ai_exception_required),
"unit_package_basis": evidence.get("unit_package_basis"),
"enrichment_steps": enrichment_steps,
}
def build_pchome_evidence_enrichment_preview(
payload: dict[str, Any],
batch_size: int = 5,
) -> dict[str, Any]:
"""Build a read-only evidence enrichment package for mapping targets."""
operator_preview = build_pchome_mapping_operator_preview(
payload,
batch_size=batch_size,
)
operator_batch = operator_preview.get("operator_batch") or {}
direct_targets = list(operator_batch.get("direct_mapping_targets") or [])
review_targets = list(operator_batch.get("review_candidate_targets") or [])
evidence_tasks = [
*[
_build_evidence_task(target, "direct_mapping")
for target in direct_targets
],
*[
_build_evidence_task(target, "review_candidate")
for target in review_targets
],
]
tasks_with_blockers = [
task for task in evidence_tasks if task.get("blocking_missing_fields")
]
missing_field_counts: dict[str, int] = {}
for task in evidence_tasks:
for field in task.get("missing_fields") or []:
missing_field_counts[field] = missing_field_counts.get(field, 0) + 1
if tasks_with_blockers:
result = "NEEDS_EVIDENCE_ENRICHMENT"
elif evidence_tasks:
result = "EVIDENCE_PREVIEW_READY"
else:
result = "NO_TARGETS"
return {
"policy": EVIDENCE_ENRICHMENT_PREVIEW_POLICY,
"result": result,
"success": bool(operator_preview.get("success")),
"generated_at": operator_preview.get("generated_at"),
"source_policy": operator_preview.get("policy"),
"stats": operator_preview.get("stats") or {},
"summary": {
"task_count": len(evidence_tasks),
"tasks_with_blockers": len(tasks_with_blockers),
"missing_field_counts": missing_field_counts,
"auto_accept_ready_count": sum(
1 for task in evidence_tasks if task.get("auto_accept_ready")
),
LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY: 0,
"primary_human_gate_count": 0,
"ai_exception_required_count": sum(
1
for task in evidence_tasks
if task.get("ai_exception_required")
),
},
"evidence_tasks": evidence_tasks,
"external_benchmark_alignment": (
operator_preview.get("external_benchmark_alignment") or {}
),
"ai_automation_plan": {
"policy": "ollama_first_read_only_evidence_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"can_execute_write": False,
"recommended_next_ai_task": "Generate deterministic identity anchors and evidence gap summaries after image and availability sources are wired.",
"blocked_until": [
"image evidence source is wired",
"availability evidence source is wired",
"unit/package parser preview is compared against production title samples",
],
},
"safety": {
"read_only_preview": True,
"fetches_external_sites": False,
"writes_database": False,
"executes_search": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"requires_operator_write_approval": True,
},
"next_actions": [
"Wire read-only image and availability enrichment before expanding auto-accept.",
"Validate deterministic unit/package basis parsing for bundle-sensitive items.",
"Keep DB writes behind the existing /api/ai/pchome-growth/backfill-momo-candidates write gate.",
],
}
def build_pchome_evidence_source_preview(
payload: dict[str, Any],
batch_size: int = 5,
) -> dict[str, Any]:
"""Build a read-only source wiring preview for missing product evidence fields."""
enrichment_preview = build_pchome_evidence_enrichment_preview(
payload,
batch_size=batch_size,
)
tasks = list(enrichment_preview.get("evidence_tasks") or [])
fields = ["image", "availability", "price", "unit_price_or_package_basis"]
field_counts = {}
source_plans = {}
for field in fields:
missing_tasks = [
task
for task in tasks
if field in (task.get("missing_fields") or [])
]
field_counts[field] = {
"missing_count": len(missing_tasks),
"present_count": max(len(tasks) - len(missing_tasks), 0),
"sample_missing_targets": [
{
"pchome_product_id": task.get("pchome_product_id") or "",
"product_name": task.get("product_name") or "",
"product_url": task.get("product_url"),
"lane": task.get("lane"),
}
for task in missing_tasks[:3]
],
}
plan = _source_plan_for_field(field, len(missing_tasks))
if plan:
source_plans[field] = plan
fetch_gate_candidates = _build_fetch_gate_candidates(tasks)
if (
field_counts["image"]["missing_count"]
or field_counts["availability"]["missing_count"]
):
result = "NEEDS_SOURCE_WIRING"
elif field_counts["price"]["missing_count"]:
result = "NEEDS_PAYLOAD_MAPPING_PROBE"
elif tasks:
result = "SOURCE_PREVIEW_READY"
else:
result = "NO_TARGETS"
return {
"policy": EVIDENCE_SOURCE_PREVIEW_POLICY,
"result": result,
"success": bool(enrichment_preview.get("success")),
"generated_at": enrichment_preview.get("generated_at"),
"source_policy": enrichment_preview.get("policy"),
"stats": enrichment_preview.get("stats") or {},
"summary": {
"task_count": len(tasks),
"field_counts": field_counts,
"fetch_gate_candidate_count": len(fetch_gate_candidates),
},
"source_plans": source_plans,
"fetch_gate_candidates": fetch_gate_candidates,
"external_benchmark_alignment": (
enrichment_preview.get("external_benchmark_alignment") or {}
),
"ai_automation_plan": {
"policy": "ollama_first_read_only_source_wiring_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"can_execute_fetch": False,
"can_execute_write": False,
"recommended_next_ai_task": "Generate schema-aware parsers for image and availability after fetch-gate tests are accepted.",
},
"safety": {
"read_only_preview": True,
"fetches_external_sites": False,
"writes_database": False,
"executes_search": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"requires_production_version_truth": True,
"requires_fetch_gate_implementation_before_external_get": True,
},
"next_actions": [
"Wire image source preview from existing payload keys before adding a controlled product-page fetch gate.",
"Wire availability source preview from existing payload keys before adding a controlled Offer availability parser.",
"Probe missing price rows through current payload mapping before any external fetch or write.",
],
}

318
services/pchome_mapping_backlog_service.py
View File

@@ -162,6 +162,14 @@ from services.pchome_mapping_backlog.evidence import (
parse_pchome_product_page_evidence_html,
parse_unit_package_basis,
)
from services.pchome_mapping_backlog.enrichment import (
_build_evidence_task,
_build_fetch_gate_candidates,
_field_enrichment_sources,
_source_plan_for_field,
build_pchome_evidence_enrichment_preview,
build_pchome_evidence_source_preview,
)
from services.pchome_mapping_backlog.reporter import (
_build_ai_automation_plan,
_build_external_benchmark_alignment,
@@ -172,165 +180,6 @@ from services.pchome_mapping_backlog.reporter import (
)
def _field_enrichment_sources(field: str) -> list[dict[str, Any]]:
source_map = {
"image": [
{
"source": "PChome product page structured data",
"mode": "future_read_only_fetch",
"writes_database": False,
"expected_output": "primary product image URL",
},
{
"source": "existing marketplace catalog payload",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "cached image_url",
},
],
"availability": [
{
"source": "PChome product page offer availability",
"mode": "future_read_only_fetch",
"writes_database": False,
"expected_output": "in_stock / out_of_stock / unknown",
},
{
"source": "merchant listing structured data",
"mode": "future_read_only_parse",
"writes_database": False,
"expected_output": "schema.org Offer availability",
},
],
"unit_price_or_package_basis": [
{
"source": "deterministic product title parser",
"mode": "local_parse_preview",
"writes_database": False,
"expected_output": "size, count, unit label, package basis",
},
{
"source": "external_price.price_basis",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "total_price / unit_price",
},
],
"price": [
{
"source": "growth opportunity payload",
"mode": "reuse_if_present",
"writes_database": False,
"expected_output": "PChome listed price",
},
],
}
return source_map.get(field, [])
def _source_plan_for_field(field: str, missing_count: int) -> dict[str, Any]:
plans = {
"image": {
"payload_keys_checked": ["image_url", "image", "product_image_url"],
"preferred_current_source": "existing marketplace catalog payload",
"future_read_only_fetch_gate": {
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"product_url_required": True,
"parse_targets": ["schema.org Product.image", "og:image", "primary product image"],
"check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
"availability": {
"payload_keys_checked": ["availability", "stock_status", "is_available"],
"preferred_current_source": "existing marketplace catalog payload",
"future_read_only_fetch_gate": {
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"product_url_required": True,
"parse_targets": ["schema.org Offer.availability", "merchant listing offer availability"],
"check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY,
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
"price": {
"payload_keys_checked": ["pchome_price", "external_price.pchome_price", "review_candidate.pchome_price"],
"preferred_current_source": "growth opportunity payload",
"future_read_only_fetch_gate": None,
"payload_mapping_probe": {
"goal": "Confirm whether missing price is a source payload gap or summary field mapping gap.",
"fetches_external_sites_in_preview": False,
"writes_database": False,
},
},
}
plan = dict(plans.get(field, {}))
if not plan:
return {}
plan["field"] = field
plan["status"] = "missing_in_current_payload" if missing_count else "covered_by_current_payload"
plan["missing_count"] = missing_count
return plan
def _build_fetch_gate_candidates(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]:
candidates = []
for task in tasks:
missing_fields = set(task.get("missing_fields") or [])
fetch_fields = [field for field in ("image", "availability") if field in missing_fields]
if not fetch_fields:
continue
candidates.append(
{
"pchome_product_id": task.get("pchome_product_id") or "",
"product_name": task.get("product_name") or "",
"product_url": task.get("product_url"),
"fields": fetch_fields,
"method": "GET",
"allowed_domain": "24h.pchome.com.tw",
"executes_fetch_in_preview": False,
"writes_database": False,
}
)
return candidates[:PCHOME_FETCH_MAX_BATCH_SIZE]
def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]:
evidence = target.get("evidence_completeness") or {}
missing_fields = list(evidence.get("missing_fields") or [])
blocking_missing_fields = list(evidence.get("blocking_missing_fields") or [])
enrichment_steps = [
{
"field": field,
"blocking": field in blocking_missing_fields,
"sources": _field_enrichment_sources(field),
}
for field in missing_fields
]
ai_exception_required = _evidence_requires_ai_exception(evidence)
return {
"lane": lane,
"pchome_product_id": target.get("pchome_product_id") or "",
"product_name": target.get("product_name") or "",
"product_url": target.get("product_url") or evidence.get("product_url"),
"sales_7d": target.get("sales_7d"),
"priority_score": target.get("priority_score"),
"action_code": target.get("action_code"),
"evidence_score": evidence.get("score"),
"present_fields": list(evidence.get("present_fields") or []),
"missing_fields": missing_fields,
"blocking_missing_fields": blocking_missing_fields,
"auto_accept_ready": bool(evidence.get("auto_accept_ready")),
**_ai_exception_compatibility_fields(ai_exception_required),
"unit_package_basis": evidence.get("unit_package_basis"),
"enrichment_steps": enrichment_steps,
}
def _build_direct_mapping_search_terms(product_name: str, max_terms: int) -> list[str]:
try:
from services.momo_crawler import build_targeted_momo_search_terms
@@ -4486,157 +4335,6 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollb
}
def build_pchome_evidence_enrichment_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
"""Build a read-only evidence enrichment package for mapping targets."""
operator_preview = build_pchome_mapping_operator_preview(payload, batch_size=batch_size)
operator_batch = operator_preview.get("operator_batch") or {}
direct_targets = list(operator_batch.get("direct_mapping_targets") or [])
review_targets = list(operator_batch.get("review_candidate_targets") or [])
evidence_tasks = [
*[_build_evidence_task(target, "direct_mapping") for target in direct_targets],
*[_build_evidence_task(target, "review_candidate") for target in review_targets],
]
tasks_with_blockers = [
task
for task in evidence_tasks
if task.get("blocking_missing_fields")
]
missing_field_counts: dict[str, int] = {}
for task in evidence_tasks:
for field in task.get("missing_fields") or []:
missing_field_counts[field] = missing_field_counts.get(field, 0) + 1
if tasks_with_blockers:
result = "NEEDS_EVIDENCE_ENRICHMENT"
elif evidence_tasks:
result = "EVIDENCE_PREVIEW_READY"
else:
result = "NO_TARGETS"
return {
"policy": EVIDENCE_ENRICHMENT_PREVIEW_POLICY,
"result": result,
"success": bool(operator_preview.get("success")),
"generated_at": operator_preview.get("generated_at"),
"source_policy": operator_preview.get("policy"),
"stats": operator_preview.get("stats") or {},
"summary": {
"task_count": len(evidence_tasks),
"tasks_with_blockers": len(tasks_with_blockers),
"missing_field_counts": missing_field_counts,
"auto_accept_ready_count": sum(1 for task in evidence_tasks if task.get("auto_accept_ready")),
LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY: 0,
"primary_human_gate_count": 0,
"ai_exception_required_count": sum(1 for task in evidence_tasks if task.get("ai_exception_required")),
},
"evidence_tasks": evidence_tasks,
"external_benchmark_alignment": operator_preview.get("external_benchmark_alignment") or {},
"ai_automation_plan": {
"policy": "ollama_first_read_only_evidence_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"can_execute_write": False,
"recommended_next_ai_task": "Generate deterministic identity anchors and evidence gap summaries after image and availability sources are wired.",
"blocked_until": [
"image evidence source is wired",
"availability evidence source is wired",
"unit/package parser preview is compared against production title samples",
],
},
"safety": {
"read_only_preview": True,
"fetches_external_sites": False,
"writes_database": False,
"executes_search": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"requires_operator_write_approval": True,
},
"next_actions": [
"Wire read-only image and availability enrichment before expanding auto-accept.",
"Validate deterministic unit/package basis parsing for bundle-sensitive items.",
"Keep DB writes behind the existing /api/ai/pchome-growth/backfill-momo-candidates write gate.",
],
}
def build_pchome_evidence_source_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
"""Build a read-only source wiring preview for missing product evidence fields."""
enrichment_preview = build_pchome_evidence_enrichment_preview(payload, batch_size=batch_size)
tasks = list(enrichment_preview.get("evidence_tasks") or [])
fields = ["image", "availability", "price", "unit_price_or_package_basis"]
field_counts = {}
source_plans = {}
for field in fields:
missing_tasks = [task for task in tasks if field in (task.get("missing_fields") or [])]
field_counts[field] = {
"missing_count": len(missing_tasks),
"present_count": max(len(tasks) - len(missing_tasks), 0),
"sample_missing_targets": [
{
"pchome_product_id": task.get("pchome_product_id") or "",
"product_name": task.get("product_name") or "",
"product_url": task.get("product_url"),
"lane": task.get("lane"),
}
for task in missing_tasks[:3]
],
}
plan = _source_plan_for_field(field, len(missing_tasks))
if plan:
source_plans[field] = plan
fetch_gate_candidates = _build_fetch_gate_candidates(tasks)
if field_counts["image"]["missing_count"] or field_counts["availability"]["missing_count"]:
result = "NEEDS_SOURCE_WIRING"
elif field_counts["price"]["missing_count"]:
result = "NEEDS_PAYLOAD_MAPPING_PROBE"
elif tasks:
result = "SOURCE_PREVIEW_READY"
else:
result = "NO_TARGETS"
return {
"policy": EVIDENCE_SOURCE_PREVIEW_POLICY,
"result": result,
"success": bool(enrichment_preview.get("success")),
"generated_at": enrichment_preview.get("generated_at"),
"source_policy": enrichment_preview.get("policy"),
"stats": enrichment_preview.get("stats") or {},
"summary": {
"task_count": len(tasks),
"field_counts": field_counts,
"fetch_gate_candidate_count": len(fetch_gate_candidates),
},
"source_plans": source_plans,
"fetch_gate_candidates": fetch_gate_candidates,
"external_benchmark_alignment": enrichment_preview.get("external_benchmark_alignment") or {},
"ai_automation_plan": {
"policy": "ollama_first_read_only_source_wiring_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"can_execute_fetch": False,
"can_execute_write": False,
"recommended_next_ai_task": "Generate schema-aware parsers for image and availability after fetch-gate tests are accepted.",
},
"safety": {
"read_only_preview": True,
"fetches_external_sites": False,
"writes_database": False,
"executes_search": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"requires_production_version_truth": True,
"requires_fetch_gate_implementation_before_external_get": True,
},
"next_actions": [
"Wire image source preview from existing payload keys before adding a controlled product-page fetch gate.",
"Wire availability source preview from existing payload keys before adding a controlled Offer availability parser.",
"Probe missing price rows through current payload mapping before any external fetch or write.",
],
}
def build_pchome_evidence_fetch_gate(
payload: dict[str, Any],
batch_size: int = 3,

48
tests/test_pchome_mapping_backlog_modularization.py
View File

@@ -7,6 +7,7 @@ from services.pchome_mapping_backlog import (
controlled_apply_executor,
controlled_apply_receipt,
controlled_apply_verifier,
enrichment,
evidence,
policies,
reporter,
@@ -201,7 +202,50 @@ def test_reporter_keeps_legacy_identity_and_read_only_preview():
assert preview["safety"]["writes_database"] is False
def test_legacy_facade_is_smaller_after_fifth_p0_extraction():
def test_enrichment_keeps_legacy_identity_and_no_write_preview():
assert (
compatibility_facade.build_pchome_evidence_enrichment_preview
is enrichment.build_pchome_evidence_enrichment_preview
)
assert (
compatibility_facade.build_pchome_evidence_source_preview
is enrichment.build_pchome_evidence_source_preview
)
assert (
compatibility_facade._build_evidence_task
is enrichment._build_evidence_task
)
assert (
compatibility_facade._build_fetch_gate_candidates
is enrichment._build_fetch_gate_candidates
)
assert (
compatibility_facade._field_enrichment_sources
is enrichment._field_enrichment_sources
)
assert (
compatibility_facade._source_plan_for_field
is enrichment._source_plan_for_field
)
payload = {
"success": True,
"generated_at": "2026-07-11T15:30:00+08:00",
"stats": {},
"opportunities": [],
}
preview = enrichment.build_pchome_evidence_enrichment_preview(payload)
source_preview = enrichment.build_pchome_evidence_source_preview(payload)
assert preview["result"] == "NO_TARGETS"
assert source_preview["result"] == "NO_TARGETS"
assert preview["safety"]["fetches_external_sites"] is False
assert preview["safety"]["writes_database"] is False
assert source_preview["safety"]["fetches_external_sites"] is False
assert source_preview["safety"]["writes_database"] is False
def test_legacy_facade_is_smaller_after_sixth_p0_extraction():
facade = ROOT / "services/pchome_mapping_backlog_service.py"
assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_200
assert sum(1 for _ in facade.open(encoding="utf-8")) < 42_900