fangke/docs/superpowers/plans/2026-09-15-m1-data-collection-foundation.md

M1 数据采集底座 · 实施计划（JS SDK + Django 采集后端 + Art Design Pro 后台）

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 打通端到端最小闭环——网站接入 JS SDK → 批量上报事件 → Django 采集入库（MySQL）→ Art Design Pro 后台展示访客跨天行为时间线。

Architecture: Monorepo 三组件：server/（Django+DRF 单应用 tracker，采集公开接口 + JWT 管理接口）、sdk/（原生 TS 写的浏览器 SDK，esbuild 打包 IIFE，≤15KB gzip）、admin/（Art Design Pro 模板精简后开发 5 个页面）。事件全量入库（M1 不分区，PK/索引设计为将来按月分区留路），会话权威在 SDK、服务端兜底校验。

Tech Stack: Python 3.11+ / Django 5.x / DRF / MySQL 8（测试用 SQLite）/ simplejwt / django-cors-headers / ip2region xdb；Node 20+ / TypeScript / esbuild / vitest + happy-dom；Vue3 + Element Plus（Art Design Pro 模板）。

Spec: docs/superpowers/specs/2026-09-15-customer-behavior-platform-design.md（M1 范围：§13 M1；红线：§4.6、§9）

---

文件结构总览

cms-cj/
├── server/                       # Django 后端
│   ├── manage.py
│   ├── requirements.txt
│   ├── pytest.ini
│   ├── config/                   # Django 工程（settings/urls/wsgi）
│   ├── tracker/                  # 唯一业务应用
│   │   ├── models.py             # Site/Visitor/VisitSession/Event/DailyStat
│   │   ├── constants.py          # 事件名白名单、限制、敏感键黑名单
│   │   ├── services/
│   │   │   ├── validation.py     # 事件结构/敏感键/时间窗校验
│   │   │   ├── ingest.py         # 批量入库：幂等/访客/会话/首触
│   │   │   ├── geo.py            # ip2region 离线解析
│   │   │   └── stats.py          # 总览聚合查询
│   │   ├── api/
│   │   │   ├── collect.py        # POST /api/collect/batch
│   │   │   ├── auth.py           # JWT 登录/刷新/me
│   │   │   ├── sites.py          # 站点 CRUD
│   │   │   ├── dashboard.py      # 总览数据
│   │   │   └── visitors.py       # 访客列表/详情/时间线/会话
│   │   ├── management/commands/refresh_stats.py
│   │   └── tests/                # pytest 测试
│   ├── Dockerfile
│   └── tracker/data/ip2region.xdb  # 离线 IP 库（下载，不进 git，部署时 volume 挂载）
├── sdk/                          # JS SDK
│   ├── package.json  vitest.config.ts  build.mjs
│   ├── src/{index,config,identity,context,transport,events,util}.ts
│   ├── test/*.test.ts
│   └── demo/index.html           # 本地联调演示页
├── admin/                        # Art Design Pro 前端（Task 14 克隆）
│   └── src/{api,views,router/modules}/...（新增 tracker 相关文件）
├── deploy/
│   ├── docker-compose.yml  nginx.conf
│   └── README-deploy.md          # 1Panel 部署步骤
└── docs/superpowers/{specs,plans}/

全局约定

• 响应包：所有 /api/admin/* 返回 {code: 200, data: ..., message: "ok"}（适配 Art Design Pro 拦截器默认约定）；错误 code 为业务码（401 未登录/403 无权/4xx 参数/500 内部）。
• 时间：全库 UTC 存储（USE_TZ=True），展示层按 settings.TIME_ZONE='Asia/Shanghai'。
• 采集跨域：SDK 用 Content-Type: text/plain 发 JSON（简单请求免预检），服务端用 PlainJSONParser 解析；django-cors-headers + 动态白名单（DB 中 Site.domain_whitelist，60s 缓存）回 Access-Control-Allow-Origin。
• 前端安全红线：全项目禁用 v-html；访客来源数据（path/title/referrer/utm）不得直接绑定 href/src——如需渲染成链接，协议白名单仅 http/https（Vue 不拦截 javascript: URI）。
• 测试：后端 pytest（SQLite 内存库，pytest.ini 指定 --ds=config.settings_test）；SDK vitest + happy-dom + fake timers。
• 提交：每任务一提交，feat:/test:/chore: 前缀。
• M1 明确不做（防执行者自作主张加码）：SPA 路由、点击/滚动/表单/意图事件、客户(contact)体系、Redis、事件表分区迁移、npm 发布。

---

Phase 0 · 骨架

Task 1: Django 工程骨架

Files:

• Create: server/requirements.txt, server/pytest.ini, server/manage.py, server/config/__init__.py, server/config/settings.py, server/config/urls.py, server/config/wsgi.py, server/tracker/__init__.py, server/tracker/apps.py, server/tracker/constants.py

• Step 1: 建目录与依赖文件

server/requirements.txt:

Django>=5.1,<5.3
djangorestframework>=3.15
djangorestframework-simplejwt>=5.3
django-cors-headers>=4.4
PyMySQL>=1.1
py-ip2region>=3.0
gunicorn>=22.0

server/config/__init__.py（MySQL 驱动适配，Windows/容器免编译）：

try:
    import pymysql
    pymysql.install_as_MySQLdb()
except ImportError:
    pass

server/pytest.ini:

[pytest]
DJANGO_SETTINGS_MODULE = config.settings_test
python_files = test_*.py
addopts = -q

• Step 2: 写 settings

server/config/settings.py 关键内容（完整写出）：

from pathlib import Path
import os

BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_KEY = os.environ.get("DJANGO_SECRET_KEY", "dev-insecure")
DEBUG = os.environ.get("DJANGO_DEBUG", "0" if os.environ.get("DJANGO_ENV") == "prod" else "1") == "1"
ALLOWED_HOSTS = os.environ.get("DJANGO_ALLOWED_HOSTS", "*").split(",")
if os.environ.get("DJANGO_ENV") == "prod":   # 生产 fail-fast：漏配即启动崩溃，不带病上线
    assert SECRET_KEY != "dev-insecure", "生产必须配置 DJANGO_SECRET_KEY"
    assert os.environ.get("DJANGO_ALLOWED_HOSTS"), "生产必须显式配置 DJANGO_ALLOWED_HOSTS"

INSTALLED_APPS = [
    "django.contrib.contenttypes",
    "django.contrib.auth",
    "corsheaders",
    "rest_framework",
    "tracker",
]
MIDDLEWARE = [
    "corsheaders.middleware.CorsMiddleware",
    "django.middleware.common.CommonMiddleware",
]
ROOT_URLCONF = "config.urls"
WSGI_APPLICATION = "config.wsgi.application"
TEMPLATES = [{
    "BACKEND": "django.template.backends.django.DjangoTemplates",
    "DIRS": [], "APP_DIRS": True, "OPTIONS": {"context_processors": []},
}]

if os.environ.get("DB_ENGINE", "mysql") == "sqlite":   # 本地无 MySQL 的开发模式
    DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3",
                             "NAME": BASE_DIR / "db.sqlite3"}}
else:
    DATABASES = {
        "default": {
            "ENGINE": "django.db.backends.mysql",
            "NAME": os.environ.get("DB_NAME", "tracker"),
            "USER": os.environ.get("DB_USER", "root"),
            "PASSWORD": os.environ.get("DB_PASSWORD", ""),
            "HOST": os.environ.get("DB_HOST", "127.0.0.1"),
            "PORT": os.environ.get("DB_PORT", "3306"),
            "OPTIONS": {"charset": "utf8mb4"},
        }
    }

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ("rest_framework_simplejwt.authentication.JWTAuthentication",),
    "DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
    "DEFAULT_THROTTLE_RATES": {"collect": "120/min", "admin": "600/min", "login": "10/min"},
    "NUM_PROXIES": 1,
}
SIMPLE_JWT = {"ACCESS_TOKEN_LIFETIME": timedelta(hours=12)}
CORS_ALLOW_ALL_ORIGINS = False
CACHES = {"default": {"BACKEND": "django.core.cache.backends.filebased.FileBasedCache",
                      "LOCATION": str(BASE_DIR / ".cache")}}   # 限流计数跨 gunicorn worker 共享（不共享则限额×3）
DATA_UPLOAD_MAX_MEMORY_SIZE = 262144                          # 256KB 应用层请求体兜底
TIME_ZONE = "Asia/Shanghai"
USE_TZ = True
LANGUAGE_CODE = "zh-hans"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
IP2REGION_XDB = str(BASE_DIR / "tracker" / "data" / "ip2region.xdb")

（from datetime import timedelta 放文件头；无 admin/session/contrib.staticfiles——管理台是独立 SPA。）

server/config/settings_test.py:

from .settings import *
DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}}

server/config/urls.py:

urlpatterns = []   # Task 7 起逐任务填充

server/tracker/constants.py:

EVENT_NAMES = {"page_view", "page_stay", "sdk_loaded"}
MAX_EVENTS_PER_BATCH = 50
MAX_PROP_KEYS = 32
MAX_STR_LEN = 500
MAX_PATH_LEN = 500
MAX_TITLE_LEN = 200
CLOCK_SKEW_LATE_SEC = 120        # 超过即标 late
CLOCK_SKEW_REJECT_SEC = 86400    # 超过即拒绝
SENSITIVE_KEY_PATTERNS = ("password", "passwd", "token", "secret", "card", "verif", "otp", "idcard")
URL_PARAM_DROP_PATTERNS = ("email", "phone", "mobile", "token", "key", "password", "verif")
SESSION_IDLE_MS = 30 * 60 * 1000
BOT_UA_KEYWORDS = ("bot", "spider", "crawl", "slurp", "bingpreview", "headless", "lighthouse")
EVENT_PROP_WHITELIST = {          # spec §9.1：未知字段拒绝。免费文本/任意属性不得入库
    "page_view": {"referrer", "referrer_domain", "utm"},
    "page_stay": {"active_ms", "path"},
    "sdk_loaded": {"sdk_version"},
}
MAX_PROP_DEPTH = 2
MAX_PROP_JSON_BYTES = 4096

• Step 3: 建 Django 工程与 app 骨架

按上面文件清单创建 manage.py（标准模板）、config/wsgi.py（标准）、tracker/apps.py（name="tracker"）。

• Step 4: 验证

cd server && python -m venv .venv && .venv/Scripts/pip install -r requirements.txt pytest pytest-django
.venv/Scripts/python manage.py check

Expected: System check identified no issues。pip freeze > requirements.lock.txt（不含 pytest 两条则手工剔除 pytest 行，锁定运行依赖）。

• Step 5: Commit chore: Django 工程骨架（config+tracker app+常量）

Task 2: SDK 工程骨架

Files:

• Create: sdk/package.json, sdk/tsconfig.json, sdk/vitest.config.ts, sdk/build.mjs, sdk/src/util.ts, sdk/test/util.test.ts

• Step 1: 初始化

sdk/package.json:

{
  "name": "ctk-sdk",
  "version": "0.1.0",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "node build.mjs",
    "test": "vitest run",
    "checksize": "node build.mjs && gzip -c dist/v1.js | wc -c"
  },
  "devDependencies": {
    "esbuild": "^0.24.0",
    "happy-dom": "^15.0.0",
    "typescript": "^5.6.0",
    "vitest": "^2.1.0"
  }
}

sdk/vitest.config.ts:

import { defineConfig } from "vitest/config";
export default defineConfig({
  test: { environment: "happy-dom", setupFiles: [] },
});

sdk/build.mjs:

import { build } from "esbuild";
await build({
  entryPoints: ["src/index.ts"],
  bundle: true, minify: true, format: "iife",
  outfile: "dist/v1.js", target: ["es2018"],
});

sdk/src/util.ts:

export function uuid(): string {
  if (globalThis.crypto?.randomUUID) return crypto.randomUUID();
  return "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx".replace(/[xy]/g, (c) => {
    const r = (Math.random() * 16) | 0;
    return (c === "x" ? r : (r & 0x3) | 0x8).toString(16);
  });
}

• Step 2: TDD uuid

sdk/test/util.test.ts:

import { describe, it, expect } from "vitest";
import { uuid } from "../src/util";
describe("uuid", () => {
  it("生成 36 位且不重复", () => {
    const a = uuid(), b = uuid();
    expect(a).toMatch(/^[0-9a-f-]{36}$/);
    expect(a).not.toBe(b);
  });
});

cd sdk && pnpm install && pnpm test

Expected: 1 passed。

• Step 3: Commit chore: SDK 工程骨架（esbuild+vitest）

---

Phase 1 · 数据模型与采集后端

Task 3: 五个核心模型

Files:

• Create: server/tracker/models.py, server/tracker/tests/test_models.py, server/tracker/migrations/__init__.py

• Step 1: 写失败测试

server/tracker/tests/test_models.py:

import pytest
from tracker.models import Site, Visitor, VisitSession, Event

pytestmark = pytest.mark.django_db

def test_create_site_generates_public_site_id():
    site = Site.objects.create(name="官网", domains=["www.example.com"])
    assert site.site_id.startswith("site_") and len(site.site_id) == 21
    assert site.is_active is True

def test_visitor_first_touch_fields():
    site = Site.objects.create(name="s", domains=["a.com"])
    v = Visitor.objects.create(site=site, visitor_id="vid1",
        first_referrer_domain="google.com", first_utm={"utm_source": "google"},
        first_landing_path="/p", country="美国", device_type="PC")
    assert v.visit_count == 0 and v.active_days == 0

def test_event_unique_site_event_id():
    site = Site.objects.create(name="s", domains=["a.com"])
    v = Visitor.objects.create(site=site, visitor_id="vid1")
    s = VisitSession.objects.create(site=site, visitor=v, session_id="sid1")
    Event.objects.create(site=site, visitor=v, session=s, event_id="e1",
        event_name="page_view", occurred_at="2026-09-15T10:00:00Z", path="/")
    with pytest.raises(Exception):
        Event.objects.create(site=site, visitor=v, session=s, event_id="e1",
            event_name="page_view", occurred_at="2026-09-15T10:01:00Z", path="/")

• Step 2: 跑测试确认失败 cd server && .venv/Scripts/python -m pytest tracker/tests/test_models.py Expected: ImportError（models 不存在）。

• Step 3: 实现 models.py

from django.db import models
import secrets

def short_public_id() -> str:
    return "site_" + secrets.token_urlsafe(12)[:16]

class Site(models.Model):
    name = models.CharField(max_length=100)
    site_id = models.CharField(max_length=32, unique=True, default=short_public_id)
    domains = models.JSONField(default=list)          # 域名白名单 ["www.example.com"]
    allow_no_origin = models.BooleanField(default=False)  # 本地/demo 调试用
    is_active = models.BooleanField(default=True)
    timezone = models.CharField(max_length=64, default="Asia/Shanghai")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    class Meta:
        db_table = "site"

class Visitor(models.Model):
    site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="visitors")
    visitor_id = models.CharField(max_length=64)
    first_seen_at = models.DateTimeField(auto_now_add=True)
    last_seen_at = models.DateTimeField(auto_now_add=True)
    first_referrer_domain = models.CharField(max_length=200, blank=True, default="")
    first_utm = models.JSONField(default=dict, blank=True)
    first_landing_path = models.CharField(max_length=500, blank=True, default="")
    device_type = models.CharField(max_length=16, blank=True, default="")
    os = models.CharField(max_length=64, blank=True, default="")
    browser = models.CharField(max_length=64, blank=True, default="")
    language = models.CharField(max_length=32, blank=True, default="")
    timezone_name = models.CharField(max_length=64, blank=True, default="")
    screen = models.CharField(max_length=32, blank=True, default="")
    last_ip = models.GenericIPAddressField(null=True, blank=True)
    country = models.CharField(max_length=64, blank=True, default="")
    province = models.CharField(max_length=64, blank=True, default="")
    city = models.CharField(max_length=64, blank=True, default="")
    visit_count = models.PositiveIntegerField(default=0)   # 会话数（服务端按新会话递增）
    active_days = models.PositiveIntegerField(default=0)   # 活跃天数
    last_active_date = models.DateField(null=True, blank=True)
    is_bot = models.BooleanField(default=False)      # UA 爬虫标记，报表默认排除
    class Meta:
        db_table = "visitor"
        constraints = [models.UniqueConstraint(fields=["site", "visitor_id"], name="uq_site_visitor")]
        indexes = [
            models.Index(fields=["site", "-last_seen_at"]),
            models.Index(fields=["site", "country"]),
        ]

class VisitSession(models.Model):
    site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="sessions")
    visitor = models.ForeignKey(Visitor, on_delete=models.CASCADE, related_name="sessions")
    session_id = models.CharField(max_length=64)
    started_at = models.DateTimeField()
    ended_at = models.DateTimeField()
    referrer_domain = models.CharField(max_length=200, blank=True, default="")
    utm = models.JSONField(default=dict, blank=True)
    landing_path = models.CharField(max_length=500, blank=True, default="")
    event_count = models.PositiveIntegerField(default=0)
    duration_ms = models.PositiveBigIntegerField(default=0)   # 会话内事件时间跨度
    stay_ms = models.PositiveBigIntegerField(default=0)       # 有效停留累计（page_stay active_ms 之和）
    class Meta:
        db_table = "visit_session"
        constraints = [models.UniqueConstraint(fields=["site", "session_id"], name="uq_site_session")]
        indexes = [models.Index(fields=["visitor", "-started_at"])]

class Event(models.Model):
    site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="events")
    visitor = models.ForeignKey(Visitor, on_delete=models.CASCADE, related_name="events")
    session = models.ForeignKey(VisitSession, on_delete=models.CASCADE, related_name="events", null=True)
    event_id = models.CharField(max_length=64)
    event_name = models.CharField(max_length=50)
    occurred_at = models.DateTimeField()
    received_at = models.DateTimeField(auto_now_add=True)
    path = models.CharField(max_length=500, blank=True, default="")
    title = models.CharField(max_length=200, blank=True, default="")
    properties = models.JSONField(default=dict, blank=True)
    ip = models.GenericIPAddressField(null=True, blank=True)
    country = models.CharField(max_length=64, blank=True, default="")
    province = models.CharField(max_length=64, blank=True, default="")
    city = models.CharField(max_length=64, blank=True, default="")
    late = models.BooleanField(default=False)
    is_bot = models.BooleanField(default=False)
    class Meta:
        db_table = "event"
        constraints = [models.UniqueConstraint(fields=["site", "event_id"], name="uq_site_event")]
        indexes = [
            models.Index(fields=["site", "visitor", "-occurred_at"]),
            models.Index(fields=["session", "occurred_at"]),
            models.Index(fields=["site", "event_name", "-occurred_at"]),
        ]

class DailyStat(models.Model):
    site = models.ForeignKey(Site, on_delete=models.CASCADE, related_name="daily_stats")
    date = models.DateField()
    pv = models.PositiveIntegerField(default=0)
    uv = models.PositiveIntegerField(default=0)
    sessions = models.PositiveIntegerField(default=0)
    new_visitors = models.PositiveIntegerField(default=0)
    class Meta:
        db_table = "daily_stat"
        constraints = [models.UniqueConstraint(fields=["site", "date"], name="uq_site_date")]

• Step 4: 迁移并跑测试

.venv/Scripts/python manage.py makemigrations tracker
.venv/Scripts/python -m pytest tracker/tests/test_models.py

Expected: 3 passed。

• Step 5: Commit feat: 核心数据模型（Site/Visitor/VisitSession/Event/DailyStat）

Task 4: 事件校验服务 validation.py

Files:

• Create: server/tracker/services/__init__.py, server/tracker/services/validation.py, server/tracker/tests/test_validation.py

• Step 1: 失败测试（覆盖：白名单事件名/敏感键拒绝/属性数与长度截断/URL 参数清洗/时间窗）

server/tracker/tests/test_validation.py 核心用例：

from datetime import datetime, timedelta, timezone as tz
from tracker.services.validation import validate_event, sanitize_url
from django.utils import timezone

NOW = timezone.now()

def _ev(**kw):
    base = dict(event_id="e1", event_name="page_view",
                occurred_at=NOW, path="/p", title="t", properties={})
    base.update(kw)
    return base

def test_ok_event():
    assert validate_event(_ev()) is None            # None = 通过

def test_unknown_event_name():
    assert "event_name" in validate_event(_ev(event_name="xxx"))

def test_sensitive_property_rejected():
    r = validate_event(_ev(properties={"user_password": "1"}))
    assert "sensitive" in r

def test_shape_type_and_whitelist():
    assert validate_event(["x"]) == "event_invalid_shape"
    assert validate_event(_ev(event_id=123)) == "invalid_event_id"
    assert validate_event(_ev(properties={"foo": "bar"})) == "property_not_allowed"

def test_too_many_props_and_long_string():
    r = validate_event(_ev(properties={f"k{i}": "v" for i in range(40)}))
    assert "properties" in r
    r = validate_event(_ev(properties={"a": "x" * 600}))
    assert "properties" in r

def test_future_and_stale_time():
    r = validate_event(_ev(occurred_at=NOW + timedelta(days=2)))
    assert "occurred_at" in r                        # 超窗拒绝
    assert validate_event(_ev(occurred_at=NOW - timedelta(hours=2))) is None  # 窗内迟到可收

def test_sanitize_url_drops_sensitive_params():
    u = sanitize_url("/p?email=a@b.com&token=x&id=3&utm_source=google&monkey=1")
    assert "email=" not in u and "token=" not in u and "utm_source=google" in u
    assert "monkey=1" in u          # fullmatch：含 key 子串的正常参数不误伤

• Step 2: 跑测试确认失败 → ModuleNotFoundError。

• Step 3: 实现

import json
import re
from datetime import datetime, timedelta
from urllib.parse import parse_qsl, urlencode, urlparse
from django.utils import timezone
from tracker import constants as C

_SENSITIVE = re.compile("|".join(C.SENSITIVE_KEY_PATTERNS), re.I)
_URL_DROP_KEY = re.compile("|".join(C.URL_PARAM_DROP_PATTERNS), re.I)

def sanitize_url(url: str) -> str:
    """按参数名整键删除（fullmatch 防止误伤 monkey 这类含 key 子串的参数）。"""
    p = urlparse(url)
    kept = [(k, v) for k, v in parse_qsl(p.query, keep_blank_values=True)
            if not _URL_DROP_KEY.fullmatch(k)]
    return p._replace(query=urlencode(kept)).geturl()

def _depth_ok(v, depth=0) -> bool:
    if depth > C.MAX_PROP_DEPTH:
        return False
    if isinstance(v, dict):
        return all(_depth_ok(x, depth + 1) for x in v.values())
    if isinstance(v, list):
        return all(_depth_ok(x, depth + 1) for x in v)
    return True

def validate_event(ev) -> str | None:
    """返回 None 表示通过；否则返回拒绝原因码。类型/白名单/体积/深度全在此守门。"""
    if not isinstance(ev, dict):
        return "event_invalid_shape"
    if ev.get("event_name") not in C.EVENT_NAMES:
        return "event_name_not_allowed"
    eid = ev.get("event_id")
    if not isinstance(eid, str) or not eid or len(eid) > 64:
        return "invalid_event_id"
    for k in ("path", "title"):
        if ev.get(k) is not None and not isinstance(ev[k], str):
            return "invalid_field_type"
    occ = ev.get("occurred_at")
    if not isinstance(occ, datetime):
        return "invalid_occurred_at"
    skew = (timezone.now() - occ).total_seconds()
    if abs(skew) > C.CLOCK_SKEW_REJECT_SEC:
        return "occurred_at_out_of_window"
    props = ev.get("properties") or {}
    if not isinstance(props, dict) or len(props) > C.MAX_PROP_KEYS:
        return "properties_invalid"
    if not set(props) <= C.EVENT_PROP_WHITELIST.get(ev["event_name"], set()):
        return "property_not_allowed"
    if not _depth_ok(props):
        return "properties_depth"
    if len(json.dumps(props, ensure_ascii=False, default=str)) > C.MAX_PROP_JSON_BYTES:
        return "properties_too_large"
    for k, v in props.items():
        if _SENSITIVE.search(str(k)):
            return "sensitive_field"
        if isinstance(v, str) and len(v) > C.MAX_STR_LEN:
            return "properties_too_long"
    if len(ev.get("path") or "") > C.MAX_PATH_LEN:
        ev["path"] = ev["path"][:C.MAX_PATH_LEN]
    if len(ev.get("title") or "") > C.MAX_TITLE_LEN:
        ev["title"] = ev["title"][:C.MAX_TITLE_LEN]
    if ev.get("path"):
        ev["path"] = sanitize_url(ev["path"])
    return None

• Step 4: 跑测试通过 → 6 passed。

• Step 5: Commit feat: 事件校验服务（白名单/敏感键/时间窗/URL清洗）

Task 5: geo.py 离线 IP 解析

Files:

• Create: server/tracker/services/geo.py, server/tracker/tests/test_geo.py

• Download: server/tracker/data/ip2region.xdb（ip2region 官方仓库 data/ip2region.xdb，~11MB，加入 .gitignore）

• Step 1: 失败测试

from tracker.services.geo import GeoResolver

def test_private_ip_returns_local():
    r = GeoResolver()
    assert r.resolve("192.168.1.5") is None       # 内网不解析（后台显示"本地"）
    assert r.resolve("127.0.0.1") is None

def test_public_ip_parsed_or_none(monkeypatch):
    r = GeoResolver()
    # xdb 未下载/解析失败时优雅降级返回 None，不抛异常
    out = r.resolve("8.8.8.8")
    assert out is None or set(out) == {"country", "province", "city"}

• Step 2: 实现

import functools, ipaddress
from django.conf import settings

class GeoResolver:
    """进程级单例：xdb 一次读入内存（官方 buffer 模式声明线程安全）。
    pip 包名 py-ip2region（模块名 ip2region）；包缺失或 xdb 缺失时优雅降级返回 None。"""

    def __init__(self):
        self._searcher = None
        try:
            from ip2region import searcher as ip2region_searcher
            with open(settings.IP2REGION_XDB, "rb") as f:
                self._searcher = ip2region_searcher.new_with_buffer(f.read())
        except Exception:
            self._searcher = None

    @functools.lru_cache(maxsize=10000)
    def resolve(self, ip: str):
        if not ip:
            return None
        try:
            if ipaddress.ip_address(ip).is_private:
                return None
        except ValueError:
            return None
        if not self._searcher:
            return None
        try:
            raw = self._searcher.search(ip)      # "国家|区域|省|市|ISP"
            parts = (raw or "").split("|")
            return {"country": parts[0], "province": parts[2], "city": parts[3]}
        except Exception:
            return None

注：py-ip2region 的实际调用以其 README 为准（模块 ip2region，入口 searcher.new_with_buffer(buff) 后 search(ip)；不支持 with 上下文管理器——用错会被 except 静默吞掉、地理功能永久失效且测试照样绿，务必按此实现）。解析结果格式 国家|区域|省|市|ISP，如 中国|0|广东省|深圳市|电信、美国|0|加利福尼亚|旧金山|0。

• Step 3: 跑测试通过（server/.gitignore 加 tracker/data/*.xdb）

• Step 4: Commit feat: ip2region 离线地理解析（优雅降级）

Task 6: ingest.py 批量入库服务

Files:

• Create: server/tracker/services/ingest.py, server/tracker/tests/test_ingest.py

职责：对已通过校验的批次执行——访客 get_or_create（首触来源只写一次）、会话 get_or_create（30 分钟兜底校验）、bulk_create(ignore_conflicts=True) 幂等写事件、更新访客聚合（visit_count/active_days/last_seen/last_ip/geo）。测试用 SQLite，忽略 MySQL 差异。

• Step 1: 失败测试（核心场景全列）：

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

def test_first_batch_creates_visitor_session_events():
    site = Site.objects.create(name="s", domains=["a.com"])
    r = ingest_batch(site, _batch([_ev(1), _ev(2)]), ip="8.8.8.8")
    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 == "8.8.8.8"
    assert Event.objects.filter(site=site).count() == 2

def test_duplicate_event_ids_counted_not_rewritten():
    site = Site.objects.create(name="s", domains=["a.com"])
    ingest_batch(site, _batch([_ev(1)]))
    r = ingest_batch(site, _batch([_ev(1)]))     # 重放
    assert r["duplicated"] == 1 and Event.objects.count() == 1

def test_new_session_increments_visit_count():
    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

def test_server_session_backstop_splits_idle_gap():
    # SDK 权威；服务端仅当同一 session_id 事件间隔>30min 时切分出后续段（记新 session_id 后缀）
    site = Site.objects.create(name="s", domains=["a.com"])
    ingest_batch(site, _batch([_ev(1, occurred_at=NOW - timedelta(minutes=90))], sid="s1"))
    ingest_batch(site, _batch([_ev(2, occurred_at=NOW)], sid="s1"))
    assert VisitSession.objects.count() == 2      # s1 与 s1#2

def test_rejected_events_reported_with_reason():
    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"

• Step 2: 跑测试确认失败 → ImportError。

• Step 3: 实现

from datetime import timedelta
from django.db import transaction
from django.db.models import F
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 GeoResolver

_geo = GeoResolver()

_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 _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=g["country"][:64], province=g["province"][:64],
                       city=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:
            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 同时切分出相同后缀撞唯一键——重查复用
            from django.db import IntegrityError as _IE  # 顶部统一 import IntegrityError
            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, 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", "")[:500],
            title=ev.get("title", "")[:200],
            properties=ev.get("properties") or {},
            ip=ip, late=abs((now - occ).total_seconds()) > C.CLOCK_SKEW_LATE_SEC,
            country=geo["country"] if geo else "",
            province=geo["province"] if geo else "",
            city=geo["city"] if geo else "",
            is_bot=is_bot))
    if ready:
        before = Event.objects.filter(site=site,
              event_id__in=[e.event_id for e in ready]).values_list("event_id", flat=True)
        existed = set(before)
        created = Event.objects.bulk_create(
            [e for e in ready if e.event_id not in existed], ignore_conflicts=True)
        # bulk_create ignore_conflicts 不回传；用存在集 + 计数差算 duplicated
        result["accepted"] = len(ready) - (len([e for e in ready if e.event_id in existed]))
        result["duplicated"] = len([e for e in ready if e.event_id in existed])
        for sess in set(e.session for e in ready):
            n = len([e for e in ready if e.session == sess])
            VisitSession.objects.filter(pk=sess.pk).update(
                ended_at=max(e.occurred_at for e in ready if e.session == sess),
                event_count=sess.event_count + n,
                duration_ms=int(sum(
                    (e.ended and 0) or 0 for e in [] ) or
                    max(0, (max(x.occurred_at for x in ready if x.session == sess)
                            - min(x.occurred_at for x in ready if x.session == sess)
                           ).total_seconds() * 1000))
    return result

注意（执行者必读——上面代码 duration_ms 段是占位，按以下正确版本实现）：

from django.db.models import F, Greatest

new_events = [e for e in ready if e.event_id not in existed]   # 重放命中的不参与聚合！
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)             # 有效停留单独累计，供访客"总停留"取数

• 重放批不得抬高会话聚合：event_count/stay_ms/duration_ms 只用 new_events（否则重试/重放被幂等去重后仍把会话计数刷高——test_replay_duplicated 需补断言 VisitSession.event_count == 1）。

• 同批多会话各自按上式计算。

• duplicated 计数允许并发近似（唯一约束保证绝不重复入库，仅统计口径可能差 1-2，M1 接受）。

• Step 4: 跑测试通过 → 5 passed。修到绿为止。

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

Task 7: 采集接口 POST /api/collect/batch

Files:

• Create: server/tracker/api/__init__.py, server/tracker/api/collect.py, server/tracker/tests/test_collect.py, Modify: server/config/urls.py, server/config/settings.py（CORS 动态白名单）

• Step 1: 失败测试（要点：白名单域放行/非白名单 403/限流触发 429/textplain 解析/响应最小化/幂等重放）

import json
import pytest
from django.core.cache import cache
from django.utils import timezone
from tracker.models import Site

NOW = timezone.now()

@pytest.fixture(autouse=True)
def _clear_cache():
    cache.clear()   # locmem 限流历史跨测试残留，会让 429 用例假红

def _post(client, site, body, origin="https://www.a.com", extra=None):
    return client.post("/api/collect/batch", data=json.dumps(body),
        content_type="text/plain", HTTP_ORIGIN=origin, **(extra or {}))

def _body(n=1, sid="s1", vid="v1"):
    return {"site_id": "site_x", "visitor_id": vid, "session_id": sid,
            "context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
            "events": [{"event_id": f"e{i}", "event_name": "page_view",
                        "occurred_at": NOW.isoformat(), "path": "/", "title": "t",
                        "properties": {}} for i in range(n)]}

def test_collect_ok_creates_events(db, client):
    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
    r = _post(client, None, _body(2))
    assert r.status_code == 200 and r.json() == {"code": 200, "data": {"accepted": 2, "duplicated": 0}, "message": "ok"}

def test_origin_not_whitelisted_rejected(db, client):
    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
    r = _post(client, None, _body(1), origin="https://evil.com")
    assert r.status_code == 403 and r.json()["code"] == 403

def test_inactive_site_rejected(db, client):
    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x", is_active=False)
    assert _post(client, None, _body(1)).status_code == 403

def test_unknown_site_403_not_404(db, client):   # 不泄露站点是否存在
    r = _post(client, None, _body(1))
    assert r.status_code == 403

def test_replay_duplicated(db, client):
    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
    _post(client, None, _body(1))
    r = _post(client, None, _body(1))
    assert r.json()["data"]["duplicated"] == 1

def test_throttle_429(db, client, settings):
    settings.REST_FRAMEWORK = {**settings.REST_FRAMEWORK,
        "DEFAULT_THROTTLE_RATES": {"collect": "3/min"}}
    Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
    for _ in range(3):
        assert _post(client, None, _body(1)).status_code == 200
    assert _post(client, None, _body(1)).status_code == 429

• Step 2: 确认失败 → 404（无路由）。

• Step 3: 实现

server/tracker/api/collect.py:

import json
from datetime import datetime, timezone as dt_tz
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 tracker.models import Site
from tracker.services.ingest import ingest_batch

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 _iso(dt):
    if isinstance(dt, str):
        s = dt.replace("Z", "+00:00")
        d = datetime.fromisoformat(s)
        return d if d.tzinfo else d.replace(tzinfo=dt_tz.utc)  # Django≥5 已删除 django.utils.timezone.utc
    return dt

class CollectBatchView(APIView):
    permission_classes = [AllowAny]
    parser_classes = [PlainJSONParser]
    throttle_classes = [CollectThrottle]
    authentication_classes = []

    def post(self, request):
        origin = request.headers.get("Origin") or request.headers.get("Referer") or ""
        if int(request.headers.get("Content-Length") or 0) > 262144:
            return self._deny(400, "payload_too_large")   # 应用层兜底（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")         # 三种拒绝统一 reason，防 site_id 枚举探测
        host = (origin.split("//")[-1].split("/")[0].split(":")[0].lower()
                if origin else "")
        if host and host not in [d.lower() for d in site.domains]:
            return self._deny(403, "not_allowed")
        if not host and not site.allow_no_origin:
            return self._deny(403, "not_allowed")
        payload = request.data or {}
        for ev in payload.get("events") or []:
            if isinstance(ev, dict) and "occurred_at" in ev:
                try:
                    ev["occurred_at"] = _iso(ev["occurred_at"])
                except ValueError:
                    return self._deny(400, "invalid_occurred_at")
        if not payload.get("visitor_id") or not payload.get("session_id"):
            return self._deny(400, "invalid_payload")
        ip = self._client_ip(request)
        r = ingest_batch(site, payload, ip, request.headers.get("User-Agent", ""))
        return Response({"code": 200,
            "data": {"accepted": r["accepted"], "duplicated": r["duplicated"]},
            "message": "ok"})

    def _match_site(self, request):
        sid = (request.data or {}).get("site_id") if isinstance(request.data, dict) else None
        # site_id 在 body；为匹配 Origin 需先读 body——parser 已解析到 request.data
        from tracker.models import Site as S
        return S.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", "")
        if xff:
            return xff.split(",")[-1].strip()
        return request.META.get("REMOTE_ADDR")

    def _deny(self, code, reason):
        resp = Response({"code": code, "data": None, "message": reason}, status=code)
        origin = self.request.headers.get("Origin")
        if origin:   # 拒绝也要回 ACAO，否则浏览器读不到错误体、SDK 误判为网络错误走重试
            resp["Access-Control-Allow-Origin"] = origin
        return resp

server/config/urls.py:

from django.urls import path
from tracker.api.collect import CollectBatchView
urlpatterns = [
    path("api/collect/batch", CollectBatchView.as_view()),
]

CORS 动态白名单（settings.py 追加 + MIDDLEWARE 顺序保持 corsheaders 在前）：

CORS_ALLOWED_ORIGIN_REGEXES = []   # 不用全局

并在 tracker/api/collect.py 出口处手动加头（简单直接，绕开 cors-headers 的静态配置）：

# post() 成功分支 Response 前：
resp = Response({...})
if origin:
    resp["Access-Control-Allow-Origin"] = origin
    resp["Access-Control-Allow-Headers"] = "content-type"
return resp

另加 OPTIONS 预检支持：CollectBatchView.options() 返回同头 200。（SDK 用 text/plain 免预检，此处为兜底。）

• Step 4: 跑测试通过 → 6 passed。

• Step 5: Commit feat: 采集接口（域名白名单/限流/textplain解析/幂等）

Task 8: refresh_stats 命令 + stats.py

Files:

• Create: server/tracker/services/stats.py, server/tracker/management/__init__.py, server/tracker/management/commands/__init__.py, server/tracker/management/commands/refresh_stats.py, server/tracker/tests/test_stats.py

• Step 1: 失败测试：造 2 天数据（访客 v1 两天访问、v2 一天）→ refresh(site, date) 后 DailyStat 的 pv/uv/sessions/new_visitors 正确；uv 按访客去重、new_visitors 按 first_seen_at 当日计数。

• Step 2: 实现 stats.py：

from datetime import datetime, time, timedelta
from django.utils import timezone
from tracker.models import DailyStat, Event, VisitSession

def refresh_daily(site, date) -> DailyStat:
    day_start = timezone.make_aware(datetime.combine(date, time.min))
    day_end = day_start + timedelta(days=1)
    pv = Event.objects.filter(site=site, is_bot=False,
                              occurred_at__gte=day_start, occurred_at__lt=day_end).count()
    uv = Event.objects.filter(site=site, is_bot=False, occurred_at__gte=day_start,
                              occurred_at__lt=day_end).values("visitor").distinct().count()
    sessions = VisitSession.objects.filter(site=site, visitor__is_bot=False,
                                           started_at__gte=day_start,
                                           started_at__lt=day_end).count()
    new_v = Visitor.objects.filter(site=site, first_seen_at__gte=day_start,
                                   first_seen_at__lt=day_end).count()
    obj, _ = DailyStat.objects.update_or_create(site=site, date=date,
        defaults={"pv": pv, "uv": uv, "sessions": sessions, "new_visitors": new_v})
    return obj

refresh_stats.py 命令：python manage.py refresh_stats --days 3（回刷近 N 天，含今天）。 调度：生产环境由 1Panel 计划任务每日 00:10 执行 docker exec <web容器> python manage.py refresh_stats --days 2（写入 Task 22 的 README-deploy；不做每日调度则趋势图历史全为零）。

• Step 3: 测试通过，Commit feat: 日聚合刷新命令

Task 9: JWT 认证接口

Files:

• Create: server/tracker/api/auth.py, server/tracker/tests/test_auth.py, Modify: config/urls.py

• Step 1: 失败测试：/api/admin/auth/token 用户名密码 → access/refresh；错密码 401；/api/admin/auth/me 带 token 返回 username/is_admin（is_staff）；无 token 401；/api/admin/users 仅 is_staff 可 POST 建普通成员。

• Step 2: 实现：TokenObtainPairView 子类（返回 {code:200,data:{access,refresh},message:"ok"} 包一层），me 返回 {username, is_admin: user.is_staff}；UserCreateView（IsAdminUser）username/password 创建 is_staff=False 用户。登录接口挂 ScopedRateThrottle scope=login（10/min/IP，防爆破；settings 已含 rate）。JWT 取舍（M1 决策，README-deploy 写明）：access 12h、refresh 显式配置 24h、无黑名单/轮换——token 泄露的唯一应急是更换 SECRET_KEY（全体登出）。

• Step 3: 测试通过，Commit feat: JWT 登录与用户接口

Task 10: 站点管理 API

Files:

• Create: server/tracker/api/sites.py, server/tracker/tests/test_sites.py, Modify: config/urls.py

• Step 1: 失败测试：CRUD（列表/新建自动生成 site_id/更新域名白名单/停用）；仅登录成员可访问，删除仅 admin；域名校验（去协议、转小写、非空列表）。

• Step 2: 实现：DRF APIView 或轻量 ModelViewSet + 自行包 {code,data,message}。序列化输出含 site_id/domains/is_active/allow_no_origin。

• Step 3: 测试通过，Commit feat: 站点管理接口

Task 11: 总览 Dashboard API

Files:

• Create: server/tracker/api/dashboard.py, server/tracker/tests/test_dashboard.py, Modify: config/urls.py

• Step 1: 失败测试（造数后断言）：

def test_overview_shape(db, client, django_user_model):
    # 造 2 访客/3 会话/10 事件，1 条在 5 分钟内（active_now=1）
    r = client.get("/api/admin/dashboard/overview")
    assert r.json()["data"]["today"] == {"pv": 10, "uv": 2, "sessions": 3,
        "new_visitors": 2, "active_now": 1}
    assert len(r.json()["data"]["trend"]) == 30
    assert r.json()["data"]["sources"][0]["source"] == "google.com"
    assert r.json()["data"]["top_pages"][0]["path"] == "/pricing"

• Step 2: 实现：today 用事件/会话表实时聚合（当天 Asia/Shanghai 边界）；trend 取 DailyStat 近 30 天（缺日补零；今天用实时值覆盖最后一位）；sources = 当天 VisitSession group by referrer_domain（空→"直接访问"）top8；top_pages = 当天 page_view group by path top8；active_now = Event received_at > now-5min distinct visitor。认证 + 成员权限。所有聚合排除 is_bot=True（爬虫不进 Dashboard）。
• Step 3: 测试通过，Commit feat: 总览聚合接口

Task 12: 访客列表/详情/时间线/会话 API

Files:

• Create: server/tracker/api/visitors.py, server/tracker/tests/test_visitors.py, Modify: config/urls.py

路由：

GET  /api/admin/visitors                       列表（分页/筛选/搜索）
GET  /api/admin/visitors/<visitor_id>          详情（档案+环境+统计+会话摘要列表）
GET  /api/admin/visitors/<visitor_id>/timeline  跨天时间线（按日分组→会话→事件）
GET  /api/admin/visitors/<visitor_id>/sessions/<session_id>  单会话页面路径

• Step 1: 失败测试：
  • 列表字段齐全：visitor_id/country/province/city/device_type/os/browser/first_seen_at/last_seen_at/visit_count/active_days/session_count/total_stay_ms/first_source/last_source/online；筛选 country/device_type/online=true/search=访客ID或IP前缀；按 last_seen_at 倒序分页（page_size 20）。
  • timeline 结构：[{date:"2026-09-15", sessions:[{session_id, started_at, events:[{time,event_name,path,title,active_ms}]}]}]，多天数据按日倒序、日内时间正序；默认近 30 天，?days= 可调，单次上限 1000 事件。
  • 会话详情：事件按 occurred_at 正序，含 active_ms（page_stay 的 properties.active_ms 就地并入相邻 page_view 或单列，见实现说明）。
• Step 2: 实现：ORM 查询 + Python 分组；total_stay_ms = Sum("sessions__stay_ms") 注解（有效停留口径，不用 duration_ms 跨度）；first_source = first_referrer_domain or first_utm.utm_source or "直接访问"；last_source 取最近会话同逻辑；online = last_seen_at > now-5min；列表默认过滤 is_bot=False（?show_bot=1 可查爬虫）。page_view 与 page_stay 合并展示：时间线返回时若 page_stay 的 path 与前一条 page_view 相同，则把 active_ms 合入该 page_view（服务端做，前端省事）；同 path 连续多段 page_stay（多次隐藏/恢复）的 active_ms 累加后合入。会话详情展示"时长"统一用 stay_ms 口径（duration_ms 是批内跨度累加，仅内部字段不展示）。
• Step 3: 测试通过，Commit feat: 访客列表/详情/时间线/会话接口

---

Phase 2 · JS SDK（M1：多页网站）

Task 13: identity.ts + context.ts

Files:

• Create: sdk/src/identity.ts, sdk/src/context.ts, sdk/test/identity.test.ts, sdk/test/context.test.ts

• Step 1: 失败测试（happy-dom 提供 localStorage/document；fake timers 测 30 分钟切会话）：

import { describe, it, expect, beforeEach, vi } from "vitest";
import { getVisitorId, getSessionId, touchActivity } from "../src/identity";

describe("identity", () => {
  beforeEach(() => localStorage.clear());
  it("visitor_id 持久化", () => {
    expect(getVisitorId()).toBe(getVisitorId());
    expect(getVisitorId()).toMatch(/^[0-9a-f-]{36}$/);
  });
  it("30 分钟无活动切新会话", () => {
    vi.useFakeTimers();
    const s1 = getSessionId();
    vi.advanceTimersByTime(10 * 60_000); touchActivity();
    vi.advanceTimersByTime(10 * 60_000);
    expect(getSessionId()).toBe(s1);            // 20min 内有活动→同会话
    vi.advanceTimersByTime(31 * 60_000);        // 再过 31min 无活动
    expect(getSessionId()).not.toBe(s1);
  });
});

context 测试：parseUtm("?utm_source=google&utm_medium=cpc&other=1") 只留 utm_*；sanitizeUrl 删 email/token 参数；buildEnvContext() 返回 device_type（UA 含 "Mobile" → "手机"）/os/browser/language/timezone/screen 键。

• Step 2: 实现

// identity.ts
import { uuid } from "./util";
const K = { vid: "ctk_vid", sid: "ctk_sid", act: "ctk_act" };
const IDLE = 30 * 60_000;
// localStorage 可能抛异常（隐私模式/禁存储，Safari ITP）：统一走带内存降级的 store，
// 任何路径都不得让异常冒泡到宿主页（spec §4.5 红线）
const mem: Record<string, string> = {};
const store = {
  get(k: string): string | null {
    try { return localStorage.getItem(k); } catch { return mem[k] ?? null; }
  },
  set(k: string, v: string) {
    try { localStorage.setItem(k, v); } catch { mem[k] = v; }
  },
};
export function getVisitorId(): string {
  let v = store.get(K.vid);
  if (!v) { v = uuid(); store.set(K.vid, v); }
  return v;
}
export function touchActivity() { store.set(K.act, String(Date.now())); }
export function getSessionId(): string {
  const last = +(store.get(K.act) || 0);
  let s = store.get(K.sid);
  if (!s || Date.now() - last > IDLE) { s = uuid(); store.set(K.sid, s); }
  touchActivity();
  return s;
}

context.ts：buildEnvContext()（UA 正则判 device_type/os/browser；navigator.language、Intl.DateTimeFormat().resolvedOptions().timeZone、screen.width+"x"+screen.height、navigator.hardwareConcurrency、navigator.connection?.effectiveType）；parseUtm(search)；sanitizeUrl(path)；pageContext() → {path: sanitized, title: document.title, referrer: document.referrer, referrer_domain, utm}。

• Step 3: 测试通过，Commit feat(sdk): 身份与环境上下文模块

Task 14: transport.ts 批量传输

Files:

• Create: sdk/src/transport.ts, sdk/src/config.ts, sdk/test/transport.test.ts

• Step 1: 失败测试（vi.stubGlobal fetch；fake timers）：

  • 满 10 条即发；15 秒定时器到即发；payload 含 batch_id/site_id/visitor_id/session_id/events。
  • 500 失败→退避重试（500ms×2^n 抖动）最多 3 次→丢弃计数；400 不重试直接丢。
  • 429 → 重试；403/400 → 不重试直接丢。
  • flushBeacon() 调 navigator.sendBeacon（spy 断言 URL 与 body）。happy-dom 没有 sendBeacon，用 vi.stubGlobal("navigator", { sendBeacon: spy })，测完 restore。
  • 队列上限 200（用 batchSize > queueMax 的自定义 cfg 测，默认 batchSize=10 到不了 200），超限丢最旧并计数。
  • 测试技巧：异步断言用 await vi.advanceTimersByTimeAsync()（flush/send 是 async）；fake timers 需含 Date（vitest 默认 toFake 即含）；document.hidden 是 getter，用 Object.defineProperty(document, "hidden", { value: true, configurable: true }) + dispatchEvent(new Event("visibilitychange")) 模拟；每个用例后 vi.useRealTimers()。

• Step 2: 实现

import { uuid } from "./util";
import { getVisitorId, getSessionId } from "./identity";

export interface CtkConfig { siteId: string; endpoint: string;
  batchSize?: number; flushMs?: number; maxRetry?: number; queueMax?: number }

export class Transport {
  private q: any[] = []; private timer: any = null;
  private retry = 0; dropped = 0;
  constructor(public cfg: Required<CtkConfig>) {}
  push(ev: object) {
    if (this.q.length >= this.cfg.queueMax) { this.q.shift(); this.dropped++; }
    this.q.push(ev);
    if (this.q.length >= this.cfg.batchSize) this.flush();
    else if (!this.timer)
      this.timer = setTimeout(() => this.flush(), this.cfg.flushMs);
  }
  flush(beacon = false) {
    if (this.timer) { clearTimeout(this.timer); this.timer = null; }
    if (!this.q.length) return;
    try {
      const events = this.q; this.q = [];
      const body = JSON.stringify({ batch_id: uuid(), site_id: this.cfg.siteId,
        visitor_id: getVisitorId(), session_id: getSessionId(),
        sent_at: new Date().toISOString(), context: (globalThis as any).__ctk_ctx,
        events });
      if (beacon && navigator.sendBeacon) {
        navigator.sendBeacon(this.cfg.endpoint, new Blob([body], {type: "text/plain"}));
        return;
      }
      this.send(body, events.length);
    } catch { /* 定时器/事件回调内的任何异常不得冒泡到宿主页 */ }
  }
  private async send(body: string, count: number) {
    try {
      const res = await fetch(this.cfg.endpoint, { method: "POST",
        keepalive: true, headers: { "Content-Type": "text/plain" }, body });
      if (res.status >= 400 && res.status < 500 && res.status !== 429) {
        this.retry = 0; return;                    // 4xx 参数/权限错不重试（403 重试只会掩盖配置错误）
      }
      if (!res.ok) throw new Error(String(res.status));
      this.retry = 0;
    } catch {
      if (this.retry++ < this.cfg.maxRetry) {
        const delay = 500 * 2 ** this.retry * (0.8 + Math.random() * 0.4);
        setTimeout(() => this.send(body, count).catch(() => {}), delay);
      } else { this.retry = 0; this.dropped += count; }   // 按事件数计丢弃
    }
  }
}

config.ts：mergeConfig(opts) 填默认 {batchSize:10, flushMs:15000, maxRetry:3, queueMax:200}。

• Step 3: 测试通过，Commit feat(sdk): 批量传输（批量/退避/信标/丢包计数）

Task 15: events.ts + index.ts 入口与打包

Files:

• Create: sdk/src/events.ts, sdk/src/index.ts, sdk/test/events.test.ts, sdk/demo/index.html, Modify: .gitignore（sdk/node_modules, sdk/dist）

• Step 1: 失败测试：

  • init 后 sdk_loaded 事件入队；page_view 含 path/title/referrer_domain/utm（utm 随每页上报，服务端只写首个，可接受冗余）。
  • 批次 context 含设备环境 + referrer_domain/utm/path（首触归因链路，断言 __ctk_ctx 字段齐全）。
  • 有效停留：fake timers 下 visible 且有 pointerdown → 3 秒后 active_ms≈3000；document.hidden=true 后不再累计；隐藏时自动 flush beacon（page_stay 已入队）。
  • localStorage 抛异常时 15 秒定时 flush 不产生未捕获异常（stub localStorage 全方法抛错 + await vi.advanceTimersByTimeAsync(15000)，配 vi.onUnhandledError 断言；且身份降级到内存后事件照常发出）。
  • 隐藏→恢复→再隐藏产生两段 page_stay，active_ms 各自正确（第二段从 0 重新累计）。

• Step 2: 实现

// events.ts
import { Transport } from "./transport";
import { pageContext } from "./context";
import { uuid } from "./util";

export function startPageTracking(t: Transport) {
  let activeMs = 0, lastAct = Date.now();
  const act = () => { lastAct = Date.now(); };
  ["pointerdown", "keydown", "wheel", "touchstart"].forEach(e =>
    document.addEventListener(e, act, { passive: true, capture: true }));
  setInterval(() => { try { if (!document.hidden && Date.now() - lastAct < 30_000) activeMs += 1000; } catch { /* 静默 */ } }, 1000);
  const emit = (event_name: string, properties: any = {}) => {
    const pc = pageContext();
    t.push({ event_id: uuid(), event_name, occurred_at: new Date().toISOString(),
      path: pc.path, title: pc.title, properties });
  };
  emit("page_view", { referrer: document.referrer,
    referrer_domain: pageContext().referrer_domain, utm: pageContext().utm });
  const close = () => {
    try {
      // path 必须与 page_view 同源同清洗（pageContext().path 含 query），
      // 否则带 UTM 进落地页时服务端"page_stay 合并进相邻 page_view"匹配失败
      if (activeMs > 0) emit("page_stay", { active_ms: activeMs, path: pageContext().path });
      activeMs = 0; t.flush(true);
    } catch { /* 静默 */ }
  };
  document.addEventListener("visibilitychange", () => { if (document.hidden) close(); });
  window.addEventListener("pagehide", close);
}

// index.ts —— 打包入口（IIFE 副作用）
import { mergeConfig } from "./config";
import { Transport } from "./transport";
import { startPageTracking } from "./events";
import { buildEnvContext, pageContext } from "./context";

const g = globalThis as any;
const q: any[][] = g.ctk?.q || [];
let transport: Transport | null = null;

function dispatch(cmd: string, arg?: any) {
  if (cmd === "init" && !transport) {
    const cfg = mergeConfig(arg);
    // 首触归因修复：批次 context 除设备环境外必须带 referrer_domain/utm/path——
    // 服务端 _touch_visitor/_get_session 只从 context 读首触，缺了则首触/会话来源恒空
    const pc = pageContext();
    (globalThis as any).__ctk_ctx = { ...buildEnvContext(),
      referrer_domain: pc.referrer_domain, utm: pc.utm, path: pc.path };
    transport = new Transport(cfg);
    transport.push({ event_id: uuid(), event_name: "sdk_loaded",
      occurred_at: new Date().toISOString(),
      path: location.pathname, title: document.title,
      properties: { sdk_version: "0.1.0" } });
    startPageTracking(transport);
  } else if (cmd === "track" && transport) transport.push(arg);
  else if (cmd === "flush" && transport) transport.flush();
}
g.ctk = dispatch;
q.forEach(([c, a]) => dispatch(c, a));

（uuid import 补上；sdk_loaded 后 init 前队列已重放。整个 dispatch 与 startPageTracking 必须包在 try/catch 中，任何异常静默吞掉（console.debug 级别记录，绝不冒泡到宿主页面）——这是 spec §4.5 的硬性要求，Task 15 测试补一条：让 localStorage 抛异常时 init 不抛。）

sdk/demo/index.html：

<!doctype html><html><head><meta charset="utf-8"><title>CTK Demo</title>
<script>window.ctk = window.ctk || function(){(ctk.q=ctk.q||[]).push(arguments)};
ctk("init", { siteId: "site_x", endpoint: "http://127.0.0.1:8000/api/collect/batch" });</script>
<script async src="../dist/v1.js"></script></head>
<body><h1>Demo 页 1</h1><a href="page2.html">去第 2 页</a></body></html>

（page2.html 同构再建一份，标题"Demo 页 2"，验证多页跳转的会话连续性。）

- [ ] **Step 3: 构建 + 体积门禁**
```bash
cd sdk && pnpm build && pnpm test
gzip -c dist/v1.js | wc -c   # 必须 ≤ 15360（15KB），超标则精简（去 polyfill/减正则）

• Step 4: Commit feat(sdk): 页面浏览/有效停留/事件入口，打包≤15KB

Task 16: SDK ↔ 后端本地集成验证（手动脚本）

• Step 1: 后端起服务（本地开发）：DB_ENGINE=sqlite python manage.py migrate && DJANGO_DEBUG=1 DB_ENGINE=sqlite python manage.py runserver（生成 db.sqlite3，后续 Task 19-21 验证复用同一环境）。createsuperuser 建 admin。demo 站点必须写死 site_id（demo HTML 硬编码）：python manage.py shell 执行 Site.objects.create(name="demo", site_id="site_x", domains=["127.0.0.1"], allow_no_origin=True)。
• Step 2: cd sdk && python -m http.server 8080，浏览器开 http://127.0.0.1:8080/demo/，点击、停留、切后台、跳 page2。
• Step 3: 断言：Event 表出现 page_view/page_stay/sdk_loaded；Visitor.visit_count/active_days 正确；重复刷新不产生重复 event_id；后台 Django shell 查询验证（截图/记录到 PR 描述）。
• Step 4: Commit（如有修复）fix: SDK↔后端联调修复

---

Phase 3 · Art Design Pro 管理后台

Task 17: 模板落地与请求层适配

Files:

• Create: admin/（模板克隆），Modify: admin/.env.development, admin/.env.production, admin/vite.config.ts, admin/src/utils/request/*（按模板实际结构定位）

• Step 1: 克隆模板（大陆网络备用：gitee 镜像 https://gitee.com/lingchen163/art-design-pro.git；npm 慢加 --registry=https://registry.npmmirror.com）

git clone --depth 1 https://github.com/Daymychen/art-design-pro.git admin
cd admin && rm -rf .git && pnpm install && pnpm dev   # 确认模板可跑
pnpm clean:dev    # 一键精简演示页（按提示输入 yes）

• Step 2: .env.development 设 VITE_API_URL=/api；.env.production 同样设 VITE_API_URL=/api（模板默认指向 Apifox Mock 域名，漏改则 pnpm build 的产物请求 mock，部署必挂）；vite.config.ts 加 proxy "/api": { target: "http://127.0.0.1:8000", changeOrigin: true }。
• Step 3: 改造模板 src/utils/request（或等价拦截器文件）：响应成功判定改为 res.data.code === 200 时 return res.data.data；401 跳登录页携带 redirect；token 从登录响应存入（沿用模板的 token 管理，替换其 mock 登录为真实 API）。
• Step 4: 登录页接 /api/admin/auth/token（用户名/密码 → access/refresh），登录成功进布局页。手动验证：错误密码报错、正确 token 进入首页。
• Step 5: Commit feat(admin): 模板精简+请求层适配 Django JWT

Task 18: api 模块 + 路由注册

Files:

• Create: admin/src/api/tracker/{auth,dashboard,visitors,sites}.ts, admin/src/router/modules/tracker.ts（命名按模板路由规范）

• Step 1: api 模块（示例 visitors.ts，其余同构）：

import request from "@/utils/request";
export interface VisitorRow { visitor_id: string; country: string; province: string;
  city: string; device_type: string; os: string; browser: string;
  first_seen_at: string; last_seen_at: string; visit_count: number;
  active_days: number; session_count: number; total_stay_ms: number;
  first_source: string; last_source: string; online: boolean }
export const getVisitors = (params: any) =>
  request.get<{ list: VisitorRow[]; total: number }>("/admin/visitors", { params });
export const getVisitor = (vid: string) => request.get(`/admin/visitors/${vid}`);
export const getTimeline = (vid: string, days = 30) =>
  request.get(`/admin/visitors/${vid}/timeline`, { params: { days } });
export const getSession = (vid: string, sid: string) =>
  request.get(`/admin/visitors/${vid}/sessions/${sid}`);

• Step 2: 路由模块注册 4 页（总览/访客列表/访客详情(hidden)/站点管理），菜单名中文，icon 任选模板内置。
• Step 3: Commit feat(admin): tracker api 模块与路由

Task 19: 站点管理页 + 总览 Dashboard 页

Files:

• Create: admin/src/views/tracker/site/index.vue, admin/src/views/tracker/dashboard/index.vue

• Step 1: 站点页：Element Plus 表格（名称/site_id 复制按钮/域名列表 tag/状态开关）+ 新建编辑 Dialog（域名校验转小写）+ 停用开关调 PATCH。手动验证增删改。

• Step 2: Dashboard 页：5 张统计卡（今日 PV/UV/会话/新访客/在线）、ECharts 30 天趋势折线（pv/uv/sessions 三线）、来源饼图 top8、热门页面表格。用模板自带图表组件或直接 ECharts。手动对拍后端返回（后端以 DJANGO_DEBUG=1 DB_ENGINE=sqlite python manage.py runserver 起，复用 Task 16 的 db.sqlite3）。

• Step 3: Commit feat(admin): 站点管理与总览页

Task 20: 访客列表页

Files:

• Create: admin/src/views/tracker/visitor/index.vue

• Step 1: 表格列：访客ID(截断+复制)/国旗 emoji+国家·省市/设备(PC·Win·Chrome 一行小字)/首访/最近访问/访问次数/活跃天数/总停留(格式化 x分x秒)/首触来源/最近来源/在线绿点。筛选栏：国家输入/设备下拉/来源输入/在线开关/关键词(访客ID·IP 前缀)/日期范围。分页 20/页，后端排序 last_seen_at desc。行点击 → 详情页。

• Step 2: 手动验证（用 Task 16 联调产生的真实数据）。

• Step 3: Commit feat(admin): 访客列表页

Task 21: 访客详情页（时间线）

Files:

• Create: admin/src/views/tracker/visitor/detail.vue

• Step 1: 布局：顶部档案卡（访客ID/国家·城市/IP/设备环境 grid/首触来源 utm 链/统计 4 数字）+ 时间线主体（el-timeline，日期分组倒序；组内会话块：起止时间+来源+时长，事件行：10:24 · 看了 /pricing · 停留 2分18秒，事件 icon 按类型区分 page_view/page_stay 合并行/sdk_loaded）+ 顶部"会话"Tab 切换会话列表（含页面路径视图：路径序列 + 各页停留）。

• Step 2: 停留格式化工具（formatStay(ms)）；空态/加载态；days 切换（7/30/90）。

• Step 3: 手动验证：当天数据即可验收交互；跨天分组用 Django shell 直插一条昨日事件验证分组渲染（Event.objects.create(..., occurred_at=昨天时间) 绕过 ingest 时间窗），真实跨天效果并入 Task 23 次日复验。这是 M1 验收页面。

• Step 4: Commit feat(admin): 访客详情跨天时间线

---

Phase 4 · 部署与验收

Task 22: Docker 部署产物

Files:

• Create: server/Dockerfile, deploy/docker-compose.yml, deploy/nginx.conf, deploy/README-deploy.md, Modify: .gitignore

• Step 1: server/Dockerfile：

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY . .
EXPOSE 8000
CMD ["sh", "-c", "python manage.py migrate && gunicorn config.wsgi:application -b 0.0.0.0:8000 -w 3 --forwarded-allow-ips=*"]

• Step 2: deploy/nginx.conf 关键段（compose 服务名统一为 web）：location /api/ { client_max_body_size 256k; proxy_pass http://web:8000; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; }（体积上限第一道 + 配合 DRF NUM_PROXIES=1 取真实 IP）；location /sdk/ { alias /var/www/sdk/; }（静态别名，放 sdk/dist/v1.js，不走代理）；location / { root /var/www/admin; try_files $uri $uri/ /index.html; }（SPA fallback）。
• Step 3: deploy/docker-compose.yml（全量写出）+ server/.dockerignore（内容：.venv/、db.sqlite3、__pycache__/、.cache/——否则 Windows 的 .venv（数百 MB）会被 COPY . . 打进镜像）：

services:
  web:
    build: ../server
    env_file: .env     # 必含 DJANGO_ENV=prod DJANGO_DEBUG=0 DJANGO_SECRET_KEY DJANGO_ALLOWED_HOSTS DB_*
    volumes:
      - ./data/ip2region.xdb:/app/tracker/data/ip2region.xdb:ro   # xdb 挂载进容器（不是 docker cp，重建不丢）
    extra_hosts:
      - "host.docker.internal:host-gateway"
  nginx:
    image: nginx:alpine
    ports: ["80:80", "443:443"]
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
      - ../admin/dist:/var/www/admin:ro
      - ../sdk/dist:/var/www/sdk:ro
    depends_on: [web]

连 1Panel 已有 MySQL 的三条路（任选其一，README 写清排查步骤）：

1. 上面已配的 extra_hosts: host-gateway + DB_HOST=host.docker.internal——前提：MySQL 端口发布在 0.0.0.0 且账号 host 为 %；
2. 推荐：加入 1Panel 网络——compose 末尾加 networks: { default: { external: true, name: 1panel-network } }，DB_HOST=<MySQL 容器名>；
3. DB_HOST=172.17.0.1（docker0 网关，MySQL 需监听可达）。

• Step 4: README-deploy.md：前置构建（cd admin && pnpm build、cd sdk && pnpm build 产出 dist 供挂载）→ 1Panel 步骤（建 MySQL 库 → Compose 部署 → 进容器 createsuperuser → xdb 放宿主 deploy/data/ 由 volume 挂载 → 1Panel 网站功能给 API 域名挂 HTTPS 证书 → 建站点拿 site_id → 目标网站贴接入代码）→ 目标网站接入片段（同 sdk/demo，endpoint 换正式域名）→ 生产 .env 必填项清单（DJANGO_ENV=prod、随机 SECRET_KEY：python -c "import secrets; print(secrets.token_urlsafe(48))"、实际域名的 ALLOWED_HOSTS——漏配启动即崩，这是设计好的 fail-fast）→ JWT 应急预案（token 泄露唯一吊销手段 = 换 SECRET_KEY，全体登出）→ 大陆网络备用源（模板 gitee 镜像、xdb gitee 下载、pip -i 清华源、pnpm --registry=https://registry.npmmirror.com）→ 若目标站套 CDN/WAF，X-Forwarded-For 层数变化需同步调整 _client_ip 与 NUM_PROXIES。
• Step 5: Commit chore: Docker/Nginx 部署产物与 1Panel 指南

Task 23: M1 端到端验收（真实网站）

• Step 1: 选一个真实网站贴接入代码（白名单填其域名）。前置（spec 治理底线，不装不验收）：目标站先上线隐私告知——一段话说明"本站使用第一方分析存储收集访问行为与设备信息，用于改进服务"，附联系方式（文案模板取 spec §9.4）。
• Step 2: 按验收清单逐项过（写入 PR/交付记录）：
  • 新访客首次访问 → 后台访客列表出现，环境/来源/地理正确
  • 浏览 3 个页面各停留可感知时长 → 时间线 3 条 page_view 且 active_ms 合理
  • 隔天再访（需真实次日复验——服务端拒绝 >24h 时间偏移事件，无法伪造跨天数据，这是防篡改设计；当天先验证：等 30 分钟后新访问使 visit_count+1、时间线新增会话组）→ 同 visitor_id、visit_count+1、时间线新增日期组
  • 首触来源 Google/带 UTM 链接进入 → first_referrer_domain/first_utm 正确（无 Google 环境时：无痕窗口 + ?utm_source=google&utm_medium=test 链接造新访客验证）
  • 直接刷新/断网重试/重复上报 → 无重复事件（event_id 幂等）（断网用 DevTools Network→Offline 再恢复）
  • curl 伪造非白名单 Origin 上报 → 403；超频 → 429（注意：白名单 Origin 本身可被伪造，这是归属控制不是认证——spec §11 定位）
  • Dashboard 数字与访客明细对得上；总停留=各会话之和
  • SDK 对页面加载性能无可感知影响（DevTools Network 确认 async + ≤15KB）
• Step 3: 全部通过 → git tag m1-beta；验收记录（含截图与数据对拍）写入 docs/验收记录-m1.md，Commit docs: M1 验收通过。同时记录采集开始日期：M2 的 IP 截断任务必须在其 180 天内上线，否则先跑截断脚本再继续采集。

---

执行注意

• Task 6 ingest 的 duration_ms 以"计划内正确版本"实现（F 表达式 + new_events 排除重放），不要照抄占位伪码。
• geo.py 是进程级单例（xdb 一次读入 buffer）；dropped 统一按事件数计。
• MySQL 的 INSERT IGNORE（ignore_conflicts）会连字段截断类错误一起吞——_FIELD_LEN 已按模型长度截断，属双保险；导入/迁移后抽查字段长度。
• Art Design Pro 模板内部组件 API 以克隆下来的实际代码为准（Task 17 后先读 src/views 现存示例再写页面，注意模板 request 封装签名可能是 request.get({url, params})）；本计划页面代码是基线，允许按模板约定适配，不允许砍功能。
• ip2region pip 包若与计划 API 不符，参照其 README 调整 geo.py，测试断言不变。
• 所有中文文案直接用于界面；代码注释保持中文、少量、说明意图。
• 每任务完成即提交；测试红→绿→提交，禁止跳测试。