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

refactor(pchome): extract backlog reporter
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 15:09:57 +08:00
parent 2e8ed32dad
commit c5f74f69a4
6 changed files with 504 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files

451
services/pchome_mapping_backlog/reporter.py Normal file
View File

@@ -0,0 +1,451 @@
"""Read-only PChome backlog reporting and operator preview projections."""
from __future__ import annotations
from typing import Any
from services.pchome_mapping_backlog.contracts import _ai_exception_compatibility_fields
from services.pchome_mapping_backlog.evidence import (
_action_code,
_action_label,
_first_present,
_pchome_product_url,
_to_float,
parse_unit_package_basis,
)
from services.pchome_mapping_backlog.policies import (
BACKLOG_POLICY,
EXTERNAL_BENCHMARK_REFERENCES,
OPERATOR_PREVIEW_POLICY,
)
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()
product_name = str(item.get("product_name") or "").strip()
product_url = _first_present(
item.get("product_url"),
item.get("pchome_url"),
_pchome_product_url(product_id),
)
pchome_price = _first_present(
item.get("pchome_price"),
external_price.get("pchome_price"),
review_candidate.get("pchome_price"),
)
image_url = _first_present(
item.get("image_url"),
item.get("image"),
item.get("product_image_url"),
)
availability = _first_present(
item.get("availability"),
item.get("stock_status"),
item.get("is_available"),
)
unit_package_basis = parse_unit_package_basis(product_name)
parsed_unit_basis = (
unit_package_basis
if unit_package_basis.get("package_basis") != "insufficient"
else None
)
unit_basis = _first_present(
external_price.get("price_basis"),
item.get("price_basis"),
item.get("unit_label"),
parsed_unit_basis,
)
unit_review_required = bool(unit_package_basis.get("risk_signals"))
checks = [
("stable_product_id", bool(product_id), "required"),
("product_name", bool(product_name), "required"),
("product_url", bool(product_url), "required"),
("price", pchome_price is not None, "required"),
("image", bool(image_url), "strongly_recommended"),
("availability", availability is not None, "strongly_recommended"),
(
"unit_price_or_package_basis",
bool(unit_basis),
"required_when_bundle_or_unit_sensitive",
),
]
present = [field for field, ok, _requirement in checks if ok]
missing = [field for field, ok, _requirement in checks if not ok]
blocking_missing = [
field
for field, ok, requirement in checks
if not ok and requirement in {"required", "strongly_recommended"}
]
score = round(len(present) / max(len(checks), 1) * 100, 1)
ai_exception_required = (
bool(blocking_missing)
or bool(review_candidate)
or not external_price
or unit_review_required
)
return {
"score": score,
"present_fields": present,
"missing_fields": missing,
"blocking_missing_fields": blocking_missing,
"auto_accept_ready": (
not blocking_missing and bool(external_price) and not unit_review_required
),
**_ai_exception_compatibility_fields(ai_exception_required),
"product_url": product_url,
"image_url": image_url,
"availability": availability,
"unit_package_basis": unit_package_basis,
}
def compact_mapping_item(item: dict[str, Any]) -> dict[str, Any]:
review_candidate = item.get("review_candidate") or {}
external_price = item.get("external_price") or {}
product_id = str(item.get("pchome_product_id") or "")
product_url = _first_present(
item.get("product_url"),
item.get("pchome_url"),
_pchome_product_url(product_id),
)
return {
"pchome_product_id": product_id,
"product_url": product_url,
"product_name": item.get("product_name") or "",
"sales_7d": round(_to_float(item.get("sales_7d")), 2),
"sales_delta_pct": item.get("sales_delta_pct"),
"priority_score": item.get("priority_score"),
"pchome_price": item.get("pchome_price"),
"action_code": _action_code(item),
"action_label": _action_label(item),
"review_candidate": {
"id": review_candidate.get("id"),
"momo_sku": review_candidate.get("momo_sku"),
"momo_name": review_candidate.get("momo_name"),
"quality_score": review_candidate.get("quality_score"),
"gap_pct": review_candidate.get("gap_pct"),
}
if review_candidate
else None,
"external_price": {
"momo_sku": external_price.get("momo_sku"),
"momo_name": external_price.get("momo_name"),
"price_basis": external_price.get("price_basis"),
"gap_pct": external_price.get("gap_pct"),
"data_source_label": external_price.get("data_source_label"),
"updated_at": external_price.get("updated_at"),
}
if external_price
else None,
"evidence_completeness": _evidence_completeness(
item,
review_candidate,
external_price,
),
"reason_lines": list(item.get("reason_lines") or [])[:3],
}
def _build_external_benchmark_alignment() -> dict[str, Any]:
return {
"references": EXTERNAL_BENCHMARK_REFERENCES,
"required_evidence_fields": [
{
"field": "stable_product_id",
"current_payload": "pchome_product_id",
"status": "present",
"why": "Stable IDs preserve mapping history and make post-run readback comparable.",
},
{
"field": "product_name",
"current_payload": "product_name",
"status": "present",
"why": "Exact title/name matching is the first identity anchor for operator review.",
},
{
"field": "product_url",
"current_payload": "derived_from_pchome_product_id",
"status": "present_for_pchome",
"why": "Operators need a direct product page path for visual confirmation.",
},
{
"field": "price",
"current_payload": "pchome_price/external_price",
"status": "partial",
"why": "Offer price and currency are required before a candidate can become decision-ready.",
},
{
"field": "image",
"current_payload": None,
"status": "missing_in_current_growth_payload",
"why": "Image evidence should be added before high-volume auto-accept expansion.",
},
{
"field": "availability",
"current_payload": None,
"status": "missing_in_current_growth_payload",
"why": "Availability prevents matching stale or non-purchasable offers.",
},
{
"field": "unit_price_or_package_basis",
"current_payload": (
"external_price.price_basis or deterministic title parser preview"
),
"status": "parser_preview_available",
"why": "Unit price and package basis protect bundles, variants, and volume-size comparisons.",
},
],
"operator_review_principles": [
"Separate direct mapping, review candidate, and already comparable items.",
"Do not auto-accept variants, colors, bundles, or catalog offers without explicit evidence.",
"Keep search/query support exact-title friendly so copied product names and model terms remain useful.",
],
}
def _build_ai_automation_plan(
selected_direct: list[dict[str, Any]],
selected_review: list[dict[str, Any]],
) -> dict[str, Any]:
return {
"policy": "ollama_first_read_only_ai_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"provider_order": [
"GCP-A 34.87.90.216:11434",
"GCP-B 34.21.145.224:11434",
"111 192.168.0.111:11434",
],
"automation_readiness": {
"direct_mapping_targets": len(selected_direct),
"review_candidate_targets": len(selected_review),
"can_generate_operator_summary": bool(selected_direct or selected_review),
"can_execute_write": False,
},
"steps": [
{
"name": "identity_anchor_extraction",
"mode": "deterministic_first_ollama_assist_later",
"writes_database": False,
"output": "brand/product_line/spec/package/variant anchors for each selected target",
},
{
"name": "candidate_search_plan",
"mode": "rule_based_query_pack",
"writes_database": False,
"output": "exact title, brand plus product line, and spec-preserving search terms",
},
{
"name": "operator_decision_summary",
"mode": "ollama_first_after_write_gate_only",
"writes_database": False,
"output": "plain-language review reason, evidence gaps, and post-write readback checklist",
},
{
"name": "post_write_readback",
"mode": "deterministic_metrics",
"writes_database": False,
"output": "mapping_rate, direct_mapping_count, review_candidate_count, mapped_count delta",
},
],
"ai_exception_required_for": [
"missing image or availability evidence",
"variant/color/fragrance/shade/package ambiguity",
"unit-price or bundle-sensitive comparisons",
"any candidate not meeting exact identity evidence",
],
}
def summarize_pchome_mapping_backlog(payload: dict[str, Any]) -> dict[str, Any]:
stats = payload.get("stats") or {}
opportunities = [
item
for item in payload.get("opportunities") or []
if isinstance(item, dict)
]
needs_mapping = [item for item in opportunities if not item.get("external_price")]
review_candidates = [
item for item in needs_mapping if item.get("review_candidate")
]
direct_mapping = [
item
for item in needs_mapping
if _action_code(item) == "map_external_product"
and not item.get("review_candidate")
]
mapped = [item for item in opportunities if item.get("external_price")]
action_counts: dict[str, int] = {}
sales_by_action: dict[str, float] = {}
for item in opportunities:
label = _action_label(item)
action_counts[label] = action_counts.get(label, 0) + 1
sales_by_action[label] = round(
sales_by_action.get(label, 0.0) + _to_float(item.get("sales_7d")),
2,
)
candidate_count = int(stats.get("candidate_count") or len(opportunities))
mapped_count = int(stats.get("mapped_count") or len(mapped))
needs_mapping_count = int(stats.get("needs_mapping_count") or len(needs_mapping))
mapping_rate = stats.get("mapping_rate")
if mapping_rate is None:
mapping_rate = round(mapped_count / max(candidate_count, 1) * 100, 1)
top_needs_mapping = sorted(
needs_mapping,
key=lambda item: (
_to_float(item.get("sales_7d")),
_to_float(item.get("priority_score")),
),
reverse=True,
)[:10]
top_review_candidates = sorted(
review_candidates,
key=lambda item: _to_float(
(item.get("review_candidate") or {}).get("quality_score")
),
reverse=True,
)[:10]
if not payload.get("success", False):
result = "BLOCKED"
elif needs_mapping_count > 0:
result = "NEEDS_MAPPING"
else:
result = "PASS"
return {
"policy": BACKLOG_POLICY,
"result": result,
"success": bool(payload.get("success")),
"generated_at": payload.get("generated_at"),
"cache_state": payload.get("cache_state"),
"system_name": payload.get("system_name"),
"message": payload.get("message"),
"stats": {
"candidate_count": candidate_count,
"mapped_count": mapped_count,
"mapping_rate": mapping_rate,
"needs_mapping_count": needs_mapping_count,
"review_candidate_count": int(
stats.get("review_candidate_count") or len(review_candidates)
),
"latest_sales_date": stats.get("latest_sales_date"),
"overall_latest_sales_date": stats.get("overall_latest_sales_date"),
"overall_sales_7d": stats.get("overall_sales_7d"),
"opportunity_sales_7d": stats.get("opportunity_sales_7d"),
"action_counts": dict(stats.get("action_counts") or action_counts),
"action_code_counts": dict(stats.get("action_code_counts") or {}),
"external_data_source_counts": dict(
stats.get("external_data_source_counts") or {}
),
},
"backlog": {
"direct_mapping_count": len(direct_mapping),
"review_candidate_count": len(review_candidates),
"mapped_opportunity_count": len(mapped),
"sales_by_action": sales_by_action,
"top_needs_mapping": [
compact_mapping_item(item) for item in top_needs_mapping
],
"top_review_candidates": [
compact_mapping_item(item) for item in top_review_candidates
],
},
"next_actions": [
"Run the production version truth guard before changing or deploying.",
"Handle direct mapping items first; they have no verified external price yet.",
"Review candidate items next; they already have MOMO candidates but need same-item confirmation.",
"Keep this report read-only until an explicit DB-write operator run is approved.",
],
}
def build_pchome_mapping_operator_preview(
payload: dict[str, Any],
batch_size: int = 5,
) -> dict[str, Any]:
"""Build a read-only operator run package for the direct mapping backlog."""
summary = summarize_pchome_mapping_backlog(payload)
backlog = summary.get("backlog") or {}
direct_items = [
item
for item in backlog.get("top_needs_mapping") or []
if item.get("action_code") == "map_external_product"
]
review_items = list(backlog.get("top_review_candidates") or [])
batch_size = max(1, min(int(batch_size or 5), 8))
selected_direct = direct_items[:batch_size]
selected_review = review_items[:batch_size]
if selected_direct:
result = "READY_FOR_OPERATOR_PREVIEW"
elif selected_review:
result = "REVIEW_CANDIDATES_ONLY"
else:
result = "NO_DIRECT_MAPPING_TARGETS"
return {
"policy": OPERATOR_PREVIEW_POLICY,
"result": result,
"success": bool(summary.get("success")),
"generated_at": summary.get("generated_at"),
"stats": summary.get("stats") or {},
"backlog": {
"direct_mapping_count": int(backlog.get("direct_mapping_count") or 0),
"review_candidate_count": int(
backlog.get("review_candidate_count") or 0
),
"mapped_opportunity_count": int(
backlog.get("mapped_opportunity_count") or 0
),
},
"operator_batch": {
"batch_size": batch_size,
"selected_direct_mapping_count": len(selected_direct),
"selected_review_candidate_count": len(selected_review),
"direct_mapping_targets": selected_direct,
"review_candidate_targets": selected_review,
},
"command_preview": {
"method": "POST",
"endpoint": "/api/ai/pchome-growth/backfill-momo-candidates",
"payload": {"limit": min(batch_size, 8)},
"executes_search": True,
"writes_database": True,
"write_gate_required": True,
},
"external_benchmark_alignment": _build_external_benchmark_alignment(),
"ai_automation_plan": _build_ai_automation_plan(
selected_direct,
selected_review,
),
"safety": {
"read_only_preview": True,
"executes_search": False,
"writes_database": False,
"dispatches_telegram": False,
"requires_production_version_truth": True,
"requires_operator_write_approval": True,
},
"required_before_execute": [
"Run production version truth guard and keep production /health as latest truth.",
"Confirm the selected direct mapping targets are the intended PChome products.",
"Confirm DB-write authorization for /api/ai/pchome-growth/backfill-momo-candidates.",
"Run post-write mapping backlog readback and compare direct_mapping_count / mapped_count.",
],
"acceptance_criteria": [
"direct_mapping_count decreases, or review_candidate_count increases with named MOMO candidates.",
"mapped_count or mapping_rate increases only when a verified external price is written.",
"No Gemini, Telegram dispatch, scheduler mutation, or unrelated DB write is part of this run.",
],
}

376
services/pchome_mapping_backlog_service.py
View File

@@ -162,220 +162,14 @@ from services.pchome_mapping_backlog.evidence import (
parse_pchome_product_page_evidence_html,
parse_unit_package_basis,
)
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()
product_name = str(item.get("product_name") or "").strip()
product_url = _first_present(item.get("product_url"), item.get("pchome_url"), _pchome_product_url(product_id))
pchome_price = _first_present(
item.get("pchome_price"),
external_price.get("pchome_price"),
review_candidate.get("pchome_price"),
)
image_url = _first_present(item.get("image_url"), item.get("image"), item.get("product_image_url"))
availability = _first_present(item.get("availability"), item.get("stock_status"), item.get("is_available"))
unit_package_basis = parse_unit_package_basis(product_name)
parsed_unit_basis = (
unit_package_basis
if unit_package_basis.get("package_basis") != "insufficient"
else None
)
unit_basis = _first_present(
external_price.get("price_basis"),
item.get("price_basis"),
item.get("unit_label"),
parsed_unit_basis,
)
unit_review_required = bool(unit_package_basis.get("risk_signals"))
checks = [
("stable_product_id", bool(product_id), "required"),
("product_name", bool(product_name), "required"),
("product_url", bool(product_url), "required"),
("price", pchome_price is not None, "required"),
("image", bool(image_url), "strongly_recommended"),
("availability", availability is not None, "strongly_recommended"),
(
"unit_price_or_package_basis",
bool(unit_basis),
"required_when_bundle_or_unit_sensitive",
),
]
present = [field for field, ok, _requirement in checks if ok]
missing = [field for field, ok, _requirement in checks if not ok]
blocking_missing = [
field
for field, ok, requirement in checks
if not ok and requirement in {"required", "strongly_recommended"}
]
score = round(len(present) / max(len(checks), 1) * 100, 1)
ai_exception_required = (
bool(blocking_missing)
or bool(review_candidate)
or not external_price
or unit_review_required
)
return {
"score": score,
"present_fields": present,
"missing_fields": missing,
"blocking_missing_fields": blocking_missing,
"auto_accept_ready": not blocking_missing and bool(external_price) and not unit_review_required,
**_ai_exception_compatibility_fields(ai_exception_required),
"product_url": product_url,
"image_url": image_url,
"availability": availability,
"unit_package_basis": unit_package_basis,
}
def compact_mapping_item(item: dict[str, Any]) -> dict[str, Any]:
review_candidate = item.get("review_candidate") or {}
external_price = item.get("external_price") or {}
product_id = str(item.get("pchome_product_id") or "")
product_url = _first_present(item.get("product_url"), item.get("pchome_url"), _pchome_product_url(product_id))
return {
"pchome_product_id": product_id,
"product_url": product_url,
"product_name": item.get("product_name") or "",
"sales_7d": round(_to_float(item.get("sales_7d")), 2),
"sales_delta_pct": item.get("sales_delta_pct"),
"priority_score": item.get("priority_score"),
"pchome_price": item.get("pchome_price"),
"action_code": _action_code(item),
"action_label": _action_label(item),
"review_candidate": {
"id": review_candidate.get("id"),
"momo_sku": review_candidate.get("momo_sku"),
"momo_name": review_candidate.get("momo_name"),
"quality_score": review_candidate.get("quality_score"),
"gap_pct": review_candidate.get("gap_pct"),
}
if review_candidate
else None,
"external_price": {
"momo_sku": external_price.get("momo_sku"),
"momo_name": external_price.get("momo_name"),
"price_basis": external_price.get("price_basis"),
"gap_pct": external_price.get("gap_pct"),
"data_source_label": external_price.get("data_source_label"),
"updated_at": external_price.get("updated_at"),
}
if external_price
else None,
"evidence_completeness": _evidence_completeness(item, review_candidate, external_price),
"reason_lines": list(item.get("reason_lines") or [])[:3],
}
def _build_external_benchmark_alignment() -> dict[str, Any]:
return {
"references": EXTERNAL_BENCHMARK_REFERENCES,
"required_evidence_fields": [
{
"field": "stable_product_id",
"current_payload": "pchome_product_id",
"status": "present",
"why": "Stable IDs preserve mapping history and make post-run readback comparable.",
},
{
"field": "product_name",
"current_payload": "product_name",
"status": "present",
"why": "Exact title/name matching is the first identity anchor for operator review.",
},
{
"field": "product_url",
"current_payload": "derived_from_pchome_product_id",
"status": "present_for_pchome",
"why": "Operators need a direct product page path for visual confirmation.",
},
{
"field": "price",
"current_payload": "pchome_price/external_price",
"status": "partial",
"why": "Offer price and currency are required before a candidate can become decision-ready.",
},
{
"field": "image",
"current_payload": None,
"status": "missing_in_current_growth_payload",
"why": "Image evidence should be added before high-volume auto-accept expansion.",
},
{
"field": "availability",
"current_payload": None,
"status": "missing_in_current_growth_payload",
"why": "Availability prevents matching stale or non-purchasable offers.",
},
{
"field": "unit_price_or_package_basis",
"current_payload": "external_price.price_basis or deterministic title parser preview",
"status": "parser_preview_available",
"why": "Unit price and package basis protect bundles, variants, and volume-size comparisons.",
},
],
"operator_review_principles": [
"Separate direct mapping, review candidate, and already comparable items.",
"Do not auto-accept variants, colors, bundles, or catalog offers without explicit evidence.",
"Keep search/query support exact-title friendly so copied product names and model terms remain useful.",
],
}
def _build_ai_automation_plan(selected_direct: list[dict[str, Any]], selected_review: list[dict[str, Any]]) -> dict[str, Any]:
return {
"policy": "ollama_first_read_only_ai_assist",
"llm_calls_in_preview": False,
"gemini_allowed": False,
"provider_order": [
"GCP-A 34.87.90.216:11434",
"GCP-B 34.21.145.224:11434",
"111 192.168.0.111:11434",
],
"automation_readiness": {
"direct_mapping_targets": len(selected_direct),
"review_candidate_targets": len(selected_review),
"can_generate_operator_summary": bool(selected_direct or selected_review),
"can_execute_write": False,
},
"steps": [
{
"name": "identity_anchor_extraction",
"mode": "deterministic_first_ollama_assist_later",
"writes_database": False,
"output": "brand/product_line/spec/package/variant anchors for each selected target",
},
{
"name": "candidate_search_plan",
"mode": "rule_based_query_pack",
"writes_database": False,
"output": "exact title, brand plus product line, and spec-preserving search terms",
},
{
"name": "operator_decision_summary",
"mode": "ollama_first_after_write_gate_only",
"writes_database": False,
"output": "plain-language review reason, evidence gaps, and post-write readback checklist",
},
{
"name": "post_write_readback",
"mode": "deterministic_metrics",
"writes_database": False,
"output": "mapping_rate, direct_mapping_count, review_candidate_count, mapped_count delta",
},
],
"ai_exception_required_for": [
"missing image or availability evidence",
"variant/color/fragrance/shade/package ambiguity",
"unit-price or bundle-sensitive comparisons",
"any candidate not meeting exact identity evidence",
],
}
from services.pchome_mapping_backlog.reporter import (
_build_ai_automation_plan,
_build_external_benchmark_alignment,
_evidence_completeness,
build_pchome_mapping_operator_preview,
compact_mapping_item,
summarize_pchome_mapping_backlog,
)
def _field_enrichment_sources(field: str) -> list[dict[str, Any]]:
@@ -537,160 +331,6 @@ def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]:
}
def summarize_pchome_mapping_backlog(payload: dict[str, Any]) -> dict[str, Any]:
stats = payload.get("stats") or {}
opportunities = [item for item in payload.get("opportunities") or [] if isinstance(item, dict)]
needs_mapping = [item for item in opportunities if not item.get("external_price")]
review_candidates = [item for item in needs_mapping if item.get("review_candidate")]
direct_mapping = [
item
for item in needs_mapping
if _action_code(item) == "map_external_product" and not item.get("review_candidate")
]
mapped = [item for item in opportunities if item.get("external_price")]
action_counts: dict[str, int] = {}
sales_by_action: dict[str, float] = {}
for item in opportunities:
label = _action_label(item)
action_counts[label] = action_counts.get(label, 0) + 1
sales_by_action[label] = round(sales_by_action.get(label, 0.0) + _to_float(item.get("sales_7d")), 2)
candidate_count = int(stats.get("candidate_count") or len(opportunities))
mapped_count = int(stats.get("mapped_count") or len(mapped))
needs_mapping_count = int(stats.get("needs_mapping_count") or len(needs_mapping))
mapping_rate = stats.get("mapping_rate")
if mapping_rate is None:
mapping_rate = round(mapped_count / max(candidate_count, 1) * 100, 1)
top_needs_mapping = sorted(
needs_mapping,
key=lambda item: (_to_float(item.get("sales_7d")), _to_float(item.get("priority_score"))),
reverse=True,
)[:10]
top_review_candidates = sorted(
review_candidates,
key=lambda item: _to_float((item.get("review_candidate") or {}).get("quality_score")),
reverse=True,
)[:10]
if not payload.get("success", False):
result = "BLOCKED"
elif needs_mapping_count > 0:
result = "NEEDS_MAPPING"
else:
result = "PASS"
return {
"policy": BACKLOG_POLICY,
"result": result,
"success": bool(payload.get("success")),
"generated_at": payload.get("generated_at"),
"cache_state": payload.get("cache_state"),
"system_name": payload.get("system_name"),
"message": payload.get("message"),
"stats": {
"candidate_count": candidate_count,
"mapped_count": mapped_count,
"mapping_rate": mapping_rate,
"needs_mapping_count": needs_mapping_count,
"review_candidate_count": int(stats.get("review_candidate_count") or len(review_candidates)),
"latest_sales_date": stats.get("latest_sales_date"),
"overall_latest_sales_date": stats.get("overall_latest_sales_date"),
"overall_sales_7d": stats.get("overall_sales_7d"),
"opportunity_sales_7d": stats.get("opportunity_sales_7d"),
"action_counts": dict(stats.get("action_counts") or action_counts),
"action_code_counts": dict(stats.get("action_code_counts") or {}),
"external_data_source_counts": dict(stats.get("external_data_source_counts") or {}),
},
"backlog": {
"direct_mapping_count": len(direct_mapping),
"review_candidate_count": len(review_candidates),
"mapped_opportunity_count": len(mapped),
"sales_by_action": sales_by_action,
"top_needs_mapping": [compact_mapping_item(item) for item in top_needs_mapping],
"top_review_candidates": [compact_mapping_item(item) for item in top_review_candidates],
},
"next_actions": [
"Run the production version truth guard before changing or deploying.",
"Handle direct mapping items first; they have no verified external price yet.",
"Review candidate items next; they already have MOMO candidates but need same-item confirmation.",
"Keep this report read-only until an explicit DB-write operator run is approved.",
],
}
def build_pchome_mapping_operator_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
"""Build a read-only operator run package for the direct mapping backlog."""
summary = summarize_pchome_mapping_backlog(payload)
backlog = summary.get("backlog") or {}
direct_items = [
item
for item in backlog.get("top_needs_mapping") or []
if item.get("action_code") == "map_external_product"
]
review_items = list(backlog.get("top_review_candidates") or [])
batch_size = max(1, min(int(batch_size or 5), 8))
selected_direct = direct_items[:batch_size]
selected_review = review_items[:batch_size]
if selected_direct:
result = "READY_FOR_OPERATOR_PREVIEW"
elif selected_review:
result = "REVIEW_CANDIDATES_ONLY"
else:
result = "NO_DIRECT_MAPPING_TARGETS"
return {
"policy": OPERATOR_PREVIEW_POLICY,
"result": result,
"success": bool(summary.get("success")),
"generated_at": summary.get("generated_at"),
"stats": summary.get("stats") or {},
"backlog": {
"direct_mapping_count": int(backlog.get("direct_mapping_count") or 0),
"review_candidate_count": int(backlog.get("review_candidate_count") or 0),
"mapped_opportunity_count": int(backlog.get("mapped_opportunity_count") or 0),
},
"operator_batch": {
"batch_size": batch_size,
"selected_direct_mapping_count": len(selected_direct),
"selected_review_candidate_count": len(selected_review),
"direct_mapping_targets": selected_direct,
"review_candidate_targets": selected_review,
},
"command_preview": {
"method": "POST",
"endpoint": "/api/ai/pchome-growth/backfill-momo-candidates",
"payload": {"limit": min(batch_size, 8)},
"executes_search": True,
"writes_database": True,
"write_gate_required": True,
},
"external_benchmark_alignment": _build_external_benchmark_alignment(),
"ai_automation_plan": _build_ai_automation_plan(selected_direct, selected_review),
"safety": {
"read_only_preview": True,
"executes_search": False,
"writes_database": False,
"dispatches_telegram": False,
"requires_production_version_truth": True,
"requires_operator_write_approval": True,
},
"required_before_execute": [
"Run production version truth guard and keep production /health as latest truth.",
"Confirm the selected direct mapping targets are the intended PChome products.",
"Confirm DB-write authorization for /api/ai/pchome-growth/backfill-momo-candidates.",
"Run post-write mapping backlog readback and compare direct_mapping_count / mapped_count.",
],
"acceptance_criteria": [
"direct_mapping_count decreases, or review_candidate_count increases with named MOMO candidates.",
"mapped_count or mapping_rate increases only when a verified external price is written.",
"No Gemini, Telegram dispatch, scheduler mutation, or unrelated DB write is part of this run.",
],
}
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