340 lines
15 KiB
Python
340 lines
15 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")
|
||
|
||
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 "'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:v7" 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 "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 "FROM products p\n JOIN LATERAL" in coverage_source
|
||
assert "WHERE p.status = 'ACTIVE'" in coverage_source
|
||
|
||
|
||
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.stale_matches" 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.stale_matches" 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["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_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",
|
||
"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 [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["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"])
|
||
|
||
|
||
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["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_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
|