shaguabijia-app-server/app/admin/repositories/coupon_data.py

"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。

数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
"""
from __future__ import annotations

from datetime import UTC, date as _date, datetime

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from app.core import rewards
from app.models.coupon_state import CouponSession
from app.models.user import User


def _cn_hour(dt: datetime) -> int:
    """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。"""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt.astimezone(rewards.CN_TZ).hour


def _percentile(sorted_vals: list[int], q: float) -> int | None:
    """线性插值分位(q=0..100,numpy 默认法)。sorted_vals 须已升序;空返回 None。"""
    if not sorted_vals:
        return None
    if len(sorted_vals) == 1:
        return sorted_vals[0]
    idx = (len(sorted_vals) - 1) * q / 100.0
    lo = int(idx)
    hi = min(lo + 1, len(sorted_vals) - 1)
    frac = idx - lo
    return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)


def _avg(vals: list[int]) -> int | None:
    return round(sum(vals) / len(vals)) if vals else None


def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
    """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
    return {
        "id": r.id,
        "trace_id": r.trace_id,
        "user_id": r.user_id,
        "user_phone": phone,
        "user_nickname": nickname,
        "status": r.status,
        "platforms": r.platforms,
        "origin_package": r.origin_package,
        "elapsed_ms": r.elapsed_ms,
        "platform_elapsed": r.platform_elapsed,
        "device_model": r.device_model,
        "rom": r.rom,
        "app_env": r.app_env,
        "started_at": r.started_at,
        "claimed_count": r.claimed_count,
        "trace_url": r.trace_url,
    }


def _empty_result() -> dict:
    return {
        "summary": {
            "started_count": 0, "completed_count": 0, "avg_elapsed_ms": None,
            "p5_ms": None, "p50_ms": None, "p95_ms": None, "p99_ms": None,
        },
        "daily": [],
        "hourly": [],
        "total": 0,
        "items": [],
    }


def coupon_data_report(
    db: Session,
    *,
    date_from: str,
    date_to: str,
    user: str | None = None,
    app_env: str | None = None,
    granularity: str = "day",
    limit: int = 500,
    offset: int = 0,
    sort: str = "time",
) -> dict:
    """日期区间(北京自然日 started_date,闭区间)领券数据:汇总卡 + 趋势 + 逐条明细。

    - user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。
    - app_env:prod/dev 精确;None=全部。
    - sort:time=发起时刻倒序(默认) / elapsed=全程耗时倒序(None 末尾)。
    """
    by_hour = granularity == "hour"
    d_from = _date.fromisoformat(date_from)
    d_to = _date.fromisoformat(date_to)

    # user 模糊 → 先定位匹配用户 id;匹配不到直接空结果(不全表扫)。
    user_ids: set[int] | None = None
    if user:
        like = f"%{user}%"
        user_ids = set(db.execute(
            select(User.id).where(or_(User.phone.like(like), User.nickname.like(like)))
        ).scalars().all())
        if not user_ids:
            return _empty_result()

    stmt = select(CouponSession).where(
        CouponSession.started_date >= d_from,
        CouponSession.started_date <= d_to,
    )
    if app_env is not None:
        stmt = stmt.where(CouponSession.app_env == app_env)
    if user_ids is not None:
        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
    rows = list(db.execute(stmt).scalars())

    # ── 汇总卡 ──
    completed_elapsed = sorted(
        r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
    )
    summary = {
        "started_count": len(rows),
        "completed_count": sum(1 for r in rows if r.status == "completed"),
        "avg_elapsed_ms": _avg(completed_elapsed),
        "p5_ms": _percentile(completed_elapsed, 5),
        "p50_ms": _percentile(completed_elapsed, 50),
        "p95_ms": _percentile(completed_elapsed, 95),
        "p99_ms": _percentile(completed_elapsed, 99),
    }

    # ── 按天趋势(柱=发起/完成数,线=平均耗时)──
    daily_map: dict[str, dict] = {}
    for r in rows:
        d = r.started_date.isoformat()
        b = daily_map.get(d)
        if b is None:
            b = {"date": d, "started_count": 0, "completed_count": 0, "_elapsed": []}
            daily_map[d] = b
        b["started_count"] += 1
        if r.status == "completed":
            b["completed_count"] += 1
            if r.elapsed_ms is not None:
                b["_elapsed"].append(r.elapsed_ms)
    daily = [
        {
            "date": b["date"],
            "started_count": b["started_count"],
            "completed_count": b["completed_count"],
            "avg_elapsed_ms": _avg(b["_elapsed"]),
        }
        for b in sorted(daily_map.values(), key=lambda x: x["date"])
    ]

    # ── 按小时趋势(单日 hour 粒度)──
    hourly: list[dict] = []
    if by_hour:
        hour_map: dict[int, dict] = {}
        for r in rows:
            h = _cn_hour(r.started_at)
            b = hour_map.get(h)
            if b is None:
                b = {"hour": h, "started_count": 0, "completed_count": 0, "_elapsed": []}
                hour_map[h] = b
            b["started_count"] += 1
            if r.status == "completed":
                b["completed_count"] += 1
                if r.elapsed_ms is not None:
                    b["_elapsed"].append(r.elapsed_ms)
        hourly = [
            {
                "hour": b["hour"],
                "started_count": b["started_count"],
                "completed_count": b["completed_count"],
                "avg_elapsed_ms": _avg(b["_elapsed"]),
            }
            for b in sorted(hour_map.values(), key=lambda x: x["hour"])
        ]

    # ── 明细:排序 + 分页 + 补用户手机号/昵称(批量,防 N+1)──
    if sort == "elapsed":
        rows.sort(key=lambda r: (r.elapsed_ms is None, -(r.elapsed_ms or 0)))
    else:  # time:发起时刻倒序
        rows.sort(key=lambda r: r.started_at, reverse=True)
    page = rows[offset:offset + limit]

    uids = {r.user_id for r in page if r.user_id is not None}
    user_map: dict[int, tuple[str | None, str | None]] = {}
    if uids:
        user_map = {
            uid: (phone, nickname)
            for uid, phone, nickname in db.execute(
                select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
            ).all()
        }
    items = []
    for r in page:
        phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
        items.append(_session_to_row(r, phone, nickname))

    return {
        "summary": summary,
        "daily": daily,
        "hourly": hourly,
        "total": len(rows),
        "items": items,
    }


def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
    """某用户全部领券记录(点手机号抽屉用):按发起时刻倒序、不限日期,total=该用户领券总次数。"""
    rows = list(db.execute(
        select(CouponSession)
        .where(CouponSession.user_id == user_id)
        .order_by(CouponSession.started_at.desc())
        .limit(limit)
    ).scalars())
    total = db.execute(
        select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
    ).scalar_one()
    return {"items": [_session_to_row(r) for r in rows], "total": int(total)}