合并：解决广告收益 eCPM 口径测试冲突

修复比价记录平均 TOKEN 成本采集与回填 (#182 )
## 问题原因 - 新版客户端通过服务端 harvest 落库，但完成路径没有触发 LLM 调用明细与 TOKEN 成本回填。 - 内部共享密钥不一致或 PriceBot 实例切换后，拉取失败只留下空值，后续没有自动补偿。 ## 本次改动 - harvest 完成后立即异步回填 LLM 调用、TOKEN 数及成本快照。 - 抽取统一、幂等的成本回填服务，并增加定时补偿 worker。 - 增加 PriceBot 内部鉴权预检、错误日志和多实例查找兜底。 - 仅回填当前价格配置生效后的终态记录，避免用现价误算更早历史数据。 - 补充环境配置、部署说明和单元测试。 ## 验证 - 相关测试：36 passed。 - 静态检查通过，diff check 通过。 - 本地页面显示 ¥0.0139，与数据库精确均值 0.013916 的四舍五入结果一致。 ## 上线注意部署时需确保 app-server 与 PriceBot 的 INTERNAL_API_SECRET 完全一致并重启两个服务；worker 启动后会自动补齐符合条件的历史空值。 --------- Co-authored-by: guke <guke@wonderable.ai> Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #182 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-27 16:00:17 +08:00 · 2026-07-27 15:51:19 +08:00 · 2026-07-26 19:58:00 +08:00
3 changed files with 91 additions and 0 deletions
@@ -431,6 +431,38 @@ def ad_revenue_report(
        for k, v in type_map.items()
    }

+    # 经营看板的两类 eCPM 必须按真实展示的 SDK eCPM 加权，不能用发奖状态修正后的
+    # 收益反推。Draw 包含新 draw 与历史 feed；看视频包含福利与提现视频。
+    category_map: dict[str, dict] = {}
+    for event in events:
+        category = (
+            "draw" if event["ad_type"] in {"draw", "feed"}
+            else "video" if event["ad_type"] in {"reward_video", "withdrawal_video"}
+            else None
+        )
+        if category is None:
+            continue
+        stat = category_map.setdefault(category, {
+            "impressions": 0,
+            "revenue_yuan": 0.0,
+            "ecpm_fen_sum": 0.0,
+        })
+        impressions = int(event["impressions"])
+        stat["impressions"] += impressions
+        stat["revenue_yuan"] += event["revenue_yuan"]
+        stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(event["ecpm"]) * impressions
+    category_stats = {
+        key: {
+            "impressions": value["impressions"],
+            "revenue_yuan": round(value["revenue_yuan"], 6),
+            "ecpm_yuan": round(
+                value["ecpm_fen_sum"] / value["impressions"] / 100.0,
+                6,
+            ) if value["impressions"] else 0.0,
+        }
+        for key, value in category_map.items()
+    }
+
    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
@@ -485,6 +517,7 @@ def ad_revenue_report(
        "daily": daily,
        "hourly": hourly,
        "type_stats": type_stats,
+        "category_stats": category_stats,
        "scene_stats": scene_stats,
        "dau": dau,
        "items": main_rows[offset:offset + limit],
@@ -75,6 +75,7 @@ class AdRevenueTypeStat(BaseModel):

    impressions: int = Field(..., description="该类型展示条数合计")
    revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+    ecpm_yuan: float | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")


 class AdRevenueRow(BaseModel):
@@ -144,6 +145,10 @@ class AdRevenueReportOut(BaseModel):
        default_factory=dict,
        description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
    )
+    category_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
+    )
    scene_stats: dict[str, AdRevenueTypeStat] = Field(
        default_factory=dict,
        description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"
@@ -211,6 +211,59 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
        db.close()


+def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
+    db = SessionLocal()
+    phone = "18800009993"
+    category_date = "2040-02-05"
+    try:
+        user = User(phone=phone, username="29999999993", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=category_date,
+            date_to=category_date,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        assert result["category_stats"] == {
+            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
+            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
 def test_feed_reward_details_keep_each_record_adn() -> None:
    db = SessionLocal()
    phone = "18800009994"