From 6147f2b9ff7488008b5f7fe2a988598a339e0fee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 15 Sep 2026 12:44:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=89=B9=E9=87=8F=E5=85=A5=E5=BA=93?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=EF=BC=88=E5=B9=82=E7=AD=89/=E9=A6=96?= =?UTF-8?q?=E8=A7=A6/=E4=BC=9A=E8=AF=9D=E5=85=9C=E5=BA=95/=E8=81=9A?= =?UTF-8?q?=E5=90=88=E6=9B=B4=E6=96=B0=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- server/tracker/services/ingest.py | 145 ++++++++++++++++++++++++++++ server/tracker/tests/test_ingest.py | 85 ++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 server/tracker/services/ingest.py create mode 100644 server/tracker/tests/test_ingest.py diff --git a/server/tracker/services/ingest.py b/server/tracker/services/ingest.py new file mode 100644 index 0000000..6e206df --- /dev/null +++ b/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 diff --git a/server/tracker/tests/test_ingest.py b/server/tracker/tests/test_ingest.py new file mode 100644 index 0000000..8e08dbc --- /dev/null +++ b/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"}]