fangke/server/tracker/api/visitors.py

"""访客列表/详情/跨天时间线/单会话接口（管理台）。

M1 单组织多站点：访客列表支持 ?site= 可选过滤（缺省聚合全部站点），行内随带站点归属
（site_id/site_name）——多站点下分辨访客来自哪个站；访客数据本身已按站点隔离。
访客列表查询按 (site,-last_seen_at) 索引逐站扫描或全表 order_by——M1 规模（万级访客）可接受，
量级上来后按站点分片即可（索引已就位）。
"""
from datetime import datetime, time, timedelta

from django.db.models import Count, OuterRef, Q, Subquery, Sum
from django.utils import timezone
from rest_framework.response import Response
from rest_framework.views import APIView

from tracker.api.envelope import EnvelopeMixin, ok
from tracker.models import Event, RawBatch, Site, SourceClassification, VisitSession, Visitor

PAGE_SIZE = 20
RAW_PAGE_SIZE = 10   # 原始批次 body 较大，页小一些
TIMELINE_DEFAULT_DAYS = 30
TIMELINE_MAX_DAYS = 90
TIMELINE_MAX_EVENTS = 1000
ONLINE_WINDOW = timedelta(minutes=5)   # 与 dashboard active_now 同口径
ATTR_SESSION_ENTRIES = 5               # 详情页归因块带出的最近会话入口分类数


def _source(referrer_domain, utm):
    """来源三级兜底：referrer_domain → utm.utm_source → 直接访问（访客档案/会话共用）。"""
    return referrer_domain or (utm or {}).get("utm_source") or "直接访问"


def _iso(dt):
    """输出给前端的时间统一按当前时区（上海）序列化：USE_TZ=True 下 isoformat() 是 UTC
    （"+00:00"），前端切串直显会与 Dashboard 的上海自然日口径同屏矛盾（差 8 小时）。"""
    return timezone.localtime(dt).isoformat() if dt else None


def _row(visitor, last_session, online_cutoff, ft=None):
    """列表/详情共用行字段。last_source 依赖最近会话（由调用方批量取好传入，防 N+1）。
    channel 等归因五字段来自该访客 first_touch 分类（无分类行 → 空串，前端回落 first_source 展示）。"""
    return {
        "visitor_id": visitor.visitor_id,
        # 站点归属：visitor.site 需随行 select_related("site")，否则逐行多一条查询
        "site_id": visitor.site.site_id, "site_name": visitor.site.name,
        "country": visitor.country, "province": visitor.province, "city": visitor.city,
        "device_type": visitor.device_type, "os": visitor.os,
        "browser": visitor.browser, "language": visitor.language,
        "first_seen_at": _iso(visitor.first_seen_at),
        "last_seen_at": _iso(visitor.last_seen_at),
        "visit_count": visitor.visit_count, "active_days": visitor.active_days,
        "session_count": visitor.session_count,
        "total_stay_ms": visitor.total_stay_ms or 0,   # 无会话时 Sum 为 None → 0
        "first_source": _source(visitor.first_referrer_domain, visitor.first_utm),
        "last_source": _source(last_session.referrer_domain, last_session.utm)
                       if last_session else "直接访问",   # 无会话访客（仅采集异常场景）
        "first_landing_path": visitor.first_landing_path,   # 列表首触来源 tooltip 用（详情本就输出）
        "channel": ft.channel if ft else "",
        "platform": ft.platform if ft else "",
        "variant": ft.variant if ft else "",
        "search_term": ft.search_term if ft else "",
        "resolution": ft.resolution if ft else "",
        "online": visitor.last_seen_at > online_cutoff,   # Python 侧比对（last_seen_at 已随行取出）
    }


def _classification(sc, with_session=False):
    """SourceClassification → API dict（详情归因块用，含证据可信度与规则版本）。"""
    if not sc:
        return None
    d = {"channel": sc.channel, "platform": sc.platform, "variant": sc.variant,
         "search_term": sc.search_term, "evidence_kind": sc.evidence_kind,
         "resolution": sc.resolution, "detail": sc.detail or {},
         "rule_version": sc.rule_version, "computed_at": _iso(sc.computed_at)}
    if with_session:
        d["session_id"] = sc.session.session_id if sc.session_id else None
    return d


def _session_summary(s):
    # duration_ms 仅内部字段不展示——前端时长一律用 stay_ms（有效停留口径）
    return {"session_id": s.session_id,
            "started_at": _iso(s.started_at), "ended_at": _iso(s.ended_at),
            "referrer_domain": s.referrer_domain, "utm": s.utm,
            "landing_path": s.landing_path, "event_count": s.event_count,
            "stay_ms": s.stay_ms}


def _annotate(qs):
    """会话数/有效停留/最近会话标记一次注解到位。

    last_source 不逐行 sessions.order_by("-started_at").first()（页大小 20 即 N+1）：
    先用相关子查询把每个访客最近会话的 pk 标到行上（走 (visitor,-started_at) 索引），
    分页后一次 IN 查询取回整页会话——全程只多 2 条 SQL。
    """
    latest = (VisitSession.objects.filter(visitor=OuterRef("pk"))
              .order_by("-started_at", "-pk"))   # -pk 平局次序跨请求稳定
    return qs.annotate(
        session_count=Count("sessions"),
        total_stay_ms=Sum("sessions__stay_ms"),       # 有效停留口径（非 duration_ms）
        latest_session_pk=Subquery(latest.values("pk")[:1]),
    )


def _page_visitors(qs, request):
    """过滤 + 分页 + 批量取最近会话，返回 (rows, total, page)。"""
    p = request.query_params
    if p.get("show_bot") != "1":       # 默认排除爬虫；show_bot=1 才包含
        qs = qs.filter(is_bot=False)
    if p.get("country"):
        qs = qs.filter(country=p["country"])
    if p.get("device_type"):
        qs = qs.filter(device_type=p["device_type"])
    if p.get("online") == "1":
        qs = qs.filter(last_seen_at__gt=timezone.now() - ONLINE_WINDOW)
    if p.get("is_new") == "1":         # first_seen_at 落在今天（USE_TZ 下 __date 按当前时区取日期）
        qs = qs.filter(first_seen_at__date=timezone.localdate())
    if p.get("search"):
        q = p["search"]
        qs = qs.filter(Q(visitor_id__istartswith=q) | Q(last_ip__istartswith=q))
    if p.get("channel"):
        # 渠道筛选（值=CHANNELS 枚举）：按 first_touch 分类内连接过滤。
        # distinct 防御分类表脏重复行（无唯一约束，并发极端下可能同键两行）撑大 count。
        qs = qs.filter(source_classifications__scope="first_touch",
                       source_classifications__channel=p["channel"]).distinct()
    date_from = _parse_date(p.get("date_from"))
    date_to = _parse_date(p.get("date_to"))
    if date_from:
        qs = qs.filter(last_seen_at__date__gte=date_from)
    if date_to:
        qs = qs.filter(last_seen_at__date__lte=date_to)

    total = qs.count()   # 注解前 count：不必为计数做会话表 join
    try:
        page = int(p.get("page", 1))
    except (TypeError, ValueError):
        page = 1
    page = max(1, page)

    page_qs = _annotate(qs).select_related("site").order_by("-last_seen_at", "-pk")
    visitors = list(page_qs[(page - 1) * PAGE_SIZE: page * PAGE_SIZE])
    latest = {s.pk: s for s in VisitSession.objects.filter(
        pk__in=[v.latest_session_pk for v in visitors if v.latest_session_pk])}
    # 归因字段批量取：整页一次 IN 查询（同 latest 的做法），不做 select_related N+1
    fts = {sc.visitor_id: sc for sc in SourceClassification.objects.filter(
        visitor__in=[v.pk for v in visitors], scope="first_touch")}
    cutoff = timezone.now() - ONLINE_WINDOW
    rows = [_row(v, latest.get(v.latest_session_pk), cutoff, fts.get(v.pk))
            for v in visitors]
    return rows, total, page


def _parse_date(raw):
    """YYYY-MM-DD 宽松解析：非法/缺失返回 None（该筛选不生效，不为此 400）。"""
    try:
        return datetime.strptime(raw, "%Y-%m-%d").date()
    except (TypeError, ValueError):
        return None


def _get_visitor(visitor_id):
    """visitor_id 仅 (site,visitor_id) 唯一——跨站重名时取最近活跃者（M1 单组织，SDK id 撞名概率可忽略）。"""
    return (Visitor.objects.filter(visitor_id=visitor_id)
            .order_by("-last_seen_at", "-pk").first())


def _not_found(message):
    return Response({"code": 404, "data": None, "message": message}, status=404)


def _merge_page_stay(events):
    """page_view/page_stay 服务端合并（时间线/单会话共用）。

    按 occurred_at 正序迭代：page_view 追加条目（active_ms=0）；page_stay 向前找同 session 内
    最近一条同 path 的 page_view 条目，把 active_ms 累加进去（同页多段隐藏/恢复多次累加）；
    找不到匹配（含 page_stay 早于任何 page_view 出现）则独立成条目（event_name 保持 page_stay）。
    作用域天然限定在本组事件序列（单会话/时间线单会话桶）——跨会话绝不合并。
    只出 time/event_name/path/title/active_ms 白名单字段，不透出 properties 原文（隐私）。
    """
    items = []
    for e in events:
        if e.event_name == "page_stay":
            ms = int((e.properties or {}).get("active_ms") or 0)
            for it in reversed(items):   # 逆序 = "最近一条"
                if it["event_name"] == "page_view" and it["path"] == e.path:
                    it["active_ms"] += ms
                    break
            else:
                items.append({"time": _iso(e.occurred_at), "event_name": e.event_name,
                              "path": e.path, "title": e.title, "active_ms": ms})
        else:
            # page_view 初始 active_ms=0；其余（sdk_loaded/自定义事件）无停留概念 → null
            active = 0 if e.event_name == "page_view" else None
            items.append({"time": _iso(e.occurred_at), "event_name": e.event_name,
                          "path": e.path, "title": e.title, "active_ms": active})
    return items


class VisitorListView(EnvelopeMixin, APIView):
    def get(self, request):
        qs = Visitor.objects.all()
        sid = request.query_params.get("site")   # 可选站点过滤，缺省聚合全部（与 dashboard 同款）
        if sid:
            site = Site.objects.filter(site_id=sid).first()
            if not site:
                return _not_found("site_not_found")
            qs = qs.filter(site=site)
        rows, total, page = _page_visitors(qs, request)
        return ok({"list": rows, "total": total, "page": page})


def _source_evidence_triplet(visitor, ft, entries):
    """来源证据三口径（详情页「来源证据」卡的数据面）。

    first_observable   首次可观测来源：first_touch 分类原样（含 direct/unknown——诚实展示
                       "首访只观察到直接访问"）；无分类行（历史数据未重算）→ unknown 占位。
    first_identifiable 首次可识别来源：first_touch 本身可识别（resolved/label_only/conflict，
                       即除"没认出来"外的全部状态）直接用；否则回溯最早一条可识别的会话入口
                       ——首访 direct 不算"可识别"，后续某天才带上可识别来源是常见轨迹。
                       都没有 → None（前端显示 —，不虚构）。
    latest_session     本次会话来源：最新一条会话入口分类；无 → None。
    """
    first_observable = _classification(ft) or {"channel": "unknown", "resolution": "unknown"}
    if ft and ft.resolution in ("resolved", "label_only", "conflict"):
        first_identifiable = _classification(ft)
    else:
        earliest = (SourceClassification.objects
                    .filter(visitor=visitor, scope="session_entry",
                            resolution__in=("resolved", "label_only"))
                    .select_related("session")
                    .order_by("session__started_at", "pk").first())
        first_identifiable = _classification(earliest, with_session=True)
    latest_session = _classification(entries[0], with_session=True) if entries else None
    return first_observable, first_identifiable, latest_session


class VisitorDetailView(EnvelopeMixin, APIView):
    def get(self, request, visitor_id):
        visitor = _annotate(Visitor.objects.filter(visitor_id=visitor_id)
                            .select_related("site")) \
            .order_by("-last_seen_at", "-pk").first()
        if not visitor:
            return _not_found("visitor_not_found")
        last_session = (VisitSession.objects.filter(pk=visitor.latest_session_pk).first()
                        if visitor.latest_session_pk else None)
        # 全量会话摘要（visitor,-started_at 索引）；M1 访客会话量有限不分页
        sessions = VisitSession.objects.filter(visitor=visitor).order_by("-started_at", "-pk")
        cutoff = timezone.now() - ONLINE_WINDOW
        ft = (SourceClassification.objects.filter(visitor=visitor, scope="first_touch")
              .first())
        entries = list(SourceClassification.objects
                       .filter(visitor=visitor, scope="session_entry")
                       .select_related("session")            # session_id 随行带出，免逐行查询
                       .order_by("-session__started_at", "-pk")[:ATTR_SESSION_ENTRIES])
        first_observable, first_identifiable, latest_session = \
            _source_evidence_triplet(visitor, ft, entries)
        return ok({**_row(visitor, last_session, cutoff, ft),
                   # 浏览页面数：page_view 事件计数（page_stay 不并计——它是停留信号不是浏览）
                   "pages_viewed": Event.objects.filter(
                       visitor=visitor, event_name="page_view").count(),
                   "screen": visitor.screen, "timezone_name": visitor.timezone_name,
                   "last_ip": visitor.last_ip, "first_utm": visitor.first_utm or {},
                   "first_landing_path": visitor.first_landing_path,
                   "first_referrer_domain": visitor.first_referrer_domain,
                   "attribution": {"first_touch": _classification(ft),
                                   "first_observable": first_observable,
                                   "first_identifiable": first_identifiable,
                                   "latest_session": latest_session,
                                   "session_entries": [_classification(e, with_session=True)
                                                       for e in entries]},
                   "sessions": [_session_summary(s) for s in sessions]})


class VisitorTimelineView(EnvelopeMixin, APIView):
    def get(self, request, visitor_id):
        visitor = _get_visitor(visitor_id)
        if not visitor:
            return _not_found("visitor_not_found")
        try:
            days = int(request.query_params.get("days", TIMELINE_DEFAULT_DAYS))
        except (TypeError, ValueError):
            days = TIMELINE_DEFAULT_DAYS
        days = max(1, min(days, TIMELINE_MAX_DAYS))   # 1-90 夹紧，非法值回落默认

        today = timezone.localdate()
        start = timezone.make_aware(
            datetime.combine(today - timedelta(days=days - 1), time.min))   # 上海自然日窗口
        # 倒序取满上限再翻正：截断保最新（更早的事件连同其独属会话不出现——若某会话事件被整段
        # 截掉，该会话不出组；部分截掉的 page_view 缺失时其 page_stay 按规则独立成条）。
        events = list(Event.objects.filter(visitor=visitor, occurred_at__gte=start)
                      .select_related("session")           # 会话信息随行带出，免逐事件查询
                      .order_by("-occurred_at", "-pk")[:TIMELINE_MAX_EVENTS])
        truncated = len(events) >= TIMELINE_MAX_EVENTS   # 切片取满即窗口内可能还有更多
        events.reverse()

        groups = {}   # date -> {session.pk/None: {"session":..., "events":[...]}}，插入序即正序
        for e in events:
            day = groups.setdefault(timezone.localtime(e.occurred_at).date(), {})
            key = e.session.pk if e.session else None   # session 为 NULL 的孤立事件并成一组
            bucket = day.setdefault(key, {"session": e.session, "events": []})
            bucket["events"].append(e)

        return ok({"days": [{"date": d.isoformat(),
                             "sessions": [{"session_id": b["session"].session_id if b["session"] else None,
                                           "started_at": _iso(b["session"].started_at)
                                                         if b["session"] else None,
                                           "events": _merge_page_stay(b["events"])}
                                          for b in groups[d].values()]}
                            for d in sorted(groups, reverse=True)],   # 日期组倒序，日内会话/事件正序
                   "truncated": truncated})   # 前端区分"没动静"和"被截断"


class VisitorSessionView(EnvelopeMixin, APIView):
    def get(self, request, visitor_id, session_id):
        visitor = _get_visitor(visitor_id)
        if not visitor:
            return _not_found("visitor_not_found")
        session = VisitSession.objects.filter(visitor=visitor, session_id=session_id).first()
        if not session:
            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})