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

只读。每行 = 一次广告事件(不再按用户聚合):
- **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
  直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
- **信息流(比价/领券)**:一次比价 / 一次领券 = 一条整场发奖(ad_feed_reward)一行,给出 eCPM /
  发奖金币 + 应发 / 实发 / 一致;点开看金币复算因子。⚠️ draw 的逐条展示(ad_ecpm,impressionId 各自
  独立、与整场发奖无公共键、无法归到「哪一次」)**不再单独占行**(2026-07 按「一次比价/领券放一块」调整)——
  其展示数 / eCPM / 预估收益仍进全量统计(合计 / 趋势 / 分类大盘 / 穿山甲对照),只是主表不逐条铺开。
- 兜底:激励视频有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 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"}


# 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 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:reward_video/feed 直传;**draw(前端「Draw 信息流」)映射成 feed_all**
    # ——业务已全切 Draw,把「Draw 信息流」当作整个信息流口径(含历史误标 feed/NULL),否则筛选会漏历史。
    if ad_type == "draw":
        audit_scene = "feed_all"
    elif ad_type in _AUDIT_SCENES:
        audit_scene = ad_type
    else:
        audit_scene = 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 == "draw":
        # draw = 所有信息流展示(业务已全 Draw,含历史误标 feed);展示行只进统计,不占主表行
        stmt = stmt.where(AdEcpmRecord.ad_type.in_(["draw", "feed"]))
    elif 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,
            "sub_rewards": [],
            "sub_count": 1,
        }
        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) 未被展示合并的发奖行 → 事件:
    #    - 激励视频(reward_video):逐条成「纯发奖」事件(每次一个 ad_session_id;有发奖无展示等)。
    #    - 信息流(feed/draw):同一次比价/领券的多条广告共享**整场 ad_session_id**(客户端整场复用),
    #      按 (user_id, ad_session_id) 聚成**一次比价 / 一次领券**父事件;sub_rewards 为组内逐条明细,
    #      应发/实发取组内合计;业务已全 Draw → 类型统一 "draw"。session 缺失(极少旧数据)各自单独成组。
    feed_groups: dict[tuple[int, str], list[dict]] = {}
    for row in all_reward_rows:
        if row["record_id"] in used_reward_ids:
            continue
        if row["scene"] == "reward_video":
            events.append({
                "event_key": f"rwd-{row['record_id']}",
                "report_date": row["_report_date"],
                "user_id": row["user_id"],
                "ad_type": "reward_video",
                "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),
                "sub_rewards": [],
                "sub_count": 1,
            })
        else:
            # 聚合单位 = 一次完整比价/领券流程:优先用 trace_id(比价带 comparisonTraceId、领券带 sessionTraceId,
            # 整个流程不变;即使中途点广告致浮层关闭重弹、ad_session_id 变了,trace_id 仍不变 → 全流程聚成一行)。
            # 无 trace_id(历史领券未上报 / 旧数据)回退整场 ad_session_id;再无则 record_id 各自成组、不误并。
            grp_key = row.get("trace_id") or row.get("ad_session_id") or f"_rid-{row['record_id']}"
            feed_groups.setdefault((row["user_id"], grp_key), []).append(row)

    # 信息流分组 → 「一次比价 / 一次领券」父事件(收益恒 0:收益只算展示侧,避免与展示行重复计)。
    for (uid, grp_key), group in feed_groups.items():
        group.sort(key=lambda r: (r["created_at"], r["record_id"]))
        rep = group[-1]  # 代表条(最新一条):时间/场景/应用/代码位取它
        expected_sum = sum(int(g["expected_coin"]) for g in group)
        actual_sum = sum(int(g["actual_coin"]) for g in group)
        # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
        ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
        avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
        # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
        row_revenue = round(sum(
            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
            for g in group if g.get("ecpm")
        ), 6)
        events.append({
            "event_key": f"feedgrp-{uid}-{grp_key}",
            "report_date": rep["_report_date"],
            "user_id": uid,
            "ad_type": "draw",  # 业务已全切 Draw 信息流,聚合行统一 draw
            "feed_scene": rep.get("feed_scene"),
            "app_env": rep.get("app_env"),
            "our_code_id": rep.get("our_code_id"),
            "created_at": rep["created_at"],
            "hour": _cn_hour(rep["created_at"]) if by_hour else None,
            "has_impression": False,
            "impressions": 0,
            "ecpm": avg_ecpm,
            "revenue_yuan": 0.0,
            "row_revenue_yuan": row_revenue,
            "adn": None,
            "slot_id": None,
            "has_reward": True,
            "status": rep["status"],  # 代表状态(逐条见展开)
            "expected_coin": expected_sum,
            "actual_coin": actual_sum,
            "matched": all(bool(g["matched"]) for g in group),
            "reward_detail": None,
            "sub_rewards": [_reward_detail(g) for g in group],
            "sub_count": len(group),
        })

    # 「场景」作为全局筛选(与 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()
    }

    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
    # 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
    # 旧数据)不计入任何场景桶。
    scene_map: dict[str, dict] = {}
    for e in events:
        sc = e.get("feed_scene")
        if not sc:
            continue
        s = scene_map.get(sc)
        if s is None:
            s = {"impressions": 0, "revenue_yuan": 0.0}
            scene_map[sc] = s
        s["impressions"] += e["impressions"]
        s["revenue_yuan"] += e["revenue_yuan"]
    scene_stats = {
        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
        for k, v in scene_map.items()
    }

    # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
    # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
    # 不随 user / ad_type / feed_scene / app_env 筛选变化(活跃用户口径无这些维度)。
    dau = admin_stats.period_active_dau(
        db, _date.fromisoformat(date_from), _date.fromisoformat(date_to)
    )

    # 主表「逐行」= 单次广告行为(2026-07 按「一次比价/领券放一块」聚合):激励视频 = 一次观看一行(展示+发奖
    # 按 ad_session_id 合并);一次比价 / 一次领券 = 该次整场多条广告按 ad_session_id 聚成一行(展开看逐条)。
    # 信息流(draw/feed)的逐条展示(ad_ecpm,impressionId 各自独立、与整场发奖无公共键)不再单独占行
    # ——其展示数 / eCPM / 预估收益已计入上面的全量统计(total_*、daily / hourly、type_stats、穿山甲对照),
    # 只是主表不逐条铺开;逐条明细在父行展开里看(sub_rewards)。合计 / 趋势 / 分类大盘均基于全量 events,
    # 不受此过滤影响;total / 分页只作用于主表行。
    main_rows = [
        e for e in events
        if not (e["ad_type"] in ("draw", "feed") and e["has_impression"] and not e["has_reward"])
    ]

    return {
        "total": len(main_rows),
        "truncated": len(main_rows) > 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,
        "scene_stats": scene_stats,
        "dau": dau,
        "items": main_rows[offset:offset + limit],
    }