fangke/server/tracker/api/collect.py

import json
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
from rest_framework.throttling import AnonRateThrottle
from rest_framework.parsers import BaseParser
from rest_framework.settings import api_settings
from tracker.models import RawBatch, Site, Visitor
from tracker.services.ingest import ingest_batch
from tracker.services.validation import parse_iso_dt, sanitize_payload

class PlainJSONParser(BaseParser):
    media_type = "text/plain"
    def parse(self, stream, *args, **kwargs):
        try:
            return json.loads(stream.read().decode("utf-8") or "{}")
        except (ValueError, UnicodeDecodeError):
            from rest_framework.exceptions import ParseError
            raise ParseError("invalid_json")   # DRF 对 parser 异常是 re-raise 不是包装，不自包就是 500

class CollectThrottle(AnonRateThrottle):
    scope = "collect"

    def get_rate(self):
        # SimpleRateThrottle.THROTTLE_RATES 在 import 时固化为类属性，
        # override_settings 改 DEFAULT_THROTTLE_RATES 对它不生效——改读实时 api_settings（DRF 收 setting_changed 会 reload）
        return api_settings.DEFAULT_THROTTLE_RATES[self.scope]

class CollectBatchView(APIView):
    permission_classes = [AllowAny]
    parser_classes = [PlainJSONParser]
    throttle_classes = [CollectThrottle]
    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", 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", 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", request_id)
        if not host and not site.allow_no_origin:
            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 截断等）——存的是"到达时"的原文（脱敏后）。
        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:
                    ev["occurred_at"] = parse_iso_dt(ev["occurred_at"])   # 共享解析（validation），视图错误策略=整批 400
                except (ValueError, TypeError):
                    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", 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, 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"],
                     "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 统一补
        resp = Response(status=200)
        resp["Access-Control-Allow-Methods"] = "POST, OPTIONS"
        resp["Access-Control-Allow-Headers"] = "content-type"
        return resp

    def finalize_response(self, request, response, *args, **kwargs):
        """统一收口：ACAO 回显 + 错误信封。

        错误信封部分与 tracker.api.envelope.EnvelopeMixin 同构——collect 端含 ACAO 回显
        （浏览器 SDK 跨域需读错误体），暂保留在自身视图，未来收敛至共享模块。

        429/ParseError 等由 DRF 异常处理器产出的响应（无 ACAO、body 是 {detail}）在这里补齐——
        否则浏览器 SDK 跨域读不到错误体会误判网络错误而重试，限流时反而打爆限流器。
        dispatch 对 initial() 抛出的 Throttled/ParseError 同样走 finalize_response，此钩子全覆盖。
        """
        response = super().finalize_response(request, response, *args, **kwargs)
        origin = request.headers.get("Origin")
        if origin and "Access-Control-Allow-Origin" not in response:   # 只回显真正的 Origin 头
            response["Access-Control-Allow-Origin"] = origin
        if response.status_code >= 400 and isinstance(response.data, dict) and "detail" in response.data:
            response.data = {"code": response.status_code, "data": None,
                             "message": response.data["detail"]}
        return response

    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"]
                  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, 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
        return Site.objects.filter(site_id=sid).first() if sid else None

    def _client_ip(self, request):
        """nginx 前置时取 X-Forwarded-For 最后一段（与 DRF NUM_PROXIES=1 一致；套 CDN 时按层数前移）。"""
        xff = request.META.get("HTTP_X_FORWARDED_FOR", "")
        ip = xff.split(",")[-1].strip() if xff else request.META.get("REMOTE_ADDR")
        # 归一 IPv4 映射地址（::ffff:1.2.3.4 → 1.2.3.4），防地理解析丢失
        if ip and ip.lower().startswith("::ffff:"):
            ip = ip[7:]
        return ip

    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)