Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d8276f4986 | |||
| 4aadb28c3c | |||
| 90c6fe599a | |||
| e529112a90 |
@@ -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")
|
||||
@@ -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}
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
|
||||
@@ -298,31 +298,21 @@ def dashboard_overview(
|
||||
if period_comparison_success_denominator > 0
|
||||
else None
|
||||
)
|
||||
# “平均节省金额”表示一次可计算的成功比价平均能省多少钱:
|
||||
# 原平台已经最便宜时 saved_amount_cents=0,也必须进入分母;否则只统计
|
||||
# “确实省到钱”的记录会系统性抬高大盘指标。源价/最优价缺失导致的 NULL
|
||||
# 无法判断实际节省额,仍排除在分母之外。
|
||||
period_saved_calculable_count = _count(
|
||||
period_saved_positive_count = _count(
|
||||
ComparisonRecord,
|
||||
*period_comparison_conds,
|
||||
ComparisonRecord.status == "success",
|
||||
ComparisonRecord.saved_amount_cents.is_not(None),
|
||||
ComparisonRecord.saved_amount_cents > 0,
|
||||
)
|
||||
period_saved_sum = _sum(
|
||||
case(
|
||||
(
|
||||
ComparisonRecord.saved_amount_cents > 0,
|
||||
ComparisonRecord.saved_amount_cents,
|
||||
),
|
||||
else_=0,
|
||||
),
|
||||
period_saved_positive_sum = _sum(
|
||||
ComparisonRecord.saved_amount_cents,
|
||||
*period_comparison_conds,
|
||||
ComparisonRecord.status == "success",
|
||||
ComparisonRecord.saved_amount_cents.is_not(None),
|
||||
ComparisonRecord.saved_amount_cents > 0,
|
||||
)
|
||||
period_avg_saved_cents = (
|
||||
round(period_saved_sum / period_saved_calculable_count)
|
||||
if period_saved_calculable_count
|
||||
round(period_saved_positive_sum / period_saved_positive_count)
|
||||
if period_saved_positive_count
|
||||
else None
|
||||
)
|
||||
period_avg_duration_ms = db.execute(
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@@ -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 # 信息流广告提现(金币折现)
|
||||
|
||||
|
||||
|
||||
+11
-3
@@ -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
|
||||
|
||||
@@ -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 全量),未来取数兜底
|
||||
|
||||
+101
-12
@@ -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,
|
||||
|
||||
@@ -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。
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -1,201 +0,0 @@
|
||||
"""分批修正历史比价记录的最优平台与节省金额派生列。
|
||||
|
||||
该回填刻意不放进 Alembic:生产启动迁移不应一次性读取并更新全部历史记录。
|
||||
调用方按主键游标分批执行,每批独立事务;默认 dry-run 时只计算差异、不写库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.comparison import ComparisonRecord
|
||||
|
||||
_DERIVED_FIELDS = (
|
||||
"source_price_cents",
|
||||
"best_platform_id",
|
||||
"best_platform_name",
|
||||
"best_price_cents",
|
||||
"saved_amount_cents",
|
||||
"is_source_best",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SavingsBackfillBatch:
|
||||
"""一批回填结果;last_id 可直接作为下一批 after_id。"""
|
||||
|
||||
scanned: int
|
||||
changed: int
|
||||
last_id: int
|
||||
done: bool
|
||||
|
||||
|
||||
def _json_value(value: Any, fallback: Any) -> Any:
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
return json.loads(value)
|
||||
except (TypeError, ValueError):
|
||||
return fallback
|
||||
return value
|
||||
|
||||
|
||||
def _decimal(value: Any) -> Decimal | None:
|
||||
if value is None or isinstance(value, bool):
|
||||
return None
|
||||
try:
|
||||
return Decimal(str(value))
|
||||
except (InvalidOperation, TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _yuan_to_cents(value: Any) -> int | None:
|
||||
amount = _decimal(value)
|
||||
if amount is None:
|
||||
return None
|
||||
return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
||||
|
||||
|
||||
def _rank(value: Any) -> Decimal:
|
||||
rank = _decimal(value)
|
||||
return rank if rank is not None else Decimal("1e18")
|
||||
|
||||
|
||||
def derive_historical_savings(
|
||||
source_price_cents: int | None,
|
||||
comparison_results: Any,
|
||||
raw_payload: Any,
|
||||
) -> dict[str, Any] | None:
|
||||
"""按当前完整购物篮规则,从历史原始数据重新派生结构化金额字段。"""
|
||||
results = _json_value(comparison_results, [])
|
||||
raw = _json_value(raw_payload, {})
|
||||
if not isinstance(results, list):
|
||||
return None
|
||||
platform_results = raw.get("platform_results", {}) if isinstance(raw, dict) else {}
|
||||
if not isinstance(platform_results, dict):
|
||||
platform_results = {}
|
||||
|
||||
clean: list[tuple[dict[str, Any], Decimal]] = []
|
||||
for result in results:
|
||||
if not isinstance(result, dict):
|
||||
continue
|
||||
price = _decimal(result.get("price"))
|
||||
if price is None:
|
||||
continue
|
||||
platform_id = result.get("platform_id")
|
||||
platform_info = platform_results.get(platform_id) if platform_id else None
|
||||
skipped_count = (
|
||||
_decimal(platform_info.get("skipped_dish_count"))
|
||||
if isinstance(platform_info, dict)
|
||||
else None
|
||||
)
|
||||
if skipped_count is not None and skipped_count > 0:
|
||||
continue
|
||||
clean.append((result, price))
|
||||
|
||||
if not clean:
|
||||
return None
|
||||
|
||||
best, best_price = min(
|
||||
clean,
|
||||
key=lambda item: (_rank(item[0].get("rank")), item[1]),
|
||||
)
|
||||
if source_price_cents is None:
|
||||
source = next(
|
||||
(
|
||||
result
|
||||
for result in results
|
||||
if isinstance(result, dict)
|
||||
and result.get("is_source")
|
||||
and _decimal(result.get("price")) is not None
|
||||
),
|
||||
None,
|
||||
)
|
||||
source_price_cents = _yuan_to_cents(source.get("price")) if source else None
|
||||
|
||||
best_price_cents = _yuan_to_cents(best_price)
|
||||
saved_amount_cents = (
|
||||
source_price_cents - best_price_cents
|
||||
if source_price_cents is not None and best_price_cents is not None
|
||||
else None
|
||||
)
|
||||
return {
|
||||
"source_price_cents": source_price_cents,
|
||||
"best_platform_id": best.get("platform_id"),
|
||||
"best_platform_name": best.get("platform_name"),
|
||||
"best_price_cents": best_price_cents,
|
||||
"saved_amount_cents": saved_amount_cents,
|
||||
"is_source_best": bool(best.get("is_source")),
|
||||
}
|
||||
|
||||
|
||||
def latest_success_id(db: Session) -> int:
|
||||
"""固定本次任务的扫描上界,避免执行期间新增记录让任务无限延长。"""
|
||||
return int(
|
||||
db.execute(
|
||||
select(func.coalesce(func.max(ComparisonRecord.id), 0)).where(
|
||||
ComparisonRecord.status == "success"
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
def repair_comparison_savings_batch(
|
||||
db: Session,
|
||||
*,
|
||||
after_id: int,
|
||||
max_id: int,
|
||||
batch_size: int = 500,
|
||||
apply: bool = False,
|
||||
) -> SavingsBackfillBatch:
|
||||
"""按 id 游标处理一批;apply=True 时仅更新实际发生变化的记录并提交。"""
|
||||
if after_id < 0 or max_id < 0:
|
||||
raise ValueError("after_id and max_id must be non-negative")
|
||||
if batch_size <= 0:
|
||||
raise ValueError("batch_size must be positive")
|
||||
|
||||
records = list(
|
||||
db.scalars(
|
||||
select(ComparisonRecord)
|
||||
.where(
|
||||
ComparisonRecord.status == "success",
|
||||
ComparisonRecord.id > after_id,
|
||||
ComparisonRecord.id <= max_id,
|
||||
)
|
||||
.order_by(ComparisonRecord.id)
|
||||
.limit(batch_size)
|
||||
)
|
||||
)
|
||||
changed = 0
|
||||
for record in records:
|
||||
derived = derive_historical_savings(
|
||||
record.source_price_cents,
|
||||
record.comparison_results,
|
||||
record.raw_payload,
|
||||
)
|
||||
if derived is None:
|
||||
continue
|
||||
differs = any(
|
||||
getattr(record, field) != derived[field] for field in _DERIVED_FIELDS
|
||||
)
|
||||
if not differs:
|
||||
continue
|
||||
changed += 1
|
||||
if apply:
|
||||
for field in _DERIVED_FIELDS:
|
||||
setattr(record, field, derived[field])
|
||||
|
||||
if apply:
|
||||
db.commit()
|
||||
|
||||
last_id = records[-1].id if records else after_id
|
||||
return SavingsBackfillBatch(
|
||||
scanned=len(records),
|
||||
changed=changed,
|
||||
last_id=last_id,
|
||||
done=not records or len(records) < batch_size or last_id >= max_id,
|
||||
)
|
||||
@@ -1,103 +0,0 @@
|
||||
"""分批回填历史比价节省金额。
|
||||
|
||||
默认只预览,不写库:
|
||||
python -m scripts.backfill_comparison_savings
|
||||
|
||||
确认预览后实际执行:
|
||||
python -m scripts.backfill_comparison_savings --apply
|
||||
|
||||
中断后可从日志最后一个 last_id 继续:
|
||||
python -m scripts.backfill_comparison_savings --apply --after-id 12345
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
from app.services.comparison_savings_backfill import (
|
||||
latest_success_id,
|
||||
repair_comparison_savings_batch,
|
||||
)
|
||||
|
||||
|
||||
def _positive_int(value: str) -> int:
|
||||
parsed = int(value)
|
||||
if parsed <= 0:
|
||||
raise argparse.ArgumentTypeError("必须为正整数")
|
||||
return parsed
|
||||
|
||||
|
||||
def _non_negative_int(value: str) -> int:
|
||||
parsed = int(value)
|
||||
if parsed < 0:
|
||||
raise argparse.ArgumentTypeError("不能为负数")
|
||||
return parsed
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="按完整购物篮口径分批回填历史比价节省金额(默认 dry-run)"
|
||||
)
|
||||
parser.add_argument("--apply", action="store_true", help="实际写库;默认仅预览")
|
||||
parser.add_argument("--batch-size", type=_positive_int, default=500)
|
||||
parser.add_argument(
|
||||
"--after-id",
|
||||
type=_non_negative_int,
|
||||
default=0,
|
||||
help="只处理大于该主键的记录,用于断点续跑",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-id",
|
||||
type=_non_negative_int,
|
||||
help="固定扫描上界;不传则启动时取成功记录最大主键",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-batches",
|
||||
type=_positive_int,
|
||||
help="本次最多处理多少批,便于灰度限量执行",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
with SessionLocal() as db:
|
||||
max_id = args.max_id if args.max_id is not None else latest_success_id(db)
|
||||
|
||||
mode = "APPLY" if args.apply else "DRY-RUN"
|
||||
after_id = args.after_id
|
||||
scanned = 0
|
||||
changed = 0
|
||||
batches = 0
|
||||
print(
|
||||
f"comparison savings backfill mode={mode} "
|
||||
f"after_id={after_id} max_id={max_id} batch_size={args.batch_size}"
|
||||
)
|
||||
|
||||
while after_id < max_id:
|
||||
with SessionLocal() as db:
|
||||
result = repair_comparison_savings_batch(
|
||||
db,
|
||||
after_id=after_id,
|
||||
max_id=max_id,
|
||||
batch_size=args.batch_size,
|
||||
apply=args.apply,
|
||||
)
|
||||
batches += 1
|
||||
scanned += result.scanned
|
||||
changed += result.changed
|
||||
after_id = result.last_id
|
||||
print(
|
||||
f"batch={batches} scanned={result.scanned} changed={result.changed} "
|
||||
f"last_id={result.last_id} done={result.done}"
|
||||
)
|
||||
if result.done or (
|
||||
args.max_batches is not None and batches >= args.max_batches
|
||||
):
|
||||
break
|
||||
|
||||
print(
|
||||
f"completed mode={mode} batches={batches} scanned={scanned} "
|
||||
f"changed={changed} last_id={after_id} max_id={max_id}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+109
-49
@@ -120,55 +120,6 @@ def test_dashboard_period_comparison_is_aggregated_by_backend(
|
||||
assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
|
||||
|
||||
|
||||
def test_dashboard_average_saved_includes_zero_and_excludes_uncalculable(
|
||||
admin_client: TestClient, admin_token: str
|
||||
) -> None:
|
||||
created_at = datetime(2037, 2, 15, 12)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db.add_all(
|
||||
[
|
||||
ComparisonRecord(
|
||||
trace_id="dashboard-saved-positive",
|
||||
status="success",
|
||||
saved_amount_cents=100,
|
||||
created_at=created_at,
|
||||
),
|
||||
ComparisonRecord(
|
||||
trace_id="dashboard-saved-source-best",
|
||||
status="success",
|
||||
saved_amount_cents=0,
|
||||
created_at=created_at,
|
||||
),
|
||||
ComparisonRecord(
|
||||
trace_id="dashboard-saved-uncalculable",
|
||||
status="success",
|
||||
saved_amount_cents=None,
|
||||
created_at=created_at,
|
||||
),
|
||||
ComparisonRecord(
|
||||
trace_id="dashboard-saved-failed",
|
||||
status="failed",
|
||||
saved_amount_cents=900,
|
||||
created_at=created_at,
|
||||
),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
response = admin_client.get(
|
||||
"/admin/api/stats/overview",
|
||||
params={"date_from": "2037-02-15", "date_to": "2037-02-15"},
|
||||
headers=_auth(admin_token),
|
||||
)
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
# (100 + 0) / 2;NULL 无法计算,failed 不属于成功比价。
|
||||
assert response.json()["period"]["comparison"]["average_saved_cents"] == 50
|
||||
|
||||
|
||||
def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
|
||||
uid = _seed_user_with_data("13800000002")
|
||||
r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
|
||||
@@ -268,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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -1,136 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.services.comparison_savings_backfill import (
|
||||
latest_success_id,
|
||||
repair_comparison_savings_batch,
|
||||
)
|
||||
|
||||
|
||||
def _record(trace_id: str, *, status: str = "success") -> ComparisonRecord:
|
||||
return ComparisonRecord(
|
||||
trace_id=trace_id,
|
||||
status=status,
|
||||
source_price_cents=4200,
|
||||
best_platform_id="missing-dishes",
|
||||
best_platform_name="Incomplete",
|
||||
best_price_cents=2500,
|
||||
saved_amount_cents=1700,
|
||||
is_source_best=False,
|
||||
comparison_results=[
|
||||
{
|
||||
"platform_id": "source",
|
||||
"platform_name": "Source",
|
||||
"price": 42,
|
||||
"rank": 3,
|
||||
"is_source": True,
|
||||
},
|
||||
{
|
||||
"platform_id": "missing-dishes",
|
||||
"platform_name": "Incomplete",
|
||||
"price": 25,
|
||||
"rank": 1,
|
||||
"is_source": False,
|
||||
},
|
||||
{
|
||||
"platform_id": "complete",
|
||||
"platform_name": "Complete",
|
||||
"price": 38.5,
|
||||
"rank": 2,
|
||||
"is_source": False,
|
||||
},
|
||||
],
|
||||
raw_payload={
|
||||
"platform_results": {
|
||||
"missing-dishes": {"skipped_dish_count": 1},
|
||||
"complete": {"skipped_dish_count": 0},
|
||||
}
|
||||
},
|
||||
created_at=datetime(2037, 3, 1, 12),
|
||||
)
|
||||
|
||||
|
||||
def test_comparison_savings_backfill_is_batched_dry_run_and_idempotent() -> None:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
affected = _record("savings-backfill-affected")
|
||||
unchanged = _record("savings-backfill-unchanged")
|
||||
unchanged.best_platform_id = "complete"
|
||||
unchanged.best_platform_name = "Complete"
|
||||
unchanged.best_price_cents = 3850
|
||||
unchanged.saved_amount_cents = 350
|
||||
failed = _record("savings-backfill-failed", status="failed")
|
||||
db.add_all([affected, unchanged, failed])
|
||||
db.commit()
|
||||
affected_id = affected.id
|
||||
unchanged_id = unchanged.id
|
||||
failed_id = failed.id
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
assert latest_success_id(db) >= unchanged_id
|
||||
max_id = unchanged_id
|
||||
affected = db.get(ComparisonRecord, affected_id)
|
||||
assert affected is not None
|
||||
|
||||
preview = repair_comparison_savings_batch(
|
||||
db,
|
||||
after_id=affected_id - 1,
|
||||
max_id=max_id,
|
||||
batch_size=1,
|
||||
)
|
||||
assert preview.scanned == 1
|
||||
assert preview.changed == 1
|
||||
assert preview.last_id == affected_id
|
||||
assert not preview.done
|
||||
db.refresh(affected)
|
||||
assert affected.best_platform_id == "missing-dishes"
|
||||
assert affected.saved_amount_cents == 1700
|
||||
|
||||
applied = repair_comparison_savings_batch(
|
||||
db,
|
||||
after_id=affected_id - 1,
|
||||
max_id=max_id,
|
||||
batch_size=1,
|
||||
apply=True,
|
||||
)
|
||||
assert applied.changed == 1
|
||||
db.refresh(affected)
|
||||
assert affected.best_platform_id == "complete"
|
||||
assert affected.best_platform_name == "Complete"
|
||||
assert affected.best_price_cents == 3850
|
||||
assert affected.saved_amount_cents == 350
|
||||
|
||||
second_batch = repair_comparison_savings_batch(
|
||||
db,
|
||||
after_id=applied.last_id,
|
||||
max_id=max_id,
|
||||
batch_size=1,
|
||||
apply=True,
|
||||
)
|
||||
assert second_batch.scanned == 1
|
||||
assert second_batch.changed == 0
|
||||
assert second_batch.last_id == unchanged_id
|
||||
assert second_batch.done
|
||||
|
||||
rerun = repair_comparison_savings_batch(
|
||||
db,
|
||||
after_id=affected_id - 1,
|
||||
max_id=max_id,
|
||||
batch_size=10,
|
||||
apply=True,
|
||||
)
|
||||
assert rerun.scanned == 2
|
||||
assert rerun.changed == 0
|
||||
|
||||
failed_row = db.get(ComparisonRecord, failed_id)
|
||||
assert failed_row is not None
|
||||
assert failed_row.best_platform_id == "missing-dishes"
|
||||
assert failed_row.saved_amount_cents == 1700
|
||||
finally:
|
||||
db.close()
|
||||
Reference in New Issue
Block a user