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

V10.547 強化單位價覆核洞察

Browse Source
This commit is contained in:
OoO
2026-06-01 12:19:48 +08:00
parent e5ecf5512e
commit f10d73ff83
9 changed files with 350 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
TODO_NEXT_STEPS.txt
View File

@@ -4,6 +4,7 @@
================================================================================
【已完成】
- V10.547 強化單位價覆核洞察：`manual_unit_price_required` 不再只是人工狀態，覆核隊列與商品看板會重新帶出單位價換算、MOMO/PChome 單位價方向、差距百分比與處理建議；決策信封 / OpenClaw / PPT 摘要可讀到 `unit_price_insight`。人工覆核寫回也會保留原始 `match_diagnostic_json` / comparison mode / diagnostic codes，避免後續簡報、審計或 AI 策略只剩人工文案而失去 matcher 證據鏈。
- V10.546 補近門檻舊候選回刷隊列：`run_retryable_candidate_revalidation()` 新增 `legacy_unmasked_attempt`，當最新狀態是 `no_result` / `refresh_no_result` / `expired_match` 時，可回撈同 SKU 早期近門檻候選交給最新版 matcher 重評；仍要求 candidate id、分數下限、無 hard veto、exact_identity，且不打開人工否決、單位價、identity_veto 或 protected existing match。
- V10.545 收斂 Dashboard 比價覆蓋率口徑：coverage cache 升到 v9，新增身份覆蓋、可用比價、新鮮度、待補身份、過期身份與人工閉環欄位；商品看板和 PChome 覆核頁改只把真正待處理狀態算進「比價覆核」，人工已否決 / 人工單位價 / 需補研究改列為人工閉環；PChome competitor map 只吃有效價格、新鮮、identity_v2 最新 row，資料新鮮度也改看可用比價 row。
- V10.544 收斂變體安全與 YES 指甲工具線：新增 YES 德悅氏指甲剪附除垢銼刀、腳皮銼腳板、藍寶石銼刀、三面拋光棒與 6/8cm 指甲剪的精準 total-price 線，要求同品牌、同工具名稱、同尺寸與同亮面/霧面/可收納/三面/不掉屑等款式訊號；同步接進 revalidation SQL。新增 MUJI / COCODOR 未知香味差異與 OPI 無型號不同色名 hard veto，HOOOME 暖燈材質差留人工覆核，搜尋詞也會優先帶香味/色名，提升 crawler 精準候選率。

2
config.py
View File

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

1
docs/memory/history_logs.md
View File

@@ -13,6 +13,7 @@
## 📅 詳細更新日誌 (考古存檔)
### 2026-06-01：PChome 比價新鮮度操作閉環
- **V10.547 單位價覆核洞察與證據鏈保留**: `manual_unit_price_required` 現在會和 `unit_comparable` 一樣重新產生單位價比較，並轉成 `unit_price_insight`，明確標示 PChome 或 MOMO 哪邊單位價較低、差距百分比、嚴重度與操作建議；Dashboard 覆核卡、商品列、決策信封與 OpenClaw/PPT 摘要都可讀到這個訊號。人工覆核寫回 `competitor_match_attempts` 時也會在欄位存在時保留原始 `match_diagnostic_json`、`comparison_mode`、`hard_veto`、`diagnostic_codes`，`competitor_match_reviews.candidate_diagnostic` 同步附帶 JSON 證據，避免人工閉環後只剩狀態文字。
- **V10.546 近門檻舊候選回刷隊列補漏**: `run_retryable_candidate_revalidation()` 的候選來源不再只看每個 SKU 最新一筆 attempt。新增 `legacy_unmasked_attempt`，當最新狀態是 `no_result` / `refresh_no_result` / `expired_match` 時，可回撈同 SKU 早期 `low_score`、`recoverable_low_score`、`true_low_confidence` 或 `rescore_accepted_current` 的近門檻候選，再交給最新版 matcher 重評。此入口仍要求 candidate product id、分數下限、無 hard veto、`exact_identity`，且不打開人工否決、單位價、identity_veto 或 protected existing match，避免為了覆蓋率破壞安全邊界。
- **V10.545 Dashboard 比價覆蓋率口徑收斂**: 商品看板與 PChome 覆核頁把「身份覆蓋率」與「可用比價覆蓋率」拆成明確欄位，coverage cache 更新為 v9 並回傳 `identity_coverage_*`、`pending_identity_count`、`stale_identity_count`、`last_decision_ready_crawled_at`。覆核 KPI 改只計算真正待處理狀態，人工否決 / 人工單位價 / 需補研究另列 `manual_closed_count`，避免人工閉環候選被混進待審總數。Dashboard 的 PChome competitor map 也改成只取新鮮、有效價格、identity_v2 的最新 row，資料新鮮度改看可用比價 row，不再被無效或低信心抓取紀錄撐高。
- **V10.544 變體安全與 YES 工具線收斂**: 延續近門檻 `low_score` 救回，但把安全邊界補得更硬。新增 YES 德悅氏指甲工具精準線，只有同品牌、同工具線、同尺寸且同亮面/霧面/可收納/三面等關鍵款式時才進 `total_price`，並接入 revalidation SQL。同步新增未知香味差異與無型號指彩色名差異 hard veto：MUJI / COCODOR 不同香型、OPI 無型號不同色名不再被高分誤配；HOOOME 暖燈陶瓷/玻璃/水晶/金屬等材質差保留人工覆核。搜尋詞對護手霜、擴香瓶、無型號指彩優先帶上香味/色名，提升 crawler 找到真同款候選的機率。

33
routes/dashboard_routes.py
View File

@@ -183,11 +183,18 @@ def _build_pchome_match_status(attempt=None, ineligible=None):
if status == 'manual_unit_price_required':
score = _to_float(attempt.get('best_match_score'))
score_text = f"候選 {round(score * 100)}%" if score is not None else "人工標記單位價"
unit_comparison = attempt.get('unit_comparison') or {}
unit_insight = attempt.get('unit_price_insight') or {}
summary = '人工已判定總價不可直接比較，需以每 ml / 每 g / 每入單位價與檔期條件判讀'
if unit_comparison.get('comparable') and unit_insight.get('summary'):
summary = f"{unit_insight.get('summary')}；仍不寫入正式總價差"
elif unit_comparison.get('summary'):
summary = f"已換算單位價：{unit_comparison.get('summary')}；仍不寫入正式總價差"
return {
'label': '人工標記單位價',
'tone': 'watch',
'blocks_price_gap': True,
'summary': '人工已判定總價不可直接比較，需以每 ml / 每 g / 每入單位價與檔期條件判讀',
'summary': summary,
'detail': score_text,
}
if status == 'manual_needs_research':
@@ -606,10 +613,14 @@ def _load_pchome_match_attempt_map(session, skus):
result = {}
try:
from services.competitor_intel_repository import (
_build_unit_comparison_for_attempt,
_build_unit_price_business_insight,
_extract_match_diagnostic_reasons,
_parse_json_payload,
)
except Exception:
_build_unit_comparison_for_attempt = None
_build_unit_price_business_insight = None
_extract_match_diagnostic_reasons = None
_parse_json_payload = None
for row in rows:
@@ -621,15 +632,19 @@ def _load_pchome_match_attempt_map(session, skus):
diagnostic_reasons = _extract_match_diagnostic_reasons(item.get('error_message'), diagnostic_payload)
item['diagnostic_reasons'] = diagnostic_reasons
item['diagnostic_reason_text'] = '、'.join(reason['label'] for reason in diagnostic_reasons)
if item.get('attempt_status') in {'unit_comparable', 'refresh_unit_comparable'}:
if item.get('attempt_status') in {'unit_comparable', 'refresh_unit_comparable', 'manual_unit_price_required'}:
try:
from services.marketplace_product_matcher import build_unit_price_comparison
item['unit_comparison'] = build_unit_price_comparison(
item.get('momo_product_name') or '',
item.get('best_competitor_product_name') or '',
item.get('momo_price'),
item.get('best_competitor_price'),
)
if _build_unit_comparison_for_attempt and _build_unit_price_business_insight:
item['unit_comparison'] = _build_unit_comparison_for_attempt(item)
item['unit_price_insight'] = _build_unit_price_business_insight(item.get('unit_comparison'), item)
else:
from services.marketplace_product_matcher import build_unit_price_comparison
item['unit_comparison'] = build_unit_price_comparison(
item.get('momo_product_name') or '',
item.get('best_competitor_product_name') or '',
item.get('momo_price'),
item.get('best_competitor_price'),
)
except Exception as exc:
sys_log.warning(f"[Dashboard] PChome 單位價比較資料建立略過: {exc}")
item['unit_comparison'] = {'comparable': False, 'reason': 'build_error'}

76
services/competitor_intel_repository.py
View File

@@ -25,6 +25,7 @@ from sqlalchemy import inspect, text
PCHOME_MATCH_SCORE_FLOOR = 0.76
UNIT_COMPARABLE_STATUSES = {"unit_comparable", "refresh_unit_comparable"}
UNIT_PRICE_DECISION_STATUSES = UNIT_COMPARABLE_STATUSES | {"manual_unit_price_required"}
MANUAL_CLOSED_ATTEMPT_STATUSES = {
"manual_rejected",
"manual_unit_price_required",
@@ -300,7 +301,7 @@ def _extract_match_diagnostic_reasons(
def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str, Any]]:
status = str(row.get("attempt_status") or "")
if status not in UNIT_COMPARABLE_STATUSES:
if status not in UNIT_PRICE_DECISION_STATUSES:
return None
try:
from services.marketplace_product_matcher import build_unit_price_comparison
@@ -314,10 +315,57 @@ def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str
return {"comparable": False, "reason": "build_error"}
def _build_unit_price_business_insight(
unit_comparison: Optional[dict[str, Any]],
item: dict[str, Any],
) -> dict[str, Any]:
"""Turn unit-price math into an operator-facing business signal."""
if not isinstance(unit_comparison, dict) or not unit_comparison.get("comparable"):
return {}
unit_gap_pct = _num(unit_comparison.get("unit_gap_pct"))
unit_gap_amount = _num(unit_comparison.get("unit_gap_amount"))
momo_unit_price = _num(unit_comparison.get("momo_unit_price"))
competitor_unit_price = _num(unit_comparison.get("competitor_unit_price"))
unit_label = str(unit_comparison.get("unit_label") or "單位")
abs_gap = abs(unit_gap_pct)
if abs_gap < 3:
direction = "near_parity"
label = "單位價接近"
summary = f"單位價接近，差距 {unit_gap_pct:+.1f}%/{unit_label}，先確認檔期、贈品與運費條件"
action_hint = "檢查檔期與贈品條件後再決定是否列入價格訊號"
elif unit_gap_pct > 0:
direction = "pchome_cheaper"
label = "PChome 單位價較低"
summary = f"PChome 單位價低 {abs_gap:.1f}%/{unit_label}，屬於潛在價格壓力"
action_hint = "確認同品、同單位與檔期條件後，可納入競價壓力觀察"
else:
direction = "momo_cheaper"
label = "MOMO 單位價較低"
summary = f"MOMO 單位價低 {abs_gap:.1f}%/{unit_label}，目前不應誤判為 PChome 價格壓力"
action_hint = "保留為單位價比較證據，不寫入總價型正式價差"
severity = "high" if abs_gap >= 15 else "medium" if abs_gap >= 5 else "low"
return {
"label": label,
"summary": summary,
"direction": direction,
"severity": severity,
"unit_label": unit_label,
"unit_gap_pct": round(unit_gap_pct, 2),
"unit_gap_amount": round(unit_gap_amount, 4),
"momo_unit_price": round(momo_unit_price, 4),
"competitor_unit_price": round(competitor_unit_price, 4),
"action_hint": action_hint,
"attempt_status": item.get("attempt_status") or "",
}
def _review_action_code(attempt_status: str) -> str:
if attempt_status == "rescore_accepted_current":
return "review_accept_identity"
if attempt_status in UNIT_COMPARABLE_STATUSES or attempt_status == "manual_unit_price_required":
if attempt_status in UNIT_PRICE_DECISION_STATUSES:
return "unit_price_required"
if attempt_status in {"no_result", "refresh_no_result", "manual_needs_research"}:
return "needs_research"
@@ -353,7 +401,7 @@ def _review_severity(attempt_status: str, item: dict[str, Any]) -> str:
return "P1"
if attempt_status == "rescore_accepted_current":
return "P2"
if attempt_status in UNIT_COMPARABLE_STATUSES:
if attempt_status in UNIT_PRICE_DECISION_STATUSES:
return "P2"
if attempt_status == "protected_existing_match":
conflict = item.get("existing_match_conflict")
@@ -435,6 +483,14 @@ def _build_review_decision_envelope(item: dict[str, Any]) -> dict[str, Any]:
"value": reason_text,
"basis": "match_diagnostic_json.reasons",
})
unit_price_insight = item.get("unit_price_insight")
if isinstance(unit_price_insight, dict) and unit_price_insight:
evidence.append({
"type": "unit_price",
"metric": "unit_price_gap_pct",
"value": f"{_num(unit_price_insight.get('unit_gap_pct')):+.1f}%",
"basis": unit_price_insight.get("summary") or "unit price comparison",
})
existing_conflict = item.get("existing_match_conflict")
if isinstance(existing_conflict, dict) and existing_conflict:
score_delta = _num(existing_conflict.get("score_delta"))
@@ -478,6 +534,7 @@ def _build_review_decision_envelope(item: dict[str, Any]) -> dict[str, Any]:
"expected_impact": {
"gap_amount": gap_amount,
"candidate_gap_pct": gap_pct,
"unit_price_insight": unit_price_insight if isinstance(unit_price_insight, dict) else {},
"existing_match_conflict": existing_conflict if isinstance(existing_conflict, dict) else {},
"risk_reduction": "medium" if attempt_status in {"rescore_accepted_current", "recoverable_low_score"} else "watch",
},
@@ -538,6 +595,14 @@ def summarize_review_decision_envelopes(
pchome_id = str(subject.get("competitor_product_id") or row.get("candidate_pc_id") or "")
gap_pct = expected.get("candidate_gap_pct")
gap_text = f"價差 {gap_pct:+.1f}%" if isinstance(gap_pct, (int, float)) else ""
unit_insight = expected.get("unit_price_insight")
unit_gap_pct = unit_insight.get("unit_gap_pct") if isinstance(unit_insight, dict) else None
unit_label = unit_insight.get("unit_label") if isinstance(unit_insight, dict) else ""
unit_text = (
f"單位價差 {unit_gap_pct:+.1f}%/{unit_label or '單位'}"
if isinstance(unit_gap_pct, (int, float))
else ""
)
evidence_basis = ""
for evidence_row in evidence:
if isinstance(evidence_row, dict) and evidence_row.get("metric") == "match_score":
@@ -561,6 +626,8 @@ def summarize_review_decision_envelopes(
]
if gap_text:
line_parts.append(gap_text)
if unit_text:
line_parts.append(unit_text)
if evidence_basis:
line_parts.append(evidence_basis)
line = " | ".join(part for part in line_parts if part)
@@ -578,6 +645,7 @@ def summarize_review_decision_envelopes(
"requires_hitl": requires_hitl,
"can_auto_execute": can_auto_execute,
"candidate_gap_pct": gap_pct,
"unit_price_gap_pct": unit_gap_pct,
"line": line,
})
@@ -595,6 +663,7 @@ def summarize_review_decision_envelopes(
def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
item = dict(row)
unit_comparison = _build_unit_comparison_for_attempt(item)
unit_price_insight = _build_unit_price_business_insight(unit_comparison, item)
match_diagnostic = item.get("error_message") or ""
diagnostic_payload = _parse_json_payload(item.get("match_diagnostic_json"))
tags = _parse_tag_list(item.get("tags"))
@@ -634,6 +703,7 @@ def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
"existing_match_conflict": existing_match_conflict,
"attempted_at": _date_label(item.get("attempted_at")),
"unit_comparison": unit_comparison,
"unit_price_insight": unit_price_insight,
}
formatted["decision_envelope"] = _build_review_decision_envelope(formatted)
return formatted

117
services/competitor_match_review_service.py
View File

@@ -9,7 +9,7 @@ import os
from datetime import datetime, timedelta, timezone
from typing import Any
from sqlalchemy import text
from sqlalchemy import inspect, text
VALID_REVIEW_ACTIONS = {
@@ -49,6 +49,46 @@ def _json_array_expr(conn, bind_name: str) -> str:
return f"CAST(:{bind_name} AS jsonb)" if conn.dialect.name == "postgresql" else f":{bind_name}"
def _json_object_expr(conn, bind_name: str) -> str:
return f"CAST(:{bind_name} AS jsonb)" if conn.dialect.name == "postgresql" else f":{bind_name}"
def _json_text_payload(value: Any) -> str | None:
if value is None:
return None
if isinstance(value, (dict, list)):
return json.dumps(value, ensure_ascii=False)
if isinstance(value, str):
stripped = value.strip()
if not stripped:
return None
try:
json.loads(stripped)
return stripped
except (TypeError, ValueError):
return json.dumps({"raw": stripped}, ensure_ascii=False)
return json.dumps(value, ensure_ascii=False)
def _has_table_column(conn, table_name: str, column_name: str) -> bool:
try:
columns = inspect(conn).get_columns(table_name)
return any(str(column.get("name")) == column_name for column in columns)
except Exception:
return False
def _candidate_diagnostic_text(attempt: dict[str, Any]) -> str:
parts: list[str] = []
error_message = str(attempt.get("error_message") or "").strip()
diagnostic_payload = _json_text_payload(attempt.get("match_diagnostic_json"))
if error_message:
parts.append(error_message)
if diagnostic_payload:
parts.append(f"match_diagnostic_json={diagnostic_payload}")
return "; ".join(parts)[:4000]
def _ensure_competitor_match_reviews_table(conn) -> None:
if conn.dialect.name == "postgresql":
conn.execute(text("""
@@ -173,21 +213,40 @@ def _fetch_latest_attempt(conn, sku: str, source: str) -> dict[str, Any] | None:
def _insert_manual_attempt(conn, attempt: dict[str, Any], action_meta: dict[str, str], source: str) -> None:
columns = [
"sku",
"source",
"momo_product_id",
"momo_product_name",
"momo_price",
"search_terms",
"candidate_count",
"attempt_status",
"best_competitor_product_id",
"best_competitor_product_name",
"best_competitor_price",
"best_match_score",
"error_message",
"attempted_at",
]
search_terms_expr = _json_array_expr(conn, "search_terms")
conn.execute(text(f"""
INSERT INTO competitor_match_attempts
(sku, source, momo_product_id, momo_product_name, momo_price,
search_terms, candidate_count, attempt_status,
best_competitor_product_id, best_competitor_product_name,
best_competitor_price, best_match_score, error_message,
attempted_at)
VALUES
(:sku, :source, :momo_product_id, :momo_product_name, :momo_price,
{search_terms_expr}, :candidate_count, :attempt_status,
:best_id, :best_name,
:best_price, :best_score, :error_message,
CURRENT_TIMESTAMP)
"""), {
values = [
":sku",
":source",
":momo_product_id",
":momo_product_name",
":momo_price",
search_terms_expr,
":candidate_count",
":attempt_status",
":best_id",
":best_name",
":best_price",
":best_score",
":error_message",
"CURRENT_TIMESTAMP",
]
params = {
"sku": attempt.get("sku"),
"source": source,
"momo_product_id": attempt.get("momo_product_id") or attempt.get("current_momo_product_id"),
@@ -201,7 +260,31 @@ def _insert_manual_attempt(conn, attempt: dict[str, Any], action_meta: dict[str,
"best_price": attempt.get("best_competitor_price"),
"best_score": attempt.get("best_match_score"),
"error_message": action_meta["message"],
})
}
if _has_table_column(conn, "competitor_match_attempts", "match_diagnostic_json"):
columns.append("match_diagnostic_json")
values.append(_json_object_expr(conn, "match_diagnostic_json"))
params["match_diagnostic_json"] = _json_text_payload(attempt.get("match_diagnostic_json"))
if _has_table_column(conn, "competitor_match_attempts", "comparison_mode"):
columns.append("comparison_mode")
values.append(":comparison_mode")
params["comparison_mode"] = attempt.get("comparison_mode")
if _has_table_column(conn, "competitor_match_attempts", "hard_veto"):
columns.append("hard_veto")
values.append(":hard_veto")
params["hard_veto"] = attempt.get("hard_veto")
if _has_table_column(conn, "competitor_match_attempts", "diagnostic_codes"):
columns.append("diagnostic_codes")
values.append(_json_array_expr(conn, "diagnostic_codes"))
params["diagnostic_codes"] = _json_text_payload(attempt.get("diagnostic_codes"))
conn.execute(text(f"""
INSERT INTO competitor_match_attempts
({", ".join(columns)})
VALUES
({", ".join(values)})
"""), params)
def _promote_manual_match(conn, attempt: dict[str, Any], source: str) -> None:
@@ -360,7 +443,7 @@ def record_competitor_match_review(
"candidate_name": attempt.get("best_competitor_product_name"),
"candidate_price": attempt.get("best_competitor_price"),
"candidate_score": attempt.get("best_match_score"),
"candidate_diagnostic": attempt.get("error_message"),
"candidate_diagnostic": _candidate_diagnostic_text(attempt),
"resulting_attempt_status": action_meta["attempt_status"],
})

6
templates/dashboard_v2.html
View File

@@ -229,6 +229,9 @@
{% if item.unit_comparison and item.unit_comparison.summary %}
<div class="dashboard-review-note momo-mono">{{ item.unit_comparison.summary }}</div>
{% endif %}
{% if item.unit_price_insight and item.unit_price_insight.summary %}
<div class="dashboard-review-note">{{ item.unit_price_insight.summary }}</div>
{% endif %}
<div class="dashboard-focus-row-links">
<a class="dashboard-platform-link is-momo momo-tracked-link" href="{{ build_momo_product_url(item.sku) if build_momo_product_url is defined else '#' }}" data-momo-original-url="{{ build_momo_product_url(item.sku) if build_momo_product_url is defined else '#' }}" target="_blank" rel="noopener noreferrer"
data-track-platform="momo"
@@ -639,6 +642,9 @@
{% if review.unit_comparison and review.unit_comparison.summary %}
<div class="dashboard-review-note momo-mono">{{ review.unit_comparison.summary }}</div>
{% endif %}
{% if review.unit_price_insight and review.unit_price_insight.summary %}
<div class="dashboard-review-note">{{ review.unit_price_insight.summary }}</div>
{% endif %}
<div class="dashboard-review-actions" aria-label="人工覆核決策">
{% if review.candidate_pc_id and review.candidate_pc_price %}
<button class="dashboard-review-action is-accept" type="button"

37
tests/test_competitor_intel_cache.py
View File

@@ -239,6 +239,43 @@ def test_rescore_accepted_review_item_has_actionable_decision_envelope():
assert any(evidence["metric"] == "candidate_gap_pct" for evidence in envelope["evidence"])
def test_manual_unit_price_review_item_keeps_business_insight():
from services.competitor_intel_repository import (
_format_competitor_review_item,
summarize_review_decision_envelopes,
)
item = _format_competitor_review_item({
"sku": "11223344",
"name": "理膚寶水 B5 全面修復霜 40ml x2 超值組",
"momo_price": 1199,
"attempt_status": "manual_unit_price_required",
"candidate_count": 1,
"best_competitor_product_id": "DABC01-B5",
"best_competitor_product_name": "理膚寶水 全面修復霜 B5 40ml",
"best_competitor_price": 679,
"best_match_score": 0.742,
"match_diagnostic_json": {
"match_type": "same_product_different_pack",
"price_basis": "unit_price",
"alert_tier": "unit_price_review",
"reasons": ["unit_comparable"],
},
})
assert item["unit_comparison"]["comparable"] is True
assert item["unit_price_insight"]["direction"] == "momo_cheaper"
assert "MOMO 單位價低" in item["unit_price_insight"]["summary"]
envelope = item["decision_envelope"]
assert envelope["recommended_action"]["action"] == "unit_price_required"
assert envelope["expected_impact"]["unit_price_insight"]["unit_gap_pct"] == -11.71
assert any(evidence["metric"] == "unit_price_gap_pct" for evidence in envelope["evidence"])
brief = summarize_review_decision_envelopes([item], limit=5)
assert "單位價差 -11.7%" in brief["text"]
assert brief["items"][0]["unit_price_gap_pct"] == -11.71
def test_protected_existing_match_envelope_explains_candidate_conflict():
from services.competitor_intel_repository import _format_competitor_review_item

107
tests/test_competitor_match_attempts_persistence.py
View File

@@ -509,6 +509,113 @@ def test_competitor_match_review_service_closes_human_review_loop():
assert "/api/pchome-review/" in dashboard_js
def test_manual_review_attempt_preserves_match_diagnostics_when_supported():
from sqlalchemy import create_engine, text
from services.competitor_match_review_service import record_competitor_match_review
engine = create_engine("sqlite:///:memory:")
diagnostic = {
"match_type": "same_product_different_pack",
"price_basis": "unit_price",
"alert_tier": "unit_price_review",
"reasons": ["unit_comparable"],
}
with engine.begin() as conn:
conn.execute(text("CREATE TABLE products (id INTEGER PRIMARY KEY, i_code TEXT, name TEXT)"))
conn.execute(text("CREATE TABLE price_records (id INTEGER PRIMARY KEY, product_id INTEGER, price NUMERIC, timestamp TEXT)"))
conn.execute(text("""
CREATE TABLE competitor_match_attempts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sku TEXT,
source TEXT,
momo_product_id INTEGER,
momo_product_name TEXT,
momo_price NUMERIC,
search_terms TEXT,
candidate_count INTEGER,
attempt_status TEXT,
best_competitor_product_id TEXT,
best_competitor_product_name TEXT,
best_competitor_price NUMERIC,
best_match_score NUMERIC,
error_message TEXT,
match_diagnostic_json TEXT,
comparison_mode TEXT,
hard_veto INTEGER,
diagnostic_codes TEXT,
attempted_at TEXT
)
"""))
conn.execute(text("""
CREATE TABLE competitor_prices (
sku TEXT,
source TEXT,
price NUMERIC,
competitor_product_id TEXT,
competitor_product_name TEXT,
match_score NUMERIC,
tags TEXT,
crawled_at TEXT,
expires_at TEXT
)
"""))
conn.execute(text("INSERT INTO products VALUES (1, 'A006', '理膚寶水 B5 全面修復霜 40ml x2')"))
conn.execute(text("INSERT INTO price_records VALUES (1, 1, 1199, '2026-05-20 09:00:00')"))
conn.execute(text("""
INSERT INTO competitor_match_attempts
(sku, source, momo_product_id, momo_product_name, momo_price,
search_terms, candidate_count, attempt_status,
best_competitor_product_id, best_competitor_product_name,
best_competitor_price, best_match_score, error_message,
match_diagnostic_json, comparison_mode, hard_veto, diagnostic_codes,
attempted_at)
VALUES
(:sku, 'pchome', 1, :momo_name, 1199,
'[]', 1, 'unit_comparable',
'DABC01-B5', :pc_name,
679, 0.742, 'score=0.742; reasons=unit_comparable',
:diagnostic, 'unit_comparable', 0, '["unit_comparable"]',
'2026-05-20 09:10:00')
"""), {
"sku": "A006",
"momo_name": "理膚寶水 B5 全面修復霜 40ml x2",
"pc_name": "理膚寶水 全面修復霜 B5 40ml",
"diagnostic": json.dumps(diagnostic, ensure_ascii=False),
})
result = record_competitor_match_review(
engine,
sku="A006",
review_action="unit_price_required",
reviewer_identity="pytest",
)
assert result["success"] is True
with engine.connect() as conn:
latest = conn.execute(text("""
SELECT attempt_status, match_diagnostic_json, comparison_mode, diagnostic_codes
FROM competitor_match_attempts
WHERE sku = 'A006'
ORDER BY id DESC
LIMIT 1
""")).mappings().first()
review_diagnostic = conn.execute(text("""
SELECT candidate_diagnostic
FROM competitor_match_reviews
WHERE sku = 'A006'
ORDER BY id DESC
LIMIT 1
""")).scalar_one()
payload = json.loads(latest["match_diagnostic_json"])
assert latest["attempt_status"] == "manual_unit_price_required"
assert latest["comparison_mode"] == "unit_comparable"
assert json.loads(latest["diagnostic_codes"]) == ["unit_comparable"]
assert payload["price_basis"] == "unit_price"
assert "match_diagnostic_json=" in review_diagnostic
assert "unit_comparable" in review_diagnostic
def test_reject_review_expires_current_formal_price():
from sqlalchemy import create_engine, text
from services.competitor_match_review_service import record_competitor_match_review