feat: 原始上报数据存储（raw_batch 表/脱敏/访客关联/详情页 Tab/30天清理）

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

server/tracker/api/collect.py

@@ -7,8 +7,9 @@ from rest_framework.permissions import AllowAny
 from rest_framework.throttling import AnonRateThrottle
 from rest_framework.parsers import BaseParser
 from rest_framework.settings import api_settings
-from tracker.models import Site
+from tracker.models import RawBatch, Site, Visitor
 from tracker.services.ingest import ingest_batch
+from tracker.services.validation import sanitize_payload
 
 class PlainJSONParser(BaseParser):
     media_type = "text/plain"
@@ -56,6 +57,9 @@ class CollectBatchView(APIView):
         if not host and not site.allow_no_origin:
             return self._deny(403, "not_allowed")
         payload = request.data if isinstance(request.data, dict) else {}
+        # 原始批次快照：须在 occurred_at 字符串转 datetime 之前深拷贝（datetime 不可 JSON 序列化），
+        # 且独立于 ingest/validate 的就地清洗（path 截断等）——存的是"到达时"的原文（脱敏后）。
+        snapshot = sanitize_payload(json.loads(json.dumps(payload, ensure_ascii=False)))
         for ev in payload.get("events") or []:
             if isinstance(ev, dict) and "occurred_at" in ev:
                 try:
@@ -66,6 +70,10 @@ class CollectBatchView(APIView):
             return self._deny(400, "invalid_payload")
         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)
+        except Exception:   # best-effort：原始表任何失败（含孤代理 encode 崩溃）不影响主采集响应
+            pass
         return Response({"code": 200,
             "data": {"accepted": r["accepted"], "duplicated": r["duplicated"]},
             "message": "ok"})
@@ -95,6 +103,26 @@ class CollectBatchView(APIView):
                              "message": response.data["detail"]}
         return response
 
+    def _save_raw_batch(self, site, payload, snapshot, r, ip):
+        """原始批次落库（每次接收一行，含被拒批次）。失败由调用方静默吞掉。
+
+        rejected（accepted=0 且有拒绝）整批截断 4KB——超批 51 事件的原文必然超限；
+        partial（部分拒绝）不截：正常批次本就受 256KB 请求体上限约束，TextField 可容。
+        visitor 允许 None：被拒批次（如超批）不会创建访客记录。
+        """
+        body = json.dumps(snapshot, ensure_ascii=False)
+        status = ("rejected" if not r["accepted"] and r["rejected"]
+                  else "partial" if r["rejected"] else "accepted")
+        if status == "rejected" and len(body) > 4096:
+            body = body[:4096] + "...[TRUNCATED]"
+        vis = Visitor.objects.filter(site=site, visitor_id=payload.get("visitor_id")).first()
+        RawBatch.objects.create(
+            site=site, visitor=vis,
+            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)
+
     def _match_site(self, request):
         sid = (request.data or {}).get("site_id") if isinstance(request.data, dict) else None
         return Site.objects.filter(site_id=sid).first() if sid else None

server/tracker/api/visitors.py

@@ -13,9 +13,10 @@ from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from tracker.api.envelope import EnvelopeMixin, ok
-from tracker.models import Event, Site, VisitSession, Visitor
+from tracker.models import Event, RawBatch, Site, VisitSession, Visitor
 
 PAGE_SIZE = 20
+RAW_PAGE_SIZE = 10   # 原始批次 body 较大，页小一些
 TIMELINE_DEFAULT_DAYS = 30
 TIMELINE_MAX_DAYS = 90
 TIMELINE_MAX_EVENTS = 1000
@@ -246,3 +247,30 @@ class VisitorSessionView(EnvelopeMixin, APIView):
             return _not_found("session_not_found")
         events = Event.objects.filter(session=session).order_by("occurred_at", "pk")  # (session,occurred_at) 索引
         return ok({**_session_summary(session), "events": _merge_page_stay(events)})
+
+
+class VisitorRawBatchesView(EnvelopeMixin, APIView):
+    """访客原始上报批次（raw_batch）：排查"上报了为何没进报表"。
+
+    body 是脱敏后的整包 JSON 文本（敏感值已 [FILTERED]、被拒批次已截断），
+    前端以 <pre> 文本渲染（禁 v-html）。按 -received_at 倒序分页，走 (visitor,-received_at) 索引。
+    """
+
+    def get(self, request, visitor_id):
+        visitor = _get_visitor(visitor_id)
+        if not visitor:
+            return _not_found("visitor_not_found")
+        qs = RawBatch.objects.filter(visitor=visitor)
+        total = qs.count()
+        try:
+            page = int(request.query_params.get("page", 1))
+        except (TypeError, ValueError):
+            page = 1
+        page = max(1, page)
+        batches = qs.order_by("-received_at", "-pk")[
+            (page - 1) * RAW_PAGE_SIZE: page * RAW_PAGE_SIZE]
+        return ok({"list": [{"id": b.id, "batch_id": b.batch_id, "status": b.status,
+                             "event_count": b.event_count, "rejected_count": b.rejected_count,
+                             "received_at": _iso(b.received_at), "body": b.body}
+                            for b in batches],
+                   "total": total, "page": page})

server/tracker/management/commands/cleanup_raw.py

@@ -0,0 +1,19 @@
+from datetime import timedelta
+
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+from tracker.models import RawBatch
+
+
+class Command(BaseCommand):
+    # 生产调度：1Panel 计划任务每日执行（与 refresh_stats 同款）
+    #   docker exec <web容器> python manage.py cleanup_raw --days 30
+    help = "清理 N 天前的原始上报批次（raw_batch 默认保留 30 天）"
+
+    def add_arguments(self, parser):
+        parser.add_argument("--days", type=int, default=30)
+
+    def handle(self, *args, **opts):
+        cutoff = timezone.now() - timedelta(days=opts["days"])
+        deleted, _ = RawBatch.objects.filter(received_at__lt=cutoff).delete()
+        self.stdout.write(f"deleted {deleted} raw batches")

server/tracker/migrations/0003_rawbatch.py

@@ -0,0 +1,33 @@
+# Generated by Django 5.2.17 on 2026-09-16 07:33
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tracker', '0002_alter_dailystat_site_alter_event_session_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='RawBatch',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('batch_id', models.CharField(blank=True, default='', max_length=64)),
+                ('body', models.TextField()),
+                ('event_count', models.PositiveIntegerField(default=0)),
+                ('rejected_count', models.PositiveIntegerField(default=0)),
+                ('status', models.CharField(default='accepted', max_length=16)),
+                ('received_at', models.DateTimeField(auto_now_add=True)),
+                ('ip', models.GenericIPAddressField(blank=True, null=True)),
+                ('site', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='raw_batches', to='tracker.site')),
+                ('visitor', models.ForeignKey(null=True, on_delete=django.db.models.deletion.CASCADE, related_name='raw_batches', to='tracker.visitor')),
+            ],
+            options={
+                'db_table': 'raw_batch',
+                'indexes': [models.Index(fields=['visitor', '-received_at'], name='raw_batch_visitor_9ea281_idx'), models.Index(fields=['site', '-received_at'], name='raw_batch_site_id_ad8e9f_idx')],
+            },
+        ),
+    ]

server/tracker/models.py

@@ -107,6 +107,30 @@ class Event(models.Model):
         ]
 
 
+class RawBatch(models.Model):
+    """原始上报批次存档：每次 collect/batch 接收都落一行（含被拒批次）。
+
+    用途：排查"事件为何没进报表"——原始 JSON 与入库结果（event_count/rejected_count/status）
+    并排可对照白名单/超批/幂等拒绝原因。body 是脱敏后快照（敏感键值 → [FILTERED]，结构保留）；
+    被拒批次截断 4KB。保留期 30 天由 cleanup_raw 命令管理（生产 cron 每日执行）。
+    写入是 best-effort：失败不影响主采集流程（见 collect.py）。
+    """
+    site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="raw_batches")
+    visitor = models.ForeignKey(Visitor, on_delete=models.CASCADE, related_name="raw_batches", null=True)
+    batch_id = models.CharField(max_length=64, blank=True, default="")
+    body = models.TextField()               # 脱敏后的整包 JSON 文本
+    event_count = models.PositiveIntegerField(default=0)
+    rejected_count = models.PositiveIntegerField(default=0)
+    status = models.CharField(max_length=16, default="accepted")  # accepted/partial/rejected
+    received_at = models.DateTimeField(auto_now_add=True)
+    ip = models.GenericIPAddressField(null=True, blank=True)
+
+    class Meta:
+        db_table = "raw_batch"
+        indexes = [models.Index(fields=["visitor", "-received_at"]),
+                   models.Index(fields=["site", "-received_at"])]
+
+
 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()

server/tracker/services/validation.py

@@ -51,6 +51,19 @@ def _walk_props(v, depth=0):
         return True, False, True
     return True, False, False
 
+def sanitize_payload(obj):
+    """原始批次快照脱敏：递归遍历 dict/list，键名命中 SENSITIVE_KEY_PATTERNS（大小写不敏感）
+    → 值替换为字符串 "[FILTERED]"（结构保留，便于排查时仍看清字段形态）。
+    返回新对象（容器全部重建，叶值原样共享）——调用方在 ingest 之前对 payload 深拷贝后传入，
+    与 ingest/validate_event 的就地改写互不影响。"""
+    if isinstance(obj, dict):
+        return {k: ("[FILTERED]" if _SENSITIVE.search(str(k)) else sanitize_payload(v))
+                for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [sanitize_payload(x) for x in obj]
+    return obj
+
+
 def validate_event(ev) -> str | None:
     """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。
     通过时会就地截断/清洗并写回 ev（path 先 sanitize_url 再截 MAX_PATH_LEN，title 截

server/tracker/tests/test_raw.py

@@ -0,0 +1,166 @@
+"""原始上报批次（raw_batch）：脱敏快照、拒批截断、重放留痕、API 分页、30 天清理。
+
+设计契约（与用户确认）：
+- 敏感字段值替换为 [FILTERED]（结构保留）——sanitize_payload 复用 SENSITIVE_KEY_PATTERNS；
+- 被拒批次也存（body 截断 4KB）；invalid_json（解析失败）不存；
+- 原始表写入失败不得影响主采集流程（best-effort）；
+- 每次接收都记一行（"原始记录"语义，重放不合并）。
+"""
+import json
+from datetime import timedelta
+from io import StringIO
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.core.cache import cache
+from django.core.management import call_command
+from django.utils import timezone
+from rest_framework.test import APIClient
+from tracker.models import RawBatch, Site, Visitor
+
+User = get_user_model()
+NOW = timezone.now()
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache():
+    cache.clear()   # locmem 限流历史跨测试残留，会让压测 429（与 test_collect 同款）
+
+
+@pytest.fixture
+def client():   # 覆盖 pytest-django 的 Django Client——认证用例需要 credentials()
+    return APIClient()
+
+
+@pytest.fixture
+def member_user():
+    return User.objects.create_user(username="member", password="memberpass123")
+
+
+def _auth(client, user):
+    from rest_framework_simplejwt.tokens import RefreshToken
+    token = RefreshToken.for_user(user)
+    client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}")
+
+
+def _post(client, body, origin="https://www.a.com"):
+    return client.post("/api/collect/batch", data=json.dumps(body),
+                       content_type="text/plain", HTTP_ORIGIN=origin)
+
+
+def _body(n=1, sid="s1", vid="v1", events=None):
+    return {"site_id": "site_x", "visitor_id": vid, "session_id": sid,
+            "context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
+            "events": events if events is not None else
+                      [{"event_id": f"e{i}", "event_name": "page_view",
+                        "occurred_at": NOW.isoformat(), "path": "/", "title": "t",
+                        "properties": {}} for i in range(n)]}
+
+
+def _site():
+    return Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
+
+
+# ---- a) 正常上报 ----
+
+def test_raw_batch_accepted(db, client):
+    _site()
+    r = _post(client, _body(2, vid="v9"))
+    assert r.status_code == 200
+    rb = RawBatch.objects.get()
+    assert rb.status == "accepted" and rb.event_count == 2 and rb.rejected_count == 0
+    vis = Visitor.objects.get(visitor_id="v9")
+    assert rb.visitor_id == vis.pk and rb.site_id == vis.site_id   # 访客关联
+    body = json.loads(rb.body)
+    assert body["visitor_id"] == "v9"                    # 完整结构在场
+    assert body["context"]["device_type"] == "PC"
+    assert [e["path"] for e in body["events"]] == ["/", "/"]   # 快照先于 ingest 就地清洗
+
+
+# ---- b) 敏感键脱敏 + partial ----
+
+def test_raw_batch_partial_and_filtered(db, client):
+    _site()
+    events = [
+        {"event_id": "ok1", "event_name": "page_view", "occurred_at": NOW.isoformat(),
+         "path": "/", "title": "t", "properties": {}},
+        {"event_id": "bad1", "event_name": "page_view", "occurred_at": NOW.isoformat(),
+         "path": "/login", "title": "t", "properties": {"user_password": "hunter2"}},
+    ]
+    r = _post(client, _body(events=events, vid="v1"))
+    assert r.json()["data"]["accepted"] == 1             # 事件层：敏感事件被拒
+    rb = RawBatch.objects.get()
+    assert rb.status == "partial" and rb.event_count == 1 and rb.rejected_count == 1
+    body = json.loads(rb.body)
+    assert body["events"][1]["properties"]["user_password"] == "[FILTERED]"   # 值替换结构保留
+    assert body["events"][1]["properties"] != {"user_password": "hunter2"}
+
+
+# ---- c) 超批 rejected + 截断 ----
+
+def test_raw_batch_rejected_truncated(db, client):
+    _site()
+    r = _post(client, _body(51, vid="v1"))               # >50 事件整批拒
+    assert r.json()["data"]["accepted"] == 0
+    rb = RawBatch.objects.get()
+    assert rb.status == "rejected" and rb.event_count == 0 and rb.rejected_count == 1
+    # 51 事件原始 JSON 远超 4KB → 截断到 4096 + 标记后缀（≤ ~4.1KB）
+    assert len(rb.body) == 4096 + len("...[TRUNCATED]")
+    assert rb.body.endswith("...[TRUNCATED]")
+    assert rb.visitor is None or rb.visitor.visitor_id == "v1"   # 超批不建访客，允许 None
+
+
+# ---- d) 重放每次都记 ----
+
+def test_raw_batch_replay_two_rows(db, client):
+    _site()
+    _post(client, _body(1, vid="v1"))
+    _post(client, _body(1, vid="v1"))
+    assert RawBatch.objects.count() == 2                 # 原始记录语义：每次接收一行
+
+
+# ---- e) 管理台 API：分页与字段 ----
+
+def test_raw_batches_api_pagination_and_fields(db, client, member_user):
+    _auth(client, member_user)
+    site = _site()
+    v = Visitor.objects.create(site=site, visitor_id="v1")
+    for i in range(15):
+        rb = RawBatch.objects.create(site=site, visitor=v, batch_id=f"b{i:02d}",
+                                     body='{"visitor_id":"v1"}', event_count=i % 3,
+                                     rejected_count=0, status="accepted", ip="1.2.3.4")
+        # received_at 是 auto_now_add，update 绕开制造时间差（与 test_visitors 同款手法）
+        RawBatch.objects.filter(pk=rb.pk).update(received_at=NOW - timedelta(minutes=i))
+
+    r = client.get("/api/admin/visitors/v1/raw-batches")
+    assert r.status_code == 200
+    d = r.json()["data"]
+    assert d["total"] == 15 and d["page"] == 1 and len(d["list"]) == 10   # page_size 10
+    row = d["list"][0]
+    assert set(row) == {"id", "batch_id", "status", "event_count",
+                        "rejected_count", "received_at", "body"}
+    assert row["batch_id"] == "b00"                      # -received_at：最新在前
+    d2 = client.get("/api/admin/visitors/v1/raw-batches", {"page": "2"}).json()["data"]
+    assert d2["page"] == 2 and len(d2["list"]) == 5
+    assert [x["batch_id"] for x in d2["list"]][-1] == "b14"
+
+    r = client.get("/api/admin/visitors/nope/raw-batches")
+    assert r.status_code == 404
+    body = r.json()
+    assert body["code"] == 404 and body["data"] is None and body["message"] == "visitor_not_found"
+
+
+# ---- f) 清理命令 ----
+
+def test_cleanup_raw_command(db):
+    site = _site()
+    v = Visitor.objects.create(site=site, visitor_id="v1")
+    old = RawBatch.objects.create(site=site, visitor=v, body="{}")
+    RawBatch.objects.filter(pk=old.pk).update(received_at=NOW - timedelta(days=31))
+    new = RawBatch.objects.create(site=site, visitor=v, body="{}")   # 刚收到
+
+    out = StringIO()
+    call_command("cleanup_raw", "--days", 30, stdout=out)
+    assert not RawBatch.objects.filter(pk=old.pk).exists()   # 30 天前删除
+    assert RawBatch.objects.filter(pk=new.pk).exists()       # 新的保留
+    assert "1" in out.getvalue()                             # 输出删除数