feat(ad): 广告收益支持环境与业务代码位筛选

app/admin/repositories/ad_revenue.py

@@ -33,7 +33,29 @@ 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
+from app.repositories import ad_pangle_revenue, app_config
+
+# 已上线过的正式业务代码位要永久保留，避免运营切换当前配置后，历史报表把旧业务位误判成测试流量。
+_KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
+
+# 测试应用中实际承载业务链路的代码位。广告测试 demo 的插屏/半屏/信息流测试位不在这里，
+# 避免“业务口径”把开发诊断曝光混进客户端与穿山甲对账。
+_TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
+
+
+def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
+    """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
+    prod_config = app_config.get_ad_config(db)
+    prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
+        str(prod_config.get(key) or "").strip()
+        for key in ("reward_code_id", "compare_draw_code_id", "coupon_draw_code_id")
+    }
+    prod_ids.discard("")
+    if app_env == "prod":
+        return prod_ids
+    if app_env == "test":
+        return set(_TEST_BUSINESS_CODE_IDS)
+    return prod_ids | set(_TEST_BUSINESS_CODE_IDS)
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -82,6 +104,7 @@ def ad_revenue_report(
     ad_type: str | None = None,
     feed_scene: str | None = None,
     app_env: str | None = None,
+    revenue_scope: str = "all",
     granularity: str = "day",
     limit: int = 500,
     offset: int = 0,
@@ -277,14 +300,18 @@ def ad_revenue_report(
     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 是真实小额、非客户端那种假值)。
+    # app_env 过滤:显式传 "prod"/"test" 只看该环境;不传=全部。该参数也会传给下方穿山甲聚合，
+    # 保证客户端预估与 GroMore 汇总使用同一应用环境口径。
     if app_env is not None:
         events = [e for e in events if e.get("app_env") == app_env]
 
+    # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
+    # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
+    business_code_ids: set[str] | None = None
+    if revenue_scope == "business":
+        business_code_ids = _business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_code_ids]
+
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
     if sort == "ecpm":
@@ -336,7 +363,13 @@ def ad_revenue_report(
     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)
+        pangle_aggs = ad_pangle_revenue.aggregate_by_date(
+            db,
+            date_from=date_from,
+            date_to=date_to,
+            app_env=app_env,
+            our_code_ids=business_code_ids,
+        )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}
             for d in daily:

app/admin/routers/ad_revenue.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 from datetime import date as _date
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 
@@ -63,6 +63,13 @@ def get_ad_revenue_report(
             "建议正式收益报表选 prod,避免测试应用的假 eCPM 污染收益合计/平均"
         ),
     ] = None,
+    revenue_scope: Annotated[
+        Literal["business", "all"],
+        Query(
+            description="business=仅业务代码位（用于客户端与穿山甲同口径对账）/ "
+            "all=穿山甲应用全部代码位（包含广告测试等非业务曝光）"
+        ),
+    ] = "all",
     granularity: Annotated[
         str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
     ] = "day",
@@ -83,6 +90,7 @@ def get_ad_revenue_report(
     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, feed_scene=feed_scene, app_env=app_env,
+        revenue_scope=revenue_scope,
         granularity=granularity, limit=limit, offset=offset, sort=sort,
     )
     return AdRevenueReportOut(