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.core import rewards
from app.models.ad_ecpm import AdEcpmRecord
from app.models.user import User


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,
    granularity: str = "day",
    limit: int = 500,
) -> dict:
    """日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。

    每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
    ad_type: None=全部 / reward_video / feed / draw。granularity=hour 时每行带北京小时(由各自时间算)。
    limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信。
    """
    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(每千次→单次);与发奖同源解析,口径一致。
            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 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),
        })

    events.sort(key=lambda e: (e["report_date"], e["user_id"], e["created_at"]))

    # 补手机号(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"])
    ]

    return {
        "total": len(events),
        "truncated": len(events) > limit,
        "total_impressions": total_impressions,
        "total_revenue_yuan": total_revenue_yuan,
        "total_expected_coin": total_expected_coin,
        "total_actual_coin": total_actual_coin,
        "mismatch_count": mismatch_count,
        "daily": daily,
        "items": events[:limit],
    }