"""来源归因规则引擎（tracker/services/attribution.classify）。 纯函数用例（无 DB）+ 重算命令/入库集成为带 db 标记的独立用例。 诚实状态语义：resolved=观察到 referrer 证据；label_only=仅 UTM 自述； conflict=两证据矛盾并列；unknown=直接访问不猜测。 """ import pytest from django.core.management import call_command from django.utils import timezone from tracker.models import Site, SourceClassification, VisitSession, Visitor from tracker.services.attribution import classify NOW = timezone.now() # ---------------------------------------------------------------- 引擎纯函数 def test_bing_search_term_extracted(): r = classify(referrer_url="https://www.bing.com/search?q=%E6%8C%96%E6%8E%98%E6%9C%BA&m=1", landing_path="/pricing") assert r["channel"] == "natural_search" assert r["platform"] == "Bing" assert r["search_term"] == "挖掘机" assert r["evidence_kind"] == "observed_referrer" assert r["resolution"] == "resolved" def test_google_never_fabricates_search_term(): r = classify(referrer_url="https://www.google.com/search?q=cats") assert r["channel"] == "natural_search" and r["platform"] == "Google" assert r["search_term"] == "" # Google 不在 SEARCH_TERM_PARAMS 表 = 恒空，绝不虚构 assert r["variant"] == "" # google.com → TLD "com" 不显示 def test_google_country_tld_as_variant(): r = classify(referrer_url="https://www.google.co.ru/search?q=x") assert r["platform"] == "Google" and r["variant"] == "ru" def test_baidu_wd_chinese_term_and_word_fallback(): r = classify(referrer_url="https://www.baidu.com/s?wd=挖掘机价格") assert r["platform"] == "百度" and r["search_term"] == "挖掘机价格" r2 = classify(referrer_url="https://m.baidu.com/s?word=租赁") # 子域 + word 参数兜底 assert r2["platform"] == "百度" and r2["search_term"] == "租赁" def test_chatgpt_referral(): r = classify(referrer_url="https://chatgpt.com/c/abc") assert r["channel"] == "ai_referral" and r["platform"] == "ChatGPT" assert r["resolution"] == "resolved" r2 = classify(referrer_url="https://chat.chatgpt.com/") # 子域同样命中 assert r2["channel"] == "ai_referral" and r2["platform"] == "ChatGPT" def test_social_twitter_short_link(): r = classify(referrer_url="https://t.co/AbCdEf") assert r["channel"] == "social" and r["platform"] == "Twitter/X" def test_generic_referral_keeps_host_as_platform(): r = classify(referrer_url="https://blog.example.com/post/1") assert r["channel"] == "referral" and r["platform"] == "blog.example.com" assert r["evidence_kind"] == "observed_referrer" and r["resolution"] == "resolved" def test_paid_click_param_wins_over_natural_search(): r = classify(referrer_url="https://www.bing.com/search?q=x", landing_path="/offer?gclid=EAIaIQ") assert r["channel"] == "paid_search" and r["platform"] == "Google Ads" # 付费优先于自然 assert r["detail"].get("click_param") == "gclid" r2 = classify(landing_path="/?fbclid=IwAR0") # 仅落地页带点击参数 assert r2["channel"] == "paid_search" and r2["platform"] == "Meta Ads" def test_paid_param_in_referrer_url(): r = classify(referrer_url="https://example.com/ad?msclkid=xyz", landing_path="/") assert r["channel"] == "paid_search" and r["platform"] == "Bing Ads" def test_micromessenger_ua_without_referrer_is_in_app(): r = classify(ua="Mozilla/5.0 (Linux; Android 14) MicroMessenger/8.0.47") assert r["channel"] == "in_app" and r["platform"] == "微信" assert r["resolution"] == "resolved" def test_in_app_marker_overlays_existing_source_in_detail(): r = classify(referrer_url="https://www.bing.com/search?q=x", ua="Mozilla/5.0 ... MicroMessenger/8.0.47") assert r["channel"] == "natural_search" # 来源结论不被 UA 覆盖 assert r["detail"].get("in_app") == "微信" # 应用内浏览器作为 surface 叠加进 detail def test_internal_referrer_treated_as_no_signal(): r = classify(referrer_url="https://www.a.com/other-page", internal_domains=["a.com"]) assert r["channel"] == "direct" and r["resolution"] == "unknown" r2 = classify(referrer_url="https://sub.a.com/x", utm={"utm_source": "newsletter"}, internal_domains=["a.com"]) # 本站跳转后仍带 UTM → 仅剩标签证据 assert r2["resolution"] == "label_only" and r2["platform"] == "newsletter" def test_utm_matching_referrer_upgrades_to_combined(): r = classify(referrer_url="https://www.bing.com/search?q=x", utm={"utm_source": "bing"}) assert r["channel"] == "natural_search" and r["evidence_kind"] == "combined" def test_utm_contradicting_referrer_is_conflict_with_both_evidences(): r = classify(referrer_url="https://facebook.com/l/abc", utm={"utm_source": "google"}) assert r["channel"] == "social" and r["platform"] == "Facebook" # 观察证据优先于自述标签 assert r["resolution"] == "conflict" assert r["evidence_kind"] == "observed_referrer" ref_ev, utm_ev = r["detail"]["referrer_evidence"], r["detail"]["utm_evidence"] assert ref_ev["platform"] == "Facebook" and ref_ev["channel"] == "social" assert utm_ev["platform"] == "google" and utm_ev["evidence_kind"] == "utm_label" def test_utm_only_is_label_only(): r = classify(utm={"utm_source": "newsletter"}) assert r["channel"] == "referral" # 未知标签 → 有名 referral，但状态诚实标注 assert r["platform"] == "newsletter" assert r["resolution"] == "label_only" and r["evidence_kind"] == "utm_label" r2 = classify(utm={"utm_source": "baidu"}) # 已知引擎标签 → 归入对应渠道（仍 label_only） assert r2["channel"] == "natural_search" and r2["platform"] == "baidu" assert r2["resolution"] == "label_only" def test_empty_everything_is_direct_unknown(): r = classify() assert r == {"channel": "direct", "platform": "", "variant": "", "search_term": "", "evidence_kind": "", "resolution": "unknown", "detail": {}} def test_search_term_cleaned_and_truncated(): long = "词" * 150 r = classify(referrer_url=f"https://www.bing.com/search?q=%20%20{long}%20") assert r["search_term"] == "词" * 100 # strip + 截 100 r2 = classify(referrer_url="https://www.bing.com/search?q=a%00b%01c") # 控制字符剔除 assert r2["search_term"] == "abc" def test_garbage_referrer_url_does_not_crash(): r = classify(referrer_url="http://[invalid", landing_path="/x") assert r["channel"] in ("referral", "direct") # 不崩，落到兜底分支即可 # ------------------------------------------------------- 入库 + 重算集成 @pytest.mark.django_db def test_ingest_writes_first_touch_and_session_entry(): from tracker.services.ingest import ingest_batch site = Site.objects.create(name="s", domains=["a.com"]) payload = {"visitor_id": "v1", "session_id": "s1", "context": { "referrer_domain": "www.bing.com", "utm": {}, "path": "/landing"}, "events": [{"event_id": "e1", "event_name": "page_view", "occurred_at": NOW, "path": "/landing", "title": "T", "properties": {"referrer": "https://www.bing.com/search?q=%E6%8C%96%E6%8E%98%E6%9C%BA"}}]} ingest_batch(site, payload, ip="114.114.114.114", ua="Mozilla/5.0") ft = SourceClassification.objects.get(site=site, visitor__visitor_id="v1", scope="first_touch") assert ft.session is None assert ft.channel == "natural_search" and ft.platform == "Bing" assert ft.search_term == "挖掘机" and ft.rule_version == "v1" se = SourceClassification.objects.get(site=site, scope="session_entry") assert se.session.session_id == "s1" assert se.channel == "natural_search" and se.platform == "Bing" @pytest.mark.django_db def test_recompute_attribution_idempotent(): from tracker.services.ingest import ingest_batch site = Site.objects.create(name="s", domains=["a.com"]) payload = {"visitor_id": "v1", "session_id": "s1", "context": { "referrer_domain": "www.bing.com", "utm": {}, "path": "/landing"}, "events": [{"event_id": "e1", "event_name": "page_view", "occurred_at": NOW, "path": "/landing", "title": "T", "properties": {"referrer": "https://www.bing.com/search?q=ok"}}]} ingest_batch(site, payload, ip="114.114.114.114", ua="Mozilla/5.0") SourceClassification.objects.all().delete() # 清掉增量写入，验证重算可全量重建 call_command("recompute_attribution", stdout=None) n1 = SourceClassification.objects.count() assert n1 == 2 # first_touch + session_entry 各一 ft = SourceClassification.objects.get(scope="first_touch") assert ft.channel == "natural_search" and ft.search_term == "ok" call_command("recompute_attribution", stdout=None) # 第二遍：upsert 不重复不漂移 assert SourceClassification.objects.count() == n1