feat(ad-revenue): admin 广告收益报表(按 用户/日期/类型/应用/代码位 聚合)

- 新增 GET /admin/api/ad-revenue-report:展示条数/收益 + 复用金币审计逐条复算做发奖对账
- ad_ecpm/ad_reward/ad_feed_reward 各加 app_env + our_code_id 两列(alembic 迁移)
- ecpm-report / feed-reward 接收并落库 app_env/our_code_id;激励发奖按 ad_session_id 回填
- ad_audit 抽出 audit_rows,报表与逐条审计复用同一复算口径
- 组级 matched 改「组内逐条全一致」,避免应发和==实发和的互相抵消掩盖错误
- list_feedbacks 改 offset 分页并返回 total(配合 admin 页码分页)
- 反馈正文上限 _CONTENT_MAX 2000→200
- 文档:新增 admin-ad-revenue-report,更新 ecpm/feed-reward/feedback 及对应 db docs

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/admin/main.py

@@ -14,6 +14,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.admin.routers.ad_audit import router as ad_audit_router
+from app.admin.routers.ad_revenue import router as ad_revenue_router
 from app.admin.routers.admins import router as admins_router
 from app.admin.routers.audit import router as audit_router
 from app.admin.routers.auth import router as auth_router
@@ -90,3 +91,4 @@ admin_app.include_router(admins_router)
 admin_app.include_router(audit_router)
 admin_app.include_router(config_router)
 admin_app.include_router(ad_audit_router)
+admin_app.include_router(ad_revenue_router)

app/admin/repositories/ad_audit.py

@@ -67,6 +67,8 @@ def _reward_video_rows(
                 "scene": "reward_video",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "app_env": rec.app_env,
+                "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
@@ -86,6 +88,8 @@ def _reward_video_rows(
                 "scene": "reward_video",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "app_env": rec.app_env,
+                "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
@@ -150,6 +154,8 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
                 "scene": "feed",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "app_env": rec.app_env,
+                "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
@@ -168,6 +174,8 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
                 "scene": "feed",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "app_env": rec.app_env,
+                "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
                 "status": rec.status,
                 "ecpm": rec.ecpm_raw,
@@ -184,6 +192,22 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
     return rows
 
 
+def audit_rows(
+    db: Session, *, date: str, user_id: int | None, scene: str | None = None
+) -> list[dict]:
+    """当日逐条发奖复算行(未排序)。scene: None=两类 / "reward_video" / "feed"。
+
+    每行含 `app_env`/`our_code_id`/`expected_coin`/`actual_coin` 等,供金币审计逐条对账,
+    也供广告收益报表把「应发/实发」按 用户×类型×应用×代码位 聚合(见 ad_revenue,复用同一复算口径)。
+    """
+    rows: list[dict] = []
+    if scene in (None, "reward_video"):
+        rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
+    if scene in (None, "feed"):
+        rows.extend(_feed_rows(db, date=date, user_id=user_id))
+    return rows
+
+
 def ad_coin_audit(
     db: Session,
     *,
@@ -200,11 +224,7 @@ def ad_coin_audit(
     影响;`items` 才是展示集(only_mismatch 时只取 ✗ 行)按 created_at 倒序截断到 limit。
     份序号在全天数据上已算好,limit 只影响展示条数、不影响 expected 复算正确性。
     """
-    rows: list[dict] = []
-    if scene in (None, "reward_video"):
-        rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
-    if scene in (None, "feed"):
-        rows.extend(_feed_rows(db, date=date, user_id=user_id))
+    rows = audit_rows(db, date=date, user_id=user_id, scene=scene)
     rows.sort(key=lambda r: (r["created_at"], r["record_id"]), reverse=True)
 
     total = len(rows)

app/admin/repositories/ad_revenue.py

@@ -0,0 +1,239 @@
+"""admin 广告收益报表:按 用户 / 日期 / 广告类型 / 应用 / 代码位 聚合(单表含发奖对账)。
+
+只读。聚合键 = user_id × ad_type × app_env × our_code_id;每组一行同时给出:
+- 展示条数 + 收益:`ad_ecpm_record`(每行 = 客户端一次广告展示;收益 = Σ eCPM元 ÷ 1000)。
+  激励视频每次展示上报一行;信息流轮播每条展示各上报一行(每条独立 id,不复用会话)。
+- 应发金币 / 实发金币:复用金币审计的**逐条复算**(`ad_audit.audit_rows`,与正式发奖同一公式口径,
+  不另写公式),把每条发奖记录的 expected/actual 按同维度求和;`matched` = 组内**逐条**全部一致
+  (任一条不符该组即不符,不用「应发和==实发和」以免互相抵消掩盖错误)。**不改发奖逻辑**,只读复算。
+
+展示与发奖来自不同表,做并集:有展示无发奖(用户中途关 / 未达发奖)、有发奖无展示
+(未上报 eCPM)都各自成行。app_env/our_code_id 旧数据为 NULL → 归到「来源未知」组。
+
+⚠️ 局限:① 历史 Draw 发奖混在 ad_feed_reward_record 无类型标记,金币侧统一记 `feed`(迁移后 Draw
+不再产生新数据)。② 聚合级只能看出「某组应发≠实发」,定位到具体哪条仍需逐条审计接口(ad-coin-audit)。
+"""
+from __future__ import annotations
+
+from datetime import date as _date, datetime, timedelta, timezone
+
+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
+
+
+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=timezone.utc)
+    return dt.astimezone(rewards.CN_TZ).hour
+
+
+def _key(
+    report_date: str,
+    user_id: int,
+    ad_type: str,
+    app_env: str | None,
+    our_code_id: str | None,
+    hour: int | None,
+) -> tuple:
+    return (report_date, user_id, ad_type, app_env or None, our_code_id or None, 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
+
+
+# 审计行的 scene 与报表 ad_type 一一对应
+_SCENE_TO_AD_TYPE = {"reward_video": "reward_video", "feed": "feed"}
+
+
+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。
+
+    聚合键含**日期**:report_date × user × ad_type × app_env × our_code_id(× 北京小时,granularity=hour)。
+    ad_type: None=全部 / reward_video / feed / draw。
+    granularity: "day"=按天 / "hour"=按小时(聚合键再加北京小时 0–23,每组一行)。
+    limit 只截断展示明细,total 与 total_* / daily 在全量上统计(不受 limit 影响),数字始终可信。
+
+    返回额外含 `daily`(按日期汇总的展示/收益/应发/实发,供前端按天趋势图;不受 limit 影响)。
+
+    注:按小时下,展示按 ecpm 记录的小时、金币按发奖记录的小时各自归桶——S2S 回调可能比展示晚
+    一会儿,故同一次广告的展示与金币偶尔落相邻小时(按天则一致)。
+    """
+    by_hour = granularity == "hour"
+    groups: dict[tuple, dict] = {}
+
+    def _grp(key: tuple) -> dict:
+        g = groups.get(key)
+        if g is None:
+            rdate, uid, atype, app_env, code_id, hour = key
+            g = {
+                "report_date": rdate,
+                "user_id": uid,
+                "ad_type": atype,
+                "app_env": app_env,
+                "our_code_id": code_id,
+                "hour": hour,
+                "impressions": 0,
+                "revenue_yuan": 0.0,
+                "expected_coin": 0,
+                "actual_coin": 0,
+                "adns": set(),
+                "impression_records": [],  # 该组逐条展示明细(展开下钻用)
+                "records": [],  # 该组逐条发奖复算明细(展开下钻用)
+            }
+            groups[key] = g
+        return g
+
+    # 1) 展示条数 + 收益 ← ad_ecpm_record(report_date 闭区间;字符串 YYYY-MM-DD 字典序即日期序)
+    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():
+        hour = _cn_hour(rec.created_at) if by_hour else None
+        g = _grp(_key(rec.report_date, rec.user_id, rec.ad_type, rec.app_env, rec.our_code_id, hour))
+        g["impressions"] += 1
+        # 单次展示收益(元) = eCPM元 ÷ 1000(每千次→单次);用与发奖同源的解析,口径一致。
+        rev = rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0
+        g["revenue_yuan"] += rev
+        if rec.adn:
+            g["adns"].add(rec.adn)
+        g["impression_records"].append({
+            "id": rec.id,
+            "created_at": rec.created_at,
+            "ecpm": rec.ecpm_raw,
+            "revenue_yuan": round(rev, 6),
+            "adn": rec.adn,
+            "slot_id": rec.slot_id,
+        })
+
+    # 2) 应发 / 实发金币 ← 复用金币审计逐条复算(同一公式口径),按同维度求和。
+    #    audit_rows 是单日的,区间逐日调用,每天的行归到当天 report_date(语义与单日报表完全一致)。
+    #    ad_type=draw 时审计无对应记录(scene 只有 reward_video/feed),金币侧自然为空。
+    audit_scene = _SCENE_TO_AD_TYPE.get(ad_type) if ad_type is not None 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):
+                atype = _SCENE_TO_AD_TYPE.get(row["scene"], row["scene"])
+                hour = _cn_hour(row["created_at"]) if by_hour else None
+                g = _grp(_key(d, row["user_id"], atype, row.get("app_env"), row.get("our_code_id"), hour))
+                g["expected_coin"] += int(row["expected_coin"])
+                g["actual_coin"] += int(row["actual_coin"])
+                # 逐条明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致)——前端展开该组时下钻展示。
+                g["records"].append({
+                    "record_id": row["record_id"],
+                    "created_at": row["created_at"],
+                    "status": row["status"],
+                    "ecpm": row["ecpm"],
+                    "ecpm_factor": row["ecpm_factor"],
+                    "units": row["units"],
+                    "lt_index_start": row["lt_index_start"],
+                    "lt_index_end": row["lt_index_end"],
+                    "lt_factor_start": row["lt_factor_start"],
+                    "lt_factor_end": row["lt_factor_end"],
+                    "expected_coin": row["expected_coin"],
+                    "actual_coin": row["actual_coin"],
+                    "matched": row["matched"],
+                })
+
+    rows = list(groups.values())
+    rows.sort(
+        key=lambda r: (
+            r["report_date"],
+            r["user_id"],
+            r["hour"] if r["hour"] is not None else -1,
+            r["ad_type"] or "",
+            r["our_code_id"] or "",
+        )
+    )
+
+    total_impressions = sum(r["impressions"] for r in rows)
+    total_expected_coin = sum(r["expected_coin"] for r in rows)
+    total_actual_coin = sum(r["actual_coin"] for r in rows)
+    total_revenue_yuan = round(sum(r["revenue_yuan"] for r in rows), 6)
+
+    # 按日期汇总(全量,不受 limit):供前端按天趋势图。
+    daily_map: dict[str, dict] = {}
+    for r in rows:
+        d = daily_map.get(r["report_date"])
+        if d is None:
+            d = {
+                "date": r["report_date"],
+                "impressions": 0,
+                "revenue_yuan": 0.0,
+                "expected_coin": 0,
+                "actual_coin": 0,
+            }
+            daily_map[r["report_date"]] = d
+        d["impressions"] += r["impressions"]
+        d["revenue_yuan"] += r["revenue_yuan"]
+        d["expected_coin"] += r["expected_coin"]
+        d["actual_coin"] += r["actual_coin"]
+    daily = [
+        {**d, "revenue_yuan": round(d["revenue_yuan"], 6)}
+        for d in sorted(daily_map.values(), key=lambda x: x["date"])
+    ]
+
+    items = [
+        {
+            "report_date": r["report_date"],
+            "user_id": r["user_id"],
+            "ad_type": r["ad_type"],
+            "app_env": r["app_env"],
+            "our_code_id": r["our_code_id"],
+            "hour": r["hour"],
+            "impressions": r["impressions"],
+            "revenue_yuan": round(r["revenue_yuan"], 6),
+            "expected_coin": r["expected_coin"],
+            "actual_coin": r["actual_coin"],
+            # 组内**逐条**全部一致才记一致——不能用「应发和==实发和」,否则一条多发+一条少发会互相
+            # 抵消、求和相等被误判为 ✓,掩盖真实发奖错误。纯展示无发奖记录的组 all([]) → True。
+            "matched": all(rec["matched"] for rec in r["records"]),
+            "adns": sorted(r["adns"]),
+            "impression_records": sorted(
+                r["impression_records"], key=lambda x: (x["created_at"], x["id"])
+            ),
+            "records": sorted(r["records"], key=lambda x: (x["created_at"], x["record_id"])),
+        }
+        for r in rows[:limit]
+    ]
+
+    return {
+        "total": len(rows),
+        "truncated": len(rows) > 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": sum(
+            1 for r in rows if not all(rec["matched"] for rec in r["records"])
+        ),
+        "daily": daily,
+        "items": items,
+    }

app/admin/repositories/queries.py

@@ -286,10 +286,11 @@ def list_feedbacks(
     sort_order: str = "desc",
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[Feedback], int | None]:
+) -> tuple[list[Feedback], int | None, int]:
     """反馈工单列表。支持 状态 / 用户ID / 内容模糊 / 提交时间范围 筛选,按 id·提交时间排序。
     **offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),代价是翻页期间
-    数据变动可能错位一条——admin 低频场景可接受。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
+    数据变动可能错位一条——admin 低频场景可接受。返回 (items, next_cursor, total),total 供页码分页。
+    created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
     stmt = select(Feedback)
     if status:
         stmt = stmt.where(Feedback.status == status)
@@ -306,14 +307,7 @@ def list_feedbacks(
     sort_col = sort_cols.get(sort_by, Feedback.id)
     order_fn = asc if sort_order == "asc" else desc
     id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id)
-    stmt = stmt.order_by(order_fn(sort_col), id_order)
-
-    offset = max(cursor or 0, 0)
-    rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all())
-    has_more = len(rows) > limit
-    items = rows[:limit]
-    next_cursor = offset + limit if has_more else None
-    return items, next_cursor
+    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
 
 
 def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:

app/admin/routers/ad_revenue.py

@@ -0,0 +1,75 @@
+"""admin 广告收益报表:按 用户/日期/广告类型/应用/代码位 聚合 展示条数 / 收益 / 金币。
+
+任意已登录 admin 可看(只读,不涉及资金操作)。聚合逻辑在 app/admin/repositories/ad_revenue.py。
+"""
+from __future__ import annotations
+
+from datetime import date as _date
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import ad_revenue
+from app.admin.schemas.ad_revenue import AdRevenueDaily, AdRevenueReportOut, AdRevenueRow
+from app.core.rewards import cn_today
+
+router = APIRouter(
+    prefix="/admin/api/ad-revenue-report",
+    tags=["admin-ad-revenue-report"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+# 区间最大跨度(天);超出拒绝,避免审计页一次拉过多天(逐日审计 + 大查询)拖垮接口。
+_MAX_RANGE_DAYS = 92
+
+
+def _parse_day(value: str | None, *, field: str, default: _date) -> _date:
+    if value is None:
+        return default
+    try:
+        return _date.fromisoformat(value)
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
+
+
+@router.get("", response_model=AdRevenueReportOut, summary="广告收益报表(按 日期区间/用户/类型/应用/代码位 聚合)")
+def get_ad_revenue_report(
+    db: AdminDb,
+    date_from: Annotated[str | None, Query(description="起始日 北京时间 YYYY-MM-DD,默认今天")] = None,
+    date_to: Annotated[str | None, Query(description="结束日 北京时间 YYYY-MM-DD,闭区间,默认=date_from")] = None,
+    user_id: Annotated[int | None, Query(description="只看某用户;不传=全部用户")] = None,
+    ad_type: Annotated[
+        str | None,
+        Query(description="reward_video / feed / draw;不传=全部类型"),
+    ] = None,
+    granularity: Annotated[
+        str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
+    ] = "day",
+    limit: Annotated[int, Query(ge=1, le=1000)] = 500,
+) -> AdRevenueReportOut:
+    today = cn_today()
+    d_from = _parse_day(date_from, field="date_from", default=today)
+    d_to = _parse_day(date_to, field="date_to", default=d_from)
+    if d_to < d_from:
+        raise HTTPException(status_code=422, detail="date_to 不能早于 date_from")
+    if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS:
+        raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天")
+
+    result = ad_revenue.ad_revenue_report(
+        db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
+        user_id=user_id, ad_type=ad_type, granularity=granularity, limit=limit,
+    )
+    return AdRevenueReportOut(
+        date_from=d_from.isoformat(),
+        date_to=d_to.isoformat(),
+        daily=[AdRevenueDaily(**d) for d in result["daily"]],
+        total=result["total"],
+        truncated=result["truncated"],
+        total_impressions=result["total_impressions"],
+        total_revenue_yuan=result["total_revenue_yuan"],
+        total_expected_coin=result["total_expected_coin"],
+        total_actual_coin=result["total_actual_coin"],
+        mismatch_count=result["mismatch_count"],
+        items=[AdRevenueRow(**r) for r in result["items"]],
+    )

app/admin/routers/feedback.py

@@ -34,7 +34,7 @@ def list_feedbacks(
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[FeedbackOut]:
-    items, next_cursor = queries.list_feedbacks(
+    items, next_cursor, total = queries.list_feedbacks(
         db,
         status=status,
         user_id=user_id,
@@ -47,7 +47,9 @@ def list_feedbacks(
         cursor=cursor,
     )
     return CursorPage(
-        items=[FeedbackOut.model_validate(f) for f in items], next_cursor=next_cursor,
+        items=[FeedbackOut.model_validate(f) for f in items],
+        next_cursor=next_cursor,
+        total=total,
     )
 
 

app/admin/schemas/ad_revenue.py

@@ -0,0 +1,90 @@
+"""广告收益报表 schemas。
+
+按 用户 / 日期 / 广告类型 / 应用 / 代码位 聚合的只读报表:展示条数、收益(元)、金币、来源。
+字段 snake_case;收益按元(float),金币按整数。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class AdRevenueImpression(BaseModel):
+    """聚合行下钻的单条**展示**明细(每次广告展示一条,展开该组时展示)。"""
+
+    id: int = Field(..., description="ad_ecpm_record 主键")
+    created_at: datetime
+    ecpm: str = Field(..., description="本次展示 eCPM 原始值(分/千次展示)")
+    revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000")
+    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…)")
+    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID)")
+
+
+class AdRevenueRecord(BaseModel):
+    """聚合行下钻的单条发奖复算明细(与金币审计同源,展开该组时展示)。"""
+
+    record_id: int
+    created_at: datetime
+    status: str = Field(..., description="granted / capped / ecpm_missing")
+    ecpm: str | None = Field(None, description="本次采用的 eCPM 原始值(分/千次展示)")
+    ecpm_factor: float | None = Field(None, description="因子1(eCPM 档);非 granted 为空")
+    units: int = Field(..., description="折算份数:激励视频恒 1;信息流 = 满 10 秒份数")
+    lt_index_start: int | None = Field(None, description="本条占用「账号累计第几份」的起")
+    lt_index_end: int | None = Field(None, description="本条占用「账号累计第几份」的止;激励视频 = 起")
+    lt_factor_start: float | None = Field(None, description="因子2(LT)起值")
+    lt_factor_end: float | None = Field(None, description="因子2(LT)止值;激励视频 = 起")
+    expected_coin: int = Field(..., description="按公式复算应发金币")
+    actual_coin: int = Field(..., description="实际入账金币")
+    matched: bool = Field(..., description="复算与实发是否一致")
+
+
+class AdRevenueDaily(BaseModel):
+    """按日期汇总的一天(供前端按天趋势图;全量,不受 limit 影响)。"""
+
+    date: str = Field(..., description="北京时间 YYYY-MM-DD")
+    impressions: int = Field(..., description="当天展示条数合计")
+    revenue_yuan: float = Field(..., description="当天预估收益合计(元)")
+    expected_coin: int = Field(..., description="当天应发金币合计")
+    actual_coin: int = Field(..., description="当天实发金币合计")
+
+
+class AdRevenueRow(BaseModel):
+    """一个聚合组(report_date × user × ad_type × app_env × our_code_id)的汇总。"""
+
+    report_date: str = Field(..., description="该组所属日期(北京时间 YYYY-MM-DD)")
+    user_id: int
+    ad_type: str = Field(..., description="reward_video(激励视频) / feed(信息流) / draw(历史 Draw 信息流)")
+    app_env: str | None = Field(None, description="我们的应用:prod(傻瓜比价正式) / test(测试应用);旧数据为空")
+    our_code_id: str | None = Field(None, description="我们后台配置的代码位 ID(104xxx);旧数据为空")
+    hour: int | None = Field(None, description="北京时间小时 0–23(granularity=hour 时有值;按天为 null)")
+    impressions: int = Field(..., description="展示条数(每条广告展示一条;轮播每条各计一次)")
+    revenue_yuan: float = Field(..., description="收益(元)= Σ(eCPM元 ÷ 1000);测试应用多为 0")
+    expected_coin: int = Field(..., description="应发金币(按公式复算,与金币审计同源)")
+    actual_coin: int = Field(..., description="实发金币(实际入账,按现发奖算法)")
+    matched: bool = Field(..., description="该组应发==实发(组内任一条不符则 false)")
+    adns: list[str] = Field(default_factory=list, description="实际填充的底层 ADN 子渠道集合(如 pangle/gdt)")
+    impression_records: list[AdRevenueImpression] = Field(
+        default_factory=list,
+        description="该组逐条展示明细(时间/eCPM/收益/adn);展开下钻用,无发奖也有(只要有展示)",
+    )
+    records: list[AdRevenueRecord] = Field(
+        default_factory=list,
+        description="该组逐条发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);展开下钻用,纯展示无发奖记录的组为空",
+    )
+
+
+class AdRevenueReportOut(BaseModel):
+    """报表响应:全量统计 + 按天趋势 + 聚合明细。"""
+
+    date_from: str = Field(..., description="报表起始日期(北京时间 YYYY-MM-DD)")
+    date_to: str = Field(..., description="报表结束日期(北京时间 YYYY-MM-DD,闭区间;单日时与 date_from 相同)")
+    daily: list[AdRevenueDaily] = Field(..., description="按日期汇总序列(全量,供按天趋势图)")
+    total: int = Field(..., description="聚合组总数(全量,不受 limit 影响)")
+    truncated: bool = Field(..., description="明细是否被 limit 截断")
+    total_impressions: int = Field(..., description="全量展示条数合计")
+    total_revenue_yuan: float = Field(..., description="全量收益合计(元)")
+    total_expected_coin: int = Field(..., description="全量应发金币合计")
+    total_actual_coin: int = Field(..., description="全量实发金币合计")
+    mismatch_count: int = Field(..., description="应发≠实发的组数(=0 说明全部按公式发放)")
+    items: list[AdRevenueRow] = Field(..., description="聚合明细(按 用户→类型→代码位 排序)")

app/api/v1/ad.py

@@ -242,10 +242,12 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
         ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
         ad_session_id=payload.ad_session_id,
         adn=payload.adn, slot_id=payload.slot_id,
+        app_env=payload.app_env, our_code_id=payload.our_code_id,
     )
     logger.info(
-        "ad ecpm report user_id=%d type=%s session=%s ecpm=%s adn=%s slot=%s",
+        "ad ecpm report user_id=%d type=%s session=%s ecpm=%s adn=%s slot=%s app=%s code=%s",
         user.id, payload.ad_type, payload.ad_session_id, payload.ecpm, payload.adn, payload.slot_id,
+        payload.app_env, payload.our_code_id,
     )
     return EcpmReportOut(ok=True)
 
@@ -359,6 +361,8 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         ad_session_id=payload.ad_session_id,
         adn=payload.adn,
         slot_id=payload.slot_id,
+        app_env=payload.app_env,
+        our_code_id=payload.our_code_id,
     )
     logger.info(
         "feed ad reward user_id=%d event=%s status=%s units=%d coin=%d",

app/api/v1/feedback.py

@@ -21,7 +21,7 @@ logger = logging.getLogger("shagua.feedback")
 router = APIRouter(prefix="/api/v1/feedback", tags=["feedback"])
 
 _MAX_IMAGES = 6
-_CONTENT_MAX = 2000
+_CONTENT_MAX = 200
 _CONTACT_MAX = 128
 
 

app/models/ad_ecpm.py

@@ -35,6 +35,11 @@ class AdEcpmRecord(Base):
     adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 实际展示用的代码位(底层 mediation rit,非客户端配置位)
     slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 我们的穿山甲应用环境:prod(傻瓜比价正式应用) / test(测试应用)。客户端按 AdConfig.useProductionApp 上报。
+    # 与底层 adn 不同:这是「我们用的是哪个 App」,adn 是「聚合后实际填充的子渠道」。旧数据为 NULL。
+    app_env: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    # 我们在穿山甲后台配置的代码位 ID(AdConfig.feedCodeId/rewardCodeId 返回的 104xxx,**非** slot_id 的底层 rit)。旧数据为 NULL。
+    our_code_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
     # 客户端上报的 eCPM 原始字符串(单位:分/千次展示,SDK getEcpm 原值,原样存)
     ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
     # 北京时间日期串 'YYYY-MM-DD',按它等值做"按天聚合"(不在 SQL 里做跨时区 date 比较)

app/models/ad_feed_reward.py

@@ -30,6 +30,9 @@ class AdFeedRewardRecord(Base):
     ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
     adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
     slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx,由客户端 feed-reward 上报带上。旧数据为 NULL。
+    app_env: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    our_code_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
     coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="granted")
 

app/models/ad_reward.py

@@ -33,6 +33,10 @@ class AdRewardRecord(Base):
     ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 本次发奖采用的 eCPM 原始值(回调自带或按 ad_session_id 匹配的客户端上报)
     ecpm_raw: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx。
+    # S2S 回调本身不带这俩,发奖时按 ad_session_id 匹配 ad_ecpm_record 回填(查不到为 NULL)。
+    app_env: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    our_code_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
     # 北京时间日期串 'YYYY-MM-DD',按它等值统计当日发奖次数
     reward_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
     # 穿山甲上报的奖励名(参考,不作发奖依据)

app/repositories/ad_ecpm.py

@@ -23,8 +23,14 @@ def create_ecpm_record(
     ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
+    app_env: str | None = None,
+    our_code_id: str | None = None,
 ) -> AdEcpmRecord:
-    """落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。"""
+    """落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。
+
+    app_env(prod/test)与 our_code_id(我们后台配置的 104xxx 代码位)供广告收益报表按
+    应用/代码位聚合;与 adn(实际填充子渠道)/slot_id(底层 rit)是两组不同口径。
+    """
     if ad_session_id:
         existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
         if existing is not None:
@@ -35,6 +41,8 @@ def create_ecpm_record(
         ad_session_id=ad_session_id,
         adn=adn,
         slot_id=slot_id,
+        app_env=app_env,
+        our_code_id=our_code_id,
         ecpm_raw=ecpm_raw,
         report_date=cn_today().isoformat(),
     )

app/repositories/ad_feed_reward.py

@@ -69,6 +69,8 @@ def grant_feed_reward(
     ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
+    app_env: str | None = None,
+    our_code_id: str | None = None,
 ) -> AdFeedRewardRecord:
     """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。
 
@@ -96,6 +98,8 @@ def grant_feed_reward(
             ecpm_raw=ecpm,
             adn=adn,
             slot_id=slot_id,
+            app_env=app_env,
+            our_code_id=our_code_id,
             coin=0,
             status="capped",
         )
@@ -118,6 +122,8 @@ def grant_feed_reward(
         ecpm_raw=ecpm,
         adn=adn,
         slot_id=slot_id,
+        app_env=app_env,
+        our_code_id=our_code_id,
         coin=coin,
         status="granted",
     )

app/repositories/ad_reward.py

@@ -90,6 +90,16 @@ def grant_ad_reward(
 
     today = cn_today().isoformat()
 
+    # 按 ad_session_id 匹配客户端 eCPM 上报:既用于缺 eCPM 时回退取值,也把「来源」
+    # (我们的应用 app_env + 我们配置的代码位 our_code_id)回填到发奖记录,供广告收益报表聚合。
+    # S2S 回调本身不带这俩;查不到(未上报 eCPM)则留空。
+    ecpm_rec = (
+        crud_ecpm.find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        if ad_session_id else None
+    )
+    src_app_env = ecpm_rec.app_env if ecpm_rec is not None else None
+    src_code_id = ecpm_rec.our_code_id if ecpm_rec is not None else None
+
     # #3 每日上限:当前产品只保留发奖次数上限(默认 500 次)。旧的观看时长闸保留字段,
     # 但 DAILY_AD_WATCH_SECONDS_LIMIT=0 时视为停用,不能命中 capped。
     over_time = (
@@ -102,19 +112,18 @@ def grant_ad_reward(
             trans_id=trans_id, user_id=user_id, coin=0, status="capped",
             reward_date=today, reward_name=reward_name, raw=raw,
             reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm,
+            app_env=src_app_env, our_code_id=src_code_id,
         )
         return _commit_record(db, rec, trans_id)
 
-    ecpm_raw = ecpm
-    if not ecpm_raw and ad_session_id:
-        ecpm_rec = crud_ecpm.find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
-        ecpm_raw = ecpm_rec.ecpm_raw if ecpm_rec is not None else None
+    ecpm_raw = ecpm or (ecpm_rec.ecpm_raw if ecpm_rec is not None else None)
 
     if not ecpm_raw:
         rec = AdRewardRecord(
             trans_id=trans_id, user_id=user_id, coin=0, status="ecpm_missing",
             reward_date=today, reward_name=reward_name, raw=raw,
             reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=None,
+            app_env=src_app_env, our_code_id=src_code_id,
         )
         return _commit_record(db, rec, trans_id)
 
@@ -131,6 +140,7 @@ def grant_ad_reward(
         trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
         reward_date=today, reward_name=reward_name, raw=raw,
         reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm_raw,
+        app_env=src_app_env, our_code_id=src_code_id,
     )
     return _commit_record(db, rec, trans_id)
 

app/schemas/ad.py

@@ -57,6 +57,12 @@ class EcpmReportIn(BaseModel):
     )
     adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
     slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
+    app_env: str | None = Field(
+        None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
+    )
+    our_code_id: str | None = Field(
+        None, max_length=64, description="我们后台配置的代码位 ID(AdConfig 的 104xxx,非底层 rit)"
+    )
 
 
 class EcpmReportOut(BaseModel):
@@ -128,6 +134,12 @@ class FeedRewardIn(BaseModel):
     duration_seconds: int = Field(..., ge=0, description="本条广告实际展示/播放秒数")
     adn: str | None = Field(None, description="实际投放 ADN")
     slot_id: str | None = Field(None, description="实际展示代码位")
+    app_env: str | None = Field(
+        None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)"
+    )
+    our_code_id: str | None = Field(
+        None, max_length=64, description="我们后台配置的代码位 ID(AdConfig 的 104xxx,非底层 rit)"
+    )
 
 
 class FeedRewardOut(BaseModel):