feat(guide-video): 支持领券和比价独立视频奖励配置

.env.example

@@ -139,11 +139,6 @@ PRICEBOT_COMPARE_TIMEOUT_SEC=60
 # 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
 # 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
 INTERNAL_API_SECRET=
-# 新版比价 harvest 完成后即时回填；以下 worker 再补偿短暂故障期间漏掉的记录。
-LLM_COST_BACKFILL_ENABLED=true
-LLM_COST_BACKFILL_INTERVAL_SEC=300
-LLM_COST_BACKFILL_BATCH_SIZE=100
-LLM_COST_BACKFILL_LOOKBACK_DAYS=30
 
 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类

alembic/versions/guide_video_scene_unique.py

@@ -0,0 +1,31 @@
+"""guide video play count is independent for coupon and comparison
+
+Revision ID: guide_video_scene_unique
+Revises: d8dd2106e438, risk_monitor_generic
+"""
+from alembic import op
+
+revision = "guide_video_scene_unique"
+down_revision = ("d8dd2106e438", "risk_monitor_generic")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
+    op.create_index(
+        "uq_guide_video_play_user_scene_seq",
+        "guide_video_play",
+        ["user_id", "scene", "seq"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
+    op.create_index(
+        "uq_guide_video_play_user_seq",
+        "guide_video_play",
+        ["user_id", "seq"],
+        unique=True,
+    )

app/admin/repositories/ad_revenue.py

@@ -431,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;
@@ -517,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, and_, asc, case, desc, func, or_, select
+from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -45,72 +45,6 @@ _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]:
@@ -883,7 +817,6 @@ 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
 
 

app/admin/routers/guide_video.py

@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
 """
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 
@@ -27,14 +27,21 @@ router = APIRouter(
 )
 
 
-def _out(db: AdminDb) -> GuideVideoConfigOut:
+GuideScene = Literal["coupon", "comparison"]
+
+
+def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
     """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
+    return GuideVideoConfigOut(
+        scene=scene,
+        **guide_video.get_config(db, scene),
+        **guide_video.play_stats(db, scene),
+    )
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb) -> GuideVideoConfigOut:
-    return _out(db)
+def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
+    return _out(db, scene)
 
 
 @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +50,23 @@ def update_config(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     before, after = guide_video.update_config(
         db,
         enabled=body.enabled,
         max_plays=body.max_plays,
         reward_coin=body.reward_coin,
+        scene=scene,
         admin_id=admin.id,
         commit=False,
     )
     write_audit(
         db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    return _out(db)
+    return _out(db, scene)
 
 
 @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +74,26 @@ async def upload_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    file: UploadFile = File(...),
+    file: Annotated[UploadFile, File()],
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     data = await file.read()
     try:
         url = media.save_guide_video(data)
     except media.MediaError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
-    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, url, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
+        detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
         ip=get_client_ip(request), commit=False,
     )
     db.commit()
     # 提交成功后再删旧片,避免新片没落库就把旧片丢了
     media.delete_guide_video(before.get("video_url"))
-    return _out(db)
+    return _out(db, scene)
 
 
 @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +101,16 @@ def delete_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
+    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
-    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, None, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
     media.delete_guide_video(before.get("video_url"))
-    return _out(db)
+    return _out(db, scene)

app/admin/schemas/ad_revenue.py

@@ -75,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):
@@ -145,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,10 +32,7 @@ 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/admin/schemas/guide_video.py

@@ -7,6 +7,7 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
 
 
 class GuideVideoConfigOut(BaseModel):
+    scene: str
     enabled: bool
     video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
     max_plays: int

app/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
+from fastapi import APIRouter, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,7 +36,6 @@ from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
-from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare")
 
@@ -81,7 +80,7 @@ def _harvest_running_blocking(
 def _harvest_done_blocking(
     trace_id: str, user_id: int | None, done_params: dict, business_type: str,
     device_id: str | None, device_info: dict | None, trace_url: str | None,
-) -> int:
+) -> None:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -103,7 +102,6 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
-        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -245,12 +243,7 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(
-    request: Request,
-    background_tasks: BackgroundTasks,
-    user: OptionalUser,
-    db: DbSession,
-) -> dict[str, Any]:
+async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -259,15 +252,12 @@ async def price_step(
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            record_id = await run_in_threadpool(
+            await run_in_threadpool(
                 _harvest_done_blocking, trace_id, (user.id if user else None),
                 done_params, "food",
                 meta.get("device_id"), meta.get("device_info"),
                 resp.get("trace_url") or done_params.get("trace_url"),
             )
-            background_tasks.add_task(
-                backfill_comparison_llm_cost, record_id, trace_id
-            )
         except Exception as e:  # noqa: BLE001
             logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,6 +16,8 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.db.session import SessionLocal
+from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -28,7 +30,8 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
+from app.services.llm_cost import compute_llm_cost, get_llm_prices
+from app.services.pricebot_llm_calls import fetch_llm_calls
 
 logger = logging.getLogger("shagua.compare_record")
 
@@ -118,7 +121,32 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    backfill_comparison_llm_cost(record_id, trace_id)
+    calls = fetch_llm_calls(trace_id)
+    if not calls:
+        return
+    db = SessionLocal()
+    try:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is None:
+            return
+        rec.llm_calls = calls
+        rec.llm_call_count = len(calls)
+        rec.retry_count = sum(1 for c in calls if c.get("error"))
+        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
+        # error 的调用 usage 可能为 None,or {} 兜底)
+        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
+        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
+        # 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
+        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
+        db.commit()
+        logger.info(
+            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
+            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
+        )
+    except Exception as e:  # noqa: BLE001 best-effort
+        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
+    finally:
+        db.close()
 
 
 @router.get(

app/core/config.py

@@ -377,12 +377,6 @@ class Settings(BaseSettings):
     # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
     # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
     INTERNAL_API_SECRET: str = ""
-    # Current compare clients are persisted by server-side harvest. Repair any
-    # recent terminal rows left without LLM usage by transient upstream/auth failures.
-    LLM_COST_BACKFILL_ENABLED: bool = True
-    LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
-    LLM_COST_BACKFILL_BATCH_SIZE: int = 100
-    LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
 
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。

app/core/llm_cost_backfill_worker.py

@@ -1,63 +0,0 @@
-"""Periodic repair worker for comparison records with missing LLM token cost."""
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import logging
-
-from app.core.config import settings
-from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
-from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
-
-logger = logging.getLogger("shagua.llm_cost_backfill_worker")
-
-
-async def _run_loop() -> None:
-    interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
-    logger.info(
-        "LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
-        interval,
-        settings.LLM_COST_BACKFILL_BATCH_SIZE,
-        settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
-    )
-    try:
-        while True:
-            try:
-                auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
-                if auth_ready:
-                    result = await asyncio.to_thread(
-                        repair_missing_comparison_llm_costs,
-                        limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
-                        lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
-                    )
-                    logger.info("LLM cost backfill batch result=%s", result)
-                else:
-                    logger.error(
-                        "LLM cost backfill skipped: PriceBot internal auth is not ready"
-                    )
-            except Exception:  # noqa: BLE001
-                logger.exception("LLM cost backfill batch failed")
-            await asyncio.sleep(interval)
-    except asyncio.CancelledError:
-        logger.info("LLM cost backfill worker stopped")
-        raise
-
-
-def start_llm_cost_backfill_worker() -> asyncio.Task | None:
-    if not settings.LLM_COST_BACKFILL_ENABLED:
-        logger.info("LLM cost backfill worker disabled")
-        return None
-    if not settings.INTERNAL_API_SECRET:
-        logger.warning(
-            "LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
-        )
-        return None
-    return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
-
-
-async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
-    if task is None:
-        return
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task

app/main.py

@@ -60,10 +60,6 @@ from app.core.inactivity_reset_worker import (
     start_inactivity_reset_worker,
     stop_inactivity_reset_worker,
 )
-from app.core.llm_cost_backfill_worker import (
-    start_llm_cost_backfill_worker,
-    stop_llm_cost_backfill_worker,
-)
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
@@ -107,7 +103,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
-    llm_cost_backfill_task = start_llm_cost_backfill_worker()
     try:
         yield
     finally:
@@ -117,7 +112,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
-        await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")

app/models/guide_video.py

@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
         # start_play 捕获 IntegrityError 降级成"这次不放视频"。
         # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
         # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
+        Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/repositories/guide_video.py

@@ -30,7 +30,11 @@ from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
 from app.repositories import wallet as crud_wallet
 
-_KEY = "coupon_guide_video"
+SCENES = ("coupon", "comparison")
+_KEY_BY_SCENE = {
+    "coupon": "coupon_guide_video",
+    "comparison": "comparison_guide_video",
+}
 
 #: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
 BIZ_TYPE = "guide_video"
@@ -41,7 +45,7 @@ _DEFAULTS: dict[str, Any] = {
     "enabled": True,
     "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
     "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 120,  # 每次固定金币
+    "reward_coin": 100,  # 每次固定金币
 }
 
 _FIELDS = tuple(_DEFAULTS.keys())
@@ -65,19 +69,28 @@ def _merge(raw: Any) -> dict[str, Any]:
     return out
 
 
-def get_config(db: Session) -> dict[str, Any]:
+def _config_key(scene: str) -> str:
+    if scene not in _KEY_BY_SCENE:
+        raise ValueError(f"unsupported guide video scene: {scene}")
+    return _KEY_BY_SCENE[scene]
+
+
+def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
     """完整配置 + updated_at(admin 读 / 业务读共用)。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     cfg = _merge(row.value if row is not None else None)
     cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
     return cfg
 
 
-def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
+def _write(
+    db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
+) -> dict[str, Any]:
     """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
-    row = db.get(AppConfig, _KEY)
+    key = _config_key(scene)
+    row = db.get(AppConfig, key)
     if row is None:
-        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
+        row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
         db.add(row)
     else:
         row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -98,11 +111,12 @@ def update_config(
     enabled: bool | None = None,
     max_plays: int | None = None,
     reward_coin: int | None = None,
+    scene: str = "coupon",
     admin_id: int,
     commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     if enabled is not None:
@@ -111,45 +125,52 @@ def update_config(
         new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
     if reward_coin is not None:
         new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
     return before, after
 
 
 def set_video(
-    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
+    db: Session, video_url: str | None, *, scene: str = "coupon",
+    admin_id: int, commit: bool = True
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
-    row = db.get(AppConfig, _KEY)
+    row = db.get(AppConfig, _config_key(scene))
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     new_value["video_url"] = video_url
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
     return before, after
 
 
 # ===== 播放计次 =====
 
 
-def used_plays(db: Session, user_id: int) -> int:
+def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
     """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
     return int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == user_id
+                GuideVideoPlay.user_id == user_id,
+                GuideVideoPlay.scene == scene,
             )
         ).scalar_one()
     )
 
 
-def play_stats(db: Session) -> dict[str, int]:
+def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
     """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
     total = int(
-        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.scene == scene
+            )
+        ).scalar_one()
     )
     granted = int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.status == "granted"
+                GuideVideoPlay.status == "granted",
+                GuideVideoPlay.scene == scene,
             )
         ).scalar_one()
     )
@@ -168,11 +189,12 @@ def start_play(
       reward_coin  播完/中途关闭都发的固定金币
       seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
     """
-    cfg = get_config(db)
+    _config_key(scene)
+    cfg = get_config(db, scene)
     video_url = (cfg.get("video_url") or "").strip()
     max_plays = int(cfg.get("max_plays") or 0)
     reward_coin = int(cfg.get("reward_coin") or 0)
-    used = used_plays(db, user_id)
+    used = used_plays(db, user_id, scene)
 
     def _miss(used_now: int) -> dict[str, Any]:
         return {
@@ -194,7 +216,7 @@ def start_play(
         scene=scene,
         seq=seq,
         video_url=video_url,
-        coin=0,
+        coin=reward_coin,
         status="playing",
         completed=0,
         started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -210,7 +232,7 @@ def start_play(
             db.flush()
     except IntegrityError:
         db.rollback()
-        return _miss(used_plays(db, user_id))
+        return _miss(used_plays(db, user_id, scene))
     return {
         "should_play": True,
         "video_url": video_url,
@@ -241,7 +263,8 @@ def grant_play(
     """
     token = (play_token or "").strip()
     # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    coin = int(get_config(db).get("reward_coin") or 0)
+    play = _find_play(db, user_id, token)
+    coin = int(play.coin if play is not None else 0)
 
     # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
     # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',

app/schemas/guide_video.py

@@ -1,13 +1,15 @@
 """新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
 from __future__ import annotations
 
+from typing import Literal
+
 from pydantic import BaseModel, Field
 
 
 class GuideVideoStartIn(BaseModel):
     """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
 
-    scene: str = Field(default="coupon", max_length=16)
+    scene: Literal["coupon", "comparison"] = "coupon"
 
 
 class GuideVideoStartOut(BaseModel):

app/services/comparison_llm_backfill.py

@@ -1,156 +0,0 @@
-"""Persist and repair comparison-record LLM token costs."""
-from __future__ import annotations
-
-import logging
-import time
-from datetime import UTC, datetime, timedelta
-
-from sqlalchemy import select
-
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.models.app_config import AppConfig
-from app.models.comparison import ComparisonRecord
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
-
-logger = logging.getLogger("shagua.comparison_llm_backfill")
-
-
-def _utc_to_beijing_naive(value: datetime) -> datetime:
-    """Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
-    if value.tzinfo is None:
-        value = value.replace(tzinfo=UTC)
-    return value.astimezone(CN_TZ).replace(tzinfo=None)
-
-
-def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
-    """Store calls and all derived fields atomically."""
-    with SessionLocal() as db:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is None or rec.trace_id != trace_id:
-            logger.warning(
-                "LLM cost backfill record mismatch record_id=%s trace=%s",
-                record_id,
-                trace_id,
-            )
-            return False
-
-        # Never recalculate a frozen historical cost with a newer price config.
-        if rec.llm_cost_yuan is not None and rec.llm_calls:
-            return False
-
-        rec.llm_calls = calls
-        rec.llm_call_count = len(calls)
-        rec.retry_count = sum(1 for call in calls if call.get("error"))
-        rec.input_tokens = sum(
-            (call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
-        )
-        rec.output_tokens = sum(
-            (call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
-        )
-        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
-            calls, get_llm_prices(db)
-        )
-        db.commit()
-        logger.info(
-            "LLM cost backfilled trace=%s calls=%d input_tokens=%d "
-            "output_tokens=%d cost=%s",
-            trace_id,
-            len(calls),
-            rec.input_tokens,
-            rec.output_tokens,
-            rec.llm_cost_yuan,
-        )
-        return True
-
-
-def backfill_comparison_llm_cost(
-    record_id: int,
-    trace_id: str,
-    *,
-    attempts: int = 3,
-    retry_delays: tuple[float, ...] = (1.0, 3.0),
-) -> bool:
-    """Fetch and persist one record, retrying short-lived upstream races."""
-    if not settings.INTERNAL_API_SECRET or not trace_id:
-        logger.warning(
-            "LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
-            trace_id,
-        )
-        return False
-    total_attempts = max(1, attempts)
-    for attempt in range(total_attempts):
-        calls = fetch_llm_calls(trace_id)
-        if calls:
-            try:
-                return _store_calls(record_id, trace_id, calls)
-            except Exception:  # noqa: BLE001 - background repair must stay alive
-                logger.exception(
-                    "LLM cost store failed trace=%s record_id=%s",
-                    trace_id,
-                    record_id,
-                )
-                return False
-
-        if attempt + 1 < total_attempts:
-            delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
-            if delay > 0:
-                time.sleep(delay)
-
-    logger.warning(
-        "LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
-        trace_id,
-        record_id,
-        total_attempts,
-    )
-    return False
-
-
-def repair_missing_comparison_llm_costs(
-    *,
-    limit: int = 100,
-    lookback_days: int = 30,
-) -> dict[str, int]:
-    """Repair a bounded batch of recent terminal records with missing cost."""
-    cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
-        days=max(1, lookback_days)
-    )
-    with SessionLocal() as db:
-        # app_config has no price history. Repricing a record from before the
-        # current config became effective would fabricate a historical cost, so
-        # only repair records at/after that timestamp.
-        price_config_updated_at = db.execute(
-            select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
-        ).scalar_one_or_none()
-        date_conditions = [ComparisonRecord.created_at >= cutoff]
-        if price_config_updated_at is not None:
-            date_conditions.append(
-                ComparisonRecord.created_at
-                >= _utc_to_beijing_naive(price_config_updated_at)
-            )
-        candidates = list(
-            db.execute(
-                select(ComparisonRecord.id, ComparisonRecord.trace_id)
-                .where(
-                    *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed")),
-                    ComparisonRecord.llm_cost_yuan.is_(None),
-                )
-                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
-                .limit(max(1, limit))
-            ).all()
-        )
-
-    repaired = 0
-    for record_id, trace_id in candidates:
-        if backfill_comparison_llm_cost(
-            record_id, trace_id, attempts=1, retry_delays=()
-        ):
-            repaired += 1
-    return {
-        "candidates": len(candidates),
-        "repaired": repaired,
-        "unresolved": len(candidates) - repaired,
-    }

app/services/pricebot_llm_calls.py

@@ -20,67 +20,19 @@ from app.core.pricebot_router import pick_pricebot
 logger = logging.getLogger("shagua.pricebot_llm")
 
 
-def pricebot_llm_auth_ready() -> bool:
-    """Verify every configured PriceBot instance accepts the shared secret."""
-    secret = settings.INTERNAL_API_SECRET
-    if not secret:
-        logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
-        return False
-    for base in settings.pricebot_instances:
-        url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
-        try:
-            resp = httpx.get(
-                url, headers={"X-Internal-Secret": secret}, timeout=3.0
-            )
-        except Exception as exc:  # noqa: BLE001
-            logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
-            return False
-        if resp.status_code != 200:
-            logger.error(
-                "PriceBot LLM auth check rejected base=%s status=%s; "
-                "verify both services use the same INTERNAL_API_SECRET",
-                base,
-                resp.status_code,
-            )
-            return False
-    return True
-
-
 def fetch_llm_calls(trace_id: str) -> list[dict]:
     """返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
     未配密钥 / 无 trace_id / 拉取失败 → []。"""
     secret = settings.INTERNAL_API_SECRET
     if not secret or not trace_id:
         return []
-    preferred = pick_pricebot(trace_id)
-    # LLM JSONL is instance-local. If the cluster topology changed after a
-    # historical trace was created, consistent hashing may now point elsewhere;
-    # probe the remaining configured instances only when the preferred one is empty.
-    bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
-    for base in bases:
-        url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
-        try:
-            resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
-            if resp.status_code == 200:
-                calls = resp.json().get("calls", []) or []
-                if calls:
-                    return calls
-                continue
-            if resp.status_code in (401, 403):
-                logger.error(
-                    "fetch_llm_calls rejected trace=%s base=%s status=%s; "
-                    "INTERNAL_API_SECRET differs between app-server and PriceBot",
-                    trace_id,
-                    base,
-                    resp.status_code,
-                )
-            else:
-                logger.warning(
-                    "fetch_llm_calls trace=%s base=%s status=%s",
-                    trace_id,
-                    base,
-                    resp.status_code,
-                )
-        except Exception as e:  # noqa: BLE001 — best-effort
-            logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
+    base = pick_pricebot(trace_id).rstrip("/")
+    url = f"{base}/api/internal/llm_calls/{trace_id}"
+    try:
+        resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
+        if resp.status_code == 200:
+            return resp.json().get("calls", []) or []
+        logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
+    except Exception as e:  # noqa: BLE001 — best-effort,任何异常都不该影响上报
+        logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
     return []

deploy/comparison-llm-cost.md

@@ -1,40 +0,0 @@
-# 比价 TOKEN 成本采集与补偿
-
-## 部署前置
-
-App Server 与 PriceBot 使用各自独立的 `.env`，但下面的值必须完全一致：
-
-- `/opt/shaguabijia-app-server/.env`
-- `/opt/pricebot-backend/.env`
-- 配置项：`INTERNAL_API_SECRET`
-
-不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
-
-App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
-`PriceBot LLM auth check rejected`，并跳过本轮补偿，避免对所有缺失记录重复发送失败请求。
-
-## 数据链路
-
-1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
-2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用，冻结 Token、成本和单价快照。
-3. 周期 worker 扫描近期 `success/failed` 且 `llm_cost_yuan IS NULL` 的记录进行补偿；
-   为避免用现价伪造历史成本，只处理当前单价配置生效时间之后的记录。
-4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
-
-## 上线验收（只读 SQL）
-
-```sql
-SELECT
-  (created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
-  count(*) AS records,
-  count(llm_cost_yuan) AS cost_records,
-  round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
-FROM comparison_record
-WHERE created_at >= now() - interval '3 days'
-GROUP BY 1
-ORDER BY 1 DESC;
-```
-
-新产生的正常终态比价记录应在短时间内写入 `input_tokens`、`output_tokens` 和
-`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填；
-原始调用已经清理的记录不能用估算值冒充真实成本。

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 14", android_version="14",
+           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", 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 14", android_version="13",
+           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
            created=ago(minutes=22)),
         # 比价反馈 · scene=找错商品 · 新端 · 无图
         fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
            source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS 4", android_version="14",
+           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", 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 14", android_version="14",
+           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", 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 13", android_version="13",
+           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", 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 14", android_version="13",
+           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
            created=ago(days=3)),
     ]
     db.add_all(feedbacks)

tests/test_admin_ad_revenue_scope.py

@@ -211,59 +211,6 @@ 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"

tests/test_admin_feedback_device_details.py

@@ -1,71 +0,0 @@
-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_compare_harvest.py

@@ -9,16 +9,17 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
 """
 from __future__ import annotations
 
+import json
 import uuid
 from unittest.mock import MagicMock, patch
 
 import httpx
-from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.schemas.compare_record import ComparisonRecordIn
+from sqlalchemy import select
 
 
 def _tid() -> str:
@@ -215,9 +216,7 @@ def test_price_step_done_harvests_success(client) -> None:
                   "action": {"command": "done", "params": _done_params()},
                   "trace_url": "https://price.shaguabijia.com/traces/done2/"}
     p, _cap = _mock_pricebot(done_frame)
-    with p, patch(
-        "app.api.v1.compare.backfill_comparison_llm_cost"
-    ) as backfill:
+    with p:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -225,7 +224,6 @@ def test_price_step_done_harvests_success(client) -> None:
         assert rec is not None and rec.status == "success"
         assert rec.best_platform_id == "meituan"
         assert rec.saved_amount_cents == 500
-        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_trace_finalize_harvests_abort(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -1,161 +0,0 @@
-from __future__ import annotations
-
-from datetime import UTC, datetime, timedelta
-
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.models.app_config import AppConfig
-from app.models.comparison import ComparisonRecord
-from app.services import comparison_llm_backfill
-
-
-def _record(
-    trace_id: str,
-    *,
-    status: str = "success",
-    created_at: datetime | None = None,
-) -> int:
-    with SessionLocal() as db:
-        rec = ComparisonRecord(
-            trace_id=trace_id,
-            status=status,
-            created_at=created_at or datetime.now(),
-        )
-        db.add(rec)
-        db.commit()
-        return rec.id
-
-
-def _delete(record_id: int) -> None:
-    with SessionLocal() as db:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is not None:
-            db.delete(rec)
-            db.commit()
-
-
-def test_backfill_retries_then_persists_cost(monkeypatch):
-    record_id = _record("llm-retry-1")
-    calls = [
-        {
-            "model": "unknown-model",
-            "error": None,
-            "usage": {"prompt_tokens": 1000, "completion_tokens": 500},
-        }
-    ]
-    responses = iter([[], calls])
-    monkeypatch.setattr(
-        comparison_llm_backfill,
-        "fetch_llm_calls",
-        lambda trace_id: next(responses),
-    )
-    sleeps: list[float] = []
-    monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
-
-    try:
-        assert comparison_llm_backfill.backfill_comparison_llm_cost(
-            record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
-        )
-        assert sleeps == [0.25]
-        with SessionLocal() as db:
-            rec = db.get(ComparisonRecord, record_id)
-            assert rec.input_tokens == 1000
-            assert rec.output_tokens == 500
-            assert rec.llm_cost_yuan is not None
-            assert rec.llm_calls == calls
-    finally:
-        _delete(record_id)
-
-
-def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
-    missing_id = _record("llm-repair-missing")
-    running_id = _record("llm-repair-running", status="running")
-    calls = [
-        {
-            "model": "unknown-model",
-            "error": None,
-            "usage": {"prompt_tokens": 100, "completion_tokens": 20},
-        }
-    ]
-    seen: list[str] = []
-
-    def fetch(trace_id: str) -> list[dict]:
-        seen.append(trace_id)
-        return calls
-
-    monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
-    try:
-        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
-            limit=10, lookback_days=1
-        )
-        assert result["repaired"] >= 1
-        assert "llm-repair-missing" in seen
-        assert "llm-repair-running" not in seen
-        with SessionLocal() as db:
-            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
-            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
-    finally:
-        _delete(missing_id)
-        _delete(running_id)
-
-
-def test_repair_excludes_records_before_current_price_config(monkeypatch):
-    price_changed_at = datetime.now(UTC) - timedelta(hours=1)
-    before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
-    after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
-    before_id = _record(
-        "llm-before-price-change",
-        created_at=before_change.replace(tzinfo=None),
-    )
-    after_id = _record(
-        "llm-after-price-change",
-        created_at=after_change.replace(tzinfo=None),
-    )
-    with SessionLocal() as db:
-        existing = db.get(AppConfig, "llm_token_price")
-        if existing is not None:
-            db.delete(existing)
-            db.flush()
-        db.add(
-            AppConfig(
-                key="llm_token_price",
-                value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
-                updated_at=price_changed_at,
-            )
-        )
-        db.commit()
-
-    seen: list[str] = []
-
-    def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
-        seen.append(trace_id)
-        return True
-
-    monkeypatch.setattr(
-        comparison_llm_backfill,
-        "backfill_comparison_llm_cost",
-        backfill,
-    )
-    try:
-        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
-            limit=10_000,
-            lookback_days=1,
-        )
-        assert "llm-after-price-change" in seen
-        assert "llm-before-price-change" not in seen
-        assert result["repaired"] == len(seen)
-        assert result["unresolved"] == 0
-    finally:
-        _delete(before_id)
-        _delete(after_id)
-        with SessionLocal() as db:
-            config = db.get(AppConfig, "llm_token_price")
-            if config is not None:
-                db.delete(config)
-                db.commit()

tests/test_guide_video.py

@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
     assert first["should_play"] is True and first["seq"] == 1
 
     # 本次请求读到的是过期计数 → 仍会算出 seq=1
-    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
     with SessionLocal() as db:
         result = crud_guide.start_play(db, uid)
 

tests/test_llm_cost.py

@@ -132,7 +132,6 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
     from app.models.app_config import AppConfig
     from app.models.comparison import ComparisonRecord
     from app.repositories import app_config
-    from app.services import comparison_llm_backfill
 
     sample = [
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -140,12 +139,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
     ]
-    monkeypatch.setattr(
-        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
-    )
-    monkeypatch.setattr(
-        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
-    )
+    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -1,55 +0,0 @@
-from unittest.mock import MagicMock
-
-from app.core.config import settings
-from app.services import pricebot_llm_calls
-
-
-def test_auth_probe_accepts_matching_secret(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
-    response = MagicMock(status_code=200)
-    get = MagicMock(return_value=response)
-    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
-
-    assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
-    assert get.call_count == len(settings.pricebot_instances)
-    assert all(
-        call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
-        for call in get.call_args_list
-    )
-
-
-def test_auth_probe_rejects_mismatched_secret(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
-    monkeypatch.setattr(
-        pricebot_llm_calls.httpx,
-        "get",
-        MagicMock(return_value=MagicMock(status_code=403)),
-    )
-
-    assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
-
-
-def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
-    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
-    monkeypatch.setattr(
-        settings,
-        "PRICEBOT_INSTANCES",
-        "http://pricebot-1:8000,http://pricebot-2:8000",
-    )
-    monkeypatch.setattr(
-        pricebot_llm_calls,
-        "pick_pricebot",
-        lambda trace_id: "http://pricebot-1:8000",
-    )
-    calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
-
-    def get(url, **kwargs):
-        response = MagicMock(status_code=200)
-        response.json.return_value = {
-            "calls": [] if "pricebot-1" in url else calls
-        }
-        return response
-
-    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
-
-    assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls