Compare commits

...

7 Commits

Author SHA1 Message Date
unknown 0ed1b08a85 修复:补齐用户领券记录点位统计 2026-07-29 11:17:55 +08:00
unknown 566a201354 修复:将领券计划数迁移接入最新主线 2026-07-29 11:03:06 +08:00
unknown 7871f1f6b6 Merge remote-tracking branch 'origin/main' into codex/pr192-alembic-fix 2026-07-29 11:02:12 +08:00
marco 4bd4e66678 重构比价结果页取数据逻辑 (#195)
Reviewed-on: #195
2026-07-29 01:59:31 +08:00
unknown 97613eca8e 修复:收敛领券计划数迁移链 2026-07-28 20:01:37 +08:00
linkeyu 90c6fe599a 修复:统一用户Draw信息流eCPM统计口径 (#190)
## 问题

业务收益详情的平均 Draw eCPM 仅平均成功发奖记录,会排除未发奖的真实展示,导致数值系统性偏高,且与广告收益页口径不一致。

## 修复

- `feed_avg_ecpm` 改为从 `ad_ecpm_record` 的全部 `draw/feed` 实际展示计算
- 成功发奖、未发奖展示均纳入,每次展示等权
- 日期、正式/测试环境、业务代码位、领券/比价场景支持与广告收益页对齐
- 奖励份数仍基于成功发奖表,不混用展示数据源
- 复用广告收益报表的业务代码位集合

## 线上数据复算

2026-07-25、正式业务、用户 #33:

- 旧口径(只看成功发奖):`29.9117 元/千次`
- 新口径(333 次真实展示):`19.9926 元/千次`
- 新值与广告收益报表一致

## 验证

- 新增成功/未发奖、场景、环境、业务代码位回归用例
- `tests/test_admin_read.py` + `tests/test_admin_ad_revenue_scope.py`:29 项全通过
- Ruff 改动文件检查通过

## 上线顺序

本 PR 需先于管理后台配套 PR 上线。

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #190
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-28 17:58:23 +08:00
unknown f902f3011c 修复:统一领券单券成功率分母
记录每轮计划券数,将跳过券计入逐场分母,并在中途退出或无逐券结果时返回稳定的成功率口径。
2026-07-28 15:15:52 +08:00
17 changed files with 578 additions and 42 deletions
@@ -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
@@ -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")
@@ -0,0 +1,28 @@
"""add planned coupon count to coupon session
Revision ID: coupon_session_planned_count
Revises: 6d2309208549
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "coupon_session_planned_count"
down_revision: str | None = "6d2309208549"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.add_column(
"coupon_session",
sa.Column("planned_coupon_count", sa.Integer(), nullable=True),
)
def downgrade() -> None:
op.drop_column("coupon_session", "planned_coupon_count")
+5 -5
View File
@@ -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}
+10 -3
View File
@@ -23,6 +23,7 @@ from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platfo
_SLOT_OK = ("success", "already_claimed")
_SLOT_TRIED = ("success", "already_claimed", "failed")
_SESSION_POINT_TOTAL = (*_SLOT_TRIED, "skipped")
def _cn_hour(dt: datetime) -> int:
@@ -166,6 +167,10 @@ def _session_to_row(
point_stats: dict | None = None,
) -> dict:
"""CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
succeeded = point_stats["succeeded"] if point_stats else 0
event_total = point_stats["tried"] if point_stats else 0
planned_total = r.planned_coupon_count or 0
point_total = max(event_total, planned_total)
return {
"id": r.id,
"trace_id": r.trace_id,
@@ -182,8 +187,8 @@ def _session_to_row(
"app_env": r.app_env,
"started_at": r.started_at,
"claimed_count": r.claimed_count,
"point_success_count": point_stats["succeeded"] if point_stats else None,
"point_total_count": point_stats["tried"] if point_stats else None,
"point_success_count": succeeded if point_total > 0 else None,
"point_total_count": point_total if point_total > 0 else None,
"trace_url": r.trace_url,
"ad_revenue_yuan": ad_revenue_yuan,
}
@@ -202,7 +207,7 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
)
.where(
CouponClaimEvent.trace_id.in_(trace_ids),
CouponClaimEvent.status.in_(_SLOT_TRIED),
CouponClaimEvent.status.in_(_SESSION_POINT_TOTAL),
)
.group_by(CouponClaimEvent.trace_id)
).all()
@@ -401,11 +406,13 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
).scalar_one()
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
point_stats_map = _point_scores_by_trace(db, [r.trace_id for r in rows])
return {
"items": [
_session_to_row(
r,
ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
point_stats=point_stats_map.get(r.trace_id),
)
for r in rows
],
+57 -14
View File
@@ -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),
}
+8
View File
@@ -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,
)
)
+7 -2
View File
@@ -79,10 +79,15 @@ class CouponDataRow(BaseModel):
started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
claimed_count: int | None = None
point_success_count: int | None = Field(
None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
None,
description="本次成功单券数(success+already_claimed);有计划数但尚无成功事件时为 0",
)
point_total_count: int | None = Field(
None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
None,
description=(
"本次计划单券数(max(逐券事件数,计划数),事件含 skipped);"
"无计划数且无逐券事件时为空"
),
)
trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
ad_revenue_yuan: float = Field(
+1 -1
View File
@@ -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 # 信息流广告提现(金币折现)
+19
View File
@@ -93,6 +93,13 @@ def _record_claims_blocking(
)
def _merge_planned_count_blocking(
trace_id: str | None, planned_count: int | None
) -> None:
with SessionLocal() as db:
coupon_repo.merge_session_planned_count(db, trace_id, planned_count)
def _mark_completed_blocking(
device_id: str, user_id: int | None, trace_id: str | None
) -> None:
@@ -176,6 +183,18 @@ async def coupon_step(
resp_json = resp.json()
# pricebot 每帧 status.progress.total 都带本轮计划券数。独立回写 session 后,
# 即使用户在第一张出结果前退出,admin 也能显示 0/N,而不是空值。
progress = (resp_json.get("status") or {}).get("progress") or {}
planned_count = progress.get("total")
if trace_id and isinstance(planned_count, int) and planned_count > 0:
try:
await run_in_threadpool(
_merge_planned_count_blocking, trace_id, planned_count
)
except Exception as e: # noqa: BLE001
logger.warning("coupon planned count write failed: %s", e)
# 领券结果沉淀:每日资产 + 逐次事件;中间帧和 done 全量帧均幂等写库。
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
if device_id:
+4
View File
@@ -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 全量),未来取数兜底
+3
View File
@@ -277,6 +277,9 @@ class CouponSession(Base):
platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
# 领到总张数(收尾帧带)。
claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# pricebot 本轮计划处理的券数。由 /coupon/step 每帧 status.progress.total 回写;
# 即使用户中途退出、尚无任何单券终态,admin 也能以计划数作为成功率分母。
planned_coupon_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 本次 session 至少领到一张(status∈{success,already_claimed})的平台 id 列表,如 ["meituan-waimai","jd-waimai"]。
# admin「领券数据」据此算整单成功率(②)/点位成功率(③);服务端 /step 逐帧按 trace_id 并集写入
# (见 coupon_state.merge_session_platform_success)。旧行=NULL → 视作空集。
+101 -12
View File
@@ -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,
+16
View File
@@ -157,6 +157,22 @@ def session_app_env(db: Session, trace_id: str | None) -> str | None:
).scalar_one_or_none()
def merge_session_planned_count(
db: Session, trace_id: str | None, planned_count: int | None
) -> None:
"""把 pricebot 队列总数回写 session,供中途退出/全跳过场次计算逐券分母。"""
if not trace_id or planned_count is None or planned_count <= 0:
return
row = db.execute(
select(CouponSession).where(CouponSession.trace_id == trace_id)
).scalar_one_or_none()
if row is None:
return
# 队列在一个 trace 内固定;取大值可防乱序旧帧覆盖,也兼容调度层补入预跳券。
row.planned_coupon_count = max(row.planned_coupon_count or 0, planned_count)
db.commit()
def record_claims(
db: Session,
device_id: str,
+10
View File
@@ -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。
+109
View File
@@ -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 返回 awareSQLite/历史转换可能返回 naive,二者必须可混排。"""
naive = datetime(2038, 1, 1, 8, 0)
+127 -5
View File
@@ -10,14 +10,16 @@ from app.admin.repositories.coupon_data import (
_point_scores_by_trace,
coupon_data_report,
coupon_point_details,
coupon_user_records,
)
from app.admin.security import create_admin_token
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimEvent, CouponSession
from app.repositories.coupon_state import merge_session_planned_count
def test_point_scores_by_trace() -> None:
"""已领算成功失败算尝试、跳过不进分母。"""
"""已领算成功失败和跳过均属于本轮计划券,进入分母。"""
db = SessionLocal()
trace = "point-score-trace"
try:
@@ -37,7 +39,7 @@ def test_point_scores_by_trace() -> None:
stats = _point_scores_by_trace(db, [trace])[trace]
assert stats["succeeded"] == 2
assert stats["tried"] == 3
assert stats["tried"] == 4
details = coupon_point_details(db, trace_id=trace)
assert [item["status"] for item in details] == [
"success", "already_claimed", "failed", "skipped"
@@ -48,8 +50,8 @@ def test_point_scores_by_trace() -> None:
db.close()
def test_skipped_detail_does_not_create_a_score() -> None:
"""仅有 skipped 时按需明细仍可查到,但列表没有虚假的 0/0 分数"""
def test_skipped_detail_creates_zero_score() -> None:
"""仅有 skipped 时也应显示 0/1,而不是把计划券静默成空值"""
db = SessionLocal()
trace = "point-score-skipped"
try:
@@ -63,7 +65,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
db.flush()
scores = _point_scores_by_trace(db, [trace, "missing-trace"])
assert trace not in scores
assert scores[trace] == {"succeeded": 0, "tried": 1}
assert "missing-trace" not in scores
assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
finally:
@@ -114,6 +116,126 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
db.close()
def test_report_uses_planned_count_when_events_are_partial_or_missing() -> None:
"""退出前只有部分/没有逐券终态时,计划数仍是稳定分母。"""
db = SessionLocal()
report_date = date(2020, 1, 6)
partial_trace = "point-score-planned-partial"
empty_trace = "point-score-planned-empty"
try:
db.add_all([
CouponSession(
trace_id=partial_trace,
device_id="planned-partial-device",
status="abandoned",
app_env="prod",
platforms=[],
planned_coupon_count=8,
started_at=datetime(2020, 1, 6, tzinfo=UTC),
started_date=report_date,
),
CouponSession(
trace_id=empty_trace,
device_id="planned-empty-device",
status="abandoned",
app_env="prod",
platforms=[],
planned_coupon_count=8,
started_at=datetime(2020, 1, 6, tzinfo=UTC),
started_date=report_date,
),
CouponClaimEvent(
trace_id=partial_trace,
device_id="planned-partial-device",
coupon_id="mt-planned-success",
claim_date=report_date,
status="success",
),
])
db.flush()
report = coupon_data_report(
db,
date_from=report_date.isoformat(),
date_to=report_date.isoformat(),
app_env="prod",
)
rows = {item["trace_id"]: item for item in report["items"]}
assert rows[partial_trace]["point_success_count"] == 1
assert rows[partial_trace]["point_total_count"] == 8
assert rows[empty_trace]["point_success_count"] == 0
assert rows[empty_trace]["point_total_count"] == 8
finally:
db.rollback()
db.close()
def test_user_records_include_actual_successes_with_planned_count() -> None:
"""用户领券记录也必须加载逐券统计,不能把有成功事件的场次显示成 0/N。"""
db = SessionLocal()
trace = "point-score-user-records"
user_id = 987656
report_date = date(2020, 1, 8)
try:
db.add_all([
CouponSession(
trace_id=trace,
device_id="planned-user-records-device",
user_id=user_id,
status="abandoned",
app_env="prod",
platforms=[],
planned_coupon_count=8,
started_at=datetime(2020, 1, 8, tzinfo=UTC),
started_date=report_date,
),
CouponClaimEvent(
trace_id=trace,
device_id="planned-user-records-device",
coupon_id="mt-user-records-success",
claim_date=report_date,
status="success",
),
])
db.flush()
records = coupon_user_records(db, user_id=user_id)
row = next(item for item in records["items"] if item["trace_id"] == trace)
assert row["point_success_count"] == 1
assert row["point_total_count"] == 8
finally:
db.rollback()
db.close()
def test_planned_count_only_grows_for_a_trace() -> None:
"""乱序帧不得用较小的队列总数覆盖已经观测到的计划数。"""
db = SessionLocal()
trace = "point-score-planned-upsert"
try:
row = CouponSession(
trace_id=trace,
device_id="planned-upsert-device",
status="started",
app_env="prod",
platforms=[],
started_at=datetime(2020, 1, 7, tzinfo=UTC),
started_date=date(2020, 1, 7),
)
db.add(row)
db.commit()
merge_session_planned_count(db, trace, 8)
merge_session_planned_count(db, trace, 3)
db.refresh(row)
assert row.planned_coupon_count == 8
finally:
db.rollback()
db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
db.commit()
db.close()
def test_coupon_point_details_endpoint() -> None:
"""前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
db = SessionLocal()