From 59de51c062c1f425bbd8beccfabd89a8305a07bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 16 Sep 2026 17:03:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9D=A5=E6=BA=90=E5=BD=92=E5=9B=A0?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=BC=95=E6=93=8E=EF=BC=88=E6=B8=A0=E9=81=93?= =?UTF-8?q?/AI/=E6=90=9C=E7=B4=A2=E8=AF=8D/=E8=AF=9A=E5=AE=9E=E7=8A=B6?= =?UTF-8?q?=E6=80=81/=E5=8E=86=E5=8F=B2=E9=87=8D=E7=AE=97=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - constants 规则库 v1：AI/引擎/社交域名表、搜索词参数白名单、付费点击参数、 应用内 UA 标记、CHANNELS 枚举；page_view 白名单开 nav_type/prerendering - services/attribution.classify 纯函数：付费>AI>引擎>社交>referral 分层， UTM 证据叠加（combined/conflict/label_only），应用内 UA 叠加/兜底， 引擎 TLD 国家 variant，搜索词白名单提取（Google 恒空不虚构） - SourceClassification 派生表（迁移 0004）：first_touch/session_entry 双 scope， rule_version 可追溯 - ingest 增量 best-effort：首访客/新会话时分类落库（savepoint+静默， 失败不影响采集）；validation 对 page_view referrer 完整 URL 参数清洗+截500 - recompute_attribution 命令：全量/按站幂等重算 - API：访客列表归因五字段+channel 筛选、详情 attribution 块（含最近5会话入口）、 dashboard channels 聚合（sources 原口径保留）；TS VisitorListParams 加 channel - 测试 97 → 124（引擎20 + 集成/接口7） Co-Authored-By: Claude Code --- admin/src/api/tracker/visitors.ts | 3 + server/tracker/api/dashboard.py | 19 +- server/tracker/api/visitors.py | 48 ++++- server/tracker/constants.py | 34 +++- .../commands/recompute_attribution.py | 88 +++++++++ .../migrations/0004_sourceclassification.py | 37 ++++ server/tracker/models.py | 31 +++ server/tracker/services/attribution.py | 180 +++++++++++++++++ server/tracker/services/ingest.py | 75 ++++++- server/tracker/services/validation.py | 4 + server/tracker/tests/test_attribution.py | 183 ++++++++++++++++++ server/tracker/tests/test_dashboard.py | 29 ++- server/tracker/tests/test_ingest.py | 20 ++ server/tracker/tests/test_validation.py | 15 ++ server/tracker/tests/test_visitors.py | 65 ++++++- 15 files changed, 816 insertions(+), 15 deletions(-) create mode 100644 server/tracker/management/commands/recompute_attribution.py create mode 100644 server/tracker/migrations/0004_sourceclassification.py create mode 100644 server/tracker/services/attribution.py create mode 100644 server/tracker/tests/test_attribution.py diff --git a/admin/src/api/tracker/visitors.ts b/admin/src/api/tracker/visitors.ts index 044bbd2..0c3427b 100644 --- a/admin/src/api/tracker/visitors.ts +++ b/admin/src/api/tracker/visitors.ts @@ -132,6 +132,9 @@ export interface VisitorListParams { show_bot?: '1' /** visitor_id / last_ip 前缀模糊匹配 */ search?: string + /** 渠道筛选（CHANNELS 枚举：natural_search/paid_search/ai_referral/social/in_app/referral/direct/unknown， + * 按访客 first_touch 分类过滤；无分类行的历史访客不属于任何渠道桶） */ + channel?: string /** YYYY-MM-DD（非法值后端宽松忽略） */ date_from?: string /** YYYY-MM-DD */ diff --git a/server/tracker/api/dashboard.py b/server/tracker/api/dashboard.py index 4e14550..324ac68 100644 --- a/server/tracker/api/dashboard.py +++ b/server/tracker/api/dashboard.py @@ -4,7 +4,7 @@ from django.utils import timezone from rest_framework.views import APIView from rest_framework.response import Response from tracker.api.envelope import EnvelopeMixin, ok -from tracker.models import DailyStat, Event, Site, VisitSession, Visitor +from tracker.models import DailyStat, Event, Site, SourceClassification, VisitSession, Visitor from tracker.services.stats import daily_counts METRICS = ("pv", "uv", "sessions", "new_visitors") @@ -31,6 +31,7 @@ class OverviewView(EnvelopeMixin, APIView): return ok({"today": counts, "trend": self._trend(site, today, counts), "sources": self._today_sources(site, today), + "channels": self._today_channels(site, today), "top_pages": self._today_pages(site, today)}) @staticmethod @@ -75,6 +76,22 @@ class OverviewView(EnvelopeMixin, APIView): .annotate(n=Count("id")) .order_by("-n", "referrer_domain")[:8]] # 平局时次序跨请求稳定 + def _today_channels(self, site, today): + """当天会话按来源渠道聚合（session_entry 分类；sources 的升级版，原始口径保留）。 + 从分类表正向聚合 + Count(distinct session)：无唯一约束下分类表偶发同键重复行 + 也不会把同一会话计两次。无分类行的会话（未重算的历史数据）不计入。""" + start, end = _day_bounds(today) + q = SourceClassification.objects.filter( + scope="session_entry", + session__visitor__is_bot=False, + session__started_at__gte=start, session__started_at__lt=end) + if site is not None: + q = q.filter(site=site) + return [{"channel": r["channel"] or "unknown", "sessions": r["n"]} + for r in q.values("channel") + .annotate(n=Count("session", distinct=True)) + .order_by("-n", "channel")[:8]] # 平局时次序跨请求稳定 + def _today_pages(self, site, today): start, end = _day_bounds(today) q = Event.objects.filter(is_bot=False, event_name="page_view", diff --git a/server/tracker/api/visitors.py b/server/tracker/api/visitors.py index 8ba51af..57c8210 100644 --- a/server/tracker/api/visitors.py +++ b/server/tracker/api/visitors.py @@ -13,7 +13,7 @@ from rest_framework.response import Response from rest_framework.views import APIView from tracker.api.envelope import EnvelopeMixin, ok -from tracker.models import Event, RawBatch, Site, VisitSession, Visitor +from tracker.models import Event, RawBatch, Site, SourceClassification, VisitSession, Visitor PAGE_SIZE = 20 RAW_PAGE_SIZE = 10 # 原始批次 body 较大，页小一些 @@ -21,6 +21,7 @@ TIMELINE_DEFAULT_DAYS = 30 TIMELINE_MAX_DAYS = 90 TIMELINE_MAX_EVENTS = 1000 ONLINE_WINDOW = timedelta(minutes=5) # 与 dashboard active_now 同口径 +ATTR_SESSION_ENTRIES = 5 # 详情页归因块带出的最近会话入口分类数 def _source(referrer_domain, utm): @@ -34,8 +35,9 @@ def _iso(dt): return timezone.localtime(dt).isoformat() if dt else None -def _row(visitor, last_session, online_cutoff): - """列表/详情共用行字段。last_source 依赖最近会话（由调用方批量取好传入，防 N+1）。""" +def _row(visitor, last_session, online_cutoff, ft=None): + """列表/详情共用行字段。last_source 依赖最近会话（由调用方批量取好传入，防 N+1）。 + channel 等归因五字段来自该访客 first_touch 分类（无分类行 → 空串，前端回落 first_source 展示）。""" return { "visitor_id": visitor.visitor_id, # 站点归属：visitor.site 需随行 select_related("site")，否则逐行多一条查询 @@ -51,10 +53,28 @@ def _row(visitor, last_session, online_cutoff): "first_source": _source(visitor.first_referrer_domain, visitor.first_utm), "last_source": _source(last_session.referrer_domain, last_session.utm) if last_session else "直接访问", # 无会话访客（仅采集异常场景） + "channel": ft.channel if ft else "", + "platform": ft.platform if ft else "", + "variant": ft.variant if ft else "", + "search_term": ft.search_term if ft else "", + "resolution": ft.resolution if ft else "", "online": visitor.last_seen_at > online_cutoff, # Python 侧比对（last_seen_at 已随行取出） } +def _classification(sc, with_session=False): + """SourceClassification → API dict（详情归因块用，含证据可信度与规则版本）。""" + if not sc: + return None + d = {"channel": sc.channel, "platform": sc.platform, "variant": sc.variant, + "search_term": sc.search_term, "evidence_kind": sc.evidence_kind, + "resolution": sc.resolution, "detail": sc.detail or {}, + "rule_version": sc.rule_version, "computed_at": _iso(sc.computed_at)} + if with_session: + d["session_id"] = sc.session.session_id if sc.session_id else None + return d + + def _session_summary(s): # duration_ms 仅内部字段不展示——前端时长一律用 stay_ms（有效停留口径） return {"session_id": s.session_id, @@ -96,6 +116,11 @@ def _page_visitors(qs, request): if p.get("search"): q = p["search"] qs = qs.filter(Q(visitor_id__istartswith=q) | Q(last_ip__istartswith=q)) + if p.get("channel"): + # 渠道筛选（值=CHANNELS 枚举）：按 first_touch 分类内连接过滤。 + # distinct 防御分类表脏重复行（无唯一约束，并发极端下可能同键两行）撑大 count。 + qs = qs.filter(source_classifications__scope="first_touch", + source_classifications__channel=p["channel"]).distinct() date_from = _parse_date(p.get("date_from")) date_to = _parse_date(p.get("date_to")) if date_from: @@ -114,8 +139,12 @@ def _page_visitors(qs, request): visitors = list(page_qs[(page - 1) * PAGE_SIZE: page * PAGE_SIZE]) latest = {s.pk: s for s in VisitSession.objects.filter( pk__in=[v.latest_session_pk for v in visitors if v.latest_session_pk])} + # 归因字段批量取：整页一次 IN 查询（同 latest 的做法），不做 select_related N+1 + fts = {sc.visitor_id: sc for sc in SourceClassification.objects.filter( + visitor__in=[v.pk for v in visitors], scope="first_touch")} cutoff = timezone.now() - ONLINE_WINDOW - rows = [_row(v, latest.get(v.latest_session_pk), cutoff) for v in visitors] + rows = [_row(v, latest.get(v.latest_session_pk), cutoff, fts.get(v.pk)) + for v in visitors] return rows, total, page @@ -190,11 +219,20 @@ class VisitorDetailView(EnvelopeMixin, APIView): # 全量会话摘要（visitor,-started_at 索引）；M1 访客会话量有限不分页 sessions = VisitSession.objects.filter(visitor=visitor).order_by("-started_at", "-pk") cutoff = timezone.now() - ONLINE_WINDOW - return ok({**_row(visitor, last_session, cutoff), + ft = (SourceClassification.objects.filter(visitor=visitor, scope="first_touch") + .first()) + entries = (SourceClassification.objects + .filter(visitor=visitor, scope="session_entry") + .select_related("session") # session_id 随行带出，免逐行查询 + .order_by("-session__started_at", "-pk")[:ATTR_SESSION_ENTRIES]) + return ok({**_row(visitor, last_session, cutoff, ft), "screen": visitor.screen, "timezone_name": visitor.timezone_name, "last_ip": visitor.last_ip, "first_utm": visitor.first_utm or {}, "first_landing_path": visitor.first_landing_path, "first_referrer_domain": visitor.first_referrer_domain, + "attribution": {"first_touch": _classification(ft), + "session_entries": [_classification(e, with_session=True) + for e in entries]}, "sessions": [_session_summary(s) for s in sessions]}) diff --git a/server/tracker/constants.py b/server/tracker/constants.py index e2f160d..d74190e 100644 --- a/server/tracker/constants.py +++ b/server/tracker/constants.py @@ -11,9 +11,41 @@ URL_PARAM_DROP_PATTERNS = ("email", "phone", "mobile", "token", "key", "password SESSION_IDLE_MS = 30 * 60 * 1000 BOT_UA_KEYWORDS = ("bot", "spider", "crawl", "slurp", "bingpreview", "headless", "lighthouse") EVENT_PROP_WHITELIST = { # spec §9.1：未知字段拒绝。免费文本/任意属性不得入库 - "page_view": {"referrer", "referrer_domain", "utm"}, + "page_view": {"referrer", "referrer_domain", "utm", "nav_type", "prerendering"}, "page_stay": {"active_ms", "path"}, "sdk_loaded": {"sdk_version"}, } MAX_PROP_DEPTH = 2 MAX_PROP_JSON_BYTES = 4096 + +# --------------------------------------------------------------------------- +# 来源归因规则库 v1（tracker/services/attribution.py 消费）。改规则必须 bump 版本号 +# 并跑 recompute_attribution 重算历史——已落库分类带 rule_version 可追溯。 +# --------------------------------------------------------------------------- +ATTRIBUTION_RULE_VERSION = "v1" +AI_REFERRAL_DOMAINS = { # AI 平台 → 显示名 + "chatgpt.com": "ChatGPT", "perplexity.ai": "Perplexity", + "copilot.microsoft.com": "Copilot", "gemini.google.com": "Gemini", + "deepseek.com": "DeepSeek", "claude.ai": "Claude", "poe.com": "Poe", "you.com": "You", +} +SEARCH_ENGINE_DOMAINS = { # 引擎后缀/域名 → 显示名（"google."/"yandex." 前缀匹配任意国家站 TLD） + "google.": "Google", "bing.com": "Bing", "baidu.com": "百度", "yandex.": "Yandex", + "duckduckgo.com": "DuckDuckGo", "yahoo.com": "Yahoo", "ecosia.org": "Ecosia", + "search.naver.com": "Naver", "search.brave.com": "Brave", +} +SEARCH_TERM_PARAMS = { # 引擎 → referrer URL 里的搜索词参数名（白名单提取） + "Bing": ["q"], "百度": ["wd", "word"], "DuckDuckGo": ["q"], "Yandex": ["text"], + "Ecosia": ["q"], "Yahoo": ["p"], "Brave": ["q"], "Naver": ["query"], +} # 注：Google 不提供，绝不虚构 +PAID_CLICK_PARAMS = {"gclid": "Google Ads", "fbclid": "Meta Ads", + "msclkid": "Bing Ads", "yclid": "Yandex Ads"} +SOCIAL_DOMAINS = { + "t.co": "Twitter/X", "twitter.com": "Twitter/X", "x.com": "Twitter/X", + "facebook.com": "Facebook", "linkedin.com": "LinkedIn", "instagram.com": "Instagram", + "vk.com": "VK", "reddit.com": "Reddit", "pinterest.com": "Pinterest", + "telegram.me": "Telegram", "t.me": "Telegram", +} +INAPP_UA_MARKERS = {"MicroMessenger": "微信", "Line/": "Line", "FBAN": "Facebook", + "FBAV": "Facebook", "Instagram": "Instagram", "WhatsApp": "WhatsApp"} +CHANNELS = ("natural_search", "paid_search", "ai_referral", "social", + "in_app", "referral", "direct", "unknown") diff --git a/server/tracker/management/commands/recompute_attribution.py b/server/tracker/management/commands/recompute_attribution.py new file mode 100644 index 0000000..133fc27 --- /dev/null +++ b/server/tracker/management/commands/recompute_attribution.py @@ -0,0 +1,88 @@ +"""历史来源分类全量重算（幂等可重复）。 + +规则引擎升级（constants.ATTRIBUTION_RULE_VERSION bump）后执行： + python manage.py recompute_attribution # 全部站点 + python manage.py recompute_attribution --site site_xxx + +数据源约束：重算只能用"落库时保住的信号"—— + - 访客首触：最早 page_view 的 properties.referrer/utm（无则档案字段兜底， + referrer_domain 合成 URL 后搜索词已丢失，诚实降级）+ first_landing_path；UA 未存档 → 空。 + - 会话入口：referrer_domain 合成 URL + utm + landing_path。 +逐条 upsert（first_touch 键 site+visitor+scope；session_entry 键 site+session+scope）， +跑两遍计数不变。 +""" +from django.core.management.base import BaseCommand + +from tracker import constants as C +from tracker.models import Event, Site, SourceClassification, VisitSession, Visitor +from tracker.services.attribution import classify + + +def _first_touch_signals(visitor): + """最早一条 page_view 的原始信号优先；旧数据无事件属性时回落访客档案字段。""" + ev = (Event.objects.filter(visitor=visitor, event_name="page_view") + .order_by("occurred_at", "pk").first()) + props = (ev.properties or {}) if ev else {} + referrer = props.get("referrer") if isinstance(props.get("referrer"), str) else "" + domain = props.get("referrer_domain") or visitor.first_referrer_domain + if not referrer and domain: + referrer = f"https://{str(domain).strip().strip('/')}" + utm = props.get("utm") if isinstance(props.get("utm"), dict) else None + return referrer, (utm or visitor.first_utm or {}), visitor.first_landing_path + + +def _upsert(site, visitor, session, scope, result): + vals = {"channel": (result["channel"] or "")[:32], + "platform": (result["platform"] or "")[:64], + "variant": (result["variant"] or "")[:64], + "search_term": (result["search_term"] or "")[:200], + "evidence_kind": (result["evidence_kind"] or "")[:32], + "resolution": (result["resolution"] or "")[:16], + "detail": result["detail"] or {}, "session": session, + "rule_version": C.ATTRIBUTION_RULE_VERSION} # 重算即升到当前规则版本 + if session is None: + SourceClassification.objects.update_or_create( + site=site, visitor=visitor, scope=scope, defaults=vals) + else: + SourceClassification.objects.update_or_create( + site=site, session=session, scope=scope, + defaults={**vals, "visitor": visitor}) + + +class Command(BaseCommand): + help = "按当前规则版本全量重算来源分类（first_touch + session_entry，幂等）" + + def add_arguments(self, parser): + parser.add_argument("--site", default=None, + help="只重算指定 site_id（缺省全部站点）") + + def handle(self, *args, **opts): + sites = Site.objects.all() + if opts["site"]: + sites = sites.filter(site_id=opts["site"]) + if not sites.exists(): + self.stdout.write(self.style.WARNING( + f"site not found: {opts['site']}")) + return + v_total = s_total = 0 + for site in sites: + domains = site.domains or [] + for visitor in Visitor.objects.filter(site=site).only("pk", "first_referrer_domain", + "first_utm", "first_landing_path"): + referrer, utm, landing = _first_touch_signals(visitor) + _upsert(site, visitor, None, "first_touch", + classify(referrer_url=referrer, utm=utm, + landing_path=landing or "", + internal_domains=domains)) + v_total += 1 + for sess in VisitSession.objects.filter(site=site).select_related("visitor"): + referrer = f"https://{sess.referrer_domain.strip().strip('/')}" \ + if sess.referrer_domain else "" + _upsert(site, sess.visitor, sess, "session_entry", + classify(referrer_url=referrer, utm=sess.utm or {}, + landing_path=sess.landing_path or "", + internal_domains=domains)) + s_total += 1 + self.stdout.write( + f"recomputed attribution: {v_total} first_touch, {s_total} session_entry " + f"(rule version {C.ATTRIBUTION_RULE_VERSION})") diff --git a/server/tracker/migrations/0004_sourceclassification.py b/server/tracker/migrations/0004_sourceclassification.py new file mode 100644 index 0000000..972e79d --- /dev/null +++ b/server/tracker/migrations/0004_sourceclassification.py @@ -0,0 +1,37 @@ +# Generated by Django 5.2.17 on 2026-09-16 08:52 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('tracker', '0003_rawbatch'), + ] + + operations = [ + migrations.CreateModel( + name='SourceClassification', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('scope', models.CharField(max_length=16)), + ('channel', models.CharField(blank=True, default='', max_length=32)), + ('platform', models.CharField(blank=True, default='', max_length=64)), + ('variant', models.CharField(blank=True, default='', max_length=64)), + ('search_term', models.CharField(blank=True, default='', max_length=200)), + ('evidence_kind', models.CharField(blank=True, default='', max_length=32)), + ('resolution', models.CharField(blank=True, default='', max_length=16)), + ('detail', models.JSONField(blank=True, default=dict)), + ('rule_version', models.CharField(default='v1', max_length=16)), + ('computed_at', models.DateTimeField(auto_now=True)), + ('session', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='source_classifications', to='tracker.visitsession')), + ('site', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='source_classifications', to='tracker.site')), + ('visitor', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='source_classifications', to='tracker.visitor')), + ], + options={ + 'db_table': 'source_classification', + 'indexes': [models.Index(fields=['site', 'visitor', 'scope'], name='source_clas_site_id_58ad9c_idx'), models.Index(fields=['site', 'channel'], name='source_clas_site_id_f08b23_idx')], + }, + ), + ] diff --git a/server/tracker/models.py b/server/tracker/models.py index 223f867..3b8c20e 100644 --- a/server/tracker/models.py +++ b/server/tracker/models.py @@ -1,6 +1,8 @@ from django.db import models import secrets +from tracker import constants as C + def short_public_id() -> str: return "site_" + secrets.token_urlsafe(12)[:16] @@ -131,6 +133,35 @@ class RawBatch(models.Model): models.Index(fields=["site", "-received_at"])] +class SourceClassification(models.Model): + """来源分类结果（attribution 规则引擎的派生存储）。 + + 每访客两行 scope：first_touch（首访来源，session=None）/ session_entry（每次新会话入口）。 + 写入两路：ingest 增量（新访客/新会话时 best-effort）+ recompute_attribution 全量重算 + （幂等 upsert）。resolution 表达证据可信度：resolved/combined/conflict/label_only/unknown—— + 观察到的 referrer 与 UTM 自述矛盾时并列存 detail，不武断取一边。 + rule_version 关联 constants.ATTRIBUTION_RULE_VERSION：规则升级后可识别存量行需重算。 + """ + site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="source_classifications") + visitor = models.ForeignKey(Visitor, on_delete=models.CASCADE, related_name="source_classifications", null=True) + session = models.ForeignKey(VisitSession, on_delete=models.CASCADE, related_name="source_classifications", null=True) + scope = models.CharField(max_length=16) # first_touch / session_entry + channel = models.CharField(max_length=32, blank=True, default="") + platform = models.CharField(max_length=64, blank=True, default="") + variant = models.CharField(max_length=64, blank=True, default="") + search_term = models.CharField(max_length=200, blank=True, default="") + evidence_kind = models.CharField(max_length=32, blank=True, default="") + resolution = models.CharField(max_length=16, blank=True, default="") + detail = models.JSONField(default=dict, blank=True) + rule_version = models.CharField(max_length=16, default=C.ATTRIBUTION_RULE_VERSION) + computed_at = models.DateTimeField(auto_now=True) + + class Meta: + db_table = "source_classification" + indexes = [models.Index(fields=["site", "visitor", "scope"]), + models.Index(fields=["site", "channel"])] + + class DailyStat(models.Model): site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="daily_stats", db_index=False) # uq_site_date 前缀覆盖 date = models.DateField() diff --git a/server/tracker/services/attribution.py b/server/tracker/services/attribution.py new file mode 100644 index 0000000..796ac85 --- /dev/null +++ b/server/tracker/services/attribution.py @@ -0,0 +1,180 @@ +"""来源归因规则引擎：裸域名 → 渠道/平台/搜索词/证据可信度分层识别。 + +纯函数、无 IO、无 DB——classify 是唯一入口，ingest 增量计算与 +recompute_attribution 历史重算共用同一套规则（改规则必须 bump +constants.ATTRIBUTION_RULE_VERSION 并重跑重算）。 + +诚实状态语义（resolution）： + resolved 观察到 referrer/点击参数/UA 等硬证据 + combined referrer 观察与 UTM 标签互相印证 + conflict referrer 结论与 UTM 标签矛盾——观察证据优先，标签并列存 detail + label_only 仅 UTM 自述，无观察证据 + unknown 全空（直接访问）——不猜测手输/书签 +""" +import re +from urllib.parse import parse_qsl, urlsplit + +from tracker import constants as C + +_CTRL_CHARS = re.compile(r"[\x00-\x1f\x7f]") +_GENERIC_TLDS = ("com",) # 引擎 host 落在这些 TLD 时不作为国家 variant 显示 + + +def _host(url: str) -> str: + """提取 referrer host；异常/空 → ""。hostname 自动去端口并小写。""" + try: + return (urlsplit(url).hostname or "").lower() + except ValueError: + return "" + + +def _canonical(host: str) -> str: + """去掉单个 www. 前缀后做域名匹配（www.bing.com → bing.com）。""" + return host[4:] if host.startswith("www.") else host + + +def _domain_match(host: str, key: str) -> bool: + """host 是否命中规则键。key 以 "." 结尾（google./yandex.）→ 前缀匹配任意国家站 + TLD；否则精确或子域（m.facebook.com 命中 facebook.com）。""" + h = _canonical(host) + k = key.lower() + if k.endswith("."): + return h == k[:-1] or h.startswith(k) + return h == k or h.endswith("." + k) + + +def _tld_variant(host: str) -> str: + """引擎 host 的 TLD 国家（www.google.co.ru → "ru"）；通用 TLD 不显示。""" + tld = _canonical(host).rsplit(".", 1)[-1] + return "" if tld in _GENERIC_TLDS else tld + + +def _clean_term(raw: str) -> str: + """搜索词清洗：strip → 剔除控制字符 → 截 100。""" + return _CTRL_CHARS.sub("", (raw or "").strip())[:100] + + +def _search_term(engine: str, referrer_url: str) -> str: + """按引擎参数名白名单从 referrer URL 提取搜索词。Google 不在表内 → 恒空（绝不虚构）。""" + for param in C.SEARCH_TERM_PARAMS.get(engine, []): + for k, v in parse_qsl(urlsplit(referrer_url).query, keep_blank_values=True): + if k == param and v.strip(): + return _clean_term(v) + return "" + + +def _paid_param(landing_path: str, referrer_url: str) -> tuple[str, str] | None: + """付费优先于自然：落地页（或 referrer）query 命中点击参数 → (参数名, 平台名)。""" + for source in (landing_path, referrer_url): + if not source: + continue + try: + params = [k for k, _ in parse_qsl(urlsplit(source).query, + keep_blank_values=True)] + except ValueError: + continue + for k in params: + if k in C.PAID_CLICK_PARAMS: + return k, C.PAID_CLICK_PARAMS[k] + return None + + +def _in_app_name(ua: str) -> str: + """UA 命中应用内浏览器标记（大小写不敏感）→ 应用显示名；未命中 → ""。""" + low = (ua or "").lower() + for marker, name in C.INAPP_UA_MARKERS.items(): + if marker.lower() in low: + return name + return "" + + +def _utm_channel(label: str) -> str: + """UTM 标签尽力归类：已知 AI/引擎/社交标签（裸标签 "baidu"/显示名 "Google" 皆可） + 映射到对应渠道，未知 → referral（有名来源，可信度由 label_only 单独表达）。""" + h = (label or "").lower().strip() + if not h: + return "referral" + for table, channel in ((C.AI_REFERRAL_DOMAINS, "ai_referral"), + (C.SEARCH_ENGINE_DOMAINS, "natural_search"), + (C.SOCIAL_DOMAINS, "social")): + for key, name in table.items(): + if _domain_match(h, key) or key.startswith(h + ".") or h == name.lower(): + return channel + return "referral" + + +def classify(*, referrer_url: str = "", utm: dict | None = None, + landing_path: str = "", ua: str = "", internal_domains: list | None = None) -> dict: + """把一次到访的原始信号分类为 {channel, platform, variant, search_term, + evidence_kind, resolution, detail}。各字段默认空串/空 dict，纯函数无副作用。 + + 规则序：本站域名去信号 → 付费点击参数 → AI → 搜索引擎 → 社交 → 泛 referral + → UTM 证据叠加（combined/conflict/label_only）→ 应用内 UA 叠加/兜底 → direct。 + """ + out = {"channel": "", "platform": "", "variant": "", "search_term": "", + "evidence_kind": "", "resolution": "", "detail": {}} + utm = utm or {} + utm_source = str(utm.get("utm_source") or "").strip() + + host = _host(referrer_url) + if host and any(_domain_match(host, d) for d in (internal_domains or []) if d): + host = "" # 本站跳转不是外部来源信号 + + paid = _paid_param(landing_path, referrer_url) + if paid: + out.update(channel="paid_search", platform=paid[1], + evidence_kind="observed_referrer", resolution="resolved", + detail={"click_param": paid[0]}) + elif host: + for key, name in C.AI_REFERRAL_DOMAINS.items(): + if _domain_match(host, key): + out.update(channel="ai_referral", platform=name) + break + else: + for key, name in C.SEARCH_ENGINE_DOMAINS.items(): + if _domain_match(host, key): + out.update(channel="natural_search", platform=name, + variant=_tld_variant(host), + search_term=_search_term(name, referrer_url)) + break + else: + for key, name in C.SOCIAL_DOMAINS.items(): + if _domain_match(host, key): + out.update(channel="social", platform=name) + break + else: + out.update(channel="referral", platform=host) + if not out["resolution"]: + out.update(evidence_kind="observed_referrer", resolution="resolved") + + # UTM 证据叠加：观察结论之上只做印证/并列，不覆盖 + if utm_source: + if out["resolution"] == "resolved": + if utm_source.lower() == out["platform"].lower(): + out["evidence_kind"] = "combined" + else: # 矛盾：观察证据保留在顶层，两路证据各自完整并列进 detail + out["resolution"] = "conflict" + out["detail"] = {**out["detail"], + "referrer_evidence": { + "channel": out["channel"], "platform": out["platform"], + "variant": out["variant"], "search_term": out["search_term"], + "evidence_kind": "observed_referrer", "referrer_host": host}, + "utm_evidence": { + "channel": _utm_channel(utm_source), "platform": utm_source, + "evidence_kind": "utm_label"}} + else: # 无 referrer 观察 → 仅标签：按 utm_source 归类显示，状态诚实标注 + out.update(channel=_utm_channel(utm_source), platform=utm_source, + evidence_kind="utm_label", resolution="label_only") + + # 应用内浏览器 UA：有来源 → 作为 surface 叠加；无任何来源 → 本身就是来源 + app = _in_app_name(ua) + if app: + if out["resolution"]: + out["detail"] = {**out["detail"], "in_app": app} + else: + out.update(channel="in_app", platform=app, + evidence_kind="observed_ua", resolution="resolved") + + if not out["channel"]: + out.update(channel="direct", resolution="unknown") # 不猜测手输/书签 + return out diff --git a/server/tracker/services/ingest.py b/server/tracker/services/ingest.py index e786726..30e3bef 100644 --- a/server/tracker/services/ingest.py +++ b/server/tracker/services/ingest.py @@ -21,11 +21,14 @@ from django.db.models.functions import Greatest from django.utils import timezone from tracker import constants as C -from tracker.models import Visitor, VisitSession, Event +from tracker.models import SourceClassification, Visitor, VisitSession, Event from tracker.services.validation import validate_event +from tracker.services.attribution import classify from tracker.services.geo import resolver as _geo # 模块级单例（勿实例化，11MB/份） _FIELD_LEN = {"device_type": 16, "os": 64, "browser": 64, "language": 32, "screen": 32} +_CLASS_FIELDS = ("channel", "platform", "variant", "search_term", + "evidence_kind", "resolution", "detail") def _clean_utm(d): @@ -40,6 +43,52 @@ def _geo_field(g, key): return v if v not in ("", "0") else "" +def _first_pageview_referrer(events) -> str: + """本批第一条带 referrer 的 page_view 的完整来源 URL（旧版 SDK 无此字段 → ""）。""" + for ev in events: + if isinstance(ev, dict) and ev.get("event_name") == "page_view": + r = (ev.get("properties") or {}).get("referrer") + if isinstance(r, str) and r: + return r + return "" + + +def _referrer_url(domain, full: str) -> str: + """分类入参统一为完整 URL：优先 page_view 的完整 referrer（含搜索词）， + 缺失时用 ctx 的 referrer_domain 合成（丢查询串，搜索词无法找回——诚实降级）。""" + if full: + return full + d = str(domain or "").strip().lower().strip("/") + return f"https://{d}" if d else "" + + +def _classify_and_store(*, site, visitor, session, scope, referrer_url, + utm, landing_path, ua): + """来源分类落库（best-effort）：任何失败静默吞掉，绝不影响采集主流程。 + savepoint 隔离——DB 异常不至于标记外层采集事务需回滚。""" + try: + with transaction.atomic(): + r = classify(referrer_url=referrer_url, utm=utm or {}, + landing_path=landing_path or "", ua=ua or "", + internal_domains=site.domains or []) + vals = {"channel": (r["channel"] or "")[:32], + "platform": (r["platform"] or "")[:64], + "variant": (r["variant"] or "")[:64], + "search_term": (r["search_term"] or "")[:200], + "evidence_kind": (r["evidence_kind"] or "")[:32], + "resolution": (r["resolution"] or "")[:16], + "detail": r["detail"] or {}, "session": session} + if session is None: # first_touch：键 site+visitor+scope（每访客一行） + SourceClassification.objects.update_or_create( + site=site, visitor=visitor, scope=scope, defaults=vals) + else: # session_entry：键 site+session+scope（每会话一行） + SourceClassification.objects.update_or_create( + site=site, session=session, scope=scope, + defaults={**vals, "visitor": visitor}) + except Exception: + pass + + def _touch_visitor(site, vid, ctx, ip): env = {k: str(ctx.get(k, "") or "")[:n] for k, n in _FIELD_LEN.items()} # Task16 联调发现：键名映射缺口——SDK context 发 "timezone"，模型字段是 timezone_name， @@ -64,10 +113,10 @@ def _touch_visitor(site, vid, ctx, ip): province=_geo_field(g, "province")[:64], city=_geo_field(g, "city")[:64]) Visitor.objects.filter(pk=v.pk).update(**upd) - return v # 下游只用 pk（FK/子查询），无需 refresh_from_db 多耗一条 SELECT + return v, created # created 供首触来源分类（ingest_batch 消费） -def _get_session(site, visitor, sid, ctx, occurred_at): +def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url=""): """会话权威在 SDK；服务端兜底：同 session_id 族（s1、s1#2…）距最后事件>30min 切分后缀段。 空闲比较必须取整个会话族的最大 ended_at：若只查根段（session_id=sid 精确匹配）， @@ -96,6 +145,13 @@ def _get_session(site, visitor, sid, ctx, occurred_at): if not sess: raise Visitor.objects.filter(pk=visitor.pk).update(visit_count=F("visit_count") + 1) + # 新会话入口来源分类（best-effort）：创建成功与并发撞键重查两条路径都会走到， + # 同键 upsert 幂等——赢家请求已写时这只是覆写同值。 + _classify_and_store(site=site, visitor=visitor, session=sess, + scope="session_entry", + referrer_url=_referrer_url(ctx.get("referrer_domain"), referrer_url), + utm=_clean_utm(ctx.get("utm")), + landing_path=str(ctx.get("path") or ""), ua=ua) return sess @@ -111,11 +167,19 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di if not events or len(events) > C.MAX_EVENTS_PER_BATCH: return {"accepted": 0, "duplicated": 0, "rejected": [{"reason": "batch_size_invalid"}]} - visitor = _touch_visitor(site, payload["visitor_id"], ctx, ip) + visitor, created = _touch_visitor(site, payload["visitor_id"], ctx, ip) geo = _geo.resolve(ip) if ip else None is_bot = any(k in (ua or "").lower() for k in C.BOT_UA_KEYWORDS) if is_bot: Visitor.objects.filter(pk=visitor.pk).update(is_bot=True) + # 首触来源分类（best-effort）：首批 page_view 的完整 referrer 优先（含搜索词）， + # 旧 SDK 只有 ctx.referrer_domain 时合成降级。仅在访客首次建档时写一次。 + pv_referrer = _first_pageview_referrer(events) + if created: + _classify_and_store(site=site, visitor=visitor, session=None, scope="first_touch", + referrer_url=_referrer_url(ctx.get("referrer_domain"), pv_referrer), + utm=_clean_utm(ctx.get("utm")), + landing_path=str(ctx.get("path") or ""), ua=ua or "") ready, session_cache, seen_eids = [], {}, set() for i, ev in enumerate(events): reason = validate_event(ev) @@ -131,7 +195,8 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di sid = payload["session_id"] sess = session_cache.get(sid) if not sess: - sess = _get_session(site, visitor, sid, ctx, occ) + sess = _get_session(site, visitor, sid, ctx, occ, ua=ua or "", + referrer_url=pv_referrer) session_cache[sid] = sess # 按原始 sid 存：切分后 sess.session_id 是 "s1#2"， # 若按它存，同批后续事件按 "s1" 查会 miss → 每条再切新会话 ready.append(Event( diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index a29d051..6f9439a 100644 --- a/server/tracker/services/validation.py +++ b/server/tracker/services/validation.py @@ -113,6 +113,10 @@ def validate_event(ev) -> str | None: ev["path"] = sanitize_url(ev["path"]) if len(ev.get("path") or "") > C.MAX_PATH_LEN: ev["path"] = ev["path"][:C.MAX_PATH_LEN] + # referrer 是完整 URL：同样过参数清洗（email/token 等整键剔除）再截 MAX_STR_LEN。 + # 放在 str_too_long 检查之后——超长 referrer 仍按 properties_too_long 拒，不静默截半。 + if ev["event_name"] == "page_view" and isinstance(props.get("referrer"), str) and props["referrer"]: + props["referrer"] = sanitize_url(props["referrer"])[:C.MAX_STR_LEN] if len(ev.get("title") or "") > C.MAX_TITLE_LEN: ev["title"] = ev["title"][:C.MAX_TITLE_LEN] return None diff --git a/server/tracker/tests/test_attribution.py b/server/tracker/tests/test_attribution.py new file mode 100644 index 0000000..66e5a65 --- /dev/null +++ b/server/tracker/tests/test_attribution.py @@ -0,0 +1,183 @@ +"""来源归因规则引擎（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 diff --git a/server/tracker/tests/test_dashboard.py b/server/tracker/tests/test_dashboard.py index d16d642..c70651d 100644 --- a/server/tracker/tests/test_dashboard.py +++ b/server/tracker/tests/test_dashboard.py @@ -3,7 +3,7 @@ from datetime import timedelta from django.contrib.auth import get_user_model from django.utils import timezone from rest_framework.test import APIClient -from tracker.models import Event, Site, VisitSession, Visitor +from tracker.models import Event, Site, SourceClassification, VisitSession, Visitor from tracker.services.stats import refresh_daily pytestmark = pytest.mark.django_db @@ -104,6 +104,33 @@ def test_overview_trend_uses_dailystat_history(client, member_user): assert d["trend"][-1]["pv"] == 0 # 今天无事件，实时覆盖为 0 +def test_overview_channels_aggregation(client, member_user): + """channels：当天会话按 session_entry 分类的 channel 聚合（与 sources 并存，互不影响）。""" + _auth(client, member_user) + site = Site.objects.create(name="s", domains=["a.com"]) + v1 = _visitor(site, "v1", last_seen=timezone.now()) + v2 = _visitor(site, "v2", last_seen=timezone.now()) + s1 = _session(site, v1, "s1", referrer="www.bing.com") + s2 = _session(site, v1, "s2", referrer="www.bing.com") + s3 = _session(site, v2, "s3", referrer="chatgpt.com") + s4 = _session(site, v2, "s4", referrer="") # 直接访问 + SourceClassification.objects.create(site=site, visitor=v1, session=s1, + scope="session_entry", channel="natural_search", platform="Bing") + SourceClassification.objects.create(site=site, visitor=v1, session=s2, + scope="session_entry", channel="natural_search", platform="Bing") + SourceClassification.objects.create(site=site, visitor=v2, session=s3, + scope="session_entry", channel="ai_referral", platform="ChatGPT") + SourceClassification.objects.create(site=site, visitor=v2, session=s4, + scope="session_entry", channel="direct", resolution="unknown") + _events(site, v1, s1, 2, "/", prefix="a") + + d = client.get("/api/admin/dashboard/overview").json()["data"] + assert d["channels"] == [{"channel": "natural_search", "sessions": 2}, + {"channel": "ai_referral", "sessions": 1}, + {"channel": "direct", "sessions": 1}] # 倒序 top8，平局按名稳定 + assert d["sources"][0] == {"source": "www.bing.com", "sessions": 2} # 原口径保留 + + def test_overview_site_filter_and_404(client, member_user): _auth(client, member_user) site_a = Site.objects.create(name="a", domains=["a.com"]) diff --git a/server/tracker/tests/test_ingest.py b/server/tracker/tests/test_ingest.py index 2f0bf4e..ce1639b 100644 --- a/server/tracker/tests/test_ingest.py +++ b/server/tracker/tests/test_ingest.py @@ -142,3 +142,23 @@ def test_context_timezone_persisted_to_visitor(db): b["context"]["timezone"] = "Asia/Shanghai" ingest_batch(site, b) assert Visitor.objects.get(visitor_id="v1").timezone_name == "Asia/Shanghai" + +# 15. 归因分类 best-effort：引擎抛异常不影响采集入库（first_touch/session_entry 双路） +def test_attribution_failure_never_breaks_ingest(db, monkeypatch): + from tracker.services import ingest as ingest_mod + site = Site.objects.create(name="s", domains=["a.com"]) + monkeypatch.setattr(ingest_mod, "classify", lambda **kw: (_ for _ in ()).throw(RuntimeError("boom"))) + r = ingest_batch(site, _batch([_ev(1)])) + assert r["accepted"] == 1 and Event.objects.count() == 1 + assert Visitor.objects.count() == 1 and VisitSession.objects.count() == 1 + +# 16. 首触分类用本批第一条 page_view 的完整 referrer（搜索词可恢复），非 ctx 裸域名 +def test_first_touch_classification_uses_full_pageview_referrer(db): + from tracker.models import SourceClassification + site = Site.objects.create(name="s", domains=["a.com"]) + b = _batch([_ev(1, properties={"referrer": "https://m.baidu.com/s?wd=挖掘机"})]) + b["context"]["utm"] = {} # 默认 fixture 带 utm_source=google，与百度矛盾 → conflict + ingest_batch(site, b) + ft = SourceClassification.objects.get(scope="first_touch") + assert ft.channel == "natural_search" and ft.platform == "百度" + assert ft.search_term == "挖掘机" and ft.resolution == "resolved" diff --git a/server/tracker/tests/test_validation.py b/server/tracker/tests/test_validation.py index b2f626e..ac41756 100644 --- a/server/tracker/tests/test_validation.py +++ b/server/tracker/tests/test_validation.py @@ -37,6 +37,21 @@ def test_sanitize_url_drops_sensitive_params(): assert "email=" not in u and "token=" not in u and "utm_source=google" in u assert "monkey=1" in u # fullmatch：含 key 子串的正常参数不误伤 + +def test_pageview_referrer_url_sanitized(): + """page_view 的 properties.referrer 是完整 URL：敏感参数整键剔除后写回。""" + ev = _ev(properties={"referrer": "https://www.bing.com/search?q=x&email=a@b.com&token=t"}) + assert validate_event(ev) is None + assert ev["properties"]["referrer"] == "https://www.bing.com/search?q=x" + + +def test_pageview_referrer_truncated_after_reencode(): + """清洗重编码（中文 → %XX 九倍膨胀）可能超出原长：出门统一截 MAX_STR_LEN=500。""" + ev = _ev(properties={"referrer": "https://www.bing.com/search?q=" + "词" * 200}) + assert validate_event(ev) is None # 原文 227 字符过长度门 + ref = ev["properties"]["referrer"] + assert len(ref) == 500 and ref.startswith("https://www.bing.com/search?q=%E8%AF%8D") + def test_nested_sensitive_key_rejected(): assert validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}})) == "sensitive_field" # utm 在白名单内，嵌套敏感键仍须命中 diff --git a/server/tracker/tests/test_visitors.py b/server/tracker/tests/test_visitors.py index 1ddf157..1e2bc44 100644 --- a/server/tracker/tests/test_visitors.py +++ b/server/tracker/tests/test_visitors.py @@ -3,7 +3,7 @@ from datetime import timedelta from django.contrib.auth import get_user_model from django.utils import timezone from rest_framework.test import APIClient -from tracker.models import Event, Site, VisitSession, Visitor +from tracker.models import Event, Site, SourceClassification, VisitSession, Visitor pytestmark = pytest.mark.django_db User = get_user_model() @@ -77,7 +77,8 @@ def test_list_fields_and_default_bot_excluded(client, member_user): assert set(row) == {"visitor_id", "site_id", "site_name", "country", "province", "city", "device_type", "os", "browser", "language", "first_seen_at", "last_seen_at", "visit_count", "active_days", "session_count", - "total_stay_ms", "first_source", "last_source", "online"} + "total_stay_ms", "first_source", "last_source", "online", + "channel", "platform", "variant", "search_term", "resolution"} assert row["country"] == "中国" and row["os"] == "Windows" and row["browser"] == "Chrome" assert row["session_count"] == 2 assert row["total_stay_ms"] == 150000 # 120000+30000，有效停留口径 @@ -319,3 +320,63 @@ def test_timeline_orphan_events_without_session(client, member_user): sess = r.json()["data"]["days"][0]["sessions"][0] assert sess["session_id"] is None and sess["started_at"] is None assert [e["event_name"] for e in sess["events"]] == ["sdk_loaded", "page_stay"] + + +def _classify(site, visitor, scope, channel, platform="", session=None, **kw): + return SourceClassification.objects.create( + site=site, visitor=visitor, session=session, scope=scope, + channel=channel, platform=platform, **kw) + + +def test_list_attribution_fields_and_channel_filter(client, member_user): + _auth(client, member_user) + site = Site.objects.create(name="s", domains=["a.com"]) + v1 = _visitor(site, "v1", last_seen=timezone.now()) + v2 = _visitor(site, "v2", last_seen=timezone.now() - timedelta(hours=1)) + v3 = _visitor(site, "v3", last_seen=timezone.now() - timedelta(hours=2)) # 无分类行 + _classify(site, v1, "first_touch", "natural_search", "Bing", + search_term="挖掘机", evidence_kind="observed_referrer", + resolution="resolved") + _classify(site, v2, "first_touch", "ai_referral", "ChatGPT", + evidence_kind="observed_referrer", resolution="resolved") + + d = client.get("/api/admin/visitors").json()["data"] + by_id = {x["visitor_id"]: x for x in d["list"]} + assert by_id["v1"]["channel"] == "natural_search" + assert by_id["v1"]["platform"] == "Bing" and by_id["v1"]["search_term"] == "挖掘机" + assert by_id["v1"]["variant"] == "" and by_id["v1"]["resolution"] == "resolved" + assert by_id["v3"]["channel"] == "" # 无分类行 → 空串（前端回落 first_source 展示） + + d = client.get("/api/admin/visitors", {"channel": "natural_search"}).json()["data"] + assert d["total"] == 1 and [x["visitor_id"] for x in d["list"]] == ["v1"] + d = client.get("/api/admin/visitors", {"channel": "direct"}).json()["data"] + assert d["total"] == 0 # 无分类访客不落进任何渠道桶 + d = client.get("/api/admin/visitors", {"channel": "ai_referral"}).json()["data"] + assert [x["visitor_id"] for x in d["list"]] == ["v2"] + + +def test_detail_attribution_block(client, member_user): + _auth(client, member_user) + site = Site.objects.create(name="s", domains=["a.com"]) + v = _visitor(site, "v1") + s1 = _session(site, v, "s1", at=NOW - timedelta(hours=2), + referrer_domain="www.bing.com") + s2 = _session(site, v, "s2", at=NOW, referrer_domain="chatgpt.com") + _classify(site, v, "first_touch", "natural_search", "Bing", + search_term="挖掘机", evidence_kind="observed_referrer", + resolution="resolved") + _classify(site, v, "session_entry", "natural_search", "Bing", + session=s1, evidence_kind="observed_referrer", resolution="resolved") + _classify(site, v, "session_entry", "ai_referral", "ChatGPT", + session=s2, evidence_kind="observed_referrer", resolution="resolved") + + r = client.get("/api/admin/visitors/v1") + assert r.status_code == 200 + attr = r.json()["data"]["attribution"] + ft = attr["first_touch"] + assert ft["channel"] == "natural_search" and ft["platform"] == "Bing" + assert ft["search_term"] == "挖掘机" and ft["resolution"] == "resolved" + assert ft["evidence_kind"] == "observed_referrer" and ft["rule_version"] == "v1" + assert ft["computed_at"] and ft["detail"] == {} + assert [e["session_id"] for e in attr["session_entries"]] == ["s2", "s1"] # 最近在前 + assert attr["session_entries"][0]["channel"] == "ai_referral"