修复：补齐收益明细广告网络来源

app/admin/repositories/ad_audit.py

@@ -12,10 +12,11 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
+from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
@@ -55,10 +56,21 @@ def _reward_video_rows(
     if user_id is not None:
         stmt = stmt.where(AdRewardRecord.user_id == user_id)
 
+    records = list(db.execute(stmt).scalars())
+    # S2S 发奖回调不携带实际填充 ADN；用相同用户和 ad_session_id 的展示记录回填。
+    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
+    impression_by_session = {
+        (record.user_id, record.ad_session_id): record
+        for record in db.execute(
+            select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id.in_(session_ids))
+        ).scalars()
+    } if session_ids else {}
+
     # 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
     granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
     rows: list[dict] = []
-    for rec in db.execute(stmt).scalars():
+    for rec in records:
+        impression = impression_by_session.get((rec.user_id, rec.ad_session_id))
         if rec.status == "granted":
             nth = granted_n.get(rec.user_id, 0) + 1
             granted_n[rec.user_id] = nth
@@ -68,6 +80,8 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": impression.adn if impression is not None else None,
+                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -90,6 +104,8 @@ def _reward_video_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": impression.adn if impression is not None else None,
+                "slot_id": impression.slot_id if impression is not None else None,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -149,6 +165,78 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
     return True
 
 
+def _nonblank(value: str | None) -> str | None:
+    value = value.strip() if value else None
+    return value or None
+
+
+def _unique_ad_source(records: list[AdEcpmRecord]) -> tuple[str | None, str | None]:
+    """仅在候选展示记录指向唯一 ADN 时回填来源，避免错误归因。"""
+    adns = {_nonblank(record.adn) for record in records}
+    adns.discard(None)
+    if len(adns) != 1:
+        return None, None
+
+    slots = {_nonblank(record.slot_id) for record in records}
+    slots.discard(None)
+    return next(iter(adns)), next(iter(slots)) if len(slots) == 1 else None
+
+
+def _feed_source_fallbacks(
+    db: Session, records: list[AdFeedRewardRecord]
+) -> tuple[
+    dict[tuple[int, str], tuple[str | None, str | None]],
+    dict[tuple[int, str, str], tuple[str | None, str | None]],
+]:
+    """为旧信息流发奖记录构建安全来源索引。"""
+    session_ids = {record.ad_session_id for record in records if record.ad_session_id}
+    trace_ids = {record.trace_id for record in records if record.trace_id}
+    if not session_ids and not trace_ids:
+        return {}, {}
+
+    filters = []
+    if session_ids:
+        filters.append(AdEcpmRecord.ad_session_id.in_(session_ids))
+    if trace_ids:
+        filters.append(AdEcpmRecord.trace_id.in_(trace_ids))
+    impressions = list(db.execute(select(AdEcpmRecord).where(or_(*filters))).scalars())
+
+    by_session: dict[tuple[int, str], list[AdEcpmRecord]] = {}
+    by_trace_ecpm: dict[tuple[int, str, str], list[AdEcpmRecord]] = {}
+    for impression in impressions:
+        if impression.ad_session_id:
+            by_session.setdefault((impression.user_id, impression.ad_session_id), []).append(impression)
+        if impression.trace_id:
+            by_trace_ecpm.setdefault(
+                (impression.user_id, impression.trace_id, impression.ecpm_raw), []
+            ).append(impression)
+
+    return (
+        {key: _unique_ad_source(value) for key, value in by_session.items()},
+        {key: _unique_ad_source(value) for key, value in by_trace_ecpm.items()},
+    )
+
+
+def _feed_source(
+    record: AdFeedRewardRecord,
+    *,
+    by_session: dict[tuple[int, str], tuple[str | None, str | None]],
+    by_trace_ecpm: dict[tuple[int, str, str], tuple[str | None, str | None]],
+) -> tuple[str | None, str | None]:
+    """返回本条发奖广告的来源；无唯一证据时保留原始空值。"""
+    adn, slot_id = _nonblank(record.adn), _nonblank(record.slot_id)
+    if adn and slot_id:
+        return adn, slot_id
+
+    candidate = by_session.get((record.user_id, record.ad_session_id or ""))
+    if candidate is None and record.trace_id:
+        candidate = by_trace_ecpm.get((record.user_id, record.trace_id, record.ecpm_raw))
+    if candidate is None:
+        return adn, slot_id
+    candidate_adn, candidate_slot_id = candidate
+    return adn or candidate_adn, slot_id or candidate_slot_id
+
+
 def _feed_rows(
     db: Session, *, date: str, user_id: int | None, scene: str | None = None
 ) -> list[dict]:
@@ -167,11 +255,17 @@ def _feed_rows(
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
+    records = list(db.execute(stmt).scalars())
+    by_session, by_trace_ecpm = _feed_source_fallbacks(db, records)
+
     # 本日之前的累计**条数**做起点,与发奖侧 granted_unit_total(COUNT granted)对齐
     granted_count: dict[int, int] = _feed_prior_granted_count(db, date=date, user_id=user_id)
     rows: list[dict] = []
-    for rec in db.execute(stmt).scalars():
+    for rec in records:
         keep = _feed_scene_matches(rec, scene)  # 累计照常推进,这里只决定是否展示本行
+        adn, slot_id = _feed_source(
+            rec, by_session=by_session, by_trace_ecpm=by_trace_ecpm
+        )
         if rec.status == "granted":
             # 一条广告 = 1 份(与 grant_feed_reward 同口径:看满一份即发该条满额,不按 unit_count 累加)。
             # nth = 账号累计第几**条**(含本日之前),与发奖侧 granted_unit_total+1 对齐;累计照常推进
@@ -188,6 +282,8 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": adn,
+                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
@@ -214,6 +310,8 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "adn": adn,
+                "slot_id": slot_id,
                 "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,

app/admin/repositories/ad_revenue.py

@@ -96,7 +96,11 @@ _REWARD_DETAIL_KEYS = (
 
 def _reward_detail(row: dict) -> dict:
     """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
-    return {k: row[k] for k in _REWARD_DETAIL_KEYS}
+    detail = {key: row[key] for key in _REWARD_DETAIL_KEYS}
+    # 聚合父行可能包含多个 ADN，来源必须保留在每一条发奖明细上。
+    detail["adn"] = row.get("adn")
+    detail["slot_id"] = row.get("slot_id")
+    return detail
 
 
 def ad_revenue_report(
@@ -244,8 +248,8 @@ def ad_revenue_report(
                 "impressions": 0,
                 "ecpm": row["ecpm"],
                 "revenue_yuan": 0.0,
-                "adn": None,
-                "slot_id": None,
+                "adn": row.get("adn"),
+                "slot_id": row.get("slot_id"),
                 "has_reward": True,
                 "status": row["status"],
                 "expected_coin": int(row["expected_coin"]),

app/admin/schemas/ad_revenue.py

@@ -40,6 +40,8 @@ class AdRevenueRecord(BaseModel):
     expected_coin: int = Field(..., description="按公式复算应发金币")
     actual_coin: int = Field(..., description="实际入账金币")
     matched: bool = Field(..., description="复算与实发是否一致")
+    adn: str | None = Field(None, description="本条发奖对应的实际填充 ADN 子渠道")
+    slot_id: str | None = Field(None, description="本条发奖对应的底层 mediation rit")
 
 
 class AdRevenueDaily(BaseModel):

tests/test_admin_ad_revenue_scope.py

@@ -8,6 +8,7 @@ from sqlalchemy import delete
 from app.admin.repositories import ad_revenue
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
+from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_pangle_revenue import AdPangleDailyRevenue
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
@@ -208,3 +209,112 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         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"
+    detail_date = "2040-02-06"
+    try:
+        user = User(phone=phone, username="29999999994", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdFeedRewardRecord(
+                client_event_id="detail-adn-pangle", user_id=user.id,
+                reward_date=detail_date, duration_seconds=20, unit_count=1,
+                ecpm_raw="12000", adn="pangle", slot_id="rit-pangle",
+                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
+                app_env="prod", our_code_id="104098712", coin=12, status="granted",
+                created_at=datetime(2040, 2, 6, 1, tzinfo=UTC),
+            ),
+            AdFeedRewardRecord(
+                client_event_id="detail-adn-gdt", user_id=user.id,
+                reward_date=detail_date, duration_seconds=20, unit_count=1,
+                ecpm_raw="25000", adn="gdt", slot_id="rit-gdt",
+                ad_type="draw", feed_scene="coupon", trace_id="detail-adn-trace",
+                app_env="prod", our_code_id="104098712", coin=25, status="granted",
+                created_at=datetime(2040, 2, 6, 2, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db, date_from=detail_date, date_to=detail_date,
+            user_id=user.id, app_env="prod", revenue_scope="all",
+        )
+
+        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
+        assert [detail["adn"] for detail in item["sub_rewards"]] == ["pangle", "gdt"]
+        assert [detail["slot_id"] for detail in item["sub_rewards"]] == ["rit-pangle", "rit-gdt"]
+    finally:
+        db.rollback()
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == detail_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
+def test_feed_reward_source_fallback_requires_unique_trace_and_ecpm() -> None:
+    db = SessionLocal()
+    phone = "18800009995"
+    fallback_date = "2040-02-07"
+    try:
+        user = User(phone=phone, username="29999999995", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdFeedRewardRecord(
+                client_event_id="source-fallback-unique", user_id=user.id,
+                reward_date=fallback_date, duration_seconds=3, unit_count=0,
+                ecpm_raw="4700", ad_session_id="flow-session", trace_id="source-trace",
+                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
+                ad_type="draw", feed_scene="comparison",
+                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
+            ),
+            AdFeedRewardRecord(
+                client_event_id="source-fallback-ambiguous", user_id=user.id,
+                reward_date=fallback_date, duration_seconds=3, unit_count=0,
+                ecpm_raw="4800", ad_session_id="flow-session", trace_id="source-trace",
+                app_env="prod", our_code_id="104098712", coin=0, status="too_short",
+                ad_type="draw", feed_scene="comparison",
+                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-unique",
+                trace_id="source-trace", ecpm_raw="4700", adn="baidu", slot_id="rit-baidu",
+                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                created_at=datetime(2040, 2, 7, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-a",
+                trace_id="source-trace", ecpm_raw="4800", adn="baidu", slot_id="rit-baidu",
+                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                created_at=datetime(2040, 2, 7, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="impression-ambiguous-b",
+                trace_id="source-trace", ecpm_raw="4800", adn="ks", slot_id="rit-ks",
+                app_env="prod", our_code_id="104098712", report_date=fallback_date,
+                created_at=datetime(2040, 2, 7, 2, 1, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db, date_from=fallback_date, date_to=fallback_date,
+            user_id=user.id, app_env="prod", revenue_scope="all",
+        )
+        item = next(row for row in result["items"] if row["event_key"].startswith("feedgrp-"))
+        details = {detail["ecpm"]: detail for detail in item["sub_rewards"]}
+        assert details["4700"]["adn"] == "baidu"
+        assert details["4700"]["slot_id"] == "rit-baidu"
+        assert details["4800"]["adn"] is None
+        assert details["4800"]["slot_id"] is None
+    finally:
+        db.rollback()
+        db.execute(delete(AdFeedRewardRecord).where(AdFeedRewardRecord.reward_date == fallback_date))
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == fallback_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()