Merge branch 'main' into codex/ad-revenue-detail-adn

## 问题原因
- 奖励统计接口用完整 User ORM 判断用户存在，滚动发布或表结构未同步时会因无关字段导致 500
- 金币记录合并广告与签到数据后直接排序，PostgreSQL 中 aware/naive datetime 混排会抛异常

## 修复内容
- 新增只投影 user.id 的用户存在性检查，保持不存在用户返回 404
- 金币记录排序前统一转换为 aware UTC 排序键
- 增加旧表结构投影与混合时区回归测试

## 验证结果
- tests/test_admin_read.py：18 项全部通过
- 线上只读数据库回归：近期 7 个活跃用户统计与金币明细全部正常返回；用户 #33 返回 1514 条记录
- 语法/未定义引用检查通过
- 全量测试 532 通过、9 失败；失败均在未修改的 origin/main 基线上复现，和本 PR 无关

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #180
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>

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/repositories/queries.py

@@ -1228,6 +1228,11 @@ def _cn_wall_to_utc(dt: datetime) -> datetime:
     return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
 
 
+def _coin_record_sort_key(row: dict) -> datetime:
+    """金币明细跨数据源排序键：兼容 SQLite naive 与 PostgreSQL aware 时间。"""
+    return _as_utc(row["created_at"])
+
+
 def user_coin_records(
     db: Session,
     user_id: int,
@@ -1319,7 +1324,10 @@ def user_coin_records(
             "coin": rec.amount,
         })
 
-    rows.sort(key=lambda r: r["created_at"], reverse=True)
+    # SQLite 常返回 naive datetime，PostgreSQL timestamptz 返回 aware datetime；
+    # 统一成 aware UTC 排序，避免线上合并广告记录与签到记录时抛
+    # “can't compare offset-naive and offset-aware datetimes”。
+    rows.sort(key=_coin_record_sort_key, reverse=True)
     has_more = len(rows) > offset + limit
 
     # 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)

app/admin/routers/users.py

@@ -86,7 +86,7 @@ def get_user_reward_stats(
     date_to: Annotated[datetime | None, Query()] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
-    if user_repo.get_user_by_id(db, user_id) is None:
+    if not user_repo.user_exists(db, user_id):
         raise HTTPException(status_code=404, detail="用户不存在")
     return UserRewardStats(
         **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)

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):

app/repositories/user.py

@@ -93,6 +93,11 @@ def get_user_by_id(db: Session, user_id: int) -> User | None:
     return db.get(User, user_id)
 
 
+def user_exists(db: Session, user_id: int) -> bool:
+    """只查主键判断用户是否存在，避免只读统计接口依赖完整用户表结构。"""
+    return db.scalar(select(User.id).where(User.id == user_id)) is not None
+
+
 def get_user_by_phone(db: Session, phone: str) -> User | None:
     stmt = select(User).where(User.phone == phone)
     return db.execute(stmt).scalar_one_or_none()

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()

tests/test_admin_read.py

@@ -1,7 +1,7 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import UTC, datetime
 
 import pytest
 from fastapi.testclient import TestClient
@@ -9,6 +9,7 @@ from sqlalchemy import event
 
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
+from app.admin.repositories import queries
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
@@ -144,6 +145,8 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     ) -> None:
         if "ad_reward_record.boost_round_id" in statement:
             raise AssertionError("提现详情不应查询未使用的 boost_round_id")
+        if "FROM user" in statement and "user.phone" in statement:
+            raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表")
 
     event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
     try:
@@ -162,6 +165,17 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
+def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
+    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
+    naive = datetime(2038, 1, 1, 8, 0)
+    aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC)
+
+    rows = [{"created_at": aware}, {"created_at": naive}]
+    rows.sort(key=queries._coin_record_sort_key, reverse=True)
+
+    assert rows == [{"created_at": naive}, {"created_at": aware}]
+
+
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
     _seed_user_with_data("13800000003")
     r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))