357 lines
17 KiB
Python
357 lines
17 KiB
Python
"""Bounded database executor for verified PChome mapping selectors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from services.pchome_mapping_backlog.artifacts import (
|
|
_canonical_retry_exception_artifact_bytes,
|
|
_resolve_retry_exception_artifact_path,
|
|
)
|
|
from services.pchome_mapping_backlog.policies import (
|
|
DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY,
|
|
)
|
|
|
|
|
|
def _retry_exception_controlled_apply_executor_id(preflight_package: dict[str, Any]) -> str:
|
|
preflight = preflight_package.get("controlled_apply_preflight") or {}
|
|
payload = {
|
|
"preflight_id": preflight.get("preflight_id") or "",
|
|
"run_id": preflight.get("run_id") or "",
|
|
"target_selector_count": (preflight_package.get("summary") or {}).get("target_selector_count") or 0,
|
|
"mutation_plan_count": (preflight_package.get("summary") or {}).get("mutation_plan_count") or 0,
|
|
}
|
|
digest = hashlib.sha256(
|
|
json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
|
|
).hexdigest()
|
|
return f"pchome-retry-exception-controlled-apply-executor-{digest[:16]}"
|
|
|
|
|
|
def _selector_to_pchome_match_params(selector: dict[str, Any]) -> dict[str, Any]:
|
|
target_pchome_product_id = str(selector.get("target_pchome_product_id") or "").strip()
|
|
return {
|
|
"momo_icode": str(selector.get("momo_product_id") or "").strip(),
|
|
"momo_name": str(selector.get("momo_product_name") or selector.get("momo_product_id") or "").strip(),
|
|
"momo_price": selector.get("momo_price"),
|
|
"pchome_id": target_pchome_product_id,
|
|
"pchome_name": str(selector.get("target_pchome_product_name") or target_pchome_product_id).strip(),
|
|
"pchome_url": f"https://24h.pchome.com.tw/prod/{target_pchome_product_id}" if target_pchome_product_id else None,
|
|
"similarity": selector.get("target_match_score"),
|
|
"advantage": "ai_match",
|
|
}
|
|
|
|
|
|
def _fetch_pchome_product_match_by_momo_icode(conn: Any, momo_icode: str) -> dict[str, Any] | None:
|
|
from sqlalchemy import text
|
|
|
|
row = conn.execute(text("""
|
|
SELECT id, momo_name, momo_icode, momo_price, pchome_id, pchome_name,
|
|
pchome_url, pchome_price, pchome_original, pchome_in_stock,
|
|
similarity, price_diff, price_diff_pct, advantage, last_checked
|
|
FROM pchome_product_matches
|
|
WHERE momo_icode = :momo_icode
|
|
"""), {"momo_icode": momo_icode}).mappings().first()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def _upsert_pchome_product_match(conn: Any, selector: dict[str, Any]) -> None:
|
|
from sqlalchemy import text
|
|
|
|
params = _selector_to_pchome_match_params(selector)
|
|
conn.execute(text("""
|
|
INSERT INTO pchome_product_matches (
|
|
momo_name, momo_icode, momo_price, pchome_id, pchome_name,
|
|
pchome_url, similarity, advantage, last_checked
|
|
)
|
|
VALUES (
|
|
:momo_name, :momo_icode, :momo_price, :pchome_id, :pchome_name,
|
|
:pchome_url, :similarity, :advantage, CURRENT_TIMESTAMP
|
|
)
|
|
ON CONFLICT (momo_icode) DO UPDATE SET
|
|
momo_name = EXCLUDED.momo_name,
|
|
momo_price = EXCLUDED.momo_price,
|
|
pchome_id = EXCLUDED.pchome_id,
|
|
pchome_name = EXCLUDED.pchome_name,
|
|
pchome_url = EXCLUDED.pchome_url,
|
|
similarity = EXCLUDED.similarity,
|
|
advantage = EXCLUDED.advantage,
|
|
last_checked = CURRENT_TIMESTAMP
|
|
"""), params)
|
|
|
|
|
|
def build_controlled_apply_executor_from_preflight(
|
|
preflight_package: dict[str, Any],
|
|
*,
|
|
execute_apply: bool = False,
|
|
engine: Any = None,
|
|
materialize_artifacts: bool = False,
|
|
artifact_root: str | Path | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Apply verified selectors and emit independently verifiable receipts."""
|
|
preflight = preflight_package.get("controlled_apply_preflight") or {}
|
|
selectors = list(preflight_package.get("target_selectors") or [])
|
|
executor_ready = bool(preflight.get("ready_for_controlled_apply_executor")) and bool(selectors)
|
|
executor_id = _retry_exception_controlled_apply_executor_id(preflight_package)
|
|
write_attempted = bool(execute_apply and executor_ready and engine is not None)
|
|
write_blockers: list[str] = []
|
|
if execute_apply and engine is None:
|
|
write_blockers.append("engine_required_for_execute_apply")
|
|
if execute_apply and not executor_ready:
|
|
write_blockers.append("controlled_apply_preflight_not_ready")
|
|
|
|
prewrite_snapshots: list[dict[str, Any]] = []
|
|
applied_records: list[dict[str, Any]] = []
|
|
post_apply_readbacks: list[dict[str, Any]] = []
|
|
missing_tables: list[str] = []
|
|
if write_attempted:
|
|
from sqlalchemy import inspect
|
|
|
|
with engine.begin() as conn:
|
|
inspector = inspect(conn)
|
|
if not inspector.has_table("pchome_product_matches"):
|
|
missing_tables.append("pchome_product_matches")
|
|
else:
|
|
for selector in selectors:
|
|
momo_icode = str(selector.get("momo_product_id") or "").strip()
|
|
before = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode)
|
|
prewrite_snapshots.append({
|
|
"selector_id": selector.get("selector_id"),
|
|
"momo_icode": momo_icode,
|
|
"before": before,
|
|
"row_existed_before": before is not None,
|
|
})
|
|
if not missing_tables:
|
|
for selector in selectors:
|
|
_upsert_pchome_product_match(conn, selector)
|
|
momo_icode = str(selector.get("momo_product_id") or "").strip()
|
|
after = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode)
|
|
applied_records.append({
|
|
"selector_id": selector.get("selector_id"),
|
|
"momo_icode": momo_icode,
|
|
"target_pchome_product_id": selector.get("target_pchome_product_id"),
|
|
"applied": bool(after),
|
|
})
|
|
post_apply_readbacks.append({
|
|
"selector_id": selector.get("selector_id"),
|
|
"momo_icode": momo_icode,
|
|
"expected_pchome_id": selector.get("target_pchome_product_id"),
|
|
"actual_pchome_id": (after or {}).get("pchome_id"),
|
|
"expected_momo_name": selector.get("momo_product_name"),
|
|
"actual_momo_name": (after or {}).get("momo_name"),
|
|
"passed": bool(after)
|
|
and str((after or {}).get("pchome_id") or "") == str(selector.get("target_pchome_product_id") or "")
|
|
and str((after or {}).get("momo_icode") or "") == momo_icode,
|
|
"writes_database": False,
|
|
})
|
|
|
|
readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed"))
|
|
applied_count = len(applied_records)
|
|
if execute_apply and missing_tables:
|
|
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_BLOCKED_MISSING_TABLE"
|
|
elif execute_apply and not selectors:
|
|
result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT"
|
|
elif execute_apply and selectors and applied_count == len(selectors) and readback_pass_count == len(selectors):
|
|
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED"
|
|
elif execute_apply:
|
|
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTION_INCOMPLETE"
|
|
elif executor_ready:
|
|
result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_READY"
|
|
else:
|
|
result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT"
|
|
|
|
rollback_steps = [
|
|
{
|
|
"selector_id": snapshot.get("selector_id"),
|
|
"momo_icode": snapshot.get("momo_icode"),
|
|
"action": "restore_previous_pchome_product_match" if snapshot.get("row_existed_before") else "delete_inserted_pchome_product_match",
|
|
"prewrite_snapshot_available": True,
|
|
"executes_in_executor": False,
|
|
"writes_database": False,
|
|
}
|
|
for snapshot in prewrite_snapshots
|
|
] or [
|
|
{
|
|
"selector_id": selector.get("selector_id"),
|
|
"momo_icode": selector.get("momo_product_id"),
|
|
"action": "planned_restore_or_delete_pchome_product_match",
|
|
"prewrite_snapshot_available": False,
|
|
"executes_in_executor": False,
|
|
"writes_database": False,
|
|
}
|
|
for selector in selectors
|
|
]
|
|
summary = {
|
|
"controlled_apply_executor_ready_count": 1 if executor_ready else 0,
|
|
"execute_apply_requested_count": 1 if execute_apply else 0,
|
|
"target_selector_count": len(selectors),
|
|
"prewrite_snapshot_count": len(prewrite_snapshots),
|
|
"applied_record_count": applied_count,
|
|
"post_apply_readback_count": len(post_apply_readbacks),
|
|
"post_apply_readback_pass_count": readback_pass_count,
|
|
"post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count,
|
|
"rollback_step_count": len(rollback_steps),
|
|
"missing_table_count": len(missing_tables),
|
|
"writes_database_count": applied_count,
|
|
"persists_candidate_count": applied_count,
|
|
}
|
|
executor_metadata = {
|
|
"executor_id": executor_id,
|
|
"source_preflight_id": preflight.get("preflight_id"),
|
|
"run_id": preflight.get("run_id"),
|
|
"stage": "P2_retry_exception_controlled_apply_executor",
|
|
"status": result,
|
|
"execute_apply": bool(execute_apply),
|
|
"target_table": "pchome_product_matches",
|
|
"ready_for_apply": executor_ready,
|
|
"write_attempted": write_attempted,
|
|
"missing_tables": missing_tables,
|
|
"requires_fresh_production_truth": True,
|
|
}
|
|
rollback_plan = {
|
|
"rollback_step_count": len(rollback_steps),
|
|
"rollback_steps": rollback_steps,
|
|
"executes_in_executor": False,
|
|
"writes_database": False,
|
|
}
|
|
safety = {
|
|
"ai_controlled_apply": True,
|
|
"execute_apply": bool(execute_apply),
|
|
"target_table": "pchome_product_matches",
|
|
"writes_database": bool(applied_count),
|
|
"writes_database_count": applied_count,
|
|
"persists_candidate": bool(applied_count),
|
|
"persists_candidate_count": applied_count,
|
|
"syncs_external_offers": False,
|
|
"dispatches_telegram": False,
|
|
"llm_calls_in_executor": False,
|
|
"gemini_allowed": False,
|
|
"requires_production_version_truth": True,
|
|
}
|
|
executor_receipt_ready = (
|
|
result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_EXECUTED"
|
|
and bool(applied_records)
|
|
and readback_pass_count == len(selectors)
|
|
)
|
|
root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data"
|
|
receipt_relative_path = (
|
|
f"artifacts/pchome_growth/retry_exception_closeout/"
|
|
f"controlled_apply_executor/{executor_id}.json"
|
|
)
|
|
executor_receipt_payload = {
|
|
"artifact_key": "retry_exception_controlled_apply_executor_receipt",
|
|
"executor_id": executor_id,
|
|
"source_preflight_id": preflight.get("preflight_id"),
|
|
"run_id": preflight.get("run_id"),
|
|
"source_policy": preflight_package.get("policy"),
|
|
"result": result,
|
|
"created_at": preflight_package.get("generated_at"),
|
|
"summary": summary,
|
|
"controlled_apply_executor": executor_metadata,
|
|
"target_selectors": selectors,
|
|
"prewrite_snapshots": prewrite_snapshots,
|
|
"applied_records": applied_records,
|
|
"post_apply_readbacks": post_apply_readbacks,
|
|
"rollback_plan": rollback_plan,
|
|
"safety": safety,
|
|
}
|
|
executor_receipt_bytes = _canonical_retry_exception_artifact_bytes(executor_receipt_payload)
|
|
executor_receipt_artifact = {
|
|
"key": "retry_exception_controlled_apply_executor_receipt",
|
|
"artifact_type": "controlled_apply_executor_receipt",
|
|
"relative_path": receipt_relative_path,
|
|
"payload_sha256": hashlib.sha256(executor_receipt_bytes).hexdigest(),
|
|
"byte_count": len(executor_receipt_bytes),
|
|
"payload": executor_receipt_payload,
|
|
"materialized": False,
|
|
"writes_database": False,
|
|
}
|
|
materialized_executor_artifacts: list[dict[str, Any]] = []
|
|
if materialize_artifacts and executor_receipt_ready:
|
|
target_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path)
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
target_path.write_bytes(executor_receipt_bytes)
|
|
materialized_executor_artifacts.append({
|
|
"key": executor_receipt_artifact["key"],
|
|
"relative_path": receipt_relative_path,
|
|
"absolute_path": str(target_path),
|
|
"payload_sha256": executor_receipt_artifact["payload_sha256"],
|
|
"written_byte_count": target_path.stat().st_size,
|
|
"writes_database": False,
|
|
})
|
|
executor_receipt_artifact["materialized"] = True
|
|
executor_receipt_artifact["absolute_path"] = str(target_path)
|
|
|
|
receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path)
|
|
actual_receipt_sha = ""
|
|
receipt_file_exists = receipt_path.exists()
|
|
if receipt_file_exists:
|
|
actual_receipt_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest()
|
|
receipt_checks = [
|
|
{"check": "executor_receipt_ready_after_apply", "passed": executor_receipt_ready},
|
|
{"check": "all_post_apply_readbacks_passed", "passed": readback_pass_count == len(selectors)},
|
|
{"check": "applied_record_count_matches_selectors", "passed": applied_count == len(selectors)},
|
|
{"check": "receipt_payload_hash_is_sha256", "passed": len(executor_receipt_artifact["payload_sha256"]) == 64},
|
|
{
|
|
"check": "materialized_receipt_exists_when_requested",
|
|
"passed": (not materialize_artifacts) or (executor_receipt_ready and receipt_file_exists),
|
|
},
|
|
{
|
|
"check": "materialized_receipt_hash_matches_expected",
|
|
"passed": (not materialize_artifacts)
|
|
or (bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"]),
|
|
},
|
|
{"check": "receipt_safety_blocks_side_effects", "passed": safety["syncs_external_offers"] is False and safety["dispatches_telegram"] is False},
|
|
]
|
|
post_executor_receipt_verifier = {
|
|
"ready": executor_receipt_ready,
|
|
"checks": receipt_checks,
|
|
"check_count": len(receipt_checks),
|
|
"passed": all(check.get("passed") is True for check in receipt_checks),
|
|
"expected_sha256": executor_receipt_artifact["payload_sha256"],
|
|
"actual_sha256": actual_receipt_sha,
|
|
"hash_match": bool(actual_receipt_sha) and actual_receipt_sha == executor_receipt_artifact["payload_sha256"],
|
|
"reads_artifact_files": bool(materialize_artifacts),
|
|
"writes_database": False,
|
|
}
|
|
summary["executor_receipt_ready_count"] = 1 if executor_receipt_ready else 0
|
|
summary["executor_receipt_payload_count"] = 1
|
|
summary["executor_receipt_materialized_count"] = len(materialized_executor_artifacts)
|
|
summary["executor_receipt_hash_match_count"] = 1 if post_executor_receipt_verifier["hash_match"] else 0
|
|
summary["post_executor_receipt_verifier_check_count"] = len(receipt_checks)
|
|
safety["writes_artifact_count"] = len(materialized_executor_artifacts)
|
|
return {
|
|
"policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY,
|
|
"result": result,
|
|
"success": bool(preflight_package.get("success")),
|
|
"generated_at": preflight_package.get("generated_at"),
|
|
"source_policy": preflight_package.get("policy"),
|
|
"stats": preflight_package.get("stats") or {},
|
|
"backlog": preflight_package.get("backlog") or {},
|
|
"summary": summary,
|
|
"controlled_apply_executor": executor_metadata,
|
|
"target_selectors": selectors,
|
|
"prewrite_snapshots": prewrite_snapshots,
|
|
"applied_records": applied_records,
|
|
"post_apply_readbacks": post_apply_readbacks,
|
|
"rollback_plan": rollback_plan,
|
|
"executor_receipt_artifact": executor_receipt_artifact,
|
|
"materialized_executor_artifacts": materialized_executor_artifacts,
|
|
"post_executor_receipt_verifier": post_executor_receipt_verifier,
|
|
"write_blockers": write_blockers,
|
|
"source_preflight_summary": preflight_package.get("summary") or {},
|
|
"next_actions": [
|
|
"Use the executor receipt artifact as the machine-verifiable closeout source for every applied selector.",
|
|
"Run rollback steps only if post-apply readback fails or a future verifier detects drift.",
|
|
"Keep future writes bounded to selector IDs from this executor package.",
|
|
],
|
|
"safety": safety,
|
|
}
|
|
|
|
|
|
__all__ = (
|
|
"build_controlled_apply_executor_from_preflight",
|
|
)
|