feat: 统计API（卡片/漏斗/趋势/分布/推送健康，逐库聚合禁ATTACH）

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-13 11:24:30 +08:00
parent ec55bc38ab
commit acf5777ff2
4 changed files with 765 additions and 0 deletions
@@ -0,0 +1,78 @@
+"""统计 API(Phase 2 Task 9,spec §7.6): /api/platform/stats(超管,可
+?tenant= 单租户下钻)与 /api/platform/tenants/{slug}/stats(本租户管理员)。
+
+聚合逻辑全在 app/platform/stats.py(纯查询函数);本层只做鉴权、日期解析
+(复用 data_routes 的 _parse_dt——ISO 日期/日期时间,日期-only 的 to 含
+当日全天,边界含,naive UTC)与 runtime 获取。
+
+多租户合并纪律(**禁 ATTACH**,spec §7.6: 默认上限 10 个附加库):
+逐租户库查询 → merge_stats 应用层合并。合并范围:
+- 含停用租户——Task 8 偏差 4 语义(后台可见停用租户数据)在统计同样
+  适用: 停用只是聊天面 503,历史数据仍属后台可见资产;
+- 跳过库文件不存在的未懒建租户——零数据等价,且不为一次统计主动建
+  空库(保持 engine 懒建纪律;与 main.lifespan 启动清扫同一 db_path.
+  exists() 判据,差异是清扫只扫 active、统计含停用,口径在此声明)。
+显式单租户(?tenant= 与 /tenants/{slug}/stats)沿用 data_routes 的
+_runtime_by_slug 语义: 未懒建租户首次查询触发建库(Task 8 语义,空
+统计 200 不炸)。
+"""
+from fastapi import APIRouter, Depends, Query, Request
+
+from app.platform.auth import require_super
+from app.platform.registry import TenantRuntime
+from app.platform.routes.data_routes import _parse_dt, _runtime_by_slug
+from app.platform.routes.tenant_routes import _tenant_guard
+from app.platform.stats import collect_stats, merge_stats
+
+router = APIRouter(prefix="/api/platform", tags=["platform-stats"])
+
+
+def _bounds(from_raw: str | None, to_raw: str | None) -> tuple:
+    """?from/&to= → (from_dt, to_dt) naive UTC 闭区间(解析/报错同 data_routes)。"""
+    bound_from = _parse_dt(from_raw, "from") if from_raw else None
+    bound_to = _parse_dt(to_raw, "to", end_of_day=True) if to_raw else None
+    return bound_from, bound_to
+
+
+async def _stats_for(
+        rt: TenantRuntime, bound_from, bound_to) -> dict:
+    """单个就绪 runtime 的统计(租户库 session 上跑全部聚合)。"""
+    async with rt.session_factory() as db:
+        return await collect_stats(db, from_dt=bound_from, to_dt=bound_to)
+
+
+@router.get("/stats")
+async def platform_stats(
+        request: Request,
+        tenant: str | None = None,
+        from_dt: str | None = Query(default=None, alias="from"),
+        to_dt: str | None = Query(default=None, alias="to"),
+        user=Depends(require_super)):
+    """平台统计(超管): 缺省全部租户合并(逐库查询应用层合并,禁 ATTACH),
+    ?tenant=slug 单租户下钻(未知 slug 404)。?from/&to= 闭区间筛选。"""
+    bound_from, bound_to = _bounds(from_dt, to_dt)
+    if tenant is not None:
+        rt = await _runtime_by_slug(request, tenant)
+        return await _stats_for(rt, bound_from, bound_to)
+
+    registry = request.app.state.registry
+    parts = []
+    for rt in registry.runtimes():
+        # 合并范围口径见模块 docstring: 含停用、跳过无库文件(不建空库)
+        if not rt.db_path.exists():
+            continue
+        ready = await registry.ensure_ready(rt)
+        parts.append(await _stats_for(ready, bound_from, bound_to))
+    return merge_stats(parts)
+
+
+@router.get("/tenants/{slug}/stats")
+async def tenant_stats(
+        slug: str, request: Request,
+        from_dt: str | None = Query(default=None, alias="from"),
+        to_dt: str | None = Query(default=None, alias="to"),
+        user=Depends(_tenant_guard)):
+    """单租户统计(本租户管理员或超管): 查询/懒建语义同 data_routes。"""
+    bound_from, bound_to = _bounds(from_dt, to_dt)
+    rt = await _runtime_by_slug(request, slug)
+    return await _stats_for(rt, bound_from, bound_to)
@@ -0,0 +1,275 @@
+"""统计查询纯函数(Phase 2 Task 9,spec §7.6): 五表现有数据聚合——卡片/
+漏斗/趋势/分布/推送健康,无新表。
+
+数据形态: 单租户 collect_stats(该租户 session)+ 跨租户 merge_stats(应用层
+合并)——**逐库查询、应用层合并,绝不用 SQLite ATTACH**(spec §7.6: ATTACH
+默认上限 10 个附加库,租户一多就炸;本模块零 ATTACH 调用)。
+
+指标口径(歧义处已定死,Phase 3 前端照此展示):
+- 会话数: conversations 行数,**不要求有消息**——统计口径是"发生的会话",
+  与数据视图列表"至少一条消息"(jc 列表语义)不同,两者有意并存。
+- 线索数: 该会话的 customer_profiles.raw_info 含 phone/whatsapp/email
+  (推送预检的"contact 三选一")任一**非 _ 前缀键**且值非空——wechat 不算
+  (推送侧只认三选一);与 contact_pipeline.normalize_contacts 的归一化键
+  同名(中英文别名入库前已归一)。
+- 高意向数: evaluations.intent_level == "interested" 的会话数。
+- 推送成功率: lead_push_logs 的 success/(success+failed);skipped(预检
+  不满足必拒)与 pending(未终态)不进分母;分母 0 → None(前端展示"—")。
+- 漏斗: 会话 → 留资 → 评估完成(有 evaluations 行)→ 推送成功(该会话有
+  过 success 日志;日志自身不另受时间窗约束——会话本体已在窗内)。
+- 趋势: 按日分组,**北京日界** date(created_at, '+8 hours')——created_at
+  库内为 naive UTC(models 约定),容器部署 TZ=Asia/Shanghai,业务日界以
+  北京为准(代码按 naive UTC 存储 + 固定 +8 偏移分组,不依赖进程本地时区,
+  测试环境与生产同口径);无数据的日期不产出行(前端补零)。
+- 分布: intent 三级 / source / site(client_info 的 site 键,JSON 取键,
+  与 data_routes 的 site 筛选同一姿势;无 site 键的会话不计)/ country
+  top10(IP 经 ip_service.resolve_country 逐行归国——iptocc 是内存库,
+  逐行查 N 会话在几百量级毫秒级可接受,故不在 SQL 里做)。
+- 推送健康: 四状态(pending/success/failed/skipped)计数 + duplicated
+  (outcome 字段,200 但对方判重——不是错误)计数。
+
+时间筛选: from_dt/to_dt 为 naive UTC 边界(由 stats_routes 经 data_routes
+ 的 _parse_dt 解析,日期-only 的 to 含当日全天,边界含)。会话锚定的指标滤
+conversations.created_at;推送锚定的指标(成功率/推送健康)滤
+lead_push_logs.created_at——两套锚点各自与本体表同窗。
+"""
+from collections import Counter, defaultdict
+
+from sqlalchemy import ColumnElement, and_, func, or_, select
+
+from app.engine.models import (
+    Conversation,
+    CustomerProfile,
+    Evaluation,
+    IntentLevel,
+    LeadPushLog,
+)
+from app.engine.services.ip_service import resolve_country
+
+# 线索判定的联系方式键(三选一;wechat 不算——与推送预检同口径)
+CONTACT_KEYS = ("phone", "whatsapp", "email")
+INTENT_KEYS = ("interested", "not_interested", "pending")
+PUSH_STATUS_KEYS = ("pending", "success", "failed", "skipped")
+COUNTRY_TOP_N = 10
+# 北京日界偏移(SQLite date() 修饰符): 见模块 docstring"趋势"条
+_TZ_BEIJING = "+8 hours"
+
+
+def _empty_stats() -> dict:
+    """零数据形态(无租户可查/空库)——与 collect_stats 输出同构。"""
+    return {
+        "cards": {
+            "conversations": 0, "leads": 0, "high_intent": 0,
+            "push_success_rate": None},
+        "funnel": {
+            "conversations": 0, "leads": 0, "evaluated": 0,
+            "push_succeeded": 0},
+        "trend": [],
+        "distributions": {
+            "intent": {k: 0 for k in INTENT_KEYS},
+            "source": {},
+            "site": {},
+            "country": []},
+        "push_health": {
+            "pending": 0, "success": 0, "failed": 0, "skipped": 0,
+            "duplicated": 0},
+    }
+
+
+def _contact_any() -> ColumnElement:
+    """SQL 布尔: raw_info 三选一键任一非空(NULL/空白串都不算)。"""
+    conds = []
+    for key in CONTACT_KEYS:
+        value = CustomerProfile.raw_info[key].as_string()
+        conds.append(and_(value.is_not(None), func.trim(value) != ""))
+    return or_(*conds)
+
+
+def _range(column, from_dt, to_dt) -> list:
+    """闭区间条件列表(边界含;None 边界不加条件)。"""
+    conds = []
+    if from_dt is not None:
+        conds.append(column >= from_dt)
+    if to_dt is not None:
+        conds.append(column <= to_dt)
+    return conds
+
+
+def _push_rate(health: dict) -> float | None:
+    """success/(success+failed);分母 0 → None。"""
+    done = health["success"] + health["failed"]
+    return health["success"] / done if done else None
+
+
+def _sorted_counts(counter: dict) -> dict:
+    """分布字典定序: 计数倒序、同数按 key 升序(JSON 对象保序,前端展示稳)。"""
+    return dict(sorted(counter.items(), key=lambda kv: (-kv[1], kv[0])))
+
+
+async def collect_stats(
+        session, *, from_dt=None, to_dt=None) -> dict:
+    """单租户统计: 在该租户库的 session 上跑全部聚合,输出指标 dict。
+
+    只读不写;所有计数走 SQL 聚合(例外: country 逐行 IP 归国,见模块
+    docstring)。"""
+    conv_range = _range(Conversation.created_at, from_dt, to_dt)
+    log_range = _range(LeadPushLog.created_at, from_dt, to_dt)
+
+    # ---- 卡片 + 漏斗(会话锚定) ------------------------------------------------
+    conversations = (await session.execute(
+        select(func.count()).select_from(Conversation).where(*conv_range)
+    )).scalar() or 0
+    leads = (await session.execute(
+        select(func.count())
+        .select_from(Conversation)
+        .join(CustomerProfile,
+              CustomerProfile.conversation_id == Conversation.id)
+        .where(*conv_range, _contact_any())
+    )).scalar() or 0
+    high_intent = (await session.execute(
+        select(func.count())
+        .select_from(Conversation)
+        .join(Evaluation, Evaluation.conversation_id == Conversation.id)
+        .where(*conv_range,
+               Evaluation.intent_level == IntentLevel.interested)
+    )).scalar() or 0
+    evaluated = (await session.execute(
+        select(func.count())
+        .select_from(Conversation)
+        .join(Evaluation, Evaluation.conversation_id == Conversation.id)
+        .where(*conv_range)
+    )).scalar() or 0
+    push_succeeded = (await session.execute(
+        select(func.count()).select_from(Conversation).where(
+            *conv_range,
+            Conversation.id.in_(
+                select(LeadPushLog.conversation_id)
+                .where(LeadPushLog.status == "success")))
+    )).scalar() or 0
+
+    # ---- 推送健康(日志锚定) --------------------------------------------------
+    health = {k: 0 for k in PUSH_STATUS_KEYS}
+    for status, count in (await session.execute(
+            select(LeadPushLog.status, func.count())
+            .where(*log_range).group_by(LeadPushLog.status))).all():
+        if status in health:        # 未知状态不计(防御 imported/手改数据)
+            health[status] = count
+    health["duplicated"] = (await session.execute(
+        select(func.count()).select_from(LeadPushLog).where(
+            *log_range, LeadPushLog.outcome == "duplicated")
+    )).scalar() or 0
+
+    # ---- 趋势(北京日界) ------------------------------------------------------
+    day = func.date(Conversation.created_at, _TZ_BEIJING)
+    conv_per_day = dict((await session.execute(
+        select(day, func.count()).where(*conv_range).group_by(day)
+    )).all())
+    lead_stmt = (
+        select(day, func.count())
+        .select_from(Conversation)
+        .join(CustomerProfile,
+              CustomerProfile.conversation_id == Conversation.id)
+        .where(*conv_range, _contact_any())
+        .group_by(day))
+    lead_per_day = dict((await session.execute(lead_stmt)).all())
+    trend = [
+        {"date": d, "conversations": conv_per_day[d],
+         "leads": lead_per_day.get(d, 0)}
+        for d in sorted(set(conv_per_day) | set(lead_per_day))
+    ]
+
+    # ---- 分布 -----------------------------------------------------------------
+    intent = {k: 0 for k in INTENT_KEYS}
+    for level, count in (await session.execute(
+            select(Evaluation.intent_level, func.count())
+            .join(Conversation, Conversation.id == Evaluation.conversation_id)
+            .where(*conv_range).group_by(Evaluation.intent_level))).all():
+        intent[level.value if hasattr(level, "value") else str(level)] = count
+
+    source = Counter()
+    for src, count in (await session.execute(
+            select(Conversation.source, func.count())
+            .where(*conv_range).group_by(Conversation.source))).all():
+        source[src.value if hasattr(src, "value") else str(src)] = count
+
+    site_expr = Conversation.client_info["site"].as_string()
+    site = Counter()
+    for site_slug, count in (await session.execute(
+            select(site_expr, func.count())
+            .where(*conv_range, site_expr.is_not(None)).group_by(site_expr)
+    )).all():
+        site[site_slug] = count
+
+    # country: 逐行 IP 归国(iptocc 内存库,几百会话毫秒级——见模块 docstring)
+    country = Counter()
+    for (ip,) in (await session.execute(
+            select(Conversation.client_info["ip"].as_string())
+            .where(*conv_range))).all():
+        country[resolve_country(ip or "")] += 1
+
+    return {
+        "cards": {
+            "conversations": conversations,
+            "leads": leads,
+            "high_intent": high_intent,
+            "push_success_rate": _push_rate(health)},
+        "funnel": {
+            "conversations": conversations,
+            "leads": leads,
+            "evaluated": evaluated,
+            "push_succeeded": push_succeeded},
+        "trend": trend,
+        "distributions": {
+            "intent": intent,
+            "source": _sorted_counts(source),
+            "site": _sorted_counts(site),
+            "country": [
+                {"country": name, "count": count}
+                for name, count in country.most_common(COUNTRY_TOP_N)]},
+        "push_health": health,
+    }
+
+
+def merge_stats(parts: list[dict]) -> dict:
+    """跨租户应用层合并(逐库结果求和;比率/排序合并后重算重排)。
+
+    parts 为空 → 零数据形态。合并语义: 标量求和;趋势按日期键合并;
+    intent/source/site 按 key 求和;country 计数合并后重取 top N。"""
+    if not parts:
+        return _empty_stats()
+    out = _empty_stats()
+
+    trend = defaultdict(lambda: {"conversations": 0, "leads": 0})
+    intent = {k: 0 for k in INTENT_KEYS}
+    source: Counter = Counter()
+    site: Counter = Counter()
+    country: Counter = Counter()
+
+    for part in parts:
+        for key in ("conversations", "leads", "high_intent"):
+            out["cards"][key] += part["cards"][key]
+        for key in ("conversations", "leads", "evaluated", "push_succeeded"):
+            out["funnel"][key] += part["funnel"][key]
+        for row in part["trend"]:
+            trend[row["date"]]["conversations"] += row["conversations"]
+            trend[row["date"]]["leads"] += row["leads"]
+        for key, count in part["distributions"]["intent"].items():
+            intent[key] = intent.get(key, 0) + count
+        source.update(part["distributions"]["source"])
+        site.update(part["distributions"]["site"])
+        for item in part["distributions"]["country"]:
+            country[item["country"]] += item["count"]
+        for key in out["push_health"]:
+            out["push_health"][key] += part["push_health"][key]
+
+    out["cards"]["push_success_rate"] = _push_rate(out["push_health"])
+    out["trend"] = [
+        {"date": d, **counts} for d, counts in sorted(trend.items())]
+    out["distributions"] = {
+        "intent": intent,
+        "source": _sorted_counts(source),
+        "site": _sorted_counts(site),
+        "country": [
+            {"country": name, "count": count}
+            for name, count in country.most_common(COUNTRY_TOP_N)]},
+    return out