diff --git a/alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py b/alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py new file mode 100644 index 0000000..311ebeb --- /dev/null +++ b/alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py @@ -0,0 +1,26 @@ +"""merge comparison platforms + fail_reason heads + +Revision ID: 6d2309208549 +Revises: comparison_platforms_col, comparison_record_fail_reason +Create Date: 2026-07-29 01:48:41.868083 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '6d2309208549' +down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/alembic/versions/comparison_platforms_col.py b/alembic/versions/comparison_platforms_col.py new file mode 100644 index 0000000..3b3d82e --- /dev/null +++ b/alembic/versions/comparison_platforms_col.py @@ -0,0 +1,47 @@ +"""add platforms unified array column to comparison_record + +展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带 +status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。 +纯新增列, 老记录为空 → 前端回退老 comparison_results。 + +Revision ID: comparison_platforms_col +Revises: user_manual_risk_fields +""" +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision: str = "comparison_platforms_col" +down_revision: str | None = "user_manual_risk_fields" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql") + + +def upgrade() -> None: + # 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动 + # `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb + # NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过, + # 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。 + bind = op.get_bind() + cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")} + if "platforms" in cols: + return + with op.batch_alter_table("comparison_record") as batch_op: + batch_op.add_column( + sa.Column( + "platforms", _JSON, nullable=False, + server_default=sa.text("'[]'"), + ) + ) + + +def downgrade() -> None: + with op.batch_alter_table("comparison_record") as batch_op: + batch_op.drop_column("platforms") diff --git a/app/admin/repositories/ad_revenue.py b/app/admin/repositories/ad_revenue.py index 6ce16b4..b297fbf 100644 --- a/app/admin/repositories/ad_revenue.py +++ b/app/admin/repositories/ad_revenue.py @@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"}) _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"}) -def _business_code_ids(db: Session, app_env: str | None) -> set[str]: +def business_code_ids(db: Session, app_env: str | None) -> set[str]: """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。""" prod_config = app_config.get_ad_config(db) prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | { @@ -320,10 +320,10 @@ def ad_revenue_report( # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试 # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。 - business_code_ids: set[str] | None = None + business_ids: set[str] | None = None if revenue_scope == "business": - business_code_ids = _business_code_ids(db, app_env) - events = [e for e in events if e.get("our_code_id") in business_code_ids] + business_ids = business_code_ids(db, app_env) + events = [e for e in events if e.get("our_code_id") in business_ids] # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排; # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。 @@ -381,7 +381,7 @@ def ad_revenue_report( date_from=date_from, date_to=date_to, app_env=app_env, - our_code_ids=business_code_ids, + our_code_ids=business_ids, ) if pangle_aggs: by_date = {a["date"]: a for a in pangle_aggs} diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 41f113c..a2158ee 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import Session from app.core import rewards from app.core.config import settings +from app.models.ad_ecpm import AdEcpmRecord from app.models.ad_feed_reward import AdFeedRewardRecord from app.models.ad_reward import AdRewardRecord from app.models.admin import AdminAuditLog @@ -1257,11 +1258,15 @@ def user_reward_stats( date_from: datetime | None = None, date_to: datetime | None = None, withdraw_source: str | None = None, + app_env: str | None = None, + revenue_scope: str = "all", + feed_scene: str | None = None, ) -> dict: """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。 - 口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加); - 平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。 + 口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。 + 平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算， + 不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。 传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。 """ withdraw_source_conds = ( @@ -1296,30 +1301,68 @@ def user_reward_stats( # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时， # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。 + business_ids: set[str] | None = None + if revenue_scope == "business": + # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。 + from app.admin.repositories.ad_revenue import business_code_ids + + business_ids = business_code_ids(db, app_env) + + rv_conds = [ + AdRewardRecord.user_id == user_id, + AdRewardRecord.reward_scene == "reward_video", + AdRewardRecord.status == "granted", + *_window_conds(AdRewardRecord.created_at, date_from, date_to), + ] + if app_env is not None: + rv_conds.append(AdRewardRecord.app_env == app_env) + if business_ids is not None: + rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids)) rv = db.execute( select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where( - AdRewardRecord.user_id == user_id, - AdRewardRecord.reward_scene == "reward_video", - AdRewardRecord.status == "granted", - *_window_conds(AdRewardRecord.created_at, date_from, date_to), + *rv_conds, ) ).all() rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw] rv_coins = sum(r.coin for r in rv) - feed = db.execute( + feed_reward_conds = [ + AdFeedRewardRecord.user_id == user_id, + AdFeedRewardRecord.status == "granted", + *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to), + ] + if app_env is not None: + feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env) + if feed_scene is not None: + feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene) + if business_ids is not None: + feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids)) + feed_rewards = db.execute( select( AdFeedRewardRecord.unit_count, - AdFeedRewardRecord.ecpm_raw, AdFeedRewardRecord.coin, ).where( - AdFeedRewardRecord.user_id == user_id, - AdFeedRewardRecord.status == "granted", - *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to), + *feed_reward_conds, ) ).all() - feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw] - feed_coins = sum(f.coin for f in feed) + feed_coins = sum(f.coin for f in feed_rewards) + + feed_impression_conds = [ + AdEcpmRecord.user_id == user_id, + AdEcpmRecord.ad_type.in_(("draw", "feed")), + *_window_conds(AdEcpmRecord.created_at, date_from, date_to), + ] + if app_env is not None: + feed_impression_conds.append(AdEcpmRecord.app_env == app_env) + if feed_scene is not None: + feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene) + if business_ids is not None: + feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids)) + feed_impressions = db.execute( + select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds) + ).all() + # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。 + feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions] trad_coins = db.execute( select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where( @@ -1338,7 +1381,7 @@ def user_reward_stats( "reward_video_count": len(rv), "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0, "reward_video_cash_cents": _coins_to_cents(rv_coins), - "feed_count": int(sum(f.unit_count for f in feed)), + "feed_count": int(sum(f.unit_count for f in feed_rewards)), "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0, "feed_cash_cents": _coins_to_cents(feed_coins), } diff --git a/app/admin/routers/users.py b/app/admin/routers/users.py index 1d8b056..bb044ef 100644 --- a/app/admin/routers/users.py +++ b/app/admin/routers/users.py @@ -88,6 +88,11 @@ def get_user_reward_stats( withdraw_source: Annotated[ str | None, Query(pattern="^(coin_cash|invite_cash)$") ] = None, + app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None, + revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all", + feed_scene: Annotated[ + str | None, Query(pattern="^(comparison|coupon|welfare)$") + ] = None, ) -> UserRewardStats: """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。""" if not user_repo.user_exists(db, user_id): @@ -99,6 +104,9 @@ def get_user_reward_stats( date_from=date_from, date_to=date_to, withdraw_source=withdraw_source, + app_env=app_env, + revenue_scope=revenue_scope, + feed_scene=feed_scene, ) ) diff --git a/app/admin/schemas/user.py b/app/admin/schemas/user.py index aa826da..2e5a757 100644 --- a/app/admin/schemas/user.py +++ b/app/admin/schemas/user.py @@ -60,7 +60,7 @@ class UserRewardStats(BaseModel): reward_video_avg_ecpm: float # 平均激励视频 eCPM(分/千次) reward_video_cash_cents: int # 激励视频提现(金币折现) feed_count: int # 累计信息流广告数(granted 份数,unit_count 累加) - feed_avg_ecpm: float # 平均信息流广告 eCPM(分/千次) + feed_avg_ecpm: float # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示) feed_cash_cents: int # 信息流广告提现(金币折现) diff --git a/app/api/v1/compare.py b/app/api/v1/compare.py index fcb1e03..fa8268a 100644 --- a/app/api/v1/compare.py +++ b/app/api/v1/compare.py @@ -108,7 +108,7 @@ def _harvest_done_blocking( def _harvest_abort_blocking( trace_id: str, status_hint: str, reason: str | None, trace_url: str | None, -) -> None: +) -> int | None: with SessionLocal() as db: rec = crud_compare.harvest_abort( db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url, @@ -118,6 +118,7 @@ def _harvest_abort_blocking( extra={"phase": "harvest_abort", "status": (rec.status if rec else None), "reason": reason}, ) + return rec.id if rec is not None else None async def _forward( @@ -291,7 +292,10 @@ async def trace_epilogue( @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)") async def trace_finalize( - request: Request, user: OptionalUser, db: DbSession + request: Request, + background_tasks: BackgroundTasks, + user: OptionalUser, + db: DbSession, ) -> dict[str, Any]: _ensure_compare_allowed(user, db) # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url}; @@ -302,12 +306,16 @@ async def trace_finalize( request, "/api/trace/finalize", user, harvest_first_frame=False, ) try: - await run_in_threadpool( + record_id = await run_in_threadpool( _harvest_abort_blocking, trace_id, (meta.get("status") or "cancelled"), (meta.get("reason") or meta.get("information")), (resp.get("trace_url") if isinstance(resp, dict) else None), ) + if record_id is not None: + background_tasks.add_task( + backfill_comparison_llm_cost, record_id, trace_id + ) except Exception as e: # noqa: BLE001 logger.warning("harvest_abort failed trace=%s: %s", trace_id, e) return resp diff --git a/app/models/comparison.py b/app/models/comparison.py index 2a60d30..7edbc99 100644 --- a/app/models/comparison.py +++ b/app/models/comparison.py @@ -113,6 +113,10 @@ class ComparisonRecord(Base): items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list) # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细) comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list) + # 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带 + # status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results + # + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。 + platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list) # 目标平台未找到、跳过的菜名 skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list) # 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底 diff --git a/app/repositories/comparison.py b/app/repositories/comparison.py index 948ae77..39df9f8 100644 --- a/app/repositories/comparison.py +++ b/app/repositories/comparison.py @@ -164,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict: is_source_best = best.is_source if best is not None else None - # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed - status = payload.status + # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status + # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。 + status = payload.record_status or payload.status if status is None: has_valid_target = any( (not r.is_source) and r.price is not None for r in results @@ -197,16 +198,35 @@ def upsert_record( 灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile; 新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。 """ - derived = _derive(payload) + # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源 + # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。 + if payload.platforms: + derived = _derive_from_platforms(payload.platforms, payload.record_status) + # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值 + derived["fail_reason"] = ( + _derive_fail_display(payload.information, payload.platform_results or {}) + if derived["status"] == "failed" + else None + ) + # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。 + # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived) + # 与 _derive_from_platforms 同名键撞键 TypeError)。 + for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"): + if not derived.get(_k): + derived[_k] = getattr(payload, _k) + else: + derived = _derive(payload) + # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补, + # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。 + for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"): + derived[_k] = getattr(payload, _k) items = [it.model_dump(exclude_none=True) for it in payload.items] fields = dict( device_id=payload.device_id, business_type=payload.business_type, - store_name=payload.store_name, product_names=_product_names_from_items(items), - source_platform_id=payload.source_platform_id, - source_platform_name=payload.source_platform_name, - source_package=payload.source_package, + # store_name / source_platform_id / source_platform_name / source_package 统一由 + # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。 information=payload.information, best_deeplink=payload.best_deeplink, trace_url=payload.trace_url, @@ -214,6 +234,7 @@ def upsert_record( skipped_dish_count=payload.skipped_dish_count, items=items, comparison_results=[r.model_dump() for r in payload.comparison_results], + platforms=list(payload.platforms or []), skipped_dish_names=list(payload.skipped_dish_names), # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None) device_model=payload.device_model, @@ -286,7 +307,8 @@ def upsert_record( def _derive_from_results( - results: list[dict], platform_results: dict | None = None + results: list[dict], platform_results: dict | None = None, + record_status: str | None = None, ) -> dict: """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。 等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。 @@ -334,7 +356,53 @@ def _derive_from_results( "saved_amount_cents": saved_amount_cents, "is_source_best": best.get("is_source") if best else None, "store_name": (src_row or {}).get("store_name") or None, - "status": "success" if has_valid_target else "failed", + # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed, + # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时 + # 回退老的 success/failed 二态派生, 向后兼容。 + "status": record_status or ("success" if has_valid_target else "failed"), + } + + +def _derive_from_platforms( + platforms: list, record_status: str | None = None, +) -> dict: + """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。 + + best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的 + platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优, + 消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无 + platforms 时调用方回退 _derive_from_results(向后兼容)。""" + rows = [p for p in (platforms or []) if isinstance(p, dict)] + src = next((p for p in rows if p.get("role") == "source"), None) + best = next((p for p in rows if p.get("is_best")), None) + source_price_cents = _yuan_to_cents(src.get("price")) if src else None + best_price_cents = _yuan_to_cents(best.get("price")) if best else None + saved_amount_cents = None + if source_price_cents is not None and best_price_cents is not None: + saved_amount_cents = source_price_cents - best_price_cents + has_valid_target = any( + p.get("role") != "source" and p.get("price") is not None for p in rows + ) + # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot + # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名 + # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。 + store_name = ( + (src or {}).get("store_name") + or (best or {}).get("store_name") + or next((p.get("store_name") for p in rows if p.get("store_name")), None) + ) + return { + "source_platform_id": (src or {}).get("platform_id"), + "source_platform_name": (src or {}).get("platform_name"), + "source_package": (src or {}).get("package"), + "source_price_cents": source_price_cents, + "best_platform_id": (best or {}).get("platform_id"), + "best_platform_name": (best or {}).get("platform_name"), + "best_price_cents": best_price_cents, + "saved_amount_cents": saved_amount_cents, + "is_source_best": (best.get("role") == "source") if best else None, + "store_name": store_name or None, + "status": record_status or ("success" if has_valid_target else "failed"), } @@ -493,7 +561,29 @@ def harvest_done( 返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。 行不存在(理论上帧0已建;防御)则新建。""" results = done_params.get("comparison_results") or [] - derived = _derive_from_results(results, done_params.get("platform_results")) + # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。 + # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。 + platforms = done_params.get("platforms") or [] + record_status = done_params.get("record_status") + # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status + # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无 + # platforms 时回退从 comparison_results 派生(向后兼容)。 + if platforms: + derived = _derive_from_platforms(platforms, record_status) + # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退) + _item_row = ( + next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None) + or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None) + or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None) + ) + items = (_item_row or {}).get("items") or [] + else: + derived = _derive_from_results( + results, done_params.get("platform_results"), record_status + ) + # pricebot 已把源单菜品塞进 comparison_results[源行].items + items = next((r.get("items") or [] for r in results if r.get("is_source")), []) + # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算 fail_reason = ( _derive_fail_display( done_params.get("information"), done_params.get("platform_results") @@ -501,8 +591,6 @@ def harvest_done( if derived["status"] == "failed" else None ) - # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items - items = next((r.get("items") or [] for r in results if r.get("is_source")), []) fields = dict( business_type=business_type or "food", information=done_params.get("information") or None, @@ -514,6 +602,7 @@ def harvest_done( skipped_dish_count=done_params.get("skipped_dish_count"), skipped_dish_names=list(done_params.get("skipped_dish_names") or []), comparison_results=results, + platforms=platforms, items=items, product_names=_product_names_from_items(items), raw_payload=done_params, diff --git a/app/schemas/compare_record.py b/app/schemas/compare_record.py index e6d3ed0..c30e495 100644 --- a/app/schemas/compare_record.py +++ b/app/schemas/compare_record.py @@ -107,6 +107,13 @@ class ComparisonRecordIn(BaseModel): # 明细 items: list[ComparisonItemIn] = Field(default_factory=list) comparison_results: list[ComparisonResultIn] = Field(default_factory=list) + # 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带 + # status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由 + # pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。 + platforms: list[dict] = Field(default_factory=list) + # 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再 + # 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。 + record_status: str | None = None # 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/ # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列), # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是 @@ -177,6 +184,9 @@ class ComparisonRecordOut(BaseModel): fail_reason: str | None = None items: list = [] comparison_results: list = [] + # 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此 + # 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。 + platforms: list = [] skipped_dish_names: list = [] total_ms: int | None = None # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。 diff --git a/app/services/comparison_llm_backfill.py b/app/services/comparison_llm_backfill.py index c7eba8c..1687c1b 100644 --- a/app/services/comparison_llm_backfill.py +++ b/app/services/comparison_llm_backfill.py @@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs( select(ComparisonRecord.id, ComparisonRecord.trace_id) .where( *date_conditions, - ComparisonRecord.status.in_(("success", "failed")), + ComparisonRecord.status.in_(("success", "failed", "cancelled")), ComparisonRecord.llm_cost_yuan.is_(None), ) .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc()) diff --git a/tests/test_admin_read.py b/tests/test_admin_read.py index 084caac..806faed 100644 --- a/tests/test_admin_read.py +++ b/tests/test_admin_read.py @@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account( assert invite.json()["cash_balance_cents"] == 456 +def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions( + admin_client: TestClient, admin_token: str +) -> None: + """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。""" + from app.models.ad_ecpm import AdEcpmRecord + from app.models.ad_feed_reward import AdFeedRewardRecord + + uid = _seed_user_with_data("13800000024") + created_at = datetime(2038, 1, 15, 4, tzinfo=UTC) + db = SessionLocal() + try: + db.add_all( + [ + AdFeedRewardRecord( + client_event_id="reward-stats-granted-high", + ad_session_id="reward-stats-granted-high", + user_id=uid, + reward_date="2038-01-15", + duration_seconds=10, + unit_count=1, + ecpm_raw="9000", + ad_type="draw", + feed_scene="coupon", + app_env="prod", + our_code_id="104098712", + coin=9, + status="granted", + created_at=created_at, + ), + AdEcpmRecord( + user_id=uid, + ad_type="draw", + feed_scene="coupon", + ad_session_id="reward-stats-impression-low", + app_env="prod", + our_code_id="104098712", + ecpm_raw="1000", + report_date="2038-01-15", + created_at=created_at, + ), + AdEcpmRecord( + user_id=uid, + ad_type="feed", + feed_scene="coupon", + ad_session_id="reward-stats-impression-mid", + app_env="prod", + our_code_id="104098712", + ecpm_raw="3000", + report_date="2038-01-15", + created_at=created_at, + ), + # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。 + AdEcpmRecord( + user_id=uid, + ad_type="draw", + feed_scene="comparison", + ad_session_id="reward-stats-other-scene", + app_env="prod", + our_code_id="104098712", + ecpm_raw="7000", + report_date="2038-01-15", + created_at=created_at, + ), + AdEcpmRecord( + user_id=uid, + ad_type="draw", + feed_scene="coupon", + ad_session_id="reward-stats-test-env", + app_env="test", + our_code_id="104127529", + ecpm_raw="8000", + report_date="2038-01-15", + created_at=created_at, + ), + AdEcpmRecord( + user_id=uid, + ad_type="draw", + feed_scene="coupon", + ad_session_id="reward-stats-non-business", + app_env="prod", + our_code_id="demo-slot", + ecpm_raw="9000", + report_date="2038-01-15", + created_at=created_at, + ), + ] + ) + db.commit() + finally: + db.close() + + response = admin_client.get( + f"/admin/api/users/{uid}/reward-stats", + params={ + "date_from": "2038-01-15T00:00:00Z", + "date_to": "2038-01-15T23:59:59Z", + "app_env": "prod", + "revenue_scope": "business", + "feed_scene": "coupon", + }, + headers=_auth(admin_token), + ) + assert response.status_code == 200, response.text + data = response.json() + assert data["feed_count"] == 1 + # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。 + assert data["feed_avg_ecpm"] == 2000.0 + + def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None: """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。""" naive = datetime(2038, 1, 1, 8, 0) diff --git a/tests/test_compare_harvest.py b/tests/test_compare_harvest.py index 82448f0..46a11ba 100644 --- a/tests/test_compare_harvest.py +++ b/tests/test_compare_harvest.py @@ -263,7 +263,9 @@ def test_trace_finalize_harvests_abort(client) -> None: with SessionLocal() as db: # 先有 running 行(帧0建的) crud.harvest_running(db, trace_id=tid, user_id=None) p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"}) - with p: + with p, patch( + "app.api.v1.compare.backfill_comparison_llm_cost" + ) as backfill: r = client.post("/api/v1/trace/finalize", json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"}) assert r.status_code == 200 @@ -271,6 +273,7 @@ def test_trace_finalize_harvests_abort(client) -> None: rec = _get(db, tid) assert rec is not None and rec.status == "cancelled" assert rec.trace_url.endswith("/fin/") + backfill.assert_called_once_with(rec.id, tid) def test_price_step_binds_user_when_authed(client) -> None: diff --git a/tests/test_comparison_llm_backfill.py b/tests/test_comparison_llm_backfill.py index fe9b925..89e2293 100644 --- a/tests/test_comparison_llm_backfill.py +++ b/tests/test_comparison_llm_backfill.py @@ -72,6 +72,7 @@ def test_backfill_retries_then_persists_cost(monkeypatch): def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch): missing_id = _record("llm-repair-missing") + cancelled_id = _record("llm-repair-cancelled", status="cancelled") running_id = _record("llm-repair-running", status="running") calls = [ { @@ -96,12 +97,15 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch): ) assert result["repaired"] >= 1 assert "llm-repair-missing" in seen + assert "llm-repair-cancelled" 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, cancelled_id).llm_cost_yuan is not None assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None finally: _delete(missing_id) + _delete(cancelled_id) _delete(running_id)