From 8459e6fa1391837cc489062305b8b6e0bdee098b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 18 Sep 2026 11:14:06 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20ingest=20v2=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E7=BA=A7=E5=BD=92=E5=B1=9E+session=5Fresume+stay=E5=8C=BA?= =?UTF-8?q?=E9=97=B4=E5=B9=B6=E9=9B=86=E9=87=8D=E7=AE=97+=E5=93=8D?= =?UTF-8?q?=E5=BA=94=E6=98=8E=E7=BB=86=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- server/tracker/api/collect.py | 33 +++--- server/tracker/services/ingest.py | 138 +++++++++++++++++++++++--- server/tracker/services/validation.py | 18 +++- server/tracker/tests/test_collect.py | 30 +++++- server/tracker/tests/test_ingest.py | 133 +++++++++++++++++++++++++ 5 files changed, 321 insertions(+), 31 deletions(-) diff --git a/server/tracker/api/collect.py b/server/tracker/api/collect.py index 757b87f..2ec0cf4 100644 --- a/server/tracker/api/collect.py +++ b/server/tracker/api/collect.py @@ -1,6 +1,8 @@ import json from datetime import datetime, timezone as dt_tz from urllib.parse import urlsplit +from uuid import uuid4 +from django.utils import timezone as dj_timezone from rest_framework.views import APIView from rest_framework.response import Response from rest_framework.permissions import AllowAny @@ -42,20 +44,21 @@ class CollectBatchView(APIView): authentication_classes = [] def post(self, request): + request_id = uuid4().hex[:32] # 每次接收一个：响应回传 + RawBatch 同值落库（对账/去重排查） origin = request.headers.get("Origin") or "" cl = request.headers.get("Content-Length") or "0" if not cl.isdecimal() or int(cl) > 262144: # isdecimal：isdigit 放过 "²"（int 仍崩），isdecimal 与 int() 精确对齐 - return self._deny(400, "payload_too_large") # 应用层兜底（nginx 另设 client_max_body_size） + return self._deny(400, "payload_too_large", request_id) # 应用层兜底（nginx 另设 client_max_body_size） site = self._match_site(request) if site is None or not site.is_active: - return self._deny(403, "not_allowed") # 三种拒绝统一 reason，防 site_id 枚举探测 + return self._deny(403, "not_allowed", request_id) # 三种拒绝统一 reason，防 site_id 枚举探测 # urlsplit 提取 host：split("//")[-1] 会被查询串里的白名单 URL 绕过； # "null" Origin（隐私上下文/沙箱 iframe）视为无 host，走 allow_no_origin 分支 host = (urlsplit(origin).hostname or "").lower() if origin and origin != "null" else "" if host and host not in [d.lower() for d in site.domains]: - return self._deny(403, "not_allowed") + return self._deny(403, "not_allowed", request_id) if not host and not site.allow_no_origin: - return self._deny(403, "not_allowed") + return self._deny(403, "not_allowed", request_id) payload = request.data if isinstance(request.data, dict) else {} # 原始批次快照：须在 occurred_at 字符串转 datetime 之前深拷贝（datetime 不可 JSON 序列化）， # 且独立于 ingest/validate 的就地清洗（path 截断等）——存的是"到达时"的原文（脱敏后）。 @@ -65,17 +68,20 @@ class CollectBatchView(APIView): try: ev["occurred_at"] = _iso(ev["occurred_at"]) except (ValueError, TypeError): - return self._deny(400, "invalid_occurred_at") + return self._deny(400, "invalid_occurred_at", request_id) if not payload.get("visitor_id") or not payload.get("session_id"): - return self._deny(400, "invalid_payload") + return self._deny(400, "invalid_payload", request_id) ip = self._client_ip(request) r = ingest_batch(site, payload, ip, request.headers.get("User-Agent", "")) try: - self._save_raw_batch(site, payload, snapshot, r, ip) + self._save_raw_batch(site, payload, snapshot, r, ip, request_id) except Exception: # best-effort：原始表任何失败（含孤代理 encode 崩溃）不影响主采集响应 pass + # 响应契约（spec §2.3）：rejected 明细回传，HTTP 200 不再隐含全量成功 return Response({"code": 200, - "data": {"accepted": r["accepted"], "duplicated": r["duplicated"]}, + "data": {"accepted": r["accepted"], "duplicated": r["duplicated"], + "rejected": r["rejected"], "request_id": request_id, + "server_time": dj_timezone.now().isoformat()}, "message": "ok"}) def options(self, request): # 预检兜底（SDK text/plain 简单请求通常不触发）；ACAO 由 finalize_response 统一补 @@ -103,12 +109,13 @@ class CollectBatchView(APIView): "message": response.data["detail"]} return response - def _save_raw_batch(self, site, payload, snapshot, r, ip): + def _save_raw_batch(self, site, payload, snapshot, r, ip, request_id=""): """原始批次落库（每次接收一行，含被拒批次）。失败由调用方静默吞掉。 rejected（accepted=0 且有拒绝）整批截断 4KB——超批 51 事件的原文必然超限； partial（部分拒绝）不截：正常批次本就受 256KB 请求体上限约束，TextField 可容。 visitor 允许 None：被拒批次（如超批）不会创建访客记录。 + request_id 与响应同值（对账："这批原始数据对应哪个响应"）。 """ body = json.dumps(snapshot, ensure_ascii=False) status = ("rejected" if not r["accepted"] and r["rejected"] @@ -121,7 +128,7 @@ class CollectBatchView(APIView): batch_id=str(payload.get("batch_id", ""))[:64], body=body, event_count=r["accepted"], rejected_count=len(r["rejected"]) if isinstance(r["rejected"], list) else r["rejected"], - status=status, ip=ip) + status=status, ip=ip, request_id=str(request_id or "")[:64]) def _match_site(self, request): sid = (request.data or {}).get("site_id") if isinstance(request.data, dict) else None @@ -136,5 +143,7 @@ class CollectBatchView(APIView): ip = ip[7:] return ip - def _deny(self, code, reason): - return Response({"code": code, "data": None, "message": reason}, status=code) # ACAO/信封由 finalize_response 统一补 + def _deny(self, code, reason, request_id=""): + # data 带 request_id（对账）；ACAO/信封由 finalize_response 统一补 + return Response({"code": code, "data": {"request_id": request_id}, + "message": reason}, status=code) diff --git a/server/tracker/services/ingest.py b/server/tracker/services/ingest.py index 9fe525c..702f246 100644 --- a/server/tracker/services/ingest.py +++ b/server/tracker/services/ingest.py @@ -4,16 +4,26 @@ ① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查 + 同批 event_id 去重（seen_eids）； ② 重放命中的事件不参与会话聚合（new_events 只含未入库事件）； - ③ visit_count/active_days/event_count/duration_ms/stay_ms 全走 F 表达式， - filter().update() 避免 read-modify-write 全量 save() 用旧值覆盖并发赢家； + ③ visit_count/active_days/event_count/duration_ms 走 F 表达式，filter().update() 避免 + read-modify-write 全量 save() 用旧值覆盖并发赢家；stay_ms 是唯一例外——schema v2 起 + 改为落库后按 Event 表重算（v1 active_ms 加法 + v2 page_id 区间并集，见 + _recompute_session_stay；15s 周期切分的相邻区间须并集拼回，增量累加会重复计）， + 写入仍是单行 filter().update()； ④ 会话 ended_at 用 Greatest 只前进不回退，乱序批次不改早结束时间； ⑤ 同批切分出的会话段按"原始 sid"缓存复用，后续事件不再切分。 +schema v2（阶段A）：批级 schema_version==2 时事件自带归属四元组（site/visitor/session/ +page_id），会话按事件级 session_id 解析（v1 信封不再是唯一依据）；v2 不做 30min 服务端 +兜底切分（会话切换由 SDK 显式 session_resume 驱动，避免双切）；session_resume 创建的 +新会话经 resume_of 链接前会话。v1 路径完全保留。 + 已接受的残余风险（M1 不修）：existed 预查与 bulk_create 之间存在毫秒级 TOCTOU 窗口—— 并发请求在预查后、写入前提交同一 event_id 时，本请求的 ignore_conflicts 静默吞掉冲突， 但该事件已被计入 new_events 参与聚合，聚合可能双计。窗口极窄且仅影响并发重放场景。 +（v2 附注：stay_ms 重算对"并发同会话双批"同样存在同类窗口——后提交者的 SELECT 可能 +看不到前者未提交事件而少算，与 TOCTOU 同级、只影响 stay_ms 单字段。） """ -from datetime import timedelta +from datetime import datetime, timedelta, timezone as dt_tz from django.db import transaction, IntegrityError from django.db.models import F, Q @@ -31,6 +41,35 @@ _CLASS_FIELDS = ("channel", "platform", "variant", "search_term", "evidence_kind", "resolution", "detail") +def _occ_dt(v): + """v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式；collect 视图已预转 datetime， + 直连调用方可能传原文）。解析失败抛 ValueError/TypeError——调用方放行原值， + 由 validate_event 统一拒 invalid_occurred_at。naive 视为 UTC（与 collect._iso 同口径）。""" + if isinstance(v, str): + d = datetime.fromisoformat(v.replace("Z", "+00:00")) + return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc) + return v + + +def _ms(v): + """v2 区间字段宽松 int 化：bool/非数值 → None（脏数据防御——recompute 会过滤 None 区间， + 该事件不参与并集但不拒批）。""" + if isinstance(v, bool) or not isinstance(v, (int, float)): + return None + return int(v) + + +def _time_quality(now, occ) -> str: + """v2 时间质量分类：received−occurred 差 d 秒——d>+120s 迟到 "late"、d<-120s 时钟倒偏 + "skew_backwards"、否则 "ok"。v1 事件留空（late 布尔已有，不动）。""" + d = (now - occ).total_seconds() + if d > C.CLOCK_SKEW_LATE_SEC: + return "late" + if d < -C.CLOCK_SKEW_LATE_SEC: + return "skew_backwards" + return "ok" + + def _clean_utm(d): """utm 键白名单（utm_* 前缀）+ 截断——任意字符串不得经 utm 入库渲染。""" return {k[:50]: str(v)[:100] for k, v in (d or {}).items() @@ -134,17 +173,26 @@ def _touch_visitor(site, vid, ctx, ip): return v, created # created 供首触来源分类（ingest_batch 消费） -def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url=""): +def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url="", + skip_idle_split=False): """会话权威在 SDK；服务端兜底：同 session_id 族（s1、s1#2…）距最后事件>30min 切分后缀段。 空闲比较必须取整个会话族的最大 ended_at：若只查根段（session_id=sid 精确匹配）， 切分一次后根段 ended_at 陈旧，后续小间隔批次会反复误判超时、每批新开段（visit_count 灌倍）。 + + skip_idle_split=True（schema v2）：SDK 显式管理会话（session_resume 驱动切换），服务端 + 兜底不介入（避免双切）；"复用现有段"能力保留——直接按 sid 精确查/建（uq_site_session + 保证至多一行），不复用 sid#N 族。 """ idle = timedelta(milliseconds=C.SESSION_IDLE_MS) - fam = list(VisitSession.objects.filter( - Q(session_id=sid) | Q(session_id__startswith=sid + "#"), site=site)) - last = max(fam, key=lambda s: s.ended_at) if fam else None - if last and occurred_at - last.ended_at > idle: + if skip_idle_split: + fam = [] + last = VisitSession.objects.filter(site=site, session_id=sid).first() + else: + fam = list(VisitSession.objects.filter( + Q(session_id=sid) | Q(session_id__startswith=sid + "#"), site=site)) + last = max(fam, key=lambda s: s.ended_at) if fam else None + if last and not skip_idle_split and occurred_at - last.ended_at > idle: suffix = 2 + sum(1 for s in fam if s.session_id.startswith(sid + "#")) sid, sess = f"{sid}#{suffix}", None else: @@ -173,6 +221,39 @@ def _get_session(site, visitor, sid, ctx, occurred_at, ua="", referrer_url=""): return sess +def _recompute_session_stay(sess): + """stay_ms 唯一聚合路径（invariant ③ 唯一例外）：落库后从 Event 表重算回写。 + + v1/v2 混合的单一解释：v1 事件（page_id 空）沿用 active_ms 加法；v2 事件按 page_id + 分组做 (active_from_ms, active_to_ms) 区间并集后求和；两者相加。重算式而非增量—— + SDK 每 15s 切一刀的相邻区间须并集拼回（增量累加会重复计）；uq_site_event 存储层 + 去重保证重算不重复。缺区间/倒置区间的 v2 脏事件整条跳过，不参与并集也不拒批。 + v1-only 会话重算值与旧增量语义恒等（v1_total 即历史 active_ms 全和）。 + """ + v1_total, by_page = 0, {} + for page_id, af, at, props in Event.objects.filter( + session=sess, event_name="page_stay").values_list( + "page_id", "active_from_ms", "active_to_ms", "properties"): + if page_id: + # 倒置区间（to= af: + by_page.setdefault(page_id, []).append((af, at)) + else: + v1_total += int((props or {}).get("active_ms", 0) or 0) + union_total = 0 + for ivs in by_page.values(): + ivs.sort() + cur_s, cur_e = ivs[0] + for s, e in ivs[1:]: + if s <= cur_e: # 相邻/重叠区间拼接（15s 周期切一刀的场景） + cur_e = max(cur_e, e) + else: + union_total += cur_e - cur_s + cur_s, cur_e = s, e + union_total += cur_e - cur_s + VisitSession.objects.filter(pk=sess.pk).update(stay_ms=v1_total + union_total) + + @transaction.atomic def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> dict: now = timezone.now() @@ -198,9 +279,16 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di 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 "") + v2 = payload.get("schema_version") == 2 # 批级信（缺省 = v1 旧 SDK，路径完全不变） ready, session_cache, seen_eids = [], {}, set() for i, ev in enumerate(events): - reason = validate_event(ev) + if v2 and isinstance(ev, dict) and isinstance(ev.get("occurred_at"), str): + try: # v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式）；解析失败放行原值， + ev["occurred_at"] = _occ_dt(ev["occurred_at"]) # validate_event 统一拒 invalid_occurred_at + except (ValueError, TypeError): + pass + reason = validate_event(ev, v2=v2, batch_site_id=site.site_id, + batch_visitor_id=payload["visitor_id"]) if reason: result["rejected"].append({"index": i, "event_id": ev.get("event_id", ""), "reason": reason}) @@ -210,13 +298,33 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di continue seen_eids.add(ev["event_id"]) occ = ev["occurred_at"] - sid = payload["session_id"] + sid = ev["session_id"] if v2 else payload["session_id"] # v2：归属按事件级字段解析 sess = session_cache.get(sid) if not sess: sess = _get_session(site, visitor, sid, ctx, occ, ua=ua or "", - referrer_url=pv_referrer) + referrer_url=pv_referrer, skip_idle_split=v2) session_cache[sid] = sess # 按原始 sid 存：切分后 sess.session_id 是 "s1#2"， # 若按它存，同批后续事件按 "s1" 查会 miss → 每条再切新会话 + if v2 and ev["event_name"] == "session_resume" and sess.resume_of_id is None: + # 续接链：resume_of 指向 prev_session_id 对应会话；prev 无效/不存在软处理（None，不拒）。 + # resume_of__isnull=True 条件幂等——重放/同批多条 resume 不覆写已设的链。 + prev = (ev.get("properties") or {}).get("prev_session_id") + prev_sess = (VisitSession.objects.filter(site=site, session_id=prev).first() + if isinstance(prev, str) and prev else None) + if prev_sess and prev_sess.pk != sess.pk: + VisitSession.objects.filter(pk=sess.pk, resume_of__isnull=True).update( + resume_of=prev_sess) + sess.resume_of = prev_sess # 刷新内存态：同批同 sid 后续 resume 不再重查重设 + extra = {} + if v2: + props = ev.get("properties") or {} + extra = {"page_id": str(ev.get("page_id") or "")[:64], + "active_from_ms": _ms(props.get("active_from")), + "active_to_ms": _ms(props.get("active_to")), + "flush_reason": str(props.get("flush_reason") or "")[:16], + "time_quality": _time_quality(now, occ), + "sdk_version": str(payload.get("sdk_version") or "")[:16], + "schema_version": 2} ready.append(Event( site=site, visitor=visitor, session=sess, event_id=ev["event_id"], event_name=ev["event_name"], @@ -225,7 +333,7 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di properties=ev.get("properties") or {}, ip=ip, late=abs((now - occ).total_seconds()) > C.CLOCK_SKEW_LATE_SEC, country=_geo_field(geo, "country")[:64], province=_geo_field(geo, "province")[:64], - city=_geo_field(geo, "city")[:64], is_bot=is_bot)) + city=_geo_field(geo, "city")[:64], is_bot=is_bot, **extra)) if ready: existed = set(Event.objects.filter( site=site, event_id__in=[e.event_id for e in ready] @@ -238,11 +346,9 @@ def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> di se = [e for e in new_events if e.session.pk == sess.pk] new_end = max(x.occurred_at for x in se) span_ms = int((new_end - min(x.occurred_at for x in se)).total_seconds() * 1000) - stay_add = sum(int(e.properties.get("active_ms", 0) or 0) - for e in se if e.event_name == "page_stay") VisitSession.objects.filter(pk=sess.pk).update( ended_at=Greatest(F("ended_at"), new_end), # 只前进不回退，防乱序批次改早结束时间 event_count=F("event_count") + len(se), - duration_ms=F("duration_ms") + span_ms, - stay_ms=F("stay_ms") + stay_add) # 有效停留单独累计，供访客"总停留"取数 + duration_ms=F("duration_ms") + span_ms) + _recompute_session_stay(sess) # stay_ms 走重算（v1 加法 + v2 区间并集），不再 F 增量 return result diff --git a/server/tracker/services/validation.py b/server/tracker/services/validation.py index 258c4ef..fcf4c28 100644 --- a/server/tracker/services/validation.py +++ b/server/tracker/services/validation.py @@ -77,14 +77,28 @@ def sanitize_payload(obj): return obj -def validate_event(ev) -> str | None: +def validate_event(ev, *, v2=False, batch_site_id="", batch_visitor_id="") -> str | None: """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。 通过时会就地截断/清洗并写回 ev（path 先 sanitize_url 再截 MAX_PATH_LEN，title 截 - MAX_TITLE_LEN）——Task 6 的 ingest 依赖此写回契约。""" + MAX_TITLE_LEN）——Task 6 的 ingest 依赖此写回契约。 + + v2=True（批级 schema_version==2）：额外校验事件级归属四元组——site_id/visitor_id + 必须与批级一致、session_id 非空 str（归属依据从 v1 信封转移到事件级）； + page_id 必须为非空 ≤64 的合法 UTF-8（session_resume 可空——它归属会话而非页面）。 + 不一致 → attribution_mismatch；page_id 非法 → invalid_page_id。""" if not isinstance(ev, dict): return "event_invalid_shape" if ev.get("event_name") not in C.EVENT_NAMES: return "event_name_not_allowed" + if v2: + if (ev.get("site_id") != batch_site_id + or ev.get("visitor_id") != batch_visitor_id + or not isinstance(ev.get("session_id"), str) or not ev["session_id"]): + return "attribution_mismatch" + pid = ev.get("page_id") + if ev["event_name"] != "session_resume" and ( + not isinstance(pid, str) or not pid or len(pid) > 64 or not _utf8_ok(pid)): + return "invalid_page_id" eid = ev.get("event_id") if not isinstance(eid, str) or not eid or len(eid) > 64 or not _utf8_ok(eid): return "invalid_event_id" diff --git a/server/tracker/tests/test_collect.py b/server/tracker/tests/test_collect.py index cfe9b3f..99160d9 100644 --- a/server/tracker/tests/test_collect.py +++ b/server/tracker/tests/test_collect.py @@ -24,7 +24,11 @@ def _body(n=1, sid="s1", vid="v1"): def test_collect_ok_creates_events(db, client): Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x") r = _post(client, _body(2)) - assert r.status_code == 200 and r.json() == {"code": 200, "data": {"accepted": 2, "duplicated": 0}, "message": "ok"} + assert r.status_code == 200 + body = r.json() # 包含式断言：v2 响应新增 rejected/request_id/server_time 字段 + assert body["code"] == 200 and body["message"] == "ok" + assert body["data"]["accepted"] == 2 and body["data"]["duplicated"] == 0 + assert body["data"]["rejected"] == [] and body["data"]["request_id"] assert r["Access-Control-Allow-Origin"] == "https://www.a.com" def test_origin_not_whitelisted_rejected(db, client): @@ -97,3 +101,27 @@ def test_malformed_content_length_400_not_500(db, client): Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x") r = _post(client, _body(1), extra={"CONTENT_LENGTH": "abc"}) assert r.status_code == 400 and r.json()["message"] == "payload_too_large" + + +# ---- 响应契约（阶段A Task3）---- + +def test_response_contract_rejected_details(db, client): + """部分拒绝批：rejected 明细（index/event_id/reason）回传——HTTP 200 不再隐含全量成功。""" + Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x") + b = _body(2) + b["events"][0]["event_name"] = "bad_name" + r = _post(client, b) + assert r.status_code == 200 + data = r.json()["data"] + assert data["accepted"] == 1 and data["duplicated"] == 0 + assert data["rejected"] == [{"index": 0, "event_id": "e0", + "reason": "event_name_not_allowed"}] + +def test_request_id_in_response_and_raw_batch(db, client): + """request_id（uuid4 hex 32）随响应回传，且与 RawBatch.request_id 同值落库（对账）。""" + from tracker.models import RawBatch + Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x") + r = _post(client, _body(1)) + rid = r.json()["data"]["request_id"] + assert isinstance(rid, str) and len(rid) == 32 + assert RawBatch.objects.get().request_id == rid diff --git a/server/tracker/tests/test_ingest.py b/server/tracker/tests/test_ingest.py index 58d837a..08078d3 100644 --- a/server/tracker/tests/test_ingest.py +++ b/server/tracker/tests/test_ingest.py @@ -217,3 +217,136 @@ def test_device_model_backfilled_on_later_batch(db): ingest_batch(site, b3) v.refresh_from_db() assert v.device_model == "Pixel 7" # 已有值不再覆盖 + + +# ---- schema v2（阶段A Task3：事件级归属 / session_resume / 区间并集 / 时间质量）---- + +SITE_ID = "site_v2" + +def _now_iso(): + return timezone.now().isoformat() + +def _v2_event(name, page_id, sid, seq=1, **props): + return {"event_id": f"e-{name}-{seq}", "event_name": name, + "occurred_at": _now_iso(), "path": "/p", "title": "t", + "site_id": SITE_ID, "visitor_id": "v1", "session_id": sid, + "page_id": page_id, "seq": seq, "properties": props} + +def _v2_batch(events, vid="v1", sid="s1"): + return {"schema_version": 2, "site_id": SITE_ID, "visitor_id": vid, + "session_id": sid, "sdk_version": "0.2.0", + "context": {"referrer_domain": "", "path": "/", "device_type": "PC"}, + "events": events} + +# 20. v2 事件级归属：同批多 session 各自落库（v1 信封 session_id 不再是唯一依据）； +# 事件级 site/visitor 与批级不一致 → attribution_mismatch +def test_v2_event_level_attribution(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + evs = [_v2_event("page_view", "p1", "sA", seq=1), + _v2_event("page_view", "p2", "sB", seq=2)] + r = ingest_batch(site, _v2_batch(evs, sid="sA")) + assert (r["accepted"], r["duplicated"], r["rejected"]) == (2, 0, []) + assert VisitSession.objects.count() == 2 + for sid_name, pid in (("sA", "p1"), ("sB", "p2")): + sess = VisitSession.objects.get(session_id=sid_name) + e = Event.objects.get(session=sess) + assert e.page_id == pid and e.schema_version == 2 + bad_v = _v2_event("page_view", "p3", "sA", seq=3) + bad_v["visitor_id"] = "v-other" # 事件级 visitor ≠ 批级 + bad_s = _v2_event("page_view", "p4", "sA", seq=4) + bad_s["site_id"] = "site_other" # 事件级 site ≠ 批级 + r2 = ingest_batch(site, _v2_batch( + [bad_v, bad_s, _v2_event("page_view", "p5", "sA", seq=5)], sid="sA")) + assert r2["accepted"] == 1 + assert r2["rejected"] == [ + {"index": 0, "event_id": "e-page_view-3", "reason": "attribution_mismatch"}, + {"index": 1, "event_id": "e-page_view-4", "reason": "attribution_mismatch"}] + +# 21. session_resume：新会话 resume_of 链接 prev_session_id 对应会话； +# prev 不存在软处理（resume_of=None 不拒）；v2 不做 30min 兜底切分 +def test_session_resume_creates_linked_session(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + ingest_batch(site, _v2_batch([_v2_event("page_view", "p1", "s1", seq=1)], sid="s1")) + r = ingest_batch(site, _v2_batch([ + _v2_event("session_resume", "", "s2", seq=2, prev_session_id="s1", reason="idle_timeout"), + _v2_event("page_view", "p2", "s2", seq=3)], sid="s2")) + assert (r["accepted"], r["rejected"]) == (2, []) + s1 = VisitSession.objects.get(session_id="s1") + s2 = VisitSession.objects.get(session_id="s2") + assert s2.resume_of == s1 + r2 = ingest_batch(site, _v2_batch([ + _v2_event("session_resume", "", "s3", seq=4, prev_session_id="ghost", reason="idle_timeout"), + _v2_event("page_view", "p3", "s3", seq=5)], sid="s3")) + assert (r2["accepted"], r2["rejected"]) == (2, []) # prev 无效不拒 + assert VisitSession.objects.get(session_id="s3").resume_of is None + assert Visitor.objects.get(visitor_id="v1").visit_count == 3 # s1/s2/s3 + # 35min 后同 sid 再发：不切 s2#2（v2 会话切换由 SDK 显式 resume 驱动） + late_ev = _v2_event("page_view", "p2b", "s2", seq=6) + late_ev["occurred_at"] = (timezone.now() + timedelta(minutes=35)).isoformat() + r3 = ingest_batch(site, _v2_batch([late_ev], sid="s2")) + assert r3["accepted"] == 1 + assert VisitSession.objects.filter(session_id__startswith="s2").count() == 1 + +# 22. v2 page_stay 区间并集（非加法）：[0,15][15,30][40,55]s → 0-30 ∪ 40-55 = 45s +def test_stay_interval_union_not_sum(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + secs = [(0, 15), (15, 30), (40, 55)] + evs = [_v2_event("page_stay", "p1", "s1", seq=i + 1, + active_from=a * 1000, active_to=b * 1000, flush_reason="periodic") + for i, (a, b) in enumerate(secs)] + r = ingest_batch(site, _v2_batch(evs, sid="s1")) + assert (r["accepted"], r["rejected"]) == (3, []) + e1 = Event.objects.get(event_id="e-page_stay-1") + assert (e1.active_from_ms, e1.active_to_ms) == (0, 15000) + assert e1.flush_reason == "periodic" and e1.sdk_version == "0.2.0" + assert VisitSession.objects.get(session_id="s1").stay_ms == 45000 # 非 sum=60000 + +# 23. v1 路径完全不变：stay_ms 仍为 active_ms 加法，v2 新字段留空 +def test_v1_events_unchanged_path(db): + site = Site.objects.create(name="s", domains=["a.com"]) + ingest_batch(site, _batch([_ev(1), _ev(2, "page_stay", properties={"active_ms": 5000, "path": "/p"})])) + ingest_batch(site, _batch([_ev(3, "page_stay", properties={"active_ms": 3000, "path": "/p"})])) + sess = VisitSession.objects.get() + assert sess.stay_ms == 8000 and sess.event_count == 3 + e = Event.objects.get(event_id="e1") + assert e.schema_version is None and e.page_id == "" and e.time_quality == "" + assert e.active_from_ms is None and e.active_to_ms is None + +# 24. time_quality 分类：正常 ok / >120s 过去 late / 未来小偏移 ok / 未来超阈 skew_backwards +def test_time_quality_classification(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + cases = [(1, 0, "ok"), (2, -130, "late"), (3, 5, "ok"), (4, 130, "skew_backwards")] + evs = [] + for seq, off, _ in cases: + e = _v2_event("page_view", f"p{seq}", "s1", seq=seq) + e["occurred_at"] = (timezone.now() + timedelta(seconds=off)).isoformat() + evs.append(e) + r = ingest_batch(site, _v2_batch(evs, sid="s1")) + assert (r["accepted"], r["rejected"]) == (4, []) + for seq, off, want in cases: + e = Event.objects.get(event_id=f"e-page_view-{seq}") + assert e.time_quality == want, f"seq={seq} offset={off}s" + assert e.late == (abs(off) > 120) # late 布尔语义不动 + +# 25. 混合解释：同会话 v1 active_ms 加法 + v2 区间并集 相加（单一解释） +def test_recompute_mixed_v1_v2(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + ingest_batch(site, _batch( + [_ev(1), _ev(2, "page_stay", properties={"active_ms": 10000, "path": "/p"})], sid="sm")) + evs = [_v2_event("page_stay", "px", "sm", seq=1, active_from=0, active_to=10000, flush_reason="periodic"), + _v2_event("page_stay", "py", "sm", seq=2, active_from=0, active_to=5000, flush_reason="periodic")] + r = ingest_batch(site, _v2_batch(evs, sid="sm")) + assert (r["accepted"], r["rejected"]) == (2, []) + assert VisitSession.objects.get(session_id="sm").stay_ms == 10000 + 15000 # v1 SUM + v2 并集 + +# 26. 脏区间防御：非数值/倒置区间不参与并集、不拒批（防 stay_ms 负值炸 PositiveBigInteger） +def test_dirty_intervals_soft_skipped(db): + site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID) + evs = [_v2_event("page_stay", "p1", "s1", seq=1, active_from=0, active_to=10000, flush_reason="periodic"), + _v2_event("page_stay", "p2", "s1", seq=2, active_from="x", active_to=9000, flush_reason="periodic"), + _v2_event("page_stay", "p3", "s1", seq=3, active_from=20000, active_to=10000, flush_reason="periodic")] + r = ingest_batch(site, _v2_batch(evs, sid="s1")) + assert (r["accepted"], r["rejected"]) == (3, []) # 脏数据软处理不拒批 + e2 = Event.objects.get(event_id="e-page_stay-2") + assert e2.active_from_ms is None # 非数值 int 化为 None + assert VisitSession.objects.get(session_id="s1").stay_ms == 10000 # 仅 p1 有效区间