浅念
350 lines
19 KiB
Python
350 lines
19 KiB
Python
"""批量入库服务：访客/会话 get_or_create、事件幂等写入、聚合更新。
|
||
|
||
并发不变量（四轮审查收口）：
|
||
① event 绝不重复入库：uq_site_event + bulk_create(ignore_conflicts=True) + 入库前 existed 预查
|
||
+ 同批 event_id 去重（seen_eids）；
|
||
② 重放命中的事件不参与会话聚合（new_events 只含未入库事件）；
|
||
③ 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
|
||
import math
|
||
|
||
from django.db import transaction, IntegrityError
|
||
from django.db.models import F, Q
|
||
from django.db.models.functions import Greatest
|
||
from django.utils import timezone
|
||
|
||
from tracker import constants as C
|
||
from tracker.models import SourceClassification, Visitor, VisitSession, Event
|
||
from tracker.services.validation import validate_event, parse_iso_dt
|
||
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")
|
||
_BIGINT_MAX = 2 ** 63 - 1 # BigIntegerField 8 字节量纲上限——超范围值炸列，按 None 降级
|
||
|
||
|
||
def _ms(v):
|
||
"""v2 区间字段宽松 int 化，任何脏值一律 None（recompute 过滤 None 区间，不拒批）：
|
||
bool/非数值；NaN/Infinity（json.loads 默认收这两个字面量，int() 抛 ValueError/
|
||
OverflowError）；超 BigIntegerField 量纲的有限值（DB 写入炸列）。"""
|
||
if isinstance(v, bool) or not isinstance(v, (int, float)):
|
||
return None
|
||
if not math.isfinite(v):
|
||
return None
|
||
return int(v) if -_BIGINT_MAX <= v <= _BIGINT_MAX else None
|
||
|
||
|
||
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()
|
||
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 _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 _extended_env(ctx):
|
||
"""扩展环境键快照：白名单键 STR 化截 100，空值跳过（任意键不得经 context 入库渲染）。"""
|
||
out = {}
|
||
for k in C.EXTENDED_ENV_KEYS:
|
||
v = ctx.get(k, "")
|
||
s = str(v if v is not None else "")
|
||
if s != "":
|
||
out[k] = s[:100]
|
||
return out
|
||
|
||
|
||
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，
|
||
# 同名推导的 _FIELD_LEN 取不到值，导致 timezone_name 恒空。
|
||
env["timezone_name"] = str(ctx.get("timezone", "") or "")[:64]
|
||
model = str(ctx.get("device_model", "") or "")[:64]
|
||
v, created = Visitor.objects.get_or_create(
|
||
site=site, visitor_id=vid,
|
||
defaults={**env, "device_model": model, "env_extra": _extended_env(ctx), "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)
|
||
else:
|
||
# 首触快照语义：非首批不覆盖 env_extra 等档案字段——唯一例外是 device_model
|
||
# 为空且本批有值时补填（UA-CH hints 晚于首批发到的场景，只补写一次）
|
||
if not v.device_model and model:
|
||
upd["device_model"] = model
|
||
if 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)
|
||
return v, created # created 供首触来源分类（ingest_batch 消费）
|
||
|
||
|
||
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)
|
||
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:
|
||
sess = last
|
||
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)
|
||
# 新会话入口来源分类（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
|
||
|
||
|
||
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<from）一并跳过：参与求和会得负值，炸 PositiveBigInteger 列
|
||
if af is not None and at is not None and at >= 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()
|
||
ctx = payload.get("context") or {}
|
||
events = payload.get("events") or []
|
||
result = {"accepted": 0, "duplicated": 0, "rejected": []}
|
||
if not payload.get("visitor_id") or not payload.get("session_id"):
|
||
return {"accepted": 0, "duplicated": 0,
|
||
"rejected": [{"reason": "invalid_payload"}]}
|
||
if not events or len(events) > C.MAX_EVENTS_PER_BATCH:
|
||
return {"accepted": 0, "duplicated": 0,
|
||
"rejected": [{"reason": "batch_size_invalid"}]}
|
||
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 "")
|
||
v2 = payload.get("schema_version") == 2 # 批级信（缺省 = v1 旧 SDK，路径完全不变）
|
||
ready, session_cache, seen_eids = [], {}, set()
|
||
for i, ev in enumerate(events):
|
||
if v2 and isinstance(ev, dict) and isinstance(ev.get("occurred_at"), str):
|
||
try: # v2 事件 occurred_at 允许 ISO 字符串（SDK 线格式）；解析失败放行原值，
|
||
ev["occurred_at"] = parse_iso_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})
|
||
continue
|
||
if ev["event_id"] in seen_eids: # 同批重复 event_id：只入一份，防聚合/计数灌倍
|
||
result["duplicated"] += 1
|
||
continue
|
||
seen_eids.add(ev["event_id"])
|
||
occ = ev["occurred_at"]
|
||
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, 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"],
|
||
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")[:64], province=_geo_field(geo, "province")[:64],
|
||
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]
|
||
).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)
|
||
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)
|
||
_recompute_session_stay(sess) # stay_ms 走重算（v1 加法 + v2 区间并集），不再 F 增量
|
||
return result
|