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

"""admin 广告收益报表:**逐条广告事件**列表(每行一次广告,含展示 + 发奖对账)。

只读。每行 = 一次广告事件(不再按用户聚合):
- **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
  直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id)
  与逐条展示无法对应,单独成「纯发奖」行。
- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。

展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算
(ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
不受 limit(只截断 items)影响。

每行带 ad_type(reward_video/feed/draw)与 feed_scene(comparison/coupon/welfare),供前端区分
「比价 Draw 收益」与「领券 Draw 收益」(比价/领券共用同一代码位,只能靠 feed_scene 分)。

⚠️ 局限:① 历史信息流/Draw 发奖 ad_type 为 NULL 的旧记录统一视为 feed(向后兼容);Draw 仅
ad_type=="draw" 的新记录单独成类。② 跨天 S2S 回调:同一次广告的展示与发奖偶尔落相邻日,各自按
report_date / reward_date 归日。
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta
from datetime import date as _date

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.admin.repositories import ad_audit
from app.admin.repositories import stats as admin_stats
from app.core import rewards
from app.models.ad_ecpm import AdEcpmRecord
from app.models.user import User
from app.repositories import ad_pangle_revenue


def _cn_hour(dt: datetime) -> int:
    """created_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 _date_range(date_from: str, date_to: str) -> list[str]:
    """闭区间内逐日 'YYYY-MM-DD' 串(含首尾)。date_from > date_to 时返回空。"""
    d0 = _date.fromisoformat(date_from)
    d1 = _date.fromisoformat(date_to)
    out: list[str] = []
    d = d0
    while d <= d1:
        out.append(d.isoformat())
        d += timedelta(days=1)
    return out


# 报表 ad_type 与审计 scene 取值一致(reward_video / feed / draw):feed 与 draw 同查发奖表
# ad_feed_reward_record,由 audit 内部按 ad_type 区分(feed 含历史 NULL,draw 仅 ad_type=="draw")。
_AUDIT_SCENES = {"reward_video", "feed", "draw"}


def _event_ad_type(row: dict) -> str:
    """纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed;
    激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。"""
    if row["scene"] == "reward_video":
        return "reward_video"
    return row.get("ad_type") or "feed"


# 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
_REWARD_DETAIL_KEYS = (
    "record_id", "created_at", "status", "ecpm", "ecpm_factor", "units",
    "lt_index_start", "lt_index_end", "lt_factor_start", "lt_factor_end",
    "expected_coin", "actual_coin", "matched",
)


def _reward_detail(row: dict) -> dict:
    """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
    return {k: row[k] for k in _REWARD_DETAIL_KEYS}


def ad_revenue_report(
    db: Session,
    *,
    date_from: str,
    date_to: str,
    user_id: int | None = None,
    ad_type: str | None = None,
    feed_scene: str | None = None,
    app_env: str | None = None,
    granularity: str = "day",
    limit: int = 500,
    offset: int = 0,
    sort: str = "time",
) -> dict:
    """日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。

    每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
    ad_type: None=全部 / reward_video / feed / draw。feed_scene: None=全部 /
    comparison / coupon / welfare,作为全局筛选(同时作用于明细、合计与 daily/hourly 趋势)。
    granularity=hour 时每行带北京小时(由各自时间算),并额外返回全量 hourly 序列。
    事件按时间倒序(新→旧)排列;limit/offset 对排序后的全量做分页切片(items 为当前页),
    total 与 total_* / daily / hourly 在全量上统计,不受分页影响。
    """
    by_hour = granularity == "hour"

    # 1) 发奖行(逐日 audit 复算):建 (user_id, ad_session_id) → [行] 映射用于和展示合并;
    #    同时保留全量列表,未被展示合并的成「纯发奖」事件。
    reward_by_session: dict[tuple[int, str], list[dict]] = {}
    all_reward_rows: list[dict] = []
    # 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被
    # 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。
    audit_scene = ad_type if ad_type in _AUDIT_SCENES else None
    if ad_type is None or audit_scene is not None:
        for d in _date_range(date_from, date_to):
            for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene):
                row["_report_date"] = d
                all_reward_rows.append(row)
                sid = row.get("ad_session_id")
                if sid:
                    reward_by_session.setdefault((row["user_id"], sid), []).append(row)

    used_reward_ids: set[int] = set()
    events: list[dict] = []

    def _pop_reward(uid: int, sid: str | None) -> dict | None:
        """取一条与 (uid, sid) 匹配且未被用过的发奖行(激励视频展示↔发奖按会话 1:1 合并)。"""
        if not sid:
            return None
        for r in reward_by_session.get((uid, sid), ()):
            if r["record_id"] not in used_reward_ids:
                used_reward_ids.add(r["record_id"])
                return r
        return None

    # 2) 展示记录(ad_ecpm):每条一个事件;能匹配到发奖则合并成「展示 + 发奖」一行。
    stmt = select(AdEcpmRecord).where(
        AdEcpmRecord.report_date >= date_from,
        AdEcpmRecord.report_date <= date_to,
    )
    if user_id is not None:
        stmt = stmt.where(AdEcpmRecord.user_id == user_id)
    if ad_type is not None:
        stmt = stmt.where(AdEcpmRecord.ad_type == ad_type)
    for rec in db.execute(stmt).scalars():
        rwd = _pop_reward(rec.user_id, rec.ad_session_id)
        ev = {
            "event_key": f"imp-{rec.id}",
            "report_date": rec.report_date,
            "user_id": rec.user_id,
            "ad_type": rec.ad_type,
            "feed_scene": rec.feed_scene,
            "app_env": rec.app_env,
            "our_code_id": rec.our_code_id,
            "created_at": rec.created_at,
            "hour": _cn_hour(rec.created_at) if by_hour else None,
            "has_impression": True,
            "impressions": 1,
            "ecpm": rec.ecpm_raw,
            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次)。eCPM 先钳到 AD_ECPM_MAX_FEN(¥500 CPM)
            # 再折收益,与发奖口径 [rewards.calculate_ad_reward_coin] 一致(2026-06-29 修:原裸 parse_ecpm_yuan
            # 不钳,伪造/异常天价 eCPM 会把报表预估收益冲到任意大;金币侧已钳、收益侧漏钳)。
            "revenue_yuan": round(
                min(rewards.parse_ecpm_yuan(rec.ecpm_raw), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0, 6,
            ),
            "adn": rec.adn,
            "slot_id": rec.slot_id,
        }
        if rwd is not None:
            ev.update({
                "has_reward": True,
                "status": rwd["status"],
                "expected_coin": int(rwd["expected_coin"]),
                "actual_coin": int(rwd["actual_coin"]),
                "matched": bool(rwd["matched"]),
                "reward_detail": _reward_detail(rwd),
            })
        else:
            # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
            ev.update({
                "has_reward": False, "status": None,
                "expected_coin": 0, "actual_coin": 0, "matched": True,
                "reward_detail": None,
            })
        events.append(ev)

    # 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。
    #    收益恒 0(收益只算展示侧,避免与展示行重复计)。
    for row in all_reward_rows:
        if row["record_id"] in used_reward_ids:
            continue
        events.append({
            "event_key": f"rwd-{row['record_id']}",
            "report_date": row["_report_date"],
            "user_id": row["user_id"],
            "ad_type": _event_ad_type(row),
            "feed_scene": row.get("feed_scene"),
            "app_env": row.get("app_env"),
            "our_code_id": row.get("our_code_id"),
            "created_at": row["created_at"],
            "hour": _cn_hour(row["created_at"]) if by_hour else None,
            "has_impression": False,
            "impressions": 0,
            "ecpm": row["ecpm"],
            "revenue_yuan": 0.0,
            "adn": None,
            "slot_id": None,
            "has_reward": True,
            "status": row["status"],
            "expected_coin": int(row["expected_coin"]),
            "actual_coin": int(row["actual_coin"]),
            "matched": bool(row["matched"]),
            "reward_detail": _reward_detail(row),
        })

    # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
    # feed_scene 仅信息流 / Draw 有值,激励视频与旧数据为 None;选中后只保留该场景事件。
    if feed_scene is not None:
        events = [e for e in events if e.get("feed_scene") == feed_scene]

    # app_env 过滤(2026-06-29 新增能力,修隐患:测试应用上报的假 eCPM 如 ¥678 CPM 会污染正式收益合计/平均):
    # 显式传 "prod"/"test" 只看该环境;不传=全部(维持现状)。**不擅自把默认改成排除 test**——本地 dev 库多为
    # test 数据、默认排除会使本地报表空,且「正式报表是否含 test」属产品口径。建议前端报表页加 app_env 筛选器
    # (默认选 prod),或产品确认后再把默认改成排除 test。注:穿山甲后台收益列(total_pangle_*)暂未联动此过滤
    # (它是独立对照列,且 pangle 的 test 是真实小额、非客户端那种假值)。
    if app_env is not None:
        events = [e for e in events if e.get("app_env") == app_env]

    # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
    # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
    if sort == "ecpm":
        events.sort(key=lambda e: rewards.parse_ecpm_fen(e["ecpm"]), reverse=True)
    else:
        events.sort(key=lambda e: (e["report_date"], e["created_at"]), reverse=True)

    # 补手机号(admin 展示用,完整不脱敏,与用户 / 钱包 / 比价记录页一致):批量一次查,避免 N+1。
    uids = {e["user_id"] for e in events}
    phone_map: dict[int, str] = {}
    if uids:
        phone_map = {
            uid: phone
            for uid, phone in db.execute(
                select(User.id, User.phone).where(User.id.in_(uids))
            ).all()
        }
    for e in events:
        e["user_phone"] = phone_map.get(e["user_id"])

    total_impressions = sum(e["impressions"] for e in events)
    total_revenue_yuan = round(sum(e["revenue_yuan"] for e in events), 6)
    total_expected_coin = sum(e["expected_coin"] for e in events)
    total_actual_coin = sum(e["actual_coin"] for e in events)
    mismatch_count = sum(1 for e in events if e["has_reward"] and not e["matched"])

    # 按日期汇总(全量,不受 limit):供前端按天趋势图。
    daily_map: dict[str, dict] = {}
    for e in events:
        d = daily_map.get(e["report_date"])
        if d is None:
            d = {"date": e["report_date"], "impressions": 0, "revenue_yuan": 0.0,
                 "expected_coin": 0, "actual_coin": 0}
            daily_map[e["report_date"]] = d
        d["impressions"] += e["impressions"]
        d["revenue_yuan"] += e["revenue_yuan"]
        d["expected_coin"] += e["expected_coin"]
        d["actual_coin"] += e["actual_coin"]
    daily = [
        {**d, "revenue_yuan": round(d["revenue_yuan"], 6)}
        for d in sorted(daily_map.values(), key=lambda x: x["date"])
    ]

    # 穿山甲后台收益(GroMore 数据 API,T+1 入库 ad_pangle_daily_revenue):汇总 + 按天趋势级展示,
    # 与上面客户端自报 eCPM 折算的预估并列对照(看 gap)。穿山甲数据**无用户/场景/类型维度**,故仅在
    # 「全量视图」(未按 user_id / ad_type / feed_scene 过滤)给值;一旦带这些过滤,穿山甲数无法对应口径
    # → 置 None,前端显示「-」并提示。逐条事件行不动(仍是客户端预估)。
    pangle_filterable = user_id is None and ad_type is None and feed_scene is None
    total_pangle_revenue_yuan: float | None = None
    total_pangle_api_revenue_yuan: float | None = None
    if pangle_filterable:
        pangle_aggs = ad_pangle_revenue.aggregate_by_date(db, date_from=date_from, date_to=date_to)
        if pangle_aggs:
            by_date = {a["date"]: a for a in pangle_aggs}
            for d in daily:
                pa = by_date.get(d["date"])
                d["pangle_revenue_yuan"] = pa["revenue_yuan"] if pa else None
                d["pangle_api_revenue_yuan"] = pa["api_revenue_yuan"] if pa else None
            total_pangle_revenue_yuan = round(sum(a["revenue_yuan"] for a in pangle_aggs), 6)
            api_vals = [a["api_revenue_yuan"] for a in pangle_aggs if a["api_revenue_yuan"] is not None]
            total_pangle_api_revenue_yuan = round(sum(api_vals), 6) if api_vals else None

    # 按小时汇总(全量,不受分页 limit/offset 影响):供前端按小时趋势图(单日 granularity=hour 时用)。
    # 只在 by_hour 下聚合(此时每个 event 带 hour);否则空。前端按天趋势仍用 daily。
    hourly: list[dict] = []
    if by_hour:
        hour_map: dict[int, dict] = {}
        for e in events:
            h = e["hour"]
            if h is None:
                continue
            hd = hour_map.get(h)
            if hd is None:
                hd = {"hour": h, "impressions": 0, "revenue_yuan": 0.0,
                      "expected_coin": 0, "actual_coin": 0}
                hour_map[h] = hd
            hd["impressions"] += e["impressions"]
            hd["revenue_yuan"] += e["revenue_yuan"]
            hd["expected_coin"] += e["expected_coin"]
            hd["actual_coin"] += e["actual_coin"]
        hourly = [
            {**hd, "revenue_yuan": round(hd["revenue_yuan"], 6)}
            for hd in sorted(hour_map.values(), key=lambda x: x["hour"])
        ]

    # 分广告类型小计(按 ad_type:展示条数 + 预估收益;eCPM 由前端用 收益÷展示×1000 算)。
    # 基于全量(已按 feed_scene 过滤)events;前端只取 draw / reward_video 两类展示。
    type_map: dict[str, dict] = {}
    for e in events:
        t = type_map.get(e["ad_type"])
        if t is None:
            t = {"impressions": 0, "revenue_yuan": 0.0}
            type_map[e["ad_type"]] = t
        t["impressions"] += e["impressions"]
        t["revenue_yuan"] += e["revenue_yuan"]
    type_stats = {
        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
        for k, v in type_map.items()
    }

    # DAU:复用大盘「今日活跃」口径(stats.today_dau,last_login_at)。该口径只能算今日,
    # 故仅当查询=今日单天时给值;历史 / 多天区间返回 None,前端显示「-」。
    is_today = date_from == date_to == rewards.cn_today().isoformat()
    dau = admin_stats.today_dau(db) if is_today else None

    return {
        "total": len(events),
        "truncated": len(events) > offset + limit,
        "total_impressions": total_impressions,
        "total_revenue_yuan": total_revenue_yuan,
        # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
        "total_pangle_revenue_yuan": total_pangle_revenue_yuan,
        "total_pangle_api_revenue_yuan": total_pangle_api_revenue_yuan,
        "pangle_revenue_available": total_pangle_revenue_yuan is not None,
        "total_expected_coin": total_expected_coin,
        "total_actual_coin": total_actual_coin,
        "mismatch_count": mismatch_count,
        "daily": daily,
        "hourly": hourly,
        "type_stats": type_stats,
        "dau": dau,
        "items": events[offset:offset + limit],
    }