549 lines
26 KiB
Python
549 lines
26 KiB
Python
from pathlib import Path
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def test_competitor_intel_cache_reuses_memory_and_shared_file(tmp_path, monkeypatch):
|
||
from services import competitor_intel_repository as repo
|
||
|
||
monkeypatch.setattr(repo, "_CACHE_FILE", Path(tmp_path) / "competitor_intel_cache.pkl")
|
||
repo._MEM_CACHE.clear()
|
||
|
||
calls = {"count": 0}
|
||
|
||
def producer():
|
||
calls["count"] += 1
|
||
return {"valid_matches": 7, "match_rate": 0.1}
|
||
|
||
first = repo._cached_payload("coverage:test", producer, ttl_seconds=60)
|
||
second = repo._cached_payload("coverage:test", producer, ttl_seconds=60)
|
||
repo._MEM_CACHE.clear()
|
||
third = repo._cached_payload("coverage:test", producer, ttl_seconds=60)
|
||
|
||
assert first == second == third == {"valid_matches": 7, "match_rate": 0.1}
|
||
assert calls["count"] == 1
|
||
|
||
|
||
def test_clear_competitor_intel_cache_removes_shared_file(tmp_path, monkeypatch):
|
||
from services import competitor_intel_repository as repo
|
||
|
||
cache_file = Path(tmp_path) / "competitor_intel_cache.pkl"
|
||
monkeypatch.setattr(repo, "_CACHE_FILE", cache_file)
|
||
repo._MEM_CACHE["x"] = {"time": 1, "value": {"ok": True}}
|
||
cache_file.write_bytes(b"stale")
|
||
|
||
repo.clear_competitor_intel_cache()
|
||
|
||
assert repo._MEM_CACHE == {}
|
||
assert not cache_file.exists()
|
||
|
||
|
||
def test_competitor_ppt_results_keep_pending_diagnostics_in_export():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
|
||
assert "LEFT JOIN valid_competitor" in source
|
||
assert "\"found\": found" in source
|
||
assert "\"match_status\"" in source
|
||
assert "\"candidate_count\"" in source
|
||
assert "\"unit_comparison\"" in source
|
||
assert "build_unit_price_comparison" in source
|
||
assert "(vc.pchome_price IS NULL)" in source
|
||
assert "(lm.momo_price - vc.pchome_price) AS price_diff" in source
|
||
assert "((lm.momo_price - vc.pchome_price) / vc.pchome_price * 100) AS price_diff_pct" in source
|
||
|
||
|
||
def test_competitor_ppt_results_use_history_for_dated_reports():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
route_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
|
||
comparison_source = source.split("def fetch_competitor_comparison_results", 1)[1].split(
|
||
"def build_competitor_intel_payload", 1
|
||
)[0]
|
||
|
||
assert "requested_historical_prices" in source
|
||
assert "use_history_prices" in source
|
||
assert "FROM competitor_price_history cph" in source
|
||
assert "cph.crawled_at >= DATE(:start_date)" in source
|
||
assert "cph.crawled_at < DATE(:end_date) + INTERVAL '1 day'" in source
|
||
assert "pr.timestamp < DATE(:end_date) + INTERVAL '1 day'" in source
|
||
assert "FROM products p\n JOIN LATERAL" in comparison_source
|
||
assert "WHERE pr.product_id = p.id" in comparison_source
|
||
assert "ORDER BY pr.timestamp DESC, pr.id DESC" in comparison_source
|
||
assert "ROW_NUMBER() OVER (PARTITION BY p.id" not in comparison_source
|
||
assert "WHERE lm.rn = 1" not in comparison_source
|
||
assert "'competitor_price_history' AS competitor_source" in source
|
||
assert "\"competitor_source\"" in source
|
||
assert "\"pc_crawled_at\"" in source
|
||
assert "指定期間 competitor_price_history" in route_source
|
||
|
||
|
||
def test_competitor_coverage_counts_only_active_product_intersection():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
coverage_source = source.split("def _fetch_competitor_coverage_uncached", 1)[1].split(
|
||
"def _fetch_manual_review_summary", 1
|
||
)[0]
|
||
|
||
assert "coverage:v14" in source
|
||
assert "CATALOG_COMPARABLE_SCORE_FLOOR" in source
|
||
assert "CATALOG_IDENTITY_REVIEW_SCORE_FLOOR" in source
|
||
assert "rescore_accepted_count" in coverage_source
|
||
assert "(SELECT COUNT(*) FROM valid_competitor) AS valid_matches" not in coverage_source
|
||
assert "identity_competitor AS" in coverage_source
|
||
assert "fresh_competitor AS" in coverage_source
|
||
assert "unknown_freshness_competitor AS" in coverage_source
|
||
assert "WHERE expires_at > CURRENT_TIMESTAMP" in coverage_source
|
||
assert "WHERE expires_at IS NULL" in coverage_source
|
||
assert "FROM latest_momo lm\n JOIN identity_competitor ic ON ic.sku = lm.sku" in coverage_source
|
||
assert "LEFT JOIN fresh_competitor fc ON fc.sku = lm.sku" in coverage_source
|
||
assert "WHERE fc.sku IS NULL" in coverage_source
|
||
assert "\"fresh_matches\": fresh" in coverage_source
|
||
assert "\"stale_matches\": stale" in coverage_source
|
||
assert "\"unknown_freshness_matches\": unknown_freshness" in coverage_source
|
||
assert "\"not_decision_ready_count\": pending + stale + unknown_freshness" in coverage_source
|
||
assert "\"decision_ready_matches\": fresh" in coverage_source
|
||
assert "\"decision_ready_rate\": round(fresh / max(active, 1) * 100, 1)" in coverage_source
|
||
assert "\"decision_support_count\": decision_support_count" in coverage_source
|
||
assert "\"decision_support_rate\": round(decision_support_count / max(active, 1) * 100, 1)" in coverage_source
|
||
assert "\"catalog_comparable_count\": catalog_comparable_count" in coverage_source
|
||
assert "_catalog_comparable_sql(\"la\")" in coverage_source
|
||
assert "_catalog_review_lane_sql(\"la\", \"catalog_variant_review\")" in coverage_source
|
||
assert "_catalog_review_lane_sql(\"la\", \"catalog_unit_review\")" in coverage_source
|
||
assert "_catalog_review_lane_sql(\"la\", \"catalog_identity_review\")" in coverage_source
|
||
assert "\"catalog_review_plan\": {" in coverage_source
|
||
assert "CATALOG_COMPARABLE_SIGNAL_REASONS" in source
|
||
assert "CATALOG_VARIANT_REVIEW_REASONS" in source
|
||
assert "CATALOG_UNIT_REVIEW_REASONS" in source
|
||
assert "CATALOG_COMPARABLE_IDENTITY_REASONS" in source
|
||
assert "\"catalog_identity_review_score_floor\": CATALOG_IDENTITY_REVIEW_SCORE_FLOOR" in coverage_source
|
||
|
||
|
||
def test_catalog_comparable_sql_includes_high_confidence_identity_review_lane():
|
||
from services.competitor_intel_repository import (
|
||
CATALOG_COMPARABLE_SCORE_FLOOR,
|
||
CATALOG_IDENTITY_REVIEW_SCORE_FLOOR,
|
||
_catalog_comparable_sql,
|
||
)
|
||
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
coverage_source = source.split("def _fetch_competitor_coverage_uncached", 1)[1].split(
|
||
"def _fetch_manual_review_summary", 1
|
||
)[0]
|
||
sql = _catalog_comparable_sql("la")
|
||
|
||
assert f">= {CATALOG_COMPARABLE_SCORE_FLOOR}" in sql
|
||
assert f">= {CATALOG_IDENTITY_REVIEW_SCORE_FLOOR}" in sql
|
||
assert "CATALOG_IDENTITY_REVIEW_SCORE_FLOOR" not in sql
|
||
assert "OR COALESCE(la.best_match_score, 0)" in sql
|
||
assert "CATALOG_COMPARABLE_SIGNAL_REASONS" not in sql
|
||
assert "CATALOG_COMPARABLE_BLOCK_REASONS" in source
|
||
assert "\"identity_coverage_matches\": valid" in coverage_source
|
||
assert "\"manual_closed_count\": manual_closed_count" in coverage_source
|
||
assert "\"last_decision_ready_crawled_at\": last_decision_ready_crawled_at" in coverage_source
|
||
assert "FROM products p\n JOIN LATERAL" in coverage_source
|
||
assert "WHERE p.status = 'ACTIVE'" in coverage_source
|
||
|
||
|
||
def test_competitor_decision_consumers_require_explicit_freshness():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
|
||
assert "(cp.expires_at IS NULL OR cp.expires_at > CURRENT_TIMESTAMP)" not in source
|
||
assert source.count("AND cp.expires_at > CURRENT_TIMESTAMP") >= 4
|
||
|
||
|
||
def test_competitor_ppt_and_ai_use_momo_minus_pchome_gap_direction():
|
||
ppt_source = (ROOT / "services" / "ppt_generator.py").read_text(encoding="utf-8")
|
||
route_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
|
||
|
||
assert 'pc_wins = [r for r in found if r.get("price_diff", 0) > 10]' in ppt_source
|
||
assert 'mo_wins = [r for r in found if r.get("price_diff", 0) < -10]' in ppt_source
|
||
assert "平均價差 {avg_pct:+.1f}%（momo - PChome）" in ppt_source
|
||
assert "price_diff = MOMO售價 − PChome售價" in route_source
|
||
assert "正值=MOMO較貴、PChome低價壓力" in route_source
|
||
|
||
|
||
def test_competitor_review_queue_is_canonical_unit_price_handoff():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
daily_template = (ROOT / "templates" / "daily_sales.html").read_text(encoding="utf-8")
|
||
growth_template = (ROOT / "templates" / "growth_analysis.html").read_text(encoding="utf-8")
|
||
|
||
assert "def fetch_competitor_review_queue" in source
|
||
assert "review_queue = fetch_competitor_review_queue" in source
|
||
assert "\"review_queue\": review_queue" in source
|
||
assert "\"unit_comparable_count\"" in source
|
||
assert "\"rescore_accepted_count\"" in source
|
||
assert "manual_review_summary" in source
|
||
assert "manual_accept_count" in source
|
||
assert "manual_reject_count" in source
|
||
assert "manual_unit_price_count" in source
|
||
assert "manual_rejected" in source
|
||
assert "manual_unit_price_required" in source
|
||
assert "manual_needs_research" in source
|
||
assert "manual_closed" in source
|
||
assert "competitor_match_reviews" in source
|
||
assert "\"status_label\"" in source
|
||
assert "\"action_label\"" in source
|
||
assert "build_unit_price_comparison" in source
|
||
assert "需單位價覆核" in daily_template
|
||
assert "人工採用" in daily_template
|
||
assert "人工否決" in daily_template
|
||
assert "人工單位價" in daily_template
|
||
assert "competitor_intel.review_queue" in daily_template
|
||
assert "coverage.fresh_matches" in growth_template
|
||
assert "coverage.fresh_match_rate" in growth_template
|
||
assert "coverage.decision_ready_rate" in growth_template
|
||
assert "coverage.decision_support_rate" in growth_template
|
||
assert "coverage.catalog_comparable_count" in growth_template
|
||
assert "型錄/任選可比" in growth_template
|
||
assert "coverage.stale_matches" in growth_template
|
||
assert "coverage.unknown_freshness_matches" in growth_template
|
||
assert "未形成有效身份配對" in growth_template
|
||
assert "coverage.unit_comparable_count" in growth_template
|
||
assert "coverage.rescore_accepted_count" in growth_template
|
||
assert "重算待人工覆核" in growth_template
|
||
assert "coverage.manual_accept_count" in growth_template
|
||
assert "coverage.manual_reject_count" in growth_template
|
||
assert "coverage.manual_unit_price_count" in growth_template
|
||
assert "comp_coverage.rescore_accepted_count" in daily_template
|
||
assert "重算待人工覆核" in daily_template
|
||
assert "comp_coverage.decision_support_rate" in daily_template
|
||
assert "comp_coverage.catalog_comparable_count" in daily_template
|
||
assert "精準可告警覆蓋" in daily_template
|
||
assert "comp_coverage.stale_matches" in daily_template
|
||
assert "comp_coverage.unknown_freshness_matches" in daily_template
|
||
assert "comp_coverage.decision_ready_rate" in daily_template
|
||
|
||
|
||
def test_competitor_review_filters_split_low_score_operational_buckets():
|
||
from services.competitor_intel_repository import REVIEW_STATUS_FILTER_GROUPS
|
||
|
||
assert REVIEW_STATUS_FILTER_GROUPS["catalog_comparable"] == (
|
||
"true_low_confidence",
|
||
"catalog_variant_review",
|
||
"catalog_unit_review",
|
||
"catalog_identity_review",
|
||
)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["catalog_variant_review"] == (
|
||
"true_low_confidence",
|
||
"catalog_variant_review",
|
||
)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["catalog_unit_review"] == (
|
||
"true_low_confidence",
|
||
"catalog_unit_review",
|
||
)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["catalog_identity_review"] == (
|
||
"true_low_confidence",
|
||
"catalog_identity_review",
|
||
)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["recoverable_low_score"] == ("recoverable_low_score",)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["true_low_confidence"] == ("true_low_confidence",)
|
||
assert REVIEW_STATUS_FILTER_GROUPS["legacy_low_score"] == ("low_score", "refresh_low_score")
|
||
assert set(REVIEW_STATUS_FILTER_GROUPS["low_score"]) == {
|
||
"low_score",
|
||
"refresh_low_score",
|
||
"recoverable_low_score",
|
||
"true_low_confidence",
|
||
}
|
||
|
||
|
||
def test_catalog_comparable_review_item_keeps_exact_match_guardrail():
|
||
from services.competitor_intel_repository import _format_competitor_review_item
|
||
|
||
item = _format_competitor_review_item({
|
||
"sku": "CAT-001",
|
||
"name": "DASHING DIVA Gloss Gel 美甲片 月影柔霧",
|
||
"momo_price": 699,
|
||
"attempt_status": "true_low_confidence",
|
||
"catalog_comparable": True,
|
||
"candidate_count": 3,
|
||
"best_competitor_product_id": "DABC-CATALOG",
|
||
"best_competitor_product_name": "DASHING DIVA Gloss Gel 美甲片 月影柔霧 任選",
|
||
"best_competitor_price": 599,
|
||
"best_match_score": 0.912,
|
||
"match_diagnostic_json": {
|
||
"match_type": "comparable",
|
||
"price_basis": "manual_review",
|
||
"alert_tier": "identity_review",
|
||
"reasons": [
|
||
"strong_product_line_match",
|
||
"variant_selection_review",
|
||
],
|
||
},
|
||
})
|
||
|
||
assert item["review_bucket"] == "catalog_variant_review"
|
||
assert item["status_label"] == "選項 / 色號待核"
|
||
assert item["catalog_review_guidance"]["primary_review_action"] == "needs_research"
|
||
assert "選項、色號" in item["action_label"]
|
||
envelope = item["decision_envelope"]
|
||
assert envelope["recommended_action"]["action"] == "needs_research"
|
||
assert envelope["guardrails"]["can_auto_execute"] is False
|
||
assert envelope["guardrails"]["catalog_comparable"] is True
|
||
assert envelope["guardrails"]["catalog_review_lane"] == "catalog_variant_review"
|
||
assert envelope["expected_impact"]["catalog_review_guidance"]["lane"] == "catalog_variant_review"
|
||
assert any(
|
||
evidence["metric"] == "catalog_comparable"
|
||
and evidence["value"] == "catalog_variant_review"
|
||
for evidence in envelope["evidence"]
|
||
)
|
||
assert any(evidence["metric"] == "catalog_comparable" for evidence in envelope["evidence"])
|
||
|
||
|
||
def test_competitor_review_reasons_prefer_json_payload_labels():
|
||
from services.competitor_intel_repository import _format_competitor_review_item
|
||
|
||
item = _format_competitor_review_item({
|
||
"sku": "SKU-1",
|
||
"name": "M.A.C Macximal 柔霧唇膏",
|
||
"momo_price": 990,
|
||
"attempt_status": "identity_veto",
|
||
"candidate_count": 1,
|
||
"best_competitor_product_id": "DABC123",
|
||
"best_competitor_product_name": "MAC Macximal 緞光唇膏",
|
||
"best_competitor_price": 880,
|
||
"best_match_score": 0.32,
|
||
"match_diagnostic_json": {
|
||
"match_type": "no_match",
|
||
"price_basis": "none",
|
||
"alert_tier": "suppress",
|
||
"identity_evidence": {
|
||
"version": "identity_evidence_v1",
|
||
"brand": {"momo": ["mac"], "competitor": ["mac"], "shared": ["mac"]},
|
||
"product_type": {"momo": "唇膏", "competitor": "唇膏", "matched": True},
|
||
"identity_anchor": "macximal 柔霧唇膏",
|
||
"shared_model_tokens": [],
|
||
"specs": {
|
||
"momo": {"volumes_ml": [], "weights_g": [], "dosages_mg": [], "counts": [], "total_piece_count": None},
|
||
"competitor": {"volumes_ml": [], "weights_g": [], "dosages_mg": [], "counts": [], "total_piece_count": None},
|
||
"mismatches": [],
|
||
},
|
||
"variant_guardrails": {
|
||
"hard_veto": True,
|
||
"conflict_reasons": ["makeup_finish_conflict"],
|
||
"catalog_count_omission": False,
|
||
},
|
||
},
|
||
"offer_evidence": {
|
||
"version": "offer_evidence_v1",
|
||
"price_basis": "none",
|
||
"alert_tier": "suppress",
|
||
"momo_price": 990,
|
||
"competitor_price": 880,
|
||
"gap_pct": 12.5,
|
||
"price_is_identity_evidence": False,
|
||
},
|
||
"reasons": [
|
||
"makeup_finish_conflict",
|
||
"nail_tool_function_conflict",
|
||
"schick_razor_line_conflict",
|
||
],
|
||
},
|
||
"error_message": "",
|
||
})
|
||
|
||
assert item["match_type_label"] == "非同款"
|
||
assert item["price_basis_label"] == "不可比"
|
||
assert item["alert_tier_label"] == "不告警"
|
||
assert item["diagnostic_reason_text"] == "妝效質地不同、工具功能不同、除毛刀品線不同"
|
||
assert item["identity_evidence_summary"].startswith("品牌 mac")
|
||
assert item["offer_evidence"]["price_is_identity_evidence"] is False
|
||
assert item["difference_highlights"][0]["dimension"] == "妝效/質地不同"
|
||
assert [reason["code"] for reason in item["diagnostic_reasons"]] == [
|
||
"makeup_finish_conflict",
|
||
"nail_tool_function_conflict",
|
||
"schick_razor_line_conflict",
|
||
]
|
||
envelope = item["decision_envelope"]
|
||
assert envelope["decision_type"] == "pchome_match_review"
|
||
assert envelope["subject"]["sku"] == "SKU-1"
|
||
assert envelope["subject"]["competitor_product_id"] == "DABC123"
|
||
assert envelope["guardrails"]["can_auto_execute"] is False
|
||
assert envelope["guardrails"]["data_quality"] == "partial"
|
||
assert envelope["guardrails"]["match_type"] == "no_match"
|
||
assert envelope["guardrails"]["identity_evidence_version"] == "identity_evidence_v1"
|
||
assert envelope["guardrails"]["price_is_identity_evidence"] is False
|
||
assert envelope["identity_evidence"]["brand"]["shared"] == ["mac"]
|
||
assert envelope["offer_evidence"]["gap_pct"] == 12.5
|
||
assert envelope["difference_highlights"][0]["dimension"] == "妝效/質地不同"
|
||
assert envelope["recommended_action"]["requires_hitl"] is True
|
||
assert envelope["recommended_action"]["action"] == "verify_or_reject_identity"
|
||
assert any(evidence["metric"] == "reasons" for evidence in envelope["evidence"])
|
||
assert any(evidence["metric"] == "identity_evidence" for evidence in envelope["evidence"])
|
||
assert any(evidence["metric"] == "offer_evidence" for evidence in envelope["evidence"])
|
||
assert any(evidence["metric"] == "difference_highlights" for evidence in envelope["evidence"])
|
||
|
||
|
||
def test_rescore_accepted_review_item_has_actionable_decision_envelope():
|
||
from services.competitor_intel_repository import _format_competitor_review_item
|
||
|
||
item = _format_competitor_review_item({
|
||
"sku": "10922465",
|
||
"name": "【Herbacin 德國小甘菊】小甘菊1號護手霜20ml",
|
||
"momo_price": 99,
|
||
"attempt_status": "rescore_accepted_current",
|
||
"candidate_count": 1,
|
||
"best_competitor_product_id": "DDAO4C-A79050612",
|
||
"best_competitor_product_name": "小甘菊經典護手霜20ml",
|
||
"best_competitor_price": 89,
|
||
"best_match_score": 0.872,
|
||
"match_diagnostic_json": {
|
||
"match_type": "exact",
|
||
"price_basis": "total_price",
|
||
"alert_tier": "identity_review",
|
||
"reasons": ["focused_exact_identity_herbacin_classic_hand_cream_20ml_brandless"],
|
||
},
|
||
})
|
||
|
||
envelope = item["decision_envelope"]
|
||
assert envelope["severity"] in {"P1", "P2"}
|
||
assert envelope["subject"]["momo_price"] == 99
|
||
assert envelope["subject"]["competitor_price"] == 89
|
||
assert envelope["recommended_action"]["action"] == "review_accept_identity"
|
||
assert envelope["guardrails"]["data_quality"] == "complete"
|
||
assert envelope["expected_impact"]["gap_amount"] == 10
|
||
assert envelope["expected_impact"]["candidate_gap_pct"] == 11.2
|
||
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
|
||
|
||
item = _format_competitor_review_item({
|
||
"sku": "14338675",
|
||
"name": "【Relove】胺基酸私密潔淨精華凝露120ml",
|
||
"momo_price": 399,
|
||
"attempt_status": "protected_existing_match",
|
||
"candidate_count": 2,
|
||
"best_competitor_product_id": "QEAE1O-A900A6DRS",
|
||
"best_competitor_product_name": "RELOVE胺基酸私密清潔凝露120ml",
|
||
"best_competitor_price": 329,
|
||
"best_match_score": 0.817,
|
||
"match_diagnostic_json": {
|
||
"match_type": "exact",
|
||
"price_basis": "total_price",
|
||
"alert_tier": "identity_review",
|
||
"reasons": ["strong_exact_spec_match", "spec_name_alignment"],
|
||
},
|
||
"error_message": (
|
||
"existing_match_conflict;existing_id=QEAE1O-A900A6DNN;"
|
||
"incoming_id=QEAE1O-A900A6DRS;existing_score=0.766;incoming_score=0.817"
|
||
),
|
||
})
|
||
|
||
envelope = item["decision_envelope"]
|
||
conflict = envelope["expected_impact"]["existing_match_conflict"]
|
||
assert item["existing_match_conflict"]["score_delta"] == 0.051
|
||
assert envelope["recommended_action"]["action"] == "compare_existing_identity"
|
||
assert envelope["severity"] == "P2"
|
||
assert envelope["guardrails"]["existing_match_protected"] is True
|
||
assert conflict["existing_product_id"] == "QEAE1O-A900A6DNN"
|
||
assert conflict["incoming_product_id"] == "QEAE1O-A900A6DRS"
|
||
assert any(evidence["metric"] == "existing_match_conflict" for evidence in envelope["evidence"])
|
||
|
||
|
||
def test_review_decision_brief_is_shared_by_openclaw_and_ppt():
|
||
from services.competitor_intel_repository import (
|
||
_format_competitor_review_item,
|
||
summarize_review_decision_envelopes,
|
||
)
|
||
|
||
item = _format_competitor_review_item({
|
||
"sku": "10922465",
|
||
"name": "【Herbacin 德國小甘菊】小甘菊1號護手霜20ml",
|
||
"momo_price": 99,
|
||
"attempt_status": "rescore_accepted_current",
|
||
"candidate_count": 1,
|
||
"best_competitor_product_id": "DDAO4C-A79050612",
|
||
"best_competitor_product_name": "小甘菊經典護手霜20ml",
|
||
"best_competitor_price": 89,
|
||
"best_match_score": 0.872,
|
||
"match_diagnostic_json": {
|
||
"match_type": "exact",
|
||
"price_basis": "total_price",
|
||
"alert_tier": "identity_review",
|
||
},
|
||
})
|
||
|
||
brief = summarize_review_decision_envelopes([item], limit=5)
|
||
|
||
assert brief["hitl_count"] == 1
|
||
assert brief["auto_execute_blocked_count"] == 1
|
||
assert brief["severity_counts"]
|
||
assert brief["data_quality_counts"] == {"complete": 1}
|
||
assert "人工確認身份後採用同款" in brief["text"]
|
||
assert "HITL" in brief["text"]
|
||
assert "DDAO4C-A79050612" in brief["text"]
|
||
|
||
repo_source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
openclaw_source = (ROOT / "services" / "openclaw_strategist_service.py").read_text(encoding="utf-8")
|
||
ppt_source = (ROOT / "services" / "ppt_generator.py").read_text(encoding="utf-8")
|
||
bot_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
|
||
|
||
assert "\"review_decision_brief\": summarize_review_decision_envelopes" in repo_source
|
||
assert "summarize_review_decision_envelopes" in openclaw_source
|
||
assert "review_decision_text" in openclaw_source
|
||
assert "PChome 覆核決策信封（HITL，不可自動寫正式價差）" in openclaw_source
|
||
assert "SUM(CASE WHEN attempt_status IN ('unit_comparable'" not in openclaw_source
|
||
assert "review_decision_brief" in bot_source
|
||
assert "覆核決策信封（HITL，不可自動寫正式價差）" in bot_source
|
||
assert "覆核決策信封（HITL）" in ppt_source
|
||
|
||
|
||
def test_competitor_ppt_prompt_uses_neutral_ewooc_viewpoint():
|
||
source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
|
||
|
||
assert "EwoooC 商品營運視角" in source
|
||
assert "待補資料不可當成成功配對" in source
|
||
assert "高信心比對" in source
|
||
assert "待補身份/價格" in source
|
||
assert "需單位價比較" in source
|
||
assert "單位價覆核樣本" in source
|
||
assert "我方 = PChome" not in source
|
||
assert "請以 PChome 視角" not in source
|
||
|
||
|
||
def test_top_competitor_risks_reads_latest_momo_price_after_valid_competitor_filter():
|
||
source = (ROOT / "services" / "competitor_intel_repository.py").read_text(encoding="utf-8")
|
||
risk_source = source.split("def _fetch_top_competitor_risks_uncached", 1)[1].split("def fetch_competitor_review_queue", 1)[0]
|
||
|
||
assert "FROM valid_competitor vc" in risk_source
|
||
assert "JOIN LATERAL" in risk_source
|
||
assert "WHERE pr.product_id = p.id" in risk_source
|
||
assert "ROW_NUMBER() OVER (PARTITION BY p.id" not in risk_source
|