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

Route rescore matches to manual review

Browse Source
This commit is contained in:
OoO
2026-05-24 18:20:36 +08:00
parent 710f7216d0
commit 5aa2412dee
10 changed files with 322 additions and 27 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.443 補 PChome rescore 人工覆核入隊：`audit_competitor_match_attempt_rescore.py --apply-accepted` 只追加 `rescore_accepted_current` attempt 進人工覆核隊列，不直接寫 `competitor_prices` / `competitor_price_history`；商品看板新增「重算可採用」分流與狀態文案，讓可救回候選先由人審確認再正式更新價差。
- V10.442 降噪 `/cicd` 舊 GitLab 探測：沒有明確啟用 `GITLAB_ENABLED=true` 與 token 時，不再打退役的 `192.168.0.110:8929` 或 SSH fallback，正式 responsive smoke 造訪 `/cicd` 只呈現空 pipeline 狀態，不污染 app logs。
- V10.441 補 PChome matcher re-score audit 與商品看板原因標籤：新增 read-only `competitor_match_attempt_rescore_audit` / `scripts/audit_competitor_match_attempt_rescore.py`，可用最新版 matcher 重新分類既有 `competitor_match_attempts`，預設不寫 DB、不更新正式價格；商品看板同步補蘭蔻/達特醫/hoi/Saugella/Lactacyd 等 focused matcher reason 中文標籤，讓「待對比」能拆成商品線不符、款式版本不符、可回刷或仍低信心。
- V10.439 收斂外部 BI / 資料協作入口：`/metabase`、`/grist` 正式頁維持 momo-pro 內部診斷 bridge，`.env.example` 與 bi profile Grist 預設改回 `https://mo.wooo.work/grist` / `GRIST_APP_HOME_URL`，並補測試禁止 `grist.wooo.work` / `awoooi` 回流到導覽設定；外部工具頁標題字級改用新版 token 與手機 media query。

2
config.py
View File

@@ -325,7 +325,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
SYSTEM_VERSION = "V10.442"
SYSTEM_VERSION = "V10.443"
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-05-24 (台北時間)
> **狀態**: 🟢 四 AI Agent 自動化閉環已落地；LLM 路由紅線升級為 Ollama-first 三主機級聯，Gemini 備援預設關閉
> **適用版本**: V10.442
> **適用版本**: V10.443
---

1
docs/memory/history_logs.md
View File

@@ -13,6 +13,7 @@
## 📅 詳細更新日誌 (考古存檔)
### 2026-05-24：PChome 近門檻身份回收第二輪
- **V10.443 PChome rescore 人工覆核入隊**: `scripts/audit_competitor_match_attempt_rescore.py --apply-accepted` 只會把最新版 matcher 已通過門檻的舊低信心候選追加為 `rescore_accepted_current` attempt，進入商品看板人工覆核隊列；不寫 `competitor_prices`、不寫 `competitor_price_history`，必須由操作員按「採用同款」後才正式更新 PChome 價差。Dashboard 補上「重算可採用待審」分流與狀態文案，避免安全回刷候選混在一般低信心項目裡。
- **V10.442 CI/CD legacy GitLab 探測降噪**: `/cicd` 舊 GitLab pipeline API 預設改為 disabled，除非明確設定 `GITLAB_ENABLED=true` 並提供 `GITLAB_TOKEN`，否則不再打 `192.168.0.110:8929` 或 SSH fallback；正式 responsive smoke 造訪 `/cicd` 時只呈現可診斷空狀態，不再把已退役 GitLab endpoint 的 connection refused / permission denied 寫成錯誤噪音。
- **V10.441 PChome matcher re-score audit**: 新增 read-only `competitor_match_attempt_rescore_audit` 服務與 `scripts/audit_competitor_match_attempt_rescore.py`，可針對既有 `competitor_match_attempts` 用最新版 matcher 重新分類成 `accepted_current` / `unit_comparable_current` / `identity_veto_current` / `low_score_current`，預設不寫 DB、不更新正式價格；商品看板同步補蘭蔻/達特醫/hoi/Saugella/Lactacyd 等 focused matcher reason 中文標籤。正式抽樣中 31 筆舊 `strong_exact_spec_match` 低信心資料，最新版 matcher 可讀出 10 筆 gate pass、1 筆單位價、11 筆 hard veto、9 筆仍低信心，作為後續人工覆核與批次回刷前的安全量化。
- **V10.440 Mustela 爽身潤膚乳同款 anchor**: marketplace matcher 新增 `慕之幼爽身潤膚乳` identity anchor，並讓標題中插入「加量版」時仍可抽出同一身份詞；正式樣本 `【Mustela 慕之恬廊】慕之幼 加量版爽身潤膚乳 500mlX2入` vs `【慕之恬廊】慕之幼爽身潤膚乳(500毫升X2入)` 由 0.741 提升到 0.801，維持 `hard_veto=false`、人工 review 型態，不放寬全域門檻、不寫正式 `competitor_prices`。

12
routes/dashboard_routes.py
View File

@@ -61,8 +61,10 @@ REVIEW_STATUS_OPTIONS = [
'manual_rejected',
'manual_unit_price_required',
'manual_needs_research',
'rescore_accepted_current',
),
},
{'key': 'rescore_accepted', 'label': '重算可採用', 'statuses': ('rescore_accepted_current',)},
{
'key': 'unit_comparable',
'label': '需單位價',
@@ -147,6 +149,16 @@ def _diagnostic_match_rejection_label(diagnostic_text, score_text, *, blocked=Tr
def _build_pchome_match_status(attempt=None, ineligible=None):
if attempt:
status = attempt.get('attempt_status') or 'unknown'
if status == 'rescore_accepted_current':
score = _to_float(attempt.get('best_match_score'))
score_text = f"目前信心 {int(score * 100)}%" if score is not None else '目前信心待補'
return {
'label': '重算可採用待審',
'tone': 'watch',
'score': score,
'summary': f'{score_text}，最新版 matcher 已通過身份門檻；仍需人工採用後才寫入正式 PChome 價差',
'detail': attempt.get('error_message')
}
if status == 'manual_accepted':
score = _to_float(attempt.get('best_match_score'))
score_text = f"人工採用 {round(score * 100)}%" if score is not None else "人工已採用同款"

48
scripts/audit_competitor_match_attempt_rescore.py
View File

@@ -19,6 +19,8 @@ if str(ROOT) not in sys.path:
from services.competitor_match_attempt_rescore_audit import ( # noqa: E402
DEFAULT_RESCAN_STATUSES,
build_match_attempt_rescore_audit,
fetch_match_attempt_rescore_rows,
materialize_rescore_accept_reviews,
summarize_match_attempt_rescore,
)
from services.competitor_price_feeder import MIN_MATCH_SCORE # noqa: E402
@@ -59,10 +61,17 @@ def main(argv: list[str] | None = None) -> int:
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--sample-limit", type=int, default=20)
parser.add_argument("--min-score", type=float, default=MIN_MATCH_SCORE)
parser.add_argument(
"--apply-accepted",
action="store_true",
help="Append accepted-current rows to competitor_match_attempts for manual review; never writes competitor_prices.",
)
args = parser.parse_args(argv)
statuses = tuple(args.statuses or DEFAULT_RESCAN_STATUSES)
if args.input:
if args.apply_accepted:
parser.error("--apply-accepted requires DB mode; do not combine it with --input.")
rows = [row for row in _read_jsonl(args.input) if not row.get("_invalid_json")]
summary = summarize_match_attempt_rescore(
rows,
@@ -73,15 +82,36 @@ def main(argv: list[str] | None = None) -> int:
from config import DATABASE_PATH
engine = create_engine(DATABASE_PATH)
summary = build_match_attempt_rescore_audit(
engine,
source=args.source,
statuses=statuses,
reason_filter=args.reason_filter or None,
limit=args.limit,
min_score=args.min_score,
sample_limit=args.sample_limit,
)
if args.apply_accepted:
with engine.begin() as conn:
rows = fetch_match_attempt_rescore_rows(
conn,
source=args.source,
statuses=statuses,
reason_filter=args.reason_filter or None,
limit=args.limit,
)
summary = summarize_match_attempt_rescore(
rows,
min_score=args.min_score,
sample_limit=args.sample_limit,
)
summary["materialize"] = materialize_rescore_accept_reviews(
conn,
rows,
source=args.source,
min_score=args.min_score,
)
else:
summary = build_match_attempt_rescore_audit(
engine,
source=args.source,
statuses=statuses,
reason_filter=args.reason_filter or None,
limit=args.limit,
min_score=args.min_score,
sample_limit=args.sample_limit,
)
print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))
return 0

39
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"}
ACTIONABLE_ATTEMPT_STATUSES = {
"rescore_accepted_current",
"unit_comparable",
"refresh_unit_comparable",
"identity_veto",
@@ -41,6 +42,7 @@ ACTIONABLE_ATTEMPT_STATUSES = {
"manual_needs_research",
}
REVIEW_STATUS_FILTER_GROUPS = {
"rescore_accepted": ("rescore_accepted_current",),
"unit_comparable": ("unit_comparable", "refresh_unit_comparable", "manual_unit_price_required"),
"identity_veto": ("identity_veto",),
"low_score": ("low_score", "refresh_low_score", "recoverable_low_score", "true_low_confidence"),
@@ -50,6 +52,7 @@ REVIEW_STATUS_FILTER_GROUPS = {
"manual_closed": ("manual_rejected", "manual_unit_price_required", "manual_needs_research"),
}
ATTEMPT_STATUS_LABELS = {
"rescore_accepted_current": "重算可採用待審",
"unit_comparable": "需單位價比較",
"refresh_unit_comparable": "需單位價比較",
"identity_veto": "身份否決",
@@ -68,6 +71,7 @@ ATTEMPT_STATUS_LABELS = {
"manual_needs_research": "人工要求補搜尋",
}
ATTEMPT_ACTION_LABELS = {
"rescore_accepted_current": "覆核後可人工採用同款",
"unit_comparable": "人工確認檔期、贈品與單位價",
"refresh_unit_comparable": "人工確認檔期、贈品與單位價",
"identity_veto": "確認是否為不同商品線或規格",
@@ -780,7 +784,7 @@ def fetch_competitor_review_queue(engine, limit: int = 12) -> list[dict]:
"""可行動的 PChome 比對覆核隊列，供 Dashboard / AI / PPT 共用。"""
limit = max(1, min(int(limit or 12), 50))
return _cached_payload(
f"review_queue:v1:limit={limit}:floor={PCHOME_MATCH_SCORE_FLOOR}",
f"review_queue:v2:limit={limit}:floor={PCHOME_MATCH_SCORE_FLOOR}",
lambda: _fetch_competitor_review_queue_uncached(engine, limit=limit),
)
@@ -802,7 +806,7 @@ def fetch_competitor_review_queue_page(
if status_filter not in REVIEW_STATUS_FILTER_GROUPS:
status_filter = ""
cache_key = (
"review_queue_page:v1:"
"review_queue_page:v2:"
f"page={page}:per={per_page}:q={search_query.lower()}:cat={category}:"
f"status={status_filter}:"
f"floor={PCHOME_MATCH_SCORE_FLOOR}"
@@ -904,13 +908,14 @@ def _review_queue_cte_and_filter(
la.error_message,
la.attempted_at,
CASE
WHEN la.attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 0
WHEN la.attempt_status = 'identity_veto' THEN 1
WHEN la.attempt_status IN ('recoverable_low_score', 'low_score', 'refresh_low_score') THEN 2
WHEN la.attempt_status = 'protected_existing_match' THEN 3
WHEN la.attempt_status = 'true_low_confidence' THEN 4
WHEN la.attempt_status = 'expired_match' THEN 5
ELSE 6
WHEN la.attempt_status = 'rescore_accepted_current' THEN 0
WHEN la.attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 1
WHEN la.attempt_status = 'identity_veto' THEN 2
WHEN la.attempt_status IN ('recoverable_low_score', 'low_score', 'refresh_low_score') THEN 3
WHEN la.attempt_status = 'protected_existing_match' THEN 4
WHEN la.attempt_status = 'true_low_confidence' THEN 5
WHEN la.attempt_status = 'expired_match' THEN 6
ELSE 7
END AS priority_rank
FROM latest_momo lm
JOIN latest_attempt la ON la.sku = lm.sku
@@ -1057,6 +1062,7 @@ def _fetch_competitor_review_queue_uncached(engine, limit: int = 12) -> list[dic
LEFT JOIN valid_competitor vc ON vc.sku = lm.sku
WHERE vc.sku IS NULL
AND la.attempt_status IN (
'rescore_accepted_current',
'unit_comparable',
'refresh_unit_comparable',
'identity_veto',
@@ -1071,13 +1077,14 @@ def _fetch_competitor_review_queue_uncached(engine, limit: int = 12) -> list[dic
)
ORDER BY
CASE
WHEN la.attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 0
WHEN la.attempt_status = 'identity_veto' THEN 1
WHEN la.attempt_status IN ('recoverable_low_score', 'low_score', 'refresh_low_score') THEN 2
WHEN la.attempt_status = 'protected_existing_match' THEN 3
WHEN la.attempt_status = 'true_low_confidence' THEN 4
WHEN la.attempt_status = 'expired_match' THEN 5
ELSE 6
WHEN la.attempt_status = 'rescore_accepted_current' THEN 0
WHEN la.attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 1
WHEN la.attempt_status = 'identity_veto' THEN 2
WHEN la.attempt_status IN ('recoverable_low_score', 'low_score', 'refresh_low_score') THEN 3
WHEN la.attempt_status = 'protected_existing_match' THEN 4
WHEN la.attempt_status = 'true_low_confidence' THEN 5
WHEN la.attempt_status = 'expired_match' THEN 6
ELSE 7
END,
lm.momo_price DESC NULLS LAST,
la.best_match_score DESC NULLS LAST,

162
services/competitor_match_attempt_rescore_audit.py
View File

@@ -6,6 +6,7 @@ from __future__ import annotations
from collections import Counter
from dataclasses import asdict, dataclass
import json
from typing import Any, Iterable, Sequence
from sqlalchemy import bindparam, text
@@ -20,6 +21,7 @@ DEFAULT_RESCAN_STATUSES = (
"low_score",
"refresh_low_score",
)
RESCORE_ACCEPTED_CURRENT_STATUS = "rescore_accepted_current"
@dataclass(frozen=True)
@@ -174,6 +176,156 @@ def summarize_match_attempt_rescore(
}
def _json_expr(conn, bind_name: str) -> str:
return f"CAST(:{bind_name} AS jsonb)" if conn.dialect.name == "postgresql" else f":{bind_name}"
def _diagnostic_payload(decision: MatchAttemptRescoreDecision) -> dict[str, Any]:
return {
"rescore_version": "current_matcher_v1",
"stored_status": decision.stored_status,
"stored_score": decision.stored_score,
"score": decision.current_score,
"hard_veto": decision.hard_veto,
"comparison_mode": decision.comparison_mode,
"match_type": decision.match_type,
"price_basis": decision.price_basis,
"alert_tier": decision.alert_tier,
"reasons": decision.reasons,
}
def _diagnostic_text(decision: MatchAttemptRescoreDecision) -> str:
reasons = ",".join(decision.reasons or [])
return (
"matcher_rescore=accepted_current; "
f"stored_status={decision.stored_status}; "
f"stored_score={decision.stored_score}; "
f"current_score={decision.current_score}; "
f"mode={decision.comparison_mode}; "
f"price_basis={decision.price_basis}; "
f"reasons={reasons}"
)
def _ensure_attempt_table(conn) -> None:
from services.competitor_price_feeder import CompetitorPriceFeeder
CompetitorPriceFeeder(engine=None)._ensure_competitor_match_attempts_table(conn)
def _already_materialized(conn, *, source: str, sku: str, candidate_id: str) -> bool:
row = conn.execute(text("""
SELECT 1
FROM competitor_match_attempts
WHERE sku = :sku
AND source = :source
AND attempt_status = :attempt_status
AND COALESCE(best_competitor_product_id, '') = :candidate_id
LIMIT 1
"""), {
"sku": sku,
"source": source,
"attempt_status": RESCORE_ACCEPTED_CURRENT_STATUS,
"candidate_id": candidate_id,
}).first()
return row is not None
def materialize_rescore_accept_reviews(
conn,
rows: Iterable[dict[str, Any]],
*,
source: str = "pchome",
min_score: float = MIN_MATCH_SCORE,
) -> dict[str, Any]:
"""Append accepted-current rescore rows to the manual review queue.
This never writes ``competitor_prices``. It only creates a latest
``competitor_match_attempts`` row that Dashboard can route through the
existing human review buttons.
"""
_ensure_attempt_table(conn)
stats = {
"scanned": 0,
"materialized": 0,
"skipped_not_gate_pass": 0,
"skipped_missing_candidate": 0,
"skipped_duplicate": 0,
"samples": [],
}
search_terms_expr = _json_expr(conn, "search_terms")
diagnostic_json_expr = _json_expr(conn, "match_diagnostic_json")
diagnostic_codes_expr = _json_expr(conn, "diagnostic_codes")
for row in rows:
stats["scanned"] += 1
decision = classify_match_attempt_row(row, min_score=min_score)
if not decision.gate_pass:
stats["skipped_not_gate_pass"] += 1
continue
candidate_id = str(row.get("best_competitor_product_id") or "").strip()
candidate_name = str(row.get("best_competitor_product_name") or "").strip()
candidate_price = _to_float(row.get("best_competitor_price"))
if not candidate_id or not candidate_name or not candidate_price or candidate_price <= 0:
stats["skipped_missing_candidate"] += 1
continue
if _already_materialized(conn, source=source, sku=decision.sku, candidate_id=candidate_id):
stats["skipped_duplicate"] += 1
continue
diagnostic_payload = _diagnostic_payload(decision)
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, competitor_product_url, competitor_image_url,
competitor_stock, match_diagnostic_json, comparison_mode,
hard_veto, diagnostic_codes)
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, :competitor_product_url, :competitor_image_url,
:competitor_stock, {diagnostic_json_expr}, :comparison_mode,
:hard_veto, {diagnostic_codes_expr})
"""), {
"sku": decision.sku,
"source": source,
"momo_product_id": row.get("momo_product_id"),
"momo_product_name": decision.momo_product_name,
"momo_price": _to_float(row.get("momo_price")),
"search_terms": json.dumps([
"matcher_rescore:accepted_current",
f"stored_status:{decision.stored_status}",
], ensure_ascii=False),
"candidate_count": int(row.get("candidate_count") or 1),
"attempt_status": RESCORE_ACCEPTED_CURRENT_STATUS,
"best_id": candidate_id,
"best_name": candidate_name[:300],
"best_price": candidate_price,
"best_score": decision.current_score,
"error_message": _diagnostic_text(decision)[:1000],
"competitor_product_url": row.get("competitor_product_url"),
"competitor_image_url": row.get("competitor_image_url"),
"competitor_stock": row.get("competitor_stock"),
"match_diagnostic_json": json.dumps(diagnostic_payload, ensure_ascii=False),
"comparison_mode": decision.comparison_mode,
"hard_veto": False,
"diagnostic_codes": json.dumps(decision.reasons, ensure_ascii=False),
})
stats["materialized"] += 1
if len(stats["samples"]) < 10:
stats["samples"].append(decision.to_dict())
return stats
def fetch_match_attempt_rescore_rows(
conn,
*,
@@ -191,12 +343,17 @@ def fetch_match_attempt_rescore_rows(
SELECT DISTINCT ON (sku, best_competitor_product_id)
sku,
attempt_status,
momo_product_id,
momo_product_name,
momo_price,
candidate_count,
best_competitor_product_id,
best_competitor_product_name,
best_competitor_price,
best_match_score,
competitor_product_url,
competitor_image_url,
competitor_stock,
diagnostic_codes,
attempted_at
FROM competitor_match_attempts
@@ -213,12 +370,17 @@ def fetch_match_attempt_rescore_rows(
SELECT
sku,
attempt_status,
momo_product_id,
momo_product_name,
momo_price,
candidate_count,
best_competitor_product_id,
best_competitor_product_name,
best_competitor_price,
best_match_score,
competitor_product_url,
competitor_image_url,
competitor_stock,
diagnostic_codes,
attempted_at,
ROW_NUMBER() OVER (

14
tests/test_competitor_identity_revalidator.py
View File

@@ -145,6 +145,20 @@ def test_dashboard_match_status_shows_manual_review_closure_states():
assert decision["gap_amount"] is None
def test_dashboard_match_status_shows_rescore_accepted_review_state():
from routes.dashboard_routes import _build_pchome_match_status
status = _build_pchome_match_status({
"attempt_status": "rescore_accepted_current",
"best_match_score": 0.801,
"error_message": "matcher_rescore=accepted_current; reasons=strong_exact_spec_match",
})
assert status["label"] == "重算可採用待審"
assert status["tone"] == "watch"
assert "人工採用後才寫入正式 PChome 價差" in status["summary"]
def test_dashboard_match_status_uses_specific_matcher_reason_labels():
from routes.dashboard_routes import _build_pchome_match_status

68
tests/test_competitor_match_attempt_rescore_audit.py
View File

@@ -1,3 +1,8 @@
import json
from sqlalchemy import create_engine, text
def test_match_attempt_rescore_audit_classifies_current_gate_pass_and_veto():
from services.competitor_match_attempt_rescore_audit import summarize_match_attempt_rescore
@@ -61,3 +66,66 @@ def test_match_attempt_rescore_audit_skips_missing_identity_text():
assert decision.suggested_status == "skipped_missing_identity_text"
assert decision.current_score is None
assert decision.gate_pass is False
def test_match_attempt_rescore_materializes_accepted_current_for_manual_review():
from services.competitor_match_attempt_rescore_audit import materialize_rescore_accept_reviews
engine = create_engine("sqlite:///:memory:")
rows = [{
"sku": "9031334",
"attempt_status": "true_low_confidence",
"momo_product_id": 9031334,
"momo_product_name": "【Mustela 慕之恬廊】慕之幼 加量版爽身潤膚乳 500mlX2入(寶寶 嬰兒乳液 公司貨 台灣獨家總代理)",
"momo_price": 1390,
"candidate_count": 1,
"best_competitor_product_id": "DDAXS-A900IHGY",
"best_competitor_product_name": "【慕之恬廊】慕之幼爽身潤膚乳(500毫升X2入)",
"best_competitor_price": 1210,
"best_match_score": 0.709,
}]
with engine.begin() as conn:
stats = materialize_rescore_accept_reviews(conn, rows)
duplicate_stats = materialize_rescore_accept_reviews(conn, rows)
stored = conn.execute(text("""
SELECT attempt_status, best_match_score, search_terms,
match_diagnostic_json, diagnostic_codes, error_message
FROM competitor_match_attempts
""")).mappings().all()
assert stats["materialized"] == 1
assert duplicate_stats["skipped_duplicate"] == 1
assert len(stored) == 1
assert stored[0]["attempt_status"] == "rescore_accepted_current"
assert float(stored[0]["best_match_score"]) >= 0.76
assert "matcher_rescore:accepted_current" in json.loads(stored[0]["search_terms"])
assert "shared_identity_anchor_reordered_line" in json.loads(stored[0]["diagnostic_codes"])
payload = json.loads(stored[0]["match_diagnostic_json"])
assert payload["stored_status"] == "true_low_confidence"
assert payload["price_basis"] == "manual_review"
assert "matcher_rescore=accepted_current" in stored[0]["error_message"]
def test_match_attempt_rescore_materialize_does_not_enqueue_veto():
from services.competitor_match_attempt_rescore_audit import materialize_rescore_accept_reviews
engine = create_engine("sqlite:///:memory:")
rows = [{
"sku": "12534698",
"attempt_status": "true_low_confidence",
"momo_product_name": "【LANCOME 蘭蔻】官方直營 超極光活粹晶露150ml(LANCOME/四重酸極光水/化妝水/精華水)",
"best_competitor_product_id": "LANCOME-BAD",
"best_competitor_product_name": "LANCOME 蘭蔻 超極限肌因精華露150ml 專櫃公司貨",
"momo_price": 3555,
"best_competitor_price": 1880,
"best_match_score": 0.748,
}]
with engine.begin() as conn:
stats = materialize_rescore_accept_reviews(conn, rows)
row_count = conn.execute(text("SELECT COUNT(*) FROM competitor_match_attempts")).scalar()
assert stats["materialized"] == 0
assert stats["skipped_not_gate_pass"] == 1
assert row_count == 0