fangke/server/tracker/tests/test_dashboard.py

import itertools

import pytest
from datetime import timedelta
from django.contrib.auth import get_user_model
from django.utils import timezone
from rest_framework.test import APIClient
from tracker.models import DailyStat, Event, Site, SourceClassification, VisitSession, Visitor
from tracker.services.ingest import ingest_batch
from tracker.tests.test_ingest import SITE_ID, _batch, _ev, _v2_batch, _v2_event

pytestmark = pytest.mark.django_db
User = get_user_model()

# 用上海本地时刻做基准：日界是上海自然日（make_aware/localdate），
# 直接用 UTC 的 now 在本地 00:00-08:00 之间跑测试会错一天
NOW = timezone.localtime(timezone.now())
TODAY = NOW.date()
YESTERDAY = TODAY - timedelta(days=1)
AT_3D = NOW - timedelta(days=3)    # 7d 当前窗口内（近 3 天）
AT_10D = NOW - timedelta(days=10)  # 7d 上期窗口内（前 7 天）


@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 _visitor(site, vid, last_seen=None, first_seen=None):
    v = Visitor.objects.create(site=site, visitor_id=vid)
    # first/last_seen_at 是 auto_now_add，只能 queryset.update 绕开
    fields = {}
    if last_seen is not None:
        fields["last_seen_at"] = last_seen
    if first_seen is not None:
        fields["first_seen_at"] = first_seen
    if fields:
        Visitor.objects.filter(pk=v.pk).update(**fields)
    return v


def _session(site, visitor, sid, referrer="", at=NOW, stay_ms=0):
    return VisitSession.objects.create(site=site, visitor=visitor, session_id=sid,
        started_at=at, ended_at=at + timedelta(minutes=5), referrer_domain=referrer,
        stay_ms=stay_ms)


def _events(site, visitor, session, n, path, at=NOW, prefix="e"):
    for i in range(n):
        Event.objects.create(site=site, visitor=visitor, session=session,
            event_id=f"{prefix}-{i}", event_name="page_view",
            occurred_at=at + timedelta(seconds=i), path=path)


def test_overview_requires_auth(client):
    assert client.get("/api/admin/dashboard/overview").status_code == 401


def test_overview_7d_cards_and_deltas(client, member_user):
    """7d 四卡指标 + delta 正负 + 上期零（returning）为 null；爬虫全口径排除。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    # 本期（近 7 天）：v1/v2 今天首访（新访客），v3 是 10 天前的老访客 3 天前回访
    v1 = _visitor(site, "v1", last_seen=timezone.now())
    v2 = _visitor(site, "v2", last_seen=timezone.now())
    v3 = _visitor(site, "v3", last_seen=timezone.now(), first_seen=AT_10D)
    s1 = _session(site, v1, "s1", stay_ms=60000)
    s2 = _session(site, v2, "s2", stay_ms=30000)
    s3 = _session(site, v3, "s3", at=AT_3D)                       # stay 0
    _events(site, v1, s1, 2, "/a", prefix="a")
    _events(site, v2, s2, 1, "/b", prefix="b")
    _events(site, v3, s3, 1, "/c", at=AT_3D, prefix="c")
    # 上期（前 7 天）：v3 首访 + v4 三次会话——上期 visitors=2 / visits=4 / avg=20000
    sp3 = _session(site, v3, "sp3", at=AT_10D, stay_ms=10000)
    _events(site, v3, sp3, 1, "/old3", at=AT_10D, prefix="o3")
    v4 = _visitor(site, "v4", last_seen=AT_10D, first_seen=AT_10D)
    for i, stay in enumerate((20000, 30000, 20000)):
        sp = _session(site, v4, f"sp4-{i}", at=AT_10D, stay_ms=stay)
        _events(site, v4, sp, 1, "/old4", at=AT_10D, prefix=f"o4{i}")
    # 爬虫：今天有大额停留与事件，任何口径都不得计入（事件行同步标记 is_bot，与 ingest 同口径）
    bot = _visitor(site, "bot-v", last_seen=timezone.now())
    Visitor.objects.filter(pk=bot.pk).update(is_bot=True)
    sb = _session(site, bot, "sb", stay_ms=999000)
    _events(site, bot, sb, 5, "/bot", prefix="bt")
    Event.objects.filter(visitor=bot).update(is_bot=True)

    r = client.get("/api/admin/dashboard/overview")            # 缺省 range=7d
    assert r.status_code == 200
    d = r.json()["data"]
    # 本期：visitors=3（v1/v2/v3）、visits=3、avg=(60k+30k+0)/3=30k、returning=3-2 新=1
    # 上期：visitors=2（v3/v4）、visits=4、avg=80k/4=20k、returning=2-2 新=0 → delta null
    assert d["cards"]["visitors"] == {"value": 3, "delta_pct": 50.0}       # 正
    assert d["cards"]["visits"] == {"value": 3, "delta_pct": -25.0}        # 负
    assert d["cards"]["avg_stay_ms"] == {"value": 30000, "delta_pct": 50.0}
    assert d["cards"]["returning"] == {"value": 1, "delta_pct": None}      # 上期 0 → null
    # 在线：v1/v2/v3 活跃（bot 排除、v4 十天前不在线）
    assert d["header"] == {"online": 3, "active_now": 3}   # active_now 兼容保留
    # 趋势：7 点日粒度实时聚合，无数据的日补零——3 天前 v3 回访有真实值
    # （uv=1、sessions=1，bot 排除），不依赖 DailyStat/聚合作业
    trend = d["trend"]
    assert len(trend) == 7
    assert trend[0]["date"] == (TODAY - timedelta(days=6)).isoformat()
    assert trend[-1] == {"date": TODAY.isoformat(), "uv": 2, "sessions": 2}
    assert trend[-4] == {"date": (TODAY - timedelta(days=3)).isoformat(), "uv": 1, "sessions": 1}
    assert trend[-2] == {"date": YESTERDAY.isoformat(), "uv": 0, "sessions": 0}
    assert d["channels"] == []                               # 无分类行 → 空列表


def test_overview_prev_zero_delta_null(client, member_user):
    """上期完全无数据：四卡 delta_pct 全为 null，卡值正常。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    v1 = _visitor(site, "v1", last_seen=timezone.now())
    s1 = _session(site, v1, "s1", stay_ms=50000)
    _events(site, v1, s1, 2, "/a", prefix="a")

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["cards"] == {
        "visitors": {"value": 1, "delta_pct": None},
        "visits": {"value": 1, "delta_pct": None},
        "avg_stay_ms": {"value": 50000, "delta_pct": None},
        "returning": {"value": 0, "delta_pct": None},   # 新访客 → 回访 0，同样上期零
    }


def test_overview_channels_pct(client, member_user):
    """channels：窗口内 session_entry 分类聚合，pct 占已分类会话比、总和 ≈ 1。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    v1 = _visitor(site, "v1", last_seen=timezone.now())
    v2 = _visitor(site, "v2", last_seen=timezone.now())
    s1 = _session(site, v1, "s1", referrer="www.bing.com")
    s2 = _session(site, v1, "s2", referrer="www.bing.com", at=AT_3D)
    s3 = _session(site, v2, "s3", referrer="chatgpt.com")
    s4 = _session(site, v2, "s4")                                 # 直接访问
    s5 = _session(site, v2, "s5", referrer="t.co", at=AT_10D)     # 上期：不计入
    s6 = _session(site, v2, "s6", referrer="example.com")         # 无分类行：不计入
    for s, ch, pf in ((s1, "natural_search", "Bing"), (s2, "natural_search", "Bing"),
                      (s3, "ai_referral", "ChatGPT"), (s4, "direct", "")):
        SourceClassification.objects.create(site=site, visitor=s.visitor, session=s,
            scope="session_entry", channel=ch, platform=pf)
    SourceClassification.objects.create(site=site, visitor=v2, session=s5,
        scope="session_entry", channel="social", platform="Twitter")
    _events(site, v1, s1, 1, "/", prefix="x")

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["channels"] == [
        {"channel": "natural_search", "sessions": 2, "pct": pytest.approx(0.5)},
        {"channel": "ai_referral", "sessions": 1, "pct": pytest.approx(0.25)},
        {"channel": "direct", "sessions": 1, "pct": pytest.approx(0.25)},  # 平局按名稳定
    ]
    assert sum(c["pct"] for c in d["channels"]) == pytest.approx(1.0)


def test_overview_range_today(client, member_user):
    """range=today：卡按今天口径、delta 比昨天；趋势单点；非法 range 回落 7d。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    vy = _visitor(site, "vy", last_seen=timezone.now(), first_seen=NOW - timedelta(days=1))
    sy = _session(site, vy, "sy", at=NOW - timedelta(days=1), stay_ms=60000)
    _events(site, vy, sy, 2, "/y", at=NOW - timedelta(days=1), prefix="y")
    vt = _visitor(site, "vt", last_seen=timezone.now())
    st = _session(site, vt, "st", stay_ms=30000)
    _events(site, vt, st, 1, "/t", prefix="t")

    d = client.get("/api/admin/dashboard/overview", {"range": "today"}).json()["data"]
    assert d["trend"] == [{"date": TODAY.isoformat(), "uv": 1, "sessions": 1}]
    assert d["cards"]["visitors"] == {"value": 1, "delta_pct": 0.0}     # 昨天也是 1
    assert d["cards"]["visits"] == {"value": 1, "delta_pct": 0.0}
    assert d["cards"]["avg_stay_ms"] == {"value": 30000, "delta_pct": -50.0}
    assert d["cards"]["returning"]["delta_pct"] is None                 # 两天都是全新访客

    assert len(client.get("/api/admin/dashboard/overview").json()["data"]["trend"]) == 7
    d = client.get("/api/admin/dashboard/overview", {"range": "abc"}).json()["data"]
    assert len(d["trend"]) == 7                                          # 非法值回落 7d
    d = client.get("/api/admin/dashboard/overview", {"range": "30d"}).json()["data"]
    assert len(d["trend"]) == 30


def test_overview_trend_realtime_and_selfheal(client, member_user):
    """趋势不依赖 DailyStat/聚合作业：昨天+今天的数据（不跑 refresh_daily）
    两天都有真实值（昨天跨站求和）；调用后 DailyStat 自愈出现对应行（update_or_create 幂等）。"""
    _auth(client, member_user)
    site_a = Site.objects.create(name="a", domains=["a.com"])
    site_b = Site.objects.create(name="b", domains=["b.com"])
    yn = NOW - timedelta(days=1)   # 昨天同一本地时刻，必落上海昨日窗口
    va = _visitor(site_a, "va", last_seen=yn, first_seen=yn)
    sa = _session(site_a, va, "sa", at=yn)
    _events(site_a, va, sa, 3, "/", at=yn, prefix="ya")
    vb = _visitor(site_b, "vb", last_seen=yn, first_seen=yn)
    sb = _session(site_b, vb, "sb", at=yn)
    _events(site_b, vb, sb, 4, "/", at=yn, prefix="yb")
    # 今天：site_b 一个新访客一条会话两条事件
    vt = _visitor(site_b, "vt", last_seen=timezone.now())
    st = _session(site_b, vt, "st")
    _events(site_b, vt, st, 2, "/t", prefix="t")
    assert DailyStat.objects.count() == 0                 # 前置：聚合作业从未跑过

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    trend = d["trend"]
    assert len(trend) == 7
    assert trend[-2] == {"date": YESTERDAY.isoformat(), "uv": 2, "sessions": 2}  # 跨站求和
    assert trend[-1] == {"date": TODAY.isoformat(), "uv": 1, "sessions": 1}
    # 自愈回写：按站按日出现三行，四指标口径与 daily_counts 一致
    ra = DailyStat.objects.get(site=site_a, date=YESTERDAY)
    assert (ra.pv, ra.uv, ra.sessions, ra.new_visitors) == (3, 1, 1, 1)
    rb = DailyStat.objects.get(site=site_b, date=YESTERDAY)
    assert (rb.pv, rb.uv, rb.sessions, rb.new_visitors) == (4, 1, 1, 1)
    rt = DailyStat.objects.get(site=site_b, date=TODAY)
    assert (rt.pv, rt.uv, rt.sessions, rt.new_visitors) == (2, 1, 1, 1)
    assert DailyStat.objects.count() == 3
    client.get("/api/admin/dashboard/overview")           # 再调一次：update 不重复建行
    assert DailyStat.objects.count() == 3


def test_overview_trend_empty_window_no_write(client, member_user):
    """空窗口：趋势为全零序列，且不写任何 DailyStat（自愈只写有数据的日）。"""
    _auth(client, member_user)
    Site.objects.create(name="s", domains=["a.com"])
    d = client.get("/api/admin/dashboard/overview").json()["data"]
    trend = d["trend"]
    assert len(trend) == 7
    assert trend[0]["date"] == (TODAY - timedelta(days=6)).isoformat()
    assert all(p["uv"] == 0 and p["sessions"] == 0 for p in trend)
    assert DailyStat.objects.count() == 0


def test_overview_site_filter_and_404(client, member_user):
    _auth(client, member_user)
    site_a = Site.objects.create(name="a", domains=["a.com"])
    site_b = Site.objects.create(name="b", domains=["b.com"])
    va = _visitor(site_a, "va", last_seen=timezone.now())
    vb = _visitor(site_b, "vb", last_seen=timezone.now())
    sa = _session(site_a, va, "sa", referrer="google.com")
    sb = _session(site_b, vb, "sb")
    _events(site_a, va, sa, 2, "/", prefix="a")
    _events(site_b, vb, sb, 5, "/b", prefix="b")
    SourceClassification.objects.create(site=site_b, visitor=vb, session=sb,
        scope="session_entry", channel="direct", resolution="unknown")
    # a 站昨天有数据（实时聚合）——过滤 site_b 时不得混入
    yn = NOW - timedelta(days=1)
    _events(site_a, va, sa, 3, "/", at=yn, prefix="ay")

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["cards"]["visitors"]["value"] == 2          # 缺省聚合全部站点
    assert d["header"]["online"] == 2
    d = client.get("/api/admin/dashboard/overview", {"site": site_b.site_id}).json()["data"]
    assert d["cards"]["visitors"]["value"] == 1 and d["cards"]["visits"]["value"] == 1
    assert d["header"]["online"] == 1                      # 只计本站
    assert d["channels"] == [{"channel": "direct", "sessions": 1,
                              "pct": pytest.approx(1.0)}]
    assert d["trend"][-2] == {"date": YESTERDAY.isoformat(), "uv": 0, "sessions": 0}
    assert d["trend"][-1] == {"date": TODAY.isoformat(), "uv": 1, "sessions": 1}

    r = client.get("/api/admin/dashboard/overview", {"site": "site_nope"})
    assert r.status_code == 404
    body = r.json()
    assert body["code"] == 404 and body["data"] is None and body["message"] == "site_not_found"


def test_overview_avg_stay_mixed_v1_v2_session(client, member_user):
    """v1+v2 混合会话回归（阶段A T4）：avg_stay_ms 读 VisitSession.stay_ms 重算值。
    经真实 ingest 落库——v1 page_stay active_ms=8000（加法）+ v2 同 page_id 相邻区间
    [0,15000][15000,30000]（并集=30000，非加法 60000）→ 会话 stay_ms=38000；
    单会话 avg=38000/1。聚合点不直接 SUM(Event.properties.active_ms)，
    T3 重算式回写后读数侧自动正确——本用例钉死该语义不被回归。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
    ingest_batch(site, _batch(                      # v1 批：建会话 sm + v1 停留增量
        [_ev(1), _ev(2, "page_stay", properties={"active_ms": 8000, "path": "/p"})],
        sid="sm"))
    r = ingest_batch(site, _v2_batch([              # v2 批：同会话同 page_id 两段相邻区间
        _v2_event("page_stay", "p1", "sm", seq=1,
                  active_from=0, active_to=15000, flush_reason="periodic"),
        _v2_event("page_stay", "p1", "sm", seq=2,
                  active_from=15000, active_to=30000, flush_reason="periodic")],
        sid="sm"))
    assert (r["accepted"], r["rejected"]) == (2, [])
    assert VisitSession.objects.get(session_id="sm").stay_ms == 38000   # 写侧：v1 SUM + v2 并集

    d = client.get("/api/admin/dashboard/overview", {"range": "today"}).json()["data"]
    assert d["cards"]["visits"]["value"] == 1
    assert d["cards"]["avg_stay_ms"] == {"value": 38000, "delta_pct": None}   # 38000/1；上期无会话 → None


def test_overview_trend_pv_only_page_view(client, member_user):
    """PV 口径回归（阶段A T11）：SDK 0.2.0 每 15s 一条 page_stay 增量 + sdk_loaded，
    趋势自愈回写 DailyStat.pv 只计 page_view（与 daily_counts 同口径）——
    增量停留事件不得灌倍 PV；uv 仍按任意事件 distinct visitor 计。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    v1 = _visitor(site, "v1", last_seen=timezone.now())
    s1 = _session(site, v1, "s1")
    # 模拟一次活跃访问的事件流：1 次浏览 + SDK 装载 + 3 条 15s 停留增量
    names = ["page_view", "sdk_loaded", "page_stay", "page_stay", "page_stay"]
    for i, name in enumerate(names):
        Event.objects.create(site=site, visitor=v1, session=s1,
            event_id=f"pvx-{i}", event_name=name,
            occurred_at=NOW + timedelta(seconds=i), path="/")

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["trend"][-1] == {"date": TODAY.isoformat(), "uv": 1, "sessions": 1}
    row = DailyStat.objects.get(site=site, date=TODAY)
    assert (row.pv, row.uv) == (1, 1)   # 5 条事件仅 1 条 page_view 计 PV


# ---------------- 询盘漏斗（阶段B T3：五级独立 distinct session） ----------------

FUNNEL_URL = "/api/admin/funnel-summary"
FEV_SEQ = itertools.count()   # 漏斗测试事件 id 计数器（跨用例唯一）


def _fev(site, visitor, session, name, at=NOW):
    """造任意 event_name 的事件（既有 _events 只造 page_view）。"""
    return Event.objects.create(site=site, visitor=visitor, session=session,
        event_id=f"fe-{next(FEV_SEQ)}", event_name=name, occurred_at=at, path="/")


def test_funnel_summary_requires_auth(client):
    assert client.get(FUNNEL_URL).status_code == 401


def test_funnel_summary_independent_session_counting(client, member_user):
    """五级逐级独立 distinct session（非嵌套非事件数）：A 的 2 次 PV 只算 1 个
    "浏览"到达；B 未发生 form_start 仍计入 form_submit 级（无前置/顺序约束）。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    va = _visitor(site, "fa", last_seen=timezone.now())
    vb = _visitor(site, "fb", last_seen=timezone.now())
    sa = _session(site, va, "fsa")
    sb = _session(site, vb, "fsb")
    _fev(site, va, sa, "page_view")
    _fev(site, va, sa, "page_view")            # 同会话第 2 次浏览：浏览级仍计 1
    _fev(site, va, sa, "contact_click")
    _fev(site, vb, sb, "page_view")
    _fev(site, vb, sb, "form_submit")          # 跳过 form_start：提交级照计（独立口径）

    r = client.get(FUNNEL_URL)                 # 缺省 days=7
    assert r.status_code == 200
    levels = r.json()["data"]["levels"]
    assert [lv["key"] for lv in levels] == ["page_view", "contact_click",
                                            "form_start", "form_submit",
                                            "form_submit_result"]
    assert [lv["label"] for lv in levels] == ["浏览", "咨询点击",
                                              "表单开始", "表单提交", "邮件已发"]
    assert [lv["count"] for lv in levels] == [2, 1, 0, 1, 0]


def test_funnel_summary_excludes_bots(client, member_user):
    """is_bot=True 的事件不计入任何级（与总览爬虫口径一致）。"""
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    vh = _visitor(site, "fh", last_seen=timezone.now())
    sh = _session(site, vh, "fsh")
    _fev(site, vh, sh, "page_view")
    _fev(site, vh, sh, "contact_click")
    bot = _visitor(site, "fbot", last_seen=timezone.now())
    Visitor.objects.filter(pk=bot.pk).update(is_bot=True)
    sbot = _session(site, bot, "fsbot")
    for name in ("page_view", "contact_click", "form_start",
                 "form_submit", "form_submit_result"):   # bot 五级全打：任何级都不进
        _fev(site, bot, sbot, name)
    Event.objects.filter(visitor=bot).update(is_bot=True)   # 与 ingest 同口径：事件行同步标记

    levels = client.get(FUNNEL_URL).json()["data"]["levels"]
    assert [lv["count"] for lv in levels] == [1, 1, 0, 0, 0]


def test_funnel_summary_days_window_site_filter_and_404(client, member_user):
    """days 窗口为上海自然日 [T-N+1, T+1)（与总览一致：days=1 即"当日"）；
    site 参数按条件过滤；非法 site 404 信封同总览。"""
    _auth(client, member_user)
    site_a = Site.objects.create(name="a", domains=["a.com"])
    site_b = Site.objects.create(name="b", domains=["b.com"])
    ya = NOW - timedelta(days=1)   # 昨天同一本地时刻，必落上海昨日窗口
    for site, vid, sid, at in ((site_a, "wa", "wsa", NOW), (site_a, "wy", "wsy", ya),
                               (site_b, "wb", "wsb", NOW)):
        v = _visitor(site, vid, last_seen=timezone.now())
        s = _session(site, v, sid, at=at)
        _fev(site, v, s, "page_view", at=at)

    d = client.get(FUNNEL_URL).json()["data"]                       # 缺省 7d 聚合全部站点
    assert d["levels"][0]["count"] == 3
    d = client.get(FUNNEL_URL, {"days": 1}).json()["data"]          # 今日窗口：昨天不计
    assert d["levels"][0]["count"] == 2
    d = client.get(FUNNEL_URL, {"days": "abc"}).json()["data"]      # 非法值回落 7d
    assert d["levels"][0]["count"] == 3
    d = client.get(FUNNEL_URL, {"site": site_b.site_id}).json()["data"]
    assert d["levels"][0]["count"] == 1
    r = client.get(FUNNEL_URL, {"site": "site_nope"})
    assert r.status_code == 404
    body = r.json()
    assert body["code"] == 404 and body["data"] is None and body["message"] == "site_not_found"