feat: 批量入库服务（幂等/首触/会话兜底/聚合更新）

Co-Authored-By: Claude Code <noreply@anthropic.com>

server/tracker/services/ingest.py

@@ -0,0 +1,145 @@
+"""批量入库服务：访客/会话 get_or_create、事件幂等写入、聚合更新。
+
+并发不变量（四轮审查收口）：
+  ① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查；
+  ② 重放命中的事件不参与会话聚合（new_events 只含未入库事件）；
+  ③ visit_count/active_days/event_count/duration_ms/stay_ms 全走 F 表达式，
+     filter().update() 避免 read-modify-write 全量 save() 用旧值覆盖并发赢家；
+  ④ 会话 ended_at 用 Greatest 只前进不回退，乱序批次不改早结束时间；
+  ⑤ 同批切分出的会话段按"原始 sid"缓存复用，后续事件不再切分。
+"""
+from datetime import timedelta
+
+from django.db import transaction, IntegrityError
+from django.db.models import F
+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.services.validation import validate_event
+from tracker.services.geo import resolver as _geo   # 模块级单例（勿实例化，11MB/份）
+
+_FIELD_LEN = {"device_type": 16, "os": 64, "browser": 64, "language": 32, "screen": 32}
+
+
+def _clean_utm(d):
+    """utm 键白名单（utm_* 前缀）+ 截断——任意字符串不得经 utm 入库渲染。"""
+    return {k[:50]: str(v)[:100] for k, v in (d or {}).items()
+            if isinstance(k, str) and k.startswith("utm_")}
+
+
+def _geo_field(g, key):
+    """geo 值归一：缺失段的 "0"/"" 占位统一为空串。geo dict 被 lru_cache 共享，只读勿改。"""
+    v = (g or {}).get(key, "")
+    return v if v not in ("", "0") else ""
+
+
+def _touch_visitor(site, vid, ctx, ip):
+    env = {k: str(ctx.get(k, "") or "")[:n] for k, n in _FIELD_LEN.items()}
+    v, created = Visitor.objects.get_or_create(
+        site=site, visitor_id=vid, defaults={**env, "last_ip": ip})
+    now = timezone.now()
+    upd = {"last_seen_at": now}   # 聚合字段一律 filter().update()，避免并发下全量 save() 用旧值覆盖赢家
+    if created:
+        upd.update(first_referrer_domain=(ctx.get("referrer_domain") or "")[:200],
+                   first_utm=_clean_utm(ctx.get("utm")),
+                   first_landing_path=(ctx.get("path") or "")[:500],
+                   last_active_date=now.date(), active_days=1)
+    elif not Visitor.objects.filter(pk=v.pk, last_active_date=now.date()).exists():
+        upd.update(last_active_date=now.date(), active_days=F("active_days") + 1)
+    if ip:
+        upd["last_ip"] = ip
+        g = _geo.resolve(ip)
+        if g:
+            upd.update(country=_geo_field(g, "country")[:64],
+                       province=_geo_field(g, "province")[:64],
+                       city=_geo_field(g, "city")[:64])
+    Visitor.objects.filter(pk=v.pk).update(**upd)
+    v.refresh_from_db()
+    return v
+
+
+def _get_session(site, visitor, sid, ctx, occurred_at):
+    """会话权威在 SDK；服务端兜底：同 session_id 距上事件>30min 切分后缀段。"""
+    idle = timedelta(milliseconds=C.SESSION_IDLE_MS)
+    sess = VisitSession.objects.filter(site=site, session_id=sid).order_by("-started_at").first()
+    if sess and occurred_at - sess.ended_at > idle:
+        suffix = 2 + VisitSession.objects.filter(
+            site=site, session_id__startswith=sid + "#").count()
+        sid = f"{sid}#{suffix}"
+        sess = None
+    if not sess:
+        try:
+            with transaction.atomic():   # savepoint：IntegrityError 后外层事务还能继续
+                sess = VisitSession.objects.create(
+                    site=site, visitor=visitor, session_id=sid,
+                    started_at=occurred_at, ended_at=occurred_at,
+                    referrer_domain=(ctx.get("referrer_domain") or "")[:200],
+                    utm=_clean_utm(ctx.get("utm")),
+                    landing_path=(ctx.get("path") or "")[:500])
+        except IntegrityError:   # 并发同 sid 同时切分出相同后缀撞 uq_site_session——重查复用
+            sess = VisitSession.objects.filter(site=site, session_id=sid).first()
+            if not sess:
+                raise
+        Visitor.objects.filter(pk=visitor.pk).update(visit_count=F("visit_count") + 1)
+    return sess
+
+
+@transaction.atomic
+def ingest_batch(site, payload: dict, ip: str | None = None, ua: str = "") -> dict:
+    now = timezone.now()
+    ctx = payload.get("context") or {}
+    events = payload.get("events") or []
+    result = {"accepted": 0, "duplicated": 0, "rejected": []}
+    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)
+    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)
+    ready, session_cache = [], {}
+    for i, ev in enumerate(events):
+        reason = validate_event(ev)
+        if reason:
+            result["rejected"].append({"index": i, "event_id": ev.get("event_id", ""),
+                                       "reason": reason})
+            continue
+        occ = ev["occurred_at"]
+        sid = payload["session_id"]
+        sess = session_cache.get(sid)
+        if not sess:
+            sess = _get_session(site, visitor, sid, ctx, occ)
+            session_cache[sid] = sess      # 按原始 sid 存：切分后 sess.session_id 是 "s1#2"，
+                                           # 若按它存，同批后续事件按 "s1" 查会 miss → 每条再切新会话
+        ready.append(Event(
+            site=site, visitor=visitor, session=sess,
+            event_id=ev["event_id"], event_name=ev["event_name"],
+            occurred_at=occ, path=(ev.get("path") or "")[:500],
+            title=(ev.get("title") or "")[:200],
+            properties=ev.get("properties") or {},
+            ip=ip, late=abs((now - occ).total_seconds()) > C.CLOCK_SKEW_LATE_SEC,
+            country=_geo_field(geo, "country"), province=_geo_field(geo, "province"),
+            city=_geo_field(geo, "city"), is_bot=is_bot))
+    if ready:
+        existed = set(Event.objects.filter(
+            site=site, event_id__in=[e.event_id for e in ready]
+        ).values_list("event_id", flat=True))
+        new_events = [e for e in ready if e.event_id not in existed]   # 重放命中的不参与聚合！
+        Event.objects.bulk_create(new_events, ignore_conflicts=True)
+        result["accepted"] = len(new_events)
+        result["duplicated"] = len(ready) - len(new_events)
+        for sess in {e.session for e in new_events}:
+            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)             # 有效停留单独累计，供访客"总停留"取数
+    return result

server/tracker/tests/test_ingest.py

@@ -0,0 +1,85 @@
+from datetime import timedelta
+from django.utils import timezone
+from tracker.models import Site, Visitor, VisitSession, Event
+from tracker.services.ingest import ingest_batch
+
+NOW = timezone.now()
+
+def _batch(events, vid="v1", sid="s1"):
+    return {"visitor_id": vid, "session_id": sid, "context": {
+        "referrer_domain": "google.com", "utm": {"utm_source": "google"},
+        "path": "/landing", "device_type": "PC", "os": "Windows",
+        "browser": "Chrome", "language": "en-US", "screen": "1920x1080"},
+        "events": events}
+
+def _ev(i, name="page_view", **kw):
+    d = dict(event_id=f"e{i}", event_name=name, occurred_at=NOW,
+             path="/p", title="T", properties={})
+    d.update(kw)
+    return d
+
+# 1. 首批建档
+def test_first_batch_creates_visitor_session_events(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    r = ingest_batch(site, _batch([_ev(1), _ev(2)]), ip="114.114.114.114")
+    assert (r["accepted"], r["duplicated"], r["rejected"]) == (2, 0, [])
+    v = Visitor.objects.get(site=site, visitor_id="v1")
+    assert v.first_referrer_domain == "google.com" and v.visit_count == 1
+    assert v.active_days == 1 and v.last_ip == "114.114.114.114"
+    assert v.country == "中国" and v.city == "南京市"    # 真实 xdb（114 DNS，江苏南京）；无 xdb 环境改为断言不抛异常即可
+    assert Event.objects.filter(site=site).count() == 2
+
+# 2. 重放幂等 + 不抬高会话聚合
+def test_duplicate_events_never_inflate_aggregates(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1), _ev(2, "page_stay", properties={"active_ms": 3000, "path": "/p"})]))
+    r = ingest_batch(site, _batch([_ev(1)]))     # 重放
+    assert r["duplicated"] == 1 and Event.objects.count() == 2
+    sess = VisitSession.objects.get()
+    assert sess.event_count == 2 and sess.stay_ms == 3000   # 重放没有重复加分
+
+# 3. 新会话递增 visit_count（F 表达式）
+def test_new_session_increments_visit_count(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1)], sid="s1"))
+    ingest_batch(site, _batch([_ev(2)], sid="s2"))
+    v = Visitor.objects.get(visitor_id="v1")
+    assert v.visit_count == 2 and VisitSession.objects.count() == 2
+
+# 4. 服务端 30 分钟兜底切分（同 sid 间隔>30min 切出 s1#2，且同批后续事件复用同段）
+def test_server_backstop_splits_idle_gap(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1, occurred_at=NOW - timedelta(minutes=90))]))
+    r = ingest_batch(site, _batch([_ev(2), _ev(3)]))       # 两条同批
+    assert VisitSession.objects.count() == 2               # s1 与 s1#2，不会切成 3 段
+    assert r["accepted"] == 2
+
+# 5. 拒绝事件带原因上报
+def test_rejected_events_reported_with_reason(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    r = ingest_batch(site, _batch([_ev(1, event_name="bad"), _ev(2)]))
+    assert r["accepted"] == 1 and r["rejected"][0]["reason"] == "event_name_not_allowed"
+
+# 6. page_stay 的 active_ms 累计进 stay_ms
+def test_stay_ms_accumulates(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, occurred_at=NOW + timedelta(minutes=1)),
+        _ev(4, "page_stay", properties={"active_ms": 7000, "path": "/p"},
+            occurred_at=NOW + timedelta(minutes=1))]))
+    assert VisitSession.objects.get().stay_ms == 12000
+
+# 7. 爬虫 UA 标记（访客+事件）
+def test_bot_ua_marked(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    ingest_batch(site, _batch([_ev(1)]), ip="66.249.66.1", ua="Mozilla/5.0 (compatible; Googlebot/2.1)")
+    assert Visitor.objects.get().is_bot is True
+    assert Event.objects.get().is_bot is True
+
+# 8. 超批拒绝
+def test_batch_size_invalid(db):
+    site = Site.objects.create(name="s", domains=["a.com"])
+    r = ingest_batch(site, _batch([_ev(i) for i in range(60)]))
+    assert r["rejected"] == [{"reason": "batch_size_invalid"}]