guke
331 lines
13 KiB
Python
331 lines
13 KiB
Python
"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
|
||
|
||
数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
|
||
区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
|
||
- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
|
||
- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
|
||
- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
|
||
- 另含 coupon_slot_report(数据源 coupon_claim_record):按 coupon_id「按券成功率」表,见设计 §13。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime
|
||
from datetime import date as _date
|
||
|
||
from sqlalchemy import case, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core import rewards
|
||
from app.models.coupon_state import CouponClaimRecord, CouponSession
|
||
from app.models.user import User
|
||
from app.repositories import ad_ecpm as crud_ecpm
|
||
from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
|
||
|
||
|
||
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 _success_rates(rows: list) -> dict:
|
||
"""平台粒度成功率(见 docs/guides/领券成功率指标-设计与埋点.md §3/§12):
|
||
|
||
- sel(s) = 勾选平台(`platforms` 空 → 全领三档 DEFAULT_PLATFORMS);
|
||
- succ(s) = `platform_success` ∩ sel(至少领到一张的平台);
|
||
- ② 整单成功率 = #{sel⊆succ 且 sel≠∅} / 发起数;
|
||
- ③ 点位成功率 = Σ|succ| / Σ|sel|;per_platform[p] = 勾了 p 且成功 / 勾了 p。
|
||
基数含全部 session(started/completed/failed/abandoned),与「发起数」同基数。
|
||
"""
|
||
started = len(rows)
|
||
full_success = 0
|
||
point_success = 0
|
||
point_total = 0
|
||
per_succ = {p: 0 for p in DEFAULT_PLATFORMS}
|
||
per_total = {p: 0 for p in DEFAULT_PLATFORMS}
|
||
for r in rows:
|
||
sel = set(r.platforms) if r.platforms else set(DEFAULT_PLATFORMS)
|
||
succ = set(r.platform_success or []) & sel
|
||
point_success += len(succ)
|
||
point_total += len(sel)
|
||
if sel and succ == sel:
|
||
full_success += 1
|
||
for p in sel:
|
||
if p in per_total: # 只统计三档已知平台;未知/非法平台 id 不进 per_platform
|
||
per_total[p] += 1
|
||
if p in succ:
|
||
per_succ[p] += 1
|
||
return {
|
||
"full_success_count": full_success,
|
||
"full_success_rate": round(full_success / started, 4) if started else None,
|
||
"point_success_count": point_success,
|
||
"point_total_count": point_total,
|
||
"point_success_rate": round(point_success / point_total, 4) if point_total else None,
|
||
"per_platform": {
|
||
p: (round(per_succ[p] / per_total[p], 4) if per_total[p] else None)
|
||
for p in DEFAULT_PLATFORMS
|
||
},
|
||
}
|
||
|
||
|
||
def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad_revenue_yuan: float = 0.0) -> 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,
|
||
"ad_revenue_yuan": ad_revenue_yuan,
|
||
}
|
||
|
||
|
||
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,
|
||
statuses: list[str] | None = None,
|
||
granularity: str = "day",
|
||
limit: int = 500,
|
||
offset: int = 0,
|
||
sort: str = "time",
|
||
) -> dict:
|
||
"""日期区间(北京自然日 started_date,闭区间)领券数据:汇总卡 + 趋势 + 逐条明细。
|
||
|
||
- user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。
|
||
- app_env:prod/dev 精确;None=全部。
|
||
- statuses:领券状态多选(started/completed/failed/abandoned);None/空=全部。整个视图
|
||
(汇总/成功率/趋势/明细)按选中状态算,与 app_env 同级过滤(方案 A)。
|
||
- 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 statuses:
|
||
stmt = stmt.where(CouponSession.status.in_(statuses))
|
||
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),
|
||
**_success_rates(rows),
|
||
}
|
||
|
||
# ── 按天趋势(柱=发起/完成数,线=平均耗时)──
|
||
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()
|
||
}
|
||
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
|
||
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, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)))
|
||
|
||
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()
|
||
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
|
||
return {
|
||
"items": [_session_to_row(r, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)) for r in rows],
|
||
"total": int(total),
|
||
}
|
||
|
||
|
||
_SLOT_OK = ("success", "already_claimed")
|
||
_SLOT_TRIED = ("success", "already_claimed", "failed")
|
||
|
||
|
||
def coupon_slot_report(
|
||
db: Session, *, date_from: str, date_to: str, app_env: str | None = None
|
||
) -> dict:
|
||
"""按 coupon_id(具体券)聚合成功率(见 docs/guides/领券成功率指标-设计与埋点.md §13)。
|
||
|
||
数据源 coupon_claim_record(粒度=设备-天,唯一键 device+coupon+day)。
|
||
- 尝试 = status ∈ {success, already_claimed, failed}(skipped 排除);
|
||
- 成功 = status ∈ {success, already_claimed};成功率 = 成功/尝试;
|
||
- claim_date 区间 + app_env(None=全部)过滤;按 tried 倒序返回。
|
||
"""
|
||
d_from = _date.fromisoformat(date_from)
|
||
d_to = _date.fromisoformat(date_to)
|
||
ok = case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0)
|
||
stmt = (
|
||
select(
|
||
CouponClaimRecord.coupon_id,
|
||
func.max(CouponClaimRecord.coupon_name).label("coupon_name"),
|
||
func.count().label("tried"),
|
||
func.sum(ok).label("succeeded"),
|
||
)
|
||
.where(
|
||
CouponClaimRecord.claim_date >= d_from,
|
||
CouponClaimRecord.claim_date <= d_to,
|
||
CouponClaimRecord.status.in_(_SLOT_TRIED),
|
||
)
|
||
.group_by(CouponClaimRecord.coupon_id)
|
||
)
|
||
if app_env is not None:
|
||
stmt = stmt.where(CouponClaimRecord.app_env == app_env)
|
||
items = []
|
||
for coupon_id, coupon_name, tried, succeeded in db.execute(stmt).all():
|
||
tried = int(tried or 0)
|
||
succeeded = int(succeeded or 0)
|
||
items.append({
|
||
"coupon_id": coupon_id,
|
||
"coupon_name": coupon_name,
|
||
"platform": coupon_id_to_platform(coupon_id),
|
||
"tried": tried,
|
||
"succeeded": succeeded,
|
||
"success_rate": round(succeeded / tried, 4) if tried else None,
|
||
})
|
||
items.sort(key=lambda x: (-x["tried"], x["coupon_id"]))
|
||
return {"items": items}
|