修复用户反馈机型与系统版本补全 (#181)

## 改动说明
- 反馈列表按同一用户、同一设备编码和反馈提交时间，补全可读机型、厂商及 ROM 大版本
- 补充常见线上机型编码映射
- 更新本地 mock 数据并增加时间边界回归测试

## 验证
- 反馈相关测试：3 项通过
- Ruff 检查通过

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #181
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"]),
@@ -427,38 +431,6 @@ 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;
@@ -513,7 +485,6 @@ 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],

app/admin/repositories/queries.py

@@ -9,7 +9,7 @@ from datetime import date, datetime, time, timedelta, timezone
 from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import Select, asc, case, desc, func, or_, select
+from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -45,6 +45,72 @@ _FEED_SCENE_LABEL = {
 }
 
 
+_DEVICE_MARKETING_NAMES = {
+    "23078RKD5C": "Redmi K60 至尊版",
+    "M2012K11AC": "Redmi K40",
+    "PJA110": "一加 Ace 2 Pro",
+    "PPG-AN00": "荣耀 GT Pro",
+    "V2166BA": "vivo Y77e",
+    "V2309A": "vivo X100",
+}
+
+
+def _device_marketing_name(model: str | None) -> str | None:
+    """把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
+    if not model:
+        return None
+    return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
+
+
+def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
+    """按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
+    candidates = [
+        item for item in feedbacks if item.device_model and item.device_model.strip()
+    ]
+    for item in candidates:
+        item.device_model_name = _device_marketing_name(item.device_model)
+        item.device_manufacturer = None
+        item.rom_version = None
+    if not candidates:
+        return
+
+    ranked = (
+        select(
+            Feedback.id.label("feedback_id"),
+            ComparisonRecord.device_manufacturer.label("device_manufacturer"),
+            ComparisonRecord.rom_version.label("rom_version"),
+            func.row_number()
+            .over(
+                partition_by=Feedback.id,
+                order_by=(
+                    ComparisonRecord.created_at.desc(),
+                    ComparisonRecord.id.desc(),
+                ),
+            )
+            .label("row_num"),
+        )
+        .join(
+            ComparisonRecord,
+            and_(
+                ComparisonRecord.user_id == Feedback.user_id,
+                ComparisonRecord.device_model == Feedback.device_model,
+                ComparisonRecord.created_at <= Feedback.created_at,
+            ),
+        )
+        .where(Feedback.id.in_([item.id for item in candidates]))
+        .subquery()
+    )
+    details = {
+        row.feedback_id: row
+        for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
+    }
+
+    for item in candidates:
+        detail = details.get(item.id)
+        item.device_manufacturer = detail.device_manufacturer if detail else None
+        item.rom_version = detail.rom_version if detail else None
+
+
 def cursor_paginate(
     db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
 ) -> tuple[list, int | None]:
@@ -817,6 +883,7 @@ def list_feedbacks(
         db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
     )
     _attach_user_info(db, items)  # 列表展示完整手机号(点手机号查该用户全部反馈)
+    _attach_feedback_device_details(db, items)
     return items, next_cursor, total
 
 
@@ -1228,6 +1295,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 +1391,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):
@@ -73,7 +75,6 @@ class AdRevenueTypeStat(BaseModel):
 
     impressions: int = Field(..., description="该类型展示条数合计")
     revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
-    ecpm_yuan: float | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -143,10 +144,6 @@ 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}};"

app/admin/schemas/feedback.py

@@ -32,7 +32,10 @@ class FeedbackOut(BaseModel):
     # 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
     app_version: str | None = None
     device_model: str | None = None
+    device_model_name: str | None = None
+    device_manufacturer: str | None = None
     rom_name: str | None = None
+    rom_version: int | None = None
     android_version: str | None = None
     # 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
     phone: str | None = None

app/repositories/comparison.py

@@ -55,13 +55,22 @@ def _product_names_from_items(items: list | None) -> str | None:
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
+    _pr = payload.platform_results or {}
 
-    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
-    best = None
+    def _is_short(r) -> bool:
+        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
+        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
+        info = _pr.get(r.platform_id) if r.platform_id else None
+        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
+
+    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
+    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
     priced = [r for r in results if r.price is not None]
-    if priced:
+    clean = [r for r in priced if not _is_short(r)]
+    best = None
+    if clean:
         best = min(
-            priced,
+            clean,
             key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
         )
 
@@ -196,14 +205,29 @@ def upsert_record(
 # ============================================================
 
 
-def _derive_from_results(results: list[dict]) -> dict:
+def _derive_from_results(
+    results: list[dict], platform_results: dict | None = None
+) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
+
+    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
+    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
+    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
+    _pr = platform_results or {}
+
+    def _is_short(r: dict) -> bool:
+        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
+        pid = r.get("platform_id")
+        info = _pr.get(pid) if pid else None
+        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
+
     priced = [r for r in results if r.get("price") is not None]
+    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
     best = None
-    if priced:
+    if clean:
         best = min(
-            priced,
+            clean,
             key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
         )
     src_row = next((r for r in results if r.get("is_source")), None)
@@ -389,7 +413,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results)
+    derived = _derive_from_results(results, done_params.get("platform_results"))
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(

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

scripts/seed_mock_feedback.py

@@ -175,22 +175,22 @@ def seed(db) -> list[Feedback]:
         # 普通反馈 · 新端(带环境快照)· 无图
         fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
            source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
+           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS 14", android_version="14",
            created=ago(minutes=6)),
         # 比价反馈 · scene=优惠不对 · 新端 · 2 图
         fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
            source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
-           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
+           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
            created=ago(minutes=22)),
         # 比价反馈 · scene=找错商品 · 新端 · 无图
         fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
            source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
+           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", android_version="14",
            created=ago(hours=1)),
         # 普通反馈 · 新端 · 1 图(表扬 + 小问题)
         fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
            source="profile", images=[imgs[2]],
-           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
+           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI 14", android_version="14",
            created=ago(hours=3)),
         # 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
         fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
@@ -206,7 +206,7 @@ def seed(db) -> list[Feedback]:
            source="profile", images=[imgs[3]],
            status="adopted", reward_coins=2000,
            review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
-           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
+           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS 13", android_version="13",
            created=ago(days=2)),
 
         # ===== 未采纳 rejected(带原因 + 回复)=====
@@ -214,7 +214,7 @@ def seed(db) -> list[Feedback]:
            source="comparison", scene="价格不准",
            status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
            admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
-           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
+           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI 14", android_version="13",
            created=ago(days=3)),
     ]
     db.add_all(feedbacks)

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
@@ -210,54 +211,110 @@ 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:
+def test_feed_reward_details_keep_each_record_adn() -> None:
     db = SessionLocal()
-    phone = "18800009993"
-    category_date = "2040-02-05"
+    phone = "18800009994"
+    detail_date = "2040-02-06"
     try:
-        user = User(phone=phone, username="29999999993", register_channel="sms")
+        user = User(phone=phone, username="29999999994", 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),
+            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),
             ),
-            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),
+            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=category_date,
-            date_to=category_date,
-            user_id=user.id,
-            app_env="prod",
-            revenue_scope="all",
+            db, date_from=detail_date, date_to=detail_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},
-        }
+        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(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        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_feedback_device_details.py

@@ -0,0 +1,71 @@
+from datetime import datetime, timedelta, timezone
+
+from app.admin.repositories import queries
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
+from app.models.feedback import Feedback
+from app.repositories import user as user_repo
+
+
+def test_feedback_device_details_use_same_user_and_model() -> None:
+    with SessionLocal() as db:
+        submitted_at = datetime.now(timezone.utc)
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone="13800009876",
+            register_channel="sms",
+        )
+        db.add_all(
+            [
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-v2166ba",
+                    device_model="V2166BA",
+                    device_manufacturer="vivo",
+                    rom_name="OriginOS",
+                    rom_version=13,
+                    created_at=submitted_at - timedelta(minutes=5),
+                ),
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-other",
+                    device_model="OTHER-CODE",
+                    device_manufacturer="Other",
+                    rom_name="OtherOS",
+                    rom_version=99,
+                    created_at=submitted_at - timedelta(minutes=5),
+                ),
+                ComparisonRecord(
+                    user_id=user.id,
+                    trace_id="feedback-device-future-upgrade",
+                    device_model="V2166BA",
+                    device_manufacturer="vivo-new",
+                    rom_name="OriginOS",
+                    rom_version=99,
+                    created_at=submitted_at + timedelta(minutes=5),
+                ),
+                Feedback(
+                    user_id=user.id,
+                    content="设备信息补全测试",
+                    contact="",
+                    status="pending",
+                    device_model="V2166BA",
+                    rom_name="OriginOS",
+                    android_version="13",
+                    created_at=submitted_at,
+                ),
+            ]
+        )
+        db.commit()
+
+        items, _next_cursor, total = queries.list_feedbacks(
+            db,
+            user_id=user.id,
+            limit=20,
+        )
+
+        assert total == 1
+        assert items[0].device_model == "V2166BA"
+        assert items[0].device_model_name == "vivo Y77e"
+        assert items[0].device_manufacturer == "vivo"
+        assert items[0].rom_version == 13

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

tests/test_derive_best_excludes_short.py

@@ -0,0 +1,102 @@
+"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
+
+回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
+会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
+修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
+纯函数, 不碰 DB。
+"""
+from app.repositories.comparison import _derive, _derive_from_results
+from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
+
+
+def test_derive_from_results_excludes_short_ordered_from_best():
+    # jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
+    results = [
+        {"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
+        {"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
+        {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
+    ]
+    platform_results = {
+        "jd_waimai": {"skipped_dish_count": 2},
+        "taobao_flash": {"skipped_dish_count": 0},
+        "meituan": {"is_source": True},
+    }
+    d = _derive_from_results(results, platform_results)
+    assert d["best_platform_id"] == "taobao_flash"
+    assert d["best_price_cents"] == 3850
+    assert d["saved_amount_cents"] == 4200 - 3850   # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
+    assert d["is_source_best"] is False
+
+
+def test_derive_from_results_all_targets_short_falls_back_to_source():
+    # 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
+    d = _derive_from_results(results, platform_results)
+    assert d["best_platform_id"] == "meituan"
+    assert d["is_source_best"] is True
+    assert d["saved_amount_cents"] == 0
+
+
+def test_derive_from_results_no_platform_results_keeps_old_behavior():
+    # 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    d = _derive_from_results(results)
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500
+
+
+def test_derive_from_results_malformed_platform_results_no_crash():
+    # 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    d = _derive_from_results(results, {"jd_waimai": "oops"})
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500
+
+
+def test_derive_pydantic_excludes_short():
+    # _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
+    payload = ComparisonRecordIn(
+        trace_id="t-short-pyd",
+        source_price=42.0,
+        source_platform_id="meituan",
+        comparison_results=[
+            ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
+            ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
+            ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
+        ],
+        platform_results={
+            "jd_waimai": {"skipped_dish_count": 2},
+            "taobao_flash": {"skipped_dish_count": 0},
+        },
+    )
+    d = _derive(payload)
+    assert d["best_platform_id"] == "taobao_flash"
+    assert d["best_price_cents"] == 3850
+    assert d["saved_amount_cents"] == 4200 - 3850
+    assert d["is_source_best"] is False
+
+
+def test_derive_pydantic_malformed_platform_results_no_crash():
+    # _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
+    payload = ComparisonRecordIn(
+        trace_id="t-malformed-pyd",
+        source_price=42.0,
+        comparison_results=[
+            ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
+            ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
+        ],
+        platform_results={"jd_waimai": "oops"},
+    )
+    d = _derive(payload)  # 不抛 AttributeError
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500