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