Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 44eb310bf8 | |||
| 84251770b4 | |||
| f1dd0f4a3d | |||
| 7dee829cfc | |||
| 2e91c9f72f | |||
| a69b7d777d | |||
| 0fc8521c3b | |||
| 6003d91fc1 |
@@ -0,0 +1,76 @@
|
||||
"""normalize granular comparison outcomes into terminal record statuses
|
||||
|
||||
Revision ID: comparison_below_min_success
|
||||
Revises: limit_policy_global_bundle
|
||||
"""
|
||||
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_below_min_success"
|
||||
down_revision: str | Sequence[str] | None = "limit_policy_global_bundle"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
|
||||
_FAILED_OUTCOMES = (
|
||||
"store_closed",
|
||||
"store_not_found",
|
||||
"items_not_found",
|
||||
"no_delivery",
|
||||
"unsupported",
|
||||
)
|
||||
|
||||
|
||||
def _comparison_record() -> sa.TableClause:
|
||||
return sa.table(
|
||||
"comparison_record",
|
||||
sa.column("status", sa.String(16)),
|
||||
sa.column("fail_reason", sa.String(256)),
|
||||
sa.column("raw_payload", _JSON),
|
||||
)
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
comparison_record = _comparison_record()
|
||||
op.execute(
|
||||
comparison_record.update()
|
||||
.where(comparison_record.c.status == "below_minimum")
|
||||
.values(status="success", fail_reason=None)
|
||||
)
|
||||
op.execute(
|
||||
comparison_record.update()
|
||||
.where(comparison_record.c.status.in_(_FAILED_OUTCOMES))
|
||||
.values(status="failed")
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
comparison_record = _comparison_record()
|
||||
# The write paths deliberately preserve the granular outcome. Restore only
|
||||
# rows that this change normalized, without touching ordinary successes.
|
||||
raw_outcome = sa.func.coalesce(
|
||||
comparison_record.c.raw_payload["record_status"].as_string(),
|
||||
comparison_record.c.raw_payload["status"].as_string(),
|
||||
)
|
||||
op.execute(
|
||||
comparison_record.update()
|
||||
.where(
|
||||
comparison_record.c.status == "success",
|
||||
raw_outcome == "below_minimum",
|
||||
)
|
||||
.values(status="below_minimum")
|
||||
)
|
||||
op.execute(
|
||||
comparison_record.update()
|
||||
.where(
|
||||
comparison_record.c.status == "failed",
|
||||
raw_outcome.in_(_FAILED_OUTCOMES),
|
||||
)
|
||||
.values(status=raw_outcome)
|
||||
)
|
||||
@@ -39,6 +39,34 @@ from app.repositories import activity, ad_ecpm
|
||||
# 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
|
||||
_NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
|
||||
|
||||
# comparison_record historically persisted a few granular business outcomes as
|
||||
# top-level statuses. Admin filters and metrics expose lifecycle buckets while
|
||||
# retaining the raw values until the data migration has run everywhere.
|
||||
_COMPARISON_STATUS_ALIASES = {
|
||||
"success": ("success", "below_minimum"),
|
||||
"failed": (
|
||||
"failed",
|
||||
"store_closed",
|
||||
"store_not_found",
|
||||
"items_not_found",
|
||||
"no_delivery",
|
||||
"unsupported",
|
||||
),
|
||||
"cancelled": ("cancelled",),
|
||||
"running": ("running",),
|
||||
}
|
||||
_COMPARISON_SUCCESS_STATUSES = _COMPARISON_STATUS_ALIASES["success"]
|
||||
_COMPARISON_FAILED_STATUSES = _COMPARISON_STATUS_ALIASES["failed"]
|
||||
_COMPARISON_COMPLETED_STATUSES = (
|
||||
*_COMPARISON_SUCCESS_STATUSES,
|
||||
*_COMPARISON_FAILED_STATUSES,
|
||||
)
|
||||
|
||||
|
||||
def _comparison_status_condition(status: str):
|
||||
values = _COMPARISON_STATUS_ALIASES.get(status, (status,))
|
||||
return ComparisonRecord.status.in_(values)
|
||||
|
||||
# 信息流点位场景 → 金币记录「赚取途径」展示名;NULL/未知 = 历史未分类。
|
||||
_FEED_SCENE_LABEL = {
|
||||
"comparison": "比价信息流",
|
||||
@@ -334,7 +362,7 @@ def _comparison_conditions(
|
||||
)
|
||||
)
|
||||
if status:
|
||||
conditions.append(ComparisonRecord.status == status)
|
||||
conditions.append(_comparison_status_condition(status))
|
||||
if business_type:
|
||||
conditions.append(ComparisonRecord.business_type == business_type)
|
||||
if store:
|
||||
@@ -419,7 +447,7 @@ def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles
|
||||
),
|
||||
).where(
|
||||
*conditions,
|
||||
ComparisonRecord.status == status,
|
||||
_comparison_status_condition(status),
|
||||
ComparisonRecord.total_ms.is_not(None),
|
||||
)
|
||||
|
||||
@@ -444,7 +472,7 @@ def _comparison_duration_aggregates(
|
||||
select(ComparisonRecord.total_ms)
|
||||
.where(
|
||||
*conditions,
|
||||
ComparisonRecord.status == status,
|
||||
_comparison_status_condition(status),
|
||||
ComparisonRecord.total_ms.is_not(None),
|
||||
)
|
||||
.order_by(ComparisonRecord.total_ms)
|
||||
@@ -474,11 +502,11 @@ def comparison_records_summary(
|
||||
row = db.execute(
|
||||
select(
|
||||
func.count(ComparisonRecord.id),
|
||||
func.sum(case((ComparisonRecord.status.in_(("success", "failed")), 1), else_=0)),
|
||||
func.sum(case((ComparisonRecord.status == "success", 1), else_=0)),
|
||||
func.sum(case((ComparisonRecord.status.in_(_COMPARISON_COMPLETED_STATUSES), 1), else_=0)),
|
||||
func.sum(case((ComparisonRecord.status.in_(_COMPARISON_SUCCESS_STATUSES), 1), else_=0)),
|
||||
func.avg(ComparisonRecord.llm_cost_yuan),
|
||||
func.sum(case((
|
||||
(ComparisonRecord.status == "success")
|
||||
ComparisonRecord.status.in_(_COMPARISON_SUCCESS_STATUSES)
|
||||
& (ComparisonRecord.saved_amount_cents > 0), 1
|
||||
), else_=0)),
|
||||
func.sum(case((ComparisonRecord.status == "cancelled", 1), else_=0)),
|
||||
|
||||
@@ -35,7 +35,7 @@ def list_comparison_records(
|
||||
db: AdminDb,
|
||||
user_id: Annotated[int | None, Query()] = None,
|
||||
phone: Annotated[str | None, Query(description="手机号前缀")] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(success|failed|cancelled)$")] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(success|failed|cancelled|running)$")] = None,
|
||||
business_type: Annotated[str | None, Query()] = None,
|
||||
store: Annotated[str | None, Query(description="店名子串模糊匹配")] = None,
|
||||
product: Annotated[str | None, Query(description="商品名子串模糊匹配")] = None,
|
||||
@@ -66,7 +66,7 @@ def comparison_records_summary(
|
||||
db: AdminDb,
|
||||
user_id: Annotated[int | None, Query()] = None,
|
||||
phone: Annotated[str | None, Query(description="手机号前缀")] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(success|failed|cancelled)$")] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(success|failed|cancelled|running)$")] = None,
|
||||
business_type: Annotated[str | None, Query()] = None,
|
||||
store: Annotated[str | None, Query(description="店名子串模糊匹配")] = None,
|
||||
product: Annotated[str | None, Query(description="商品名子串模糊匹配")] = None,
|
||||
|
||||
@@ -20,7 +20,7 @@ class AdminComparisonListItem(BaseModel):
|
||||
trace_id: str
|
||||
# admin 是 debug 工具,无条件下发 trace_url(不看 user.debug_trace_enabled)
|
||||
trace_url: str | None = None
|
||||
status: str
|
||||
status: str # success / failed / cancelled / running;旧细分值由前端兼容映射
|
||||
information: str | None = None
|
||||
store_name: str | None = None
|
||||
product_names: str | None = None # 下单商品名派生串(顿号分隔;「商品」列展示 + 商品搜索)
|
||||
@@ -83,6 +83,7 @@ class AdminComparisonDetail(AdminComparisonListItem):
|
||||
skipped_dish_count: int | None = None
|
||||
device_id: str | None = None
|
||||
items: list = []
|
||||
platforms: list = [] # pricebot 渲染就绪的逐平台卡片模型(status=ok/业务失败细分)
|
||||
comparison_results: list = [] # 逐平台对比(价格/rank/coupon/打烊...)
|
||||
skipped_dish_names: list = []
|
||||
# 全量环境
|
||||
|
||||
@@ -21,7 +21,6 @@ from __future__ import annotations
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@@ -32,6 +31,7 @@ from app.api.deps import DbSession, OptionalUser
|
||||
from app.core.config import settings
|
||||
from app.core.logging import trace_id_ctx
|
||||
from app.core.pricebot_client import get_pricebot_client
|
||||
from app.core.trace_ids import new_trace_id
|
||||
from app.core.pricebot_router import pick_pricebot
|
||||
from app.db.session import SessionLocal
|
||||
from app.repositories import comparison as crud_compare
|
||||
@@ -142,7 +142,7 @@ async def _forward(
|
||||
trace_id = meta.get("trace_id")
|
||||
minted = False
|
||||
if not trace_id:
|
||||
trace_id = str(uuid.uuid4())
|
||||
trace_id = new_trace_id()
|
||||
meta["trace_id"] = trace_id
|
||||
raw = json.dumps(meta).encode() # 仅首帧重新序列化(注入 trace_id);后续帧走原始 bytes
|
||||
minted = True
|
||||
|
||||
@@ -17,6 +17,7 @@ from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
|
||||
|
||||
from app.api.deps import CurrentUser, DbSession
|
||||
from app.core import limit_policy
|
||||
from app.core.trace_ids import new_trace_id
|
||||
from app.repositories import comparison as crud_compare
|
||||
from app.repositories import risk as risk_repo
|
||||
from app.schemas.compare_record import (
|
||||
@@ -53,6 +54,10 @@ def reserve_compare_start(
|
||||
scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
|
||||
):
|
||||
raise HTTPException(status_code=403, detail="账号存在异常,该功能暂不可用")
|
||||
# trace_id 统一由服务端签发(客户端不带时):预占额度本就是任务的第一个请求,
|
||||
# 签发与建 running 行合一,此后 Phase1/Phase2/记录/前端日志全链用同一个 id。
|
||||
# 客户端带了则沿用——老客户端兼容 + 同 trace 重试幂等(reserve_daily_start 按 trace_id 去重)。
|
||||
trace_id = payload.trace_id or new_trace_id()
|
||||
try:
|
||||
policy = limit_policy.resolve(
|
||||
db,
|
||||
@@ -63,7 +68,7 @@ def reserve_compare_start(
|
||||
rec, used = crud_compare.reserve_daily_start(
|
||||
db,
|
||||
user_id=user.id,
|
||||
trace_id=payload.trace_id,
|
||||
trace_id=trace_id,
|
||||
business_type=payload.business_type,
|
||||
device_id=payload.device_id,
|
||||
limit=policy.limit,
|
||||
@@ -94,6 +99,7 @@ def reserve_compare_start(
|
||||
limit=policy.limit,
|
||||
used=used,
|
||||
remaining=max(policy.limit - used, 0) if policy.limit is not None else None,
|
||||
trace_id=trace_id,
|
||||
)
|
||||
|
||||
|
||||
|
||||
+31
-5
@@ -21,6 +21,7 @@ from fastapi.concurrency import run_in_threadpool
|
||||
from app.api.deps import CurrentUser, DbSession
|
||||
from app.core.config import settings
|
||||
from app.core.pricebot_client import get_pricebot_client
|
||||
from app.core.trace_ids import new_trace_id
|
||||
from app.core.pricebot_router import pick_pricebot
|
||||
from app.db.session import SessionLocal
|
||||
from app.repositories import coupon_state as coupon_repo
|
||||
@@ -30,6 +31,7 @@ from app.schemas.coupon_state import (
|
||||
CouponPromptShouldShowOut,
|
||||
CouponPromptShownIn,
|
||||
CouponSessionIn,
|
||||
CouponSessionOut,
|
||||
CouponStatsOut,
|
||||
)
|
||||
|
||||
@@ -175,6 +177,12 @@ async def coupon_step(
|
||||
)
|
||||
|
||||
resp_json = resp.json()
|
||||
# 每帧响应顶层回传本次任务 trace_id(对齐 compare _forward 的 setdefault):客户端任一帧
|
||||
# 都能从响应拿到全链 id。**只回显请求里带的、不 mint**——step 是循环接口,每帧签新 id
|
||||
# 会把一次任务打散;领券 trace_id 的唯一签发点在 /coupon/session (status=started)。
|
||||
# pricebot 响应顶层本无 trace_id(只有 trace_url),setdefault 不会覆盖任何上游值。
|
||||
if isinstance(resp_json, dict) and trace_id:
|
||||
resp_json.setdefault("trace_id", trace_id)
|
||||
|
||||
# 领券结果沉淀:每日资产 + 逐次事件;中间帧和 done 全量帧均幂等写库。
|
||||
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
|
||||
@@ -204,15 +212,33 @@ async def coupon_step(
|
||||
return resp_json
|
||||
|
||||
|
||||
@router.post("/session", summary="领券任务流水上报(admin 领券数据看板数据源)")
|
||||
def coupon_session(payload: CouponSessionIn, db: DbSession) -> dict[str, bool]:
|
||||
@router.post(
|
||||
"/session",
|
||||
response_model=CouponSessionOut,
|
||||
summary="领券任务流水上报(admin 领券数据看板数据源;started 兼签发本轮 trace_id)",
|
||||
)
|
||||
def coupon_session(payload: CouponSessionIn, db: DbSession) -> CouponSessionOut:
|
||||
"""客户端两段上报一次领券流水(发起 started / 收尾 completed-failed-abandoned),按 trace_id upsert
|
||||
到 coupon_session。不鉴权(同领券循环 MVP,按 device_id/trace_id);供 admin「领券数据」看板算
|
||||
发起/完成数、耗时分位、机型维度。写库失败不应连累客户端(本就 fire-and-forget),吞掉返回 ok。"""
|
||||
发起/完成数、耗时分位、机型维度。写库失败不应连累客户端(本就 fire-and-forget),吞掉返回 ok。
|
||||
|
||||
trace_id 统一由后端签发:started 不带 trace_id → 签发 uuid 并随响应返回,客户端全程用它
|
||||
(领券 step 循环 / 收尾上报 / 前端运行日志)。签发不依赖写库成功——写库失败照样返回 trace_id,
|
||||
后续收尾上报 upsert 会补建行。非 started 缺 trace_id 不签发(收尾没有 id 只能是异常调用,
|
||||
签发新 id 只会造出一行查不到发起信息的孤儿),不写库、trace_id=null 返回。
|
||||
"""
|
||||
trace_id = payload.trace_id or (
|
||||
new_trace_id() if payload.status == "started" else None
|
||||
)
|
||||
if trace_id is None:
|
||||
logger.warning(
|
||||
"coupon session missing trace_id for status=%s (skip write)", payload.status
|
||||
)
|
||||
return CouponSessionOut(ok=True, trace_id=None)
|
||||
try:
|
||||
coupon_repo.upsert_coupon_session(
|
||||
db,
|
||||
trace_id=payload.trace_id,
|
||||
trace_id=trace_id,
|
||||
device_id=payload.device_id,
|
||||
status=payload.status,
|
||||
started_at_ms=payload.started_at_ms,
|
||||
@@ -229,7 +255,7 @@ def coupon_session(payload: CouponSessionIn, db: DbSession) -> dict[str, bool]:
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("coupon session write failed: %s", e)
|
||||
return {"ok": True}
|
||||
return CouponSessionOut(ok=True, trace_id=trace_id)
|
||||
|
||||
|
||||
@router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)")
|
||||
|
||||
+8
-4
@@ -234,13 +234,17 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: i
|
||||
会铸天量金币;钳在这唯一入口,feed 与 reward_video 回退客户端 eCPM 的路径都护住,且阈值高于
|
||||
所有真实值,不影响正规发奖。
|
||||
|
||||
下限(2026-07):有真实正 eCPM 时单份至少 1 金币——低 eCPM 单份收益四舍五入成 0 时兜底为 1,
|
||||
避免用户看了广告却因数值太小被记 too_short 零发。eCPM 缺失/为 0/非法(没有真实广告价值)仍返 0,
|
||||
不凭空铸币、不破坏 ecpm_missing 语义;防刷上限仍由 AD_ECPM_MAX_FEN 钳顶把守。
|
||||
下限(2026-08,产品口径「看了就保底 1」):**任何输入**都至少 1 金币,与前端展示公式
|
||||
FeedRewardFormula.singleUnitCoin 完全对齐(那边注释:"无论 eCPM 是否为空、非法或非正数,
|
||||
单条广告最低都发 1 金币,不能出现 +0")。此前 eCPM 缺失/为 0 返 0,造成两端不一致:
|
||||
小球显示 +1、后端信息流记 too_short 零发;激励视频侧 "0" 字符串还是 truthy、绕过
|
||||
ecpm_missing 的 `if not ecpm_raw` 判定,落成 granted 0 币且白占当日额度/LT 计数。
|
||||
防刷影响:伪造 eCPM≤0 每天至多多骗 每日上限×1 金币(500 金币=0.05 元),量级可控;
|
||||
天价伪造仍由 AD_ECPM_MAX_FEN 钳顶把守。
|
||||
"""
|
||||
ecpm_yuan = min(parse_ecpm_yuan(ecpm), AD_ECPM_MAX_FEN / 100.0)
|
||||
if ecpm_yuan <= 0:
|
||||
return 0
|
||||
return 1 # 保底:缺失/为 0/非法也发 1(镜像前端 validEcpmFen 判非法 → 直接返 1)
|
||||
yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(count_after_this)
|
||||
return max(1, round(yuan * COIN_PER_YUAN))
|
||||
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
"""trace_id 签发(全后端唯一签发口径, 2026-07 起替代裸 uuid4)。
|
||||
|
||||
格式: "YYYYMMDD_HHMMSS_" + 12 位小写 hex 随机, 共 28 字符, 如
|
||||
20260731_162254_a1b2c3d4e5f6
|
||||
|
||||
Why 带时间前缀: pricebot 落盘目录名/trace_url 尾段**直接用 trace_id 本身**
|
||||
(见 pricebot app/utils/trace_ids.py), id/目录/URL 三者合一——此前 uuid trace_id
|
||||
与 {首帧时刻}_{uuid[:16]} 目录名是两套标识, URL 只有 pricebot 能拼、按前缀反查
|
||||
还有同秒歧义。时间用北京时间(CN_TZ)——不依赖各机器 TZ 配置, 与业务时区一致。
|
||||
|
||||
唯一性: 秒级前缀 + 48bit 随机(hex12), 同一秒内碰撞概率可忽略(比价/领券发起 QPS
|
||||
远低于产生生日碰撞的量级); pricebot 侧同秒多 trace 靠随机段区分(目录精确匹配、
|
||||
llm jsonl 按尾 12 分文件, 不做前缀模糊匹配)。
|
||||
|
||||
兼容: 三个签发点(compare/start、coupon/session started、compare.py _forward mint)
|
||||
统一走这里; 客户端自带 trace_id(老客户端/重试幂等)仍原样沿用——pricebot 对老
|
||||
uuid 格式保持既有目录/短标识行为, 两代 id 并行不冲突。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from app.core.rewards import CN_TZ
|
||||
|
||||
|
||||
def new_trace_id() -> str:
|
||||
"""签发一个自描述 trace_id: 北京时间前缀 + 12 位 hex 随机。"""
|
||||
return f"{datetime.now(CN_TZ):%Y%m%d_%H%M%S}_{uuid.uuid4().hex[:12]}"
|
||||
@@ -97,7 +97,7 @@ class ComparisonRecord(Base):
|
||||
total_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
skipped_dish_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
|
||||
# success(拿到有效对比)/ failed(出错或没采到目标价)
|
||||
# success(流程正常完成,含 below_minimum)/ failed(技术异常或未形成可比报价,含店铺打烊等)
|
||||
status: Mapped[str] = mapped_column(String(16), nullable=False, default="success")
|
||||
# done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
|
||||
# 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
|
||||
|
||||
@@ -76,6 +76,30 @@ _BIZ_STATUS_PRIORITY = (
|
||||
)
|
||||
|
||||
|
||||
def _normalize_record_status(status: str | None) -> str | None:
|
||||
"""Map granular business outcomes onto the record lifecycle status.
|
||||
|
||||
``record_status`` describes the business outcome, while
|
||||
``comparison_record.status`` is also the completed-comparison flag used by
|
||||
milestones, stats and idempotent rewards. ``below_minimum`` is a completed
|
||||
success because the target cart produced a trustworthy conclusion. Other
|
||||
known target-side outcomes did not produce a comparable quote and belong
|
||||
to the failed record bucket. The granular outcome remains in ``raw_payload``
|
||||
and ``platform_results`` for result rendering.
|
||||
"""
|
||||
if status == "below_minimum":
|
||||
return "success"
|
||||
if status in {
|
||||
"store_closed",
|
||||
"store_not_found",
|
||||
"items_not_found",
|
||||
"no_delivery",
|
||||
"unsupported",
|
||||
}:
|
||||
return "failed"
|
||||
return status
|
||||
|
||||
|
||||
def _store_closed_text(reason: str | None) -> str:
|
||||
"""打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
|
||||
r = reason or ""
|
||||
@@ -164,9 +188,10 @@ def _derive(payload: ComparisonRecordIn) -> dict:
|
||||
|
||||
is_source_best = best.is_source if best is not None else None
|
||||
|
||||
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
|
||||
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
|
||||
status = payload.record_status or payload.status
|
||||
# status:优先 pricebot record_status → 客户端显式 status → 兜底派生。
|
||||
# below_minimum 是已形成可信结论的正常完成态,记录级归 success;细分结局仍完整保留在
|
||||
# raw_payload/platform_results,供结果卡展示"未满起送"。
|
||||
status = _normalize_record_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
|
||||
@@ -201,7 +226,9 @@ def upsert_record(
|
||||
# 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
|
||||
# _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
|
||||
if payload.platforms:
|
||||
derived = _derive_from_platforms(payload.platforms, payload.record_status)
|
||||
derived = _derive_from_platforms(
|
||||
payload.platforms, payload.record_status or payload.status
|
||||
)
|
||||
# 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
|
||||
derived["fail_reason"] = (
|
||||
_derive_fail_display(payload.information, payload.platform_results or {})
|
||||
@@ -356,10 +383,10 @@ 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,
|
||||
# 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
|
||||
# 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
|
||||
# 回退老的 success/failed 二态派生, 向后兼容。
|
||||
"status": record_status or ("success" if has_valid_target else "failed"),
|
||||
# below_minimum 已完成到购物车并形成可信结论,记录级计 success;细分结局仍在
|
||||
# raw_payload/platform_results。旧 pricebot 未下发 record_status 时回退二态派生。
|
||||
"status": _normalize_record_status(record_status)
|
||||
or ("success" if has_valid_target else "failed"),
|
||||
}
|
||||
|
||||
|
||||
@@ -412,7 +439,8 @@ def _derive_from_platforms(
|
||||
"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"),
|
||||
"status": _normalize_record_status(record_status)
|
||||
or ("success" if has_valid_target else "failed"),
|
||||
}
|
||||
|
||||
|
||||
@@ -584,7 +612,8 @@ def harvest_done(
|
||||
行不存在(理论上帧0已建;防御)则新建。"""
|
||||
results = done_params.get("comparison_results") or []
|
||||
# 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
|
||||
# record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
|
||||
# record_status: 记录级业务结局(success/below_minimum/store_closed/failed)。其中
|
||||
# below_minimum 是正常完成态,持久化 status 归 success,原值仍随 done_params 落 raw_payload。
|
||||
platforms = done_params.get("platforms") or []
|
||||
record_status = done_params.get("record_status")
|
||||
# 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
|
||||
|
||||
@@ -25,6 +25,7 @@ from datetime import datetime, timedelta
|
||||
from sqlalchemy import delete, func, select, update
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.rewards import CN_TZ
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.ops_marquee_seed import OpsMarqueeSeed
|
||||
@@ -54,7 +55,7 @@ _SEED_MAX_CENTS = 100000
|
||||
# 展示层随机(抽样/金额/时间/名字合成)仍每次重算,缓存只省查询;新记录最多晚 30s 进轮播,可接受。
|
||||
_REAL_ROWS_TTL_SECONDS = 30
|
||||
_REAL_ROWS_FETCH_CAP = 600 # 一次多取些,够 limit≤30 去重后取数;命中缓存后复用
|
||||
_real_rows_cache: dict = {"at": None, "rows": None}
|
||||
_real_rows_cache: dict = {"at": None, "rows": None, "test_phones": None}
|
||||
|
||||
# ===== 用户标识脱敏(对齐 PRD) + 种子无真实昵称时的假名合成 =====
|
||||
# 脱敏规则(按字符数,中英文皆适用):有昵称→n≥5「首+***+末」、n=4「首+**+末」、n≤3「首+**」;
|
||||
@@ -198,10 +199,16 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
|
||||
返回纯元组(脱离 session),可安全跨请求复用。极端并发下偶尔多查一次(无锁、幂等),纯门面无副作用。
|
||||
"""
|
||||
now = datetime.now(CN_TZ)
|
||||
test_phones = tuple(sorted(settings.test_account_phones))
|
||||
cached, at = _real_rows_cache["rows"], _real_rows_cache["at"]
|
||||
if cached is not None and at is not None and (now - at).total_seconds() < _REAL_ROWS_TTL_SECONDS:
|
||||
if (
|
||||
cached is not None
|
||||
and at is not None
|
||||
and _real_rows_cache["test_phones"] == test_phones
|
||||
and (now - at).total_seconds() < _REAL_ROWS_TTL_SECONDS
|
||||
):
|
||||
return cached
|
||||
rows = db.execute(
|
||||
stmt = (
|
||||
select(ComparisonRecord.user_id, ComparisonRecord.saved_amount_cents, User.nickname)
|
||||
.join(User, User.id == ComparisonRecord.user_id)
|
||||
.where(
|
||||
@@ -211,9 +218,12 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
|
||||
)
|
||||
.order_by(ComparisonRecord.created_at.desc())
|
||||
.limit(_REAL_ROWS_FETCH_CAP)
|
||||
).all()
|
||||
)
|
||||
if test_phones:
|
||||
stmt = stmt.where(User.phone.not_in(test_phones))
|
||||
rows = db.execute(stmt).all()
|
||||
out = [(int(uid), int(sc), nick) for uid, sc, nick in rows]
|
||||
_real_rows_cache["rows"], _real_rows_cache["at"] = out, now
|
||||
_real_rows_cache.update(rows=out, at=now, test_phones=test_phones)
|
||||
return out
|
||||
|
||||
|
||||
@@ -340,7 +350,7 @@ def list_real_records(
|
||||
# pool: [(cluster_key, item)];cluster_key 供去连簇——真实=user_id、种子=各自唯一负数(互不聚簇)
|
||||
pool: list[tuple[int, dict]] = []
|
||||
if mode != "seed":
|
||||
rows = db.execute(
|
||||
stmt = (
|
||||
select(
|
||||
ComparisonRecord.user_id,
|
||||
ComparisonRecord.saved_amount_cents,
|
||||
@@ -355,7 +365,11 @@ def list_real_records(
|
||||
)
|
||||
.order_by(ComparisonRecord.created_at.desc())
|
||||
.limit(_REAL_BROWSE_CAP)
|
||||
).all()
|
||||
)
|
||||
test_phones = tuple(sorted(settings.test_account_phones))
|
||||
if test_phones:
|
||||
stmt = stmt.where(User.phone.not_in(test_phones))
|
||||
rows = db.execute(stmt).all()
|
||||
for uid, sc, nick, ca in rows:
|
||||
pool.append((
|
||||
int(uid),
|
||||
|
||||
@@ -111,8 +111,9 @@ class ComparisonRecordIn(BaseModel):
|
||||
# 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、再兜底二态派生。
|
||||
# 记录级业务结局(pricebot 下发): success/below_minimum/store_closed/failed。
|
||||
# below_minimum 表示流程正常完成,持久化主状态归 success;store_closed/items_not_found 等
|
||||
# 已知无报价结局归 failed。原值仍随 raw_payload 落库,admin/记录页从 platform_results 展示细分结论。
|
||||
record_status: str | None = None
|
||||
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
|
||||
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
|
||||
@@ -214,9 +215,13 @@ class ComparisonRecordCreatedOut(BaseModel):
|
||||
|
||||
|
||||
class CompareStartReserveIn(BaseModel):
|
||||
"""Reserve one authenticated comparison start before the agent begins."""
|
||||
"""Reserve one authenticated comparison start before the agent begins.
|
||||
|
||||
trace_id: str = Field(..., min_length=1, max_length=64)
|
||||
trace_id 可选:不带 = 请服务端签发(统一 trace_id 由后端下发,前端/SLS 日志/
|
||||
pricebot 全链用同一个 id);带 = 沿用客户端值(老客户端兼容 + 网络重试幂等)。
|
||||
"""
|
||||
|
||||
trace_id: str | None = Field(default=None, min_length=1, max_length=64)
|
||||
business_type: str = Field(default="food", min_length=1, max_length=16)
|
||||
device_id: str | None = Field(default=None, max_length=64)
|
||||
|
||||
@@ -225,6 +230,9 @@ class CompareStartReserveOut(BaseModel):
|
||||
limit: int | None
|
||||
used: int
|
||||
remaining: int | None
|
||||
# 本次比价全链 trace_id(服务端签发的,或回显客户端带来的)。客户端必须以它为准,
|
||||
# 贯穿 Phase1/Phase2 step、比价记录、trace 收尾与前端运行日志上报。
|
||||
trace_id: str
|
||||
|
||||
|
||||
class CompareStatsOut(BaseModel):
|
||||
|
||||
@@ -58,9 +58,13 @@ class CouponSessionIn(BaseModel):
|
||||
- 发起(status=started):带勾选平台 + 机型/ROM/app_env + started_at_ms(发起墙钟毫秒)。
|
||||
- 收尾(completed/failed/abandoned):带 elapsed_ms(全程耗时)+ platform_elapsed(各平台耗时)+ claimed_count。
|
||||
不鉴权(同领券循环 MVP,按 device_id/trace_id),user_id 登录态带上做留痕(可空)。
|
||||
|
||||
trace_id 可选:started 不带 = 请服务端签发本轮领券 trace_id(统一 trace_id 由后端下发,
|
||||
响应 CouponSessionOut.trace_id 返回,客户端全程用它);带 = 沿用客户端值(老客户端兼容)。
|
||||
非 started 缺 trace_id 不签发(防孤儿行),返回 trace_id=null 且不写库。
|
||||
"""
|
||||
|
||||
trace_id: str
|
||||
trace_id: str | None = None
|
||||
device_id: str
|
||||
status: str # started / completed / failed / abandoned
|
||||
started_at_ms: int # 发起墙钟毫秒(客户端 System.currentTimeMillis)
|
||||
@@ -74,3 +78,16 @@ class CouponSessionIn(BaseModel):
|
||||
platform_elapsed: dict[str, int] | None = None
|
||||
claimed_count: int | None = None
|
||||
trace_url: str | None = None
|
||||
|
||||
|
||||
class CouponSessionOut(BaseModel):
|
||||
"""POST /api/v1/coupon/session 响应。
|
||||
|
||||
trace_id = 本轮领券全链 id(服务端签发的,或回显客户端带来的);客户端以它为准贯穿
|
||||
/coupon/step 循环、收尾上报与前端运行日志。⚠️ 不能沿用旧的 dict[str, bool] 返回注解——
|
||||
FastAPI 会按注解校验响应,字符串 trace_id 过 bool 校验必炸,故显式建模。
|
||||
非 started 且缺 trace_id 时为 null(不签发防孤儿行)。
|
||||
"""
|
||||
|
||||
ok: bool = True
|
||||
trace_id: str | None = None
|
||||
|
||||
@@ -1,8 +1,13 @@
|
||||
"""发奖公式下限:有真实正 eCPM 时最低发 1 金币,eCPM 缺失/为 0 仍发 0。
|
||||
"""发奖公式下限:看了就保底 1 金币,与前端展示公式完全对齐。
|
||||
|
||||
针对「eCPM 过低时公式四舍五入成 0 金币 → 被记 too_short 不发」的问题:只要广告有真实
|
||||
正 eCPM,单份金币至少 1(不再 0);但 eCPM 缺失/为 0/非法(没有真实广告价值)仍发 0,
|
||||
不凭空铸币、不破坏 ecpm_missing 语义。公式是发奖与后台审计对账的唯一口径,改这一处两边同源。
|
||||
产品口径(2026-08「看了就保底 1」):`calculate_ad_reward_coin` 对**任何输入**都至少返回 1,
|
||||
镜像客户端 FeedRewardFormula.singleUnitCoin(那边:eCPM 空/非法/非正数都返 1)。此前 eCPM
|
||||
缺失/为 0 返 0,与前端小球显示的 +1 不一致,且信息流侧把这类看满一份的广告记成 too_short 零发。
|
||||
公式是发奖与后台审计对账的唯一口径,改这一处两边同源。
|
||||
|
||||
注:激励视频(S2S 回调)路径在公式之前还有一道 `if not ecpm_raw` 早退——**完全没上报 eCPM**
|
||||
的回调仍记 ecpm_missing 零发(见 test_ad_reward.test_callback_without_ecpm_records_exception),
|
||||
那是"回调缺字段"的数据完整性闸,区别于"广告如实上报 eCPM=0"(=看了真广告 → 保底 1)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -10,19 +15,23 @@ from app.core.rewards import calculate_ad_reward_coin
|
||||
|
||||
|
||||
def test_low_positive_ecpm_floors_to_one_coin() -> None:
|
||||
"""真实但极低的 eCPM 原本四舍五入成 0 金币,现在兜底为 1(线上 record 4667 的 43 分场景)。"""
|
||||
"""真实但极低的 eCPM 原本四舍五入成 0 金币,兜底为 1(线上 record 4667 的 43 分场景)。"""
|
||||
# 43 分 = ¥0.43 CPM,因子1=0.1,重度用户 LT 第 69 条=1.0 → 0.43/1000×0.1×1.0×10000=0.43 → 旧口径 round=0
|
||||
assert calculate_ad_reward_coin("43", 69) == 1
|
||||
# 更低的 5 分同理:算出来 <0.5,旧口径也是 0
|
||||
assert calculate_ad_reward_coin("5", 11) == 1
|
||||
|
||||
|
||||
def test_zero_or_missing_ecpm_stays_zero() -> None:
|
||||
"""eCPM 缺失 / 为 0 / 非法(没有真实广告价值)不兜底,仍发 0。"""
|
||||
assert calculate_ad_reward_coin("0", 1) == 0
|
||||
assert calculate_ad_reward_coin(None, 1) == 0
|
||||
assert calculate_ad_reward_coin("", 1) == 0
|
||||
assert calculate_ad_reward_coin("abc", 1) == 0
|
||||
def test_zero_or_missing_ecpm_also_floors_to_one() -> None:
|
||||
"""eCPM 为 0 / 缺失 / 非法都保底 1(2026-08「看了就保底 1」,与前端 FeedRewardFormula 对齐)。
|
||||
|
||||
尤其 "0"(广告如实上报零价值)此前返 0,导致小球显示 +1、后端信息流记 too_short 零发的
|
||||
前后端不一致 —— 现在两端都 1。
|
||||
"""
|
||||
assert calculate_ad_reward_coin("0", 1) == 1
|
||||
assert calculate_ad_reward_coin(None, 1) == 1
|
||||
assert calculate_ad_reward_coin("", 1) == 1
|
||||
assert calculate_ad_reward_coin("abc", 1) == 1
|
||||
|
||||
|
||||
def test_normal_ecpm_value_unchanged() -> None:
|
||||
@@ -57,3 +66,31 @@ def test_feed_reward_low_ecpm_grants_one_coin_instead_of_too_short() -> None:
|
||||
assert rec.coin == 1
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_feed_reward_zero_ecpm_grants_one_coin() -> None:
|
||||
"""端到端:eCPM 如实上报 0(用户反馈的真实广告返回 eCPM=0 场景),看满一份也保底 1、
|
||||
状态 granted —— 与前端小球显示的 +1 一致,不再前显示后零发。"""
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.user import User
|
||||
from app.repositories.ad_feed_reward import grant_feed_reward
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
user = User(phone="19900000044", username="feedzero44", register_channel="sms")
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
rec = grant_feed_reward(
|
||||
db, user.id,
|
||||
client_event_id="feed-zero-ecpm-0001",
|
||||
ecpm="0", # 广告如实上报 eCPM=0
|
||||
duration_seconds=15, # 看满一份
|
||||
ad_type="draw",
|
||||
feed_scene="comparison",
|
||||
)
|
||||
assert rec.status == "granted"
|
||||
assert rec.coin == 1
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@@ -720,6 +720,16 @@ def test_comparison_records_show_readable_device_and_rom_version(
|
||||
rom_name="OriginOS",
|
||||
rom_version=4,
|
||||
android_version="14",
|
||||
platforms=[
|
||||
{
|
||||
"platform_id": "meituan",
|
||||
"platform_name": "美团",
|
||||
"status": "ok",
|
||||
"role": "target",
|
||||
"price": 18.8,
|
||||
"is_best": True,
|
||||
}
|
||||
],
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
@@ -747,6 +757,8 @@ def test_comparison_records_show_readable_device_and_rom_version(
|
||||
assert detail.status_code == 200, detail.text
|
||||
assert detail.json()["device_model_name"] == "vivo Y77e"
|
||||
assert detail.json()["rom_version"] == 4
|
||||
assert detail.json()["platforms"][0]["status"] == "ok"
|
||||
assert detail.json()["platforms"][0]["is_best"] is True
|
||||
|
||||
|
||||
def test_comparison_records_show_real_order_status(
|
||||
|
||||
@@ -48,7 +48,10 @@ def test_compare_start_is_idempotent_by_trace_id(client) -> None:
|
||||
retry = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
|
||||
|
||||
assert first.status_code == 200, first.text
|
||||
assert first.json() == {"limit": 100, "used": 1, "remaining": 99}
|
||||
# trace_id 回显客户端带来的值(老协议幂等路径)
|
||||
assert first.json() == {
|
||||
"limit": 100, "used": 1, "remaining": 99, "trace_id": payload["trace_id"],
|
||||
}
|
||||
assert retry.status_code == 200, retry.text
|
||||
assert retry.json() == first.json()
|
||||
with SessionLocal() as db:
|
||||
@@ -69,6 +72,28 @@ def test_compare_start_is_idempotent_by_trace_id(client) -> None:
|
||||
assert record.device_id == "quota-device"
|
||||
|
||||
|
||||
def test_compare_start_issues_trace_id_when_absent(client) -> None:
|
||||
"""新客户端不带 trace_id → 服务端签发并随响应返回,running 行以签发 id 建。"""
|
||||
token, user_id = _login(client)
|
||||
response = client.post(
|
||||
"/api/v1/compare/start",
|
||||
json={"business_type": "food", "device_id": "quota-device-issue"},
|
||||
headers=_headers(token),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
issued = body["trace_id"]
|
||||
assert issued # 非空签发
|
||||
assert body["used"] == 1
|
||||
with SessionLocal() as db:
|
||||
record = db.execute(
|
||||
select(ComparisonRecord).where(ComparisonRecord.trace_id == issued)
|
||||
).scalar_one()
|
||||
assert record.user_id == user_id
|
||||
assert record.status == "running"
|
||||
assert record.device_id == "quota-device-issue"
|
||||
|
||||
|
||||
def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
|
||||
token, user_id = _login(client)
|
||||
now = datetime.now(CN_TZ).replace(tzinfo=None)
|
||||
@@ -101,7 +126,9 @@ def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
|
||||
headers=_headers(token),
|
||||
)
|
||||
assert allowed.status_code == 200, allowed.text
|
||||
assert allowed.json() == {"limit": 100, "used": 100, "remaining": 0}
|
||||
assert allowed.json() == {
|
||||
"limit": 100, "used": 100, "remaining": 0, "trace_id": final_allowed_trace,
|
||||
}
|
||||
|
||||
rejected_trace = f"quota-rejected-{user_id}"
|
||||
response = client.post(
|
||||
|
||||
@@ -13,6 +13,7 @@ import uuid
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
@@ -25,6 +26,26 @@ def _tid() -> str:
|
||||
return uuid.uuid4().hex
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw_status", "record_status"),
|
||||
[
|
||||
("success", "success"),
|
||||
("below_minimum", "success"),
|
||||
("failed", "failed"),
|
||||
("store_closed", "failed"),
|
||||
("store_not_found", "failed"),
|
||||
("items_not_found", "failed"),
|
||||
("no_delivery", "failed"),
|
||||
("unsupported", "failed"),
|
||||
("cancelled", "cancelled"),
|
||||
("running", "running"),
|
||||
(None, None),
|
||||
],
|
||||
)
|
||||
def test_record_status_normalization(raw_status, record_status) -> None:
|
||||
assert crud._normalize_record_status(raw_status) == record_status
|
||||
|
||||
|
||||
def _done_params() -> dict:
|
||||
"""一份典型 done 帧 params:美团 25 元 vs 源淘宝闪购 30 元 → 省 5 元、success。"""
|
||||
return {
|
||||
@@ -111,6 +132,116 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
|
||||
assert newly2 is False
|
||||
|
||||
|
||||
def test_harvest_done_below_minimum_counts_as_completed_success(client) -> None:
|
||||
"""未达起送是可信业务结论:主状态/完成奖励归 success,细分结局仍留在 raw_payload。"""
|
||||
tid = _tid()
|
||||
done_below_minimum = {
|
||||
"record_status": "below_minimum",
|
||||
"comparison_results": [
|
||||
{
|
||||
"platform_id": "meituan",
|
||||
"platform_name": "美团",
|
||||
"package": "com.sankuai.meituan",
|
||||
"price": 59.0,
|
||||
"is_source": True,
|
||||
"rank": 1,
|
||||
"store_name": "测试店",
|
||||
"items": [{"name": "红乌苏", "qty": 1}],
|
||||
},
|
||||
],
|
||||
"platform_results": {
|
||||
"meituan": {"is_source": True, "status": "source", "price": 59.0},
|
||||
"taobao_flash": {
|
||||
"is_source": False,
|
||||
"status": "below_minimum",
|
||||
"reason": "购物车未达起送门槛(差 ¥25.2)",
|
||||
},
|
||||
},
|
||||
"information": "淘宝未达起送门槛,可加菜凑单后下单",
|
||||
}
|
||||
with SessionLocal() as db:
|
||||
crud.harvest_running(db, trace_id=tid, user_id=None)
|
||||
rec, newly = crud.harvest_done(
|
||||
db, trace_id=tid, user_id=None, done_params=done_below_minimum
|
||||
)
|
||||
|
||||
assert rec.status == "success"
|
||||
assert newly is True
|
||||
assert rec.fail_reason is None
|
||||
assert rec.raw_payload["record_status"] == "below_minimum"
|
||||
assert (
|
||||
rec.raw_payload["platform_results"]["taobao_flash"]["status"]
|
||||
== "below_minimum"
|
||||
)
|
||||
|
||||
|
||||
def test_harvest_done_platforms_below_minimum_counts_as_success(client) -> None:
|
||||
"""新 platforms 单源派生路径也必须执行同一 below_minimum → success 归一化。"""
|
||||
tid = _tid()
|
||||
with SessionLocal() as db:
|
||||
rec, newly = crud.harvest_done(
|
||||
db,
|
||||
trace_id=tid,
|
||||
user_id=None,
|
||||
done_params={
|
||||
"record_status": "below_minimum",
|
||||
"platforms": [
|
||||
{
|
||||
"role": "source",
|
||||
"platform_id": "meituan",
|
||||
"platform_name": "美团",
|
||||
"price": 59.0,
|
||||
"store_name": "测试店",
|
||||
"items": [{"name": "红乌苏", "qty": 1}],
|
||||
},
|
||||
{
|
||||
"role": "target",
|
||||
"platform_id": "taobao_flash",
|
||||
"platform_name": "淘宝",
|
||||
"status": "below_minimum",
|
||||
"price": None,
|
||||
},
|
||||
],
|
||||
"platform_results": {
|
||||
"taobao_flash": {
|
||||
"is_source": False,
|
||||
"status": "below_minimum",
|
||||
"reason": "购物车未达起送门槛",
|
||||
}
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
assert rec.status == "success"
|
||||
assert newly is True
|
||||
assert rec.raw_payload["record_status"] == "below_minimum"
|
||||
|
||||
|
||||
def test_legacy_upsert_below_minimum_counts_as_success(client) -> None:
|
||||
"""灰度期客户端直报路径无论走 status 还是 record_status 都不能落第四种主状态。"""
|
||||
tid = _tid()
|
||||
payload = ComparisonRecordIn(
|
||||
trace_id=tid,
|
||||
business_type="food",
|
||||
status="below_minimum",
|
||||
comparison_results=[],
|
||||
platform_results={
|
||||
"taobao_flash": {
|
||||
"is_source": False,
|
||||
"status": "below_minimum",
|
||||
"reason": "购物车未达起送门槛",
|
||||
}
|
||||
},
|
||||
information="淘宝未达起送门槛,可加菜凑单后下单",
|
||||
)
|
||||
with SessionLocal() as db:
|
||||
rec = crud.upsert_record(db, user_id=987654, payload=payload)
|
||||
|
||||
assert rec.status == "success"
|
||||
assert rec.fail_reason is None
|
||||
assert rec.raw_payload["status"] == "below_minimum"
|
||||
|
||||
|
||||
def test_harvest_done_failed_derives_fail_reason(client) -> None:
|
||||
"""failed 记录:记录级 information 笼统,但 fail_reason 从 platform_results 救出具体原因
|
||||
(id 3030 型:美团系统失败 + 京东 items_not_found → 展示京东那条)。"""
|
||||
|
||||
@@ -23,7 +23,7 @@ def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
|
||||
)
|
||||
|
||||
assert sql.count("percentile_cont") == 4
|
||||
assert "comparison_record.status = 'success'" in sql
|
||||
assert "comparison_record.status IN ('success', 'below_minimum')" in sql
|
||||
|
||||
|
||||
def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
|
||||
@@ -34,6 +34,9 @@ def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
|
||||
("summary-success-b", "success", 3000, 2.0, 0),
|
||||
("summary-failed", "failed", 100_000, 3.0, 0),
|
||||
("summary-cancelled", "cancelled", 5000, 4.0, 0),
|
||||
("summary-below-minimum", "below_minimum", 5000, None, 0),
|
||||
("summary-store-closed", "store_closed", 200_000, None, 0),
|
||||
("summary-running", "running", 4000, None, 0),
|
||||
]
|
||||
for trace_id, status, total_ms, cost, saved in rows:
|
||||
db.add(ComparisonRecord(
|
||||
@@ -56,26 +59,44 @@ def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
|
||||
db, date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
|
||||
)
|
||||
|
||||
assert summary["started"] == 4
|
||||
assert summary["completed"] == 3
|
||||
assert summary["success"] == 2
|
||||
assert summary["success_rate"] == pytest.approx(2 / 3)
|
||||
assert summary["started"] == 7
|
||||
assert summary["completed"] == 5
|
||||
assert summary["success"] == 3
|
||||
assert summary["success_rate"] == pytest.approx(3 / 6)
|
||||
assert summary["avg_token_cost"] == pytest.approx(2.5)
|
||||
assert summary["lower_price_rate"] == 0.5
|
||||
assert summary["avg_duration_ms"] == 2000
|
||||
assert summary["p5_duration_ms"] == 1100
|
||||
assert summary["p50_duration_ms"] == 2000
|
||||
assert summary["p95_duration_ms"] == 2900
|
||||
assert summary["p99_duration_ms"] == 2980
|
||||
assert summary["lower_price_rate"] == pytest.approx(1 / 3)
|
||||
assert summary["avg_duration_ms"] == 3000
|
||||
assert summary["p5_duration_ms"] == 1200
|
||||
assert summary["p50_duration_ms"] == 3000
|
||||
assert summary["p95_duration_ms"] == 4800
|
||||
assert summary["p99_duration_ms"] == 4960
|
||||
assert summary["cancelled"] == 1
|
||||
assert summary["cancelled_rate"] == 0.25
|
||||
assert summary["cancelled_rate"] == pytest.approx(1 / 7)
|
||||
assert summary["cancelled_p50_ms"] == 5000
|
||||
|
||||
items, _next_cursor, total = queries.list_comparison_records(
|
||||
db, date_from=date(2038, 1, 15), date_to=date(2038, 1, 15), limit=20
|
||||
)
|
||||
assert total == 4
|
||||
assert total == 7
|
||||
assert {item.trace_id for item in items} == {row[0] for row in rows}
|
||||
|
||||
success_items, _next_cursor, success_total = queries.list_comparison_records(
|
||||
db, status="success", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
|
||||
)
|
||||
assert success_total == 3
|
||||
assert {item.status for item in success_items} == {"success", "below_minimum"}
|
||||
|
||||
failed_items, _next_cursor, failed_total = queries.list_comparison_records(
|
||||
db, status="failed", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
|
||||
)
|
||||
assert failed_total == 2
|
||||
assert {item.status for item in failed_items} == {"failed", "store_closed"}
|
||||
|
||||
running_items, _next_cursor, running_total = queries.list_comparison_records(
|
||||
db, status="running", date_from=date(2038, 1, 15), date_to=date(2038, 1, 15)
|
||||
)
|
||||
assert running_total == 1
|
||||
assert running_items[0].status == "running"
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Regression coverage for comparison terminal-status normalization."""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic.migration import MigrationContext
|
||||
from alembic.operations import Operations
|
||||
|
||||
|
||||
def _migration_module():
|
||||
path = (
|
||||
Path(__file__).parents[1]
|
||||
/ "alembic"
|
||||
/ "versions"
|
||||
/ "comparison_below_minimum_as_success.py"
|
||||
)
|
||||
spec = importlib.util.spec_from_file_location(path.stem, path)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_business_status_migration_upgrade_and_downgrade() -> None:
|
||||
engine = sa.create_engine("sqlite:///:memory:")
|
||||
metadata = sa.MetaData()
|
||||
records = sa.Table(
|
||||
"comparison_record",
|
||||
metadata,
|
||||
sa.Column("id", sa.Integer, primary_key=True),
|
||||
sa.Column("status", sa.String(16), nullable=False),
|
||||
sa.Column("fail_reason", sa.String(256)),
|
||||
sa.Column("raw_payload", sa.JSON, nullable=False),
|
||||
)
|
||||
metadata.create_all(engine)
|
||||
module = _migration_module()
|
||||
|
||||
with engine.begin() as connection:
|
||||
connection.execute(
|
||||
records.insert(),
|
||||
[
|
||||
{
|
||||
"status": "below_minimum",
|
||||
"fail_reason": "旧失败原因",
|
||||
"raw_payload": {"record_status": "below_minimum"},
|
||||
},
|
||||
{
|
||||
"status": "success",
|
||||
"fail_reason": None,
|
||||
"raw_payload": {"record_status": "success"},
|
||||
},
|
||||
{
|
||||
"status": "failed",
|
||||
"fail_reason": "技术异常",
|
||||
"raw_payload": {"status": "failed"},
|
||||
},
|
||||
{
|
||||
"status": "store_closed",
|
||||
"fail_reason": "店铺打烊",
|
||||
"raw_payload": {"record_status": "store_closed"},
|
||||
},
|
||||
{
|
||||
"status": "items_not_found",
|
||||
"fail_reason": "商品未找到",
|
||||
"raw_payload": {"status": "items_not_found"},
|
||||
},
|
||||
],
|
||||
)
|
||||
module.op = Operations(MigrationContext.configure(connection))
|
||||
|
||||
module.upgrade()
|
||||
upgraded = connection.execute(
|
||||
sa.select(records.c.status, records.c.fail_reason).order_by(records.c.id)
|
||||
).all()
|
||||
assert upgraded == [
|
||||
("success", None),
|
||||
("success", None),
|
||||
("failed", "技术异常"),
|
||||
("failed", "店铺打烊"),
|
||||
("failed", "商品未找到"),
|
||||
]
|
||||
|
||||
module.downgrade()
|
||||
downgraded = connection.execute(
|
||||
sa.select(records.c.status, records.c.fail_reason).order_by(records.c.id)
|
||||
).all()
|
||||
assert downgraded == [
|
||||
("below_minimum", None),
|
||||
("success", None),
|
||||
("failed", "技术异常"),
|
||||
("store_closed", "店铺打烊"),
|
||||
("items_not_found", "商品未找到"),
|
||||
]
|
||||
@@ -8,6 +8,7 @@ mock 掉对 pricebot 的 httpx 调用,验证:
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@@ -62,7 +63,8 @@ def test_coupon_step_no_auth_required(client) -> None:
|
||||
|
||||
|
||||
def test_coupon_step_passes_body_through(client, access_token) -> None:
|
||||
"""带 token + pricebot 200 → 响应原样透传,请求 body 原样转发到 /api/coupon/step。"""
|
||||
"""带 token + pricebot 200 → 请求 body 原样转发到 /api/coupon/step;
|
||||
响应在透传基础上顶层回显本次任务 trace_id(setdefault 注入,其余字段原样)。"""
|
||||
fake_pricebot_resp = {
|
||||
"success": True,
|
||||
"action": {
|
||||
@@ -83,9 +85,12 @@ def test_coupon_step_passes_body_through(client, access_token) -> None:
|
||||
|
||||
captured: dict = {}
|
||||
|
||||
async def fake_post(self, url, json=None, **kw):
|
||||
# ⚠️ coupon_step 转发用 content=raw(原始字节透传,不重新 dumps),不是 json= ——
|
||||
# fake 必须捕 content。旧 fake 只捕 json= 导致 captured["json"] 恒 None,本测试
|
||||
# 自 content=raw 优化后一直红着(pre-existing),本次顺手修正。
|
||||
async def fake_post(self, url, content=None, **kw):
|
||||
captured["url"] = url
|
||||
captured["json"] = json
|
||||
captured["content"] = content
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json = lambda: fake_pricebot_resp
|
||||
@@ -99,9 +104,10 @@ def test_coupon_step_passes_body_through(client, access_token) -> None:
|
||||
)
|
||||
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == fake_pricebot_resp
|
||||
# 验证请求被原样转发(body 不动 + URL 指向 pricebot)
|
||||
assert captured["json"] == _stub_request_body()
|
||||
# 顶层多出 trace_id 回显(值=请求带的;不 mint,签发点唯一在 /coupon/session started)
|
||||
assert r.json() == {**fake_pricebot_resp, "trace_id": "test-trace-1"}
|
||||
# 验证请求被原样转发(body 字节不动 + URL 指向 pricebot)
|
||||
assert json.loads(captured["content"]) == _stub_request_body()
|
||||
assert captured["url"].endswith("/api/coupon/step")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
"""POST /api/v1/coupon/session 的 trace_id 签发行为(统一 trace_id 由后端下发)。
|
||||
|
||||
- started 不带 trace_id → 服务端签发并返回,行以签发 id 建;
|
||||
- started 带 trace_id → 回显沿用(老客户端兼容);
|
||||
- 非 started 缺 trace_id → 不签发不写库,trace_id=null(防孤儿行)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.coupon_state import CouponSession
|
||||
|
||||
|
||||
def _base_payload(**overrides) -> dict:
|
||||
payload = {
|
||||
"device_id": "cs-issue-device",
|
||||
"status": "started",
|
||||
"started_at_ms": 1_722_000_000_000,
|
||||
"platforms": ["meituan-waimai"],
|
||||
"app_env": "dev",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
def test_session_started_issues_trace_id_when_absent(client) -> None:
|
||||
response = client.post("/api/v1/coupon/session", json=_base_payload())
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["ok"] is True
|
||||
issued = body["trace_id"]
|
||||
assert issued
|
||||
with SessionLocal() as db:
|
||||
row = db.execute(
|
||||
select(CouponSession).where(CouponSession.trace_id == issued)
|
||||
).scalar_one()
|
||||
assert row.device_id == "cs-issue-device"
|
||||
assert row.status == "started"
|
||||
|
||||
|
||||
def test_session_started_echoes_client_trace_id(client) -> None:
|
||||
response = client.post(
|
||||
"/api/v1/coupon/session", json=_base_payload(trace_id="cs-legacy-1")
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["trace_id"] == "cs-legacy-1"
|
||||
with SessionLocal() as db:
|
||||
count = db.scalar(
|
||||
select(func.count(CouponSession.id)).where(
|
||||
CouponSession.trace_id == "cs-legacy-1"
|
||||
)
|
||||
)
|
||||
assert count == 1
|
||||
|
||||
|
||||
def test_session_terminal_without_trace_id_skips_write(client) -> None:
|
||||
response = client.post(
|
||||
"/api/v1/coupon/session",
|
||||
json=_base_payload(status="completed", elapsed_ms=1234, claimed_count=2),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert body["ok"] is True
|
||||
assert body["trace_id"] is None
|
||||
@@ -0,0 +1,77 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from app.core.config import settings
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.comparison import ComparisonRecord
|
||||
from app.models.user import User
|
||||
from app.repositories import ops_marquee
|
||||
|
||||
|
||||
def _clear_real_rows_cache() -> None:
|
||||
ops_marquee._real_rows_cache.update(at=None, rows=None, test_phones=None)
|
||||
|
||||
|
||||
def test_real_marquee_records_exclude_configured_test_account(monkeypatch) -> None:
|
||||
legacy_test_phone = "19900009991"
|
||||
listed_test_phone = "19900009992"
|
||||
real_phone = "19900009993"
|
||||
monkeypatch.setattr(settings, "TEST_ACCOUNT_PHONE", legacy_test_phone)
|
||||
monkeypatch.setattr(settings, "TEST_ACCOUNT_PHONES", listed_test_phone)
|
||||
_clear_real_rows_cache()
|
||||
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
legacy_test_user = User(
|
||||
phone=legacy_test_phone,
|
||||
username="29900009991",
|
||||
register_channel="sms",
|
||||
)
|
||||
listed_test_user = User(
|
||||
phone=listed_test_phone,
|
||||
username="29900009992",
|
||||
register_channel="sms",
|
||||
)
|
||||
real_user = User(
|
||||
phone=real_phone,
|
||||
username="29900009993",
|
||||
register_channel="sms",
|
||||
)
|
||||
db.add_all([legacy_test_user, listed_test_user, real_user])
|
||||
db.flush()
|
||||
db.add_all([
|
||||
ComparisonRecord(
|
||||
user_id=legacy_test_user.id,
|
||||
trace_id="marquee-legacy-test-account-trace",
|
||||
status="success",
|
||||
saved_amount_cents=29_991,
|
||||
items=[],
|
||||
comparison_results=[],
|
||||
),
|
||||
ComparisonRecord(
|
||||
user_id=listed_test_user.id,
|
||||
trace_id="marquee-listed-test-account-trace",
|
||||
status="success",
|
||||
saved_amount_cents=29_992,
|
||||
items=[],
|
||||
comparison_results=[],
|
||||
),
|
||||
ComparisonRecord(
|
||||
user_id=real_user.id,
|
||||
trace_id="marquee-real-account-trace",
|
||||
status="success",
|
||||
saved_amount_cents=29_993,
|
||||
items=[],
|
||||
comparison_results=[],
|
||||
),
|
||||
])
|
||||
db.flush()
|
||||
|
||||
app_rows = ops_marquee._recent_real_rows(db)
|
||||
browse_rows, _ = ops_marquee.list_real_records(db, mode="real", limit=1_000)
|
||||
finally:
|
||||
_clear_real_rows_cache()
|
||||
|
||||
assert all(saved not in {29_991, 29_992} for _uid, saved, _nickname in app_rows)
|
||||
assert any(saved == 29_993 for _uid, saved, _nickname in app_rows)
|
||||
assert all(row["saved_amount_cents"] not in {29_991, 29_992} for row in browse_rows)
|
||||
assert any(row["saved_amount_cents"] == 29_993 for row in browse_rows)
|
||||
Reference in New Issue
Block a user