Compare commits

...

8 Commits

Author SHA1 Message Date
guke 5871c02055 Merge branch 'main' into fix-bugfix0731 2026-08-01 23:23:14 +08:00
linkeyu ab2de6ec79 修复:统一比价记录状态枚举口径 (#209)
## 变更说明

- below_minimum 归入成功,保留原始业务结局
- store_closed/store_not_found/items_not_found/no_delivery/unsupported 归入失败
- running 保持进行中生命周期状态
- 历史细分状态迁移为三态终态,迁移可逆
- 后台成功/失败筛选及汇总兼容迁移前历史值

## 验证

- 相关回归:48 passed
- Ruff:通过
- Alembic:单一 head

---------

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: linkeyu <798648091@qq.com>
Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #209
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-08-01 23:09:21 +08:00
guke 14a9473323 Merge branch 'main' into fix-bugfix0731 2026-08-01 23:06:46 +08:00
linkeyu 84251770b4 修复:首页轮播排除测试账号数据 (#215)
## 背景

线上首页轮播候选记录中,测试账号 u33(11111111111)贡献约 51.8% 的真实记录,导致该用户频繁出现。

## 改动

- App 首页轮播真实记录查询排除全部已配置测试账号
- 后台首页轮播可展示记录同步采用相同过滤口径
- 同时兼容 TEST_ACCOUNT_PHONE 与 TEST_ACCOUNT_PHONES
- 测试账号配置变化时立即使轮播查询缓存失效
- 不删除历史业务数据,仅在展示查询中排除

## 验证

- ruff check:通过
- pytest tests/test_ops_marquee.py -q:1 passed

---------

Co-authored-by: linkeyu <798648091@qq.com>
Reviewed-on: #215
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-08-01 21:35:06 +08:00
marco 2e91c9f72f ecpm保底是1 2026-08-01 02:57:45 +08:00
marco a69b7d777d feat(compare/coupon): trace_id 统一由后端签发,前端不再本地生成 (#212)
Reviewed-on: #212
2026-07-31 23:52:19 +08:00
marco 0fc8521c3b feat(compare/coupon): trace_id 统一由后端签发,前端不再本地生成 (#210)
一次比价/领券的 trace_id 改由后端签发,让前端 SLS 运行日志(trace_id 索引列)、
app-server 比价记录与领券流水、pricebot trace 目录/run.log/trace_url 全链共用同一个
id 查到底(此前前端各业务自己 randomUUID,虽同链但非后端签发、也无单一签发点)。

- POST /api/v1/compare/start(预占额度,任务第一个请求,签发与建 running 行合一):
  请求 trace_id 改可选,缺省时服务端签发 uuid;响应新增 trace_id 字段返回(签发的或
  回显客户端带来的)。客户端带值则沿用——老客户端兼容 + 同 trace 重试幂等。
- POST /api/v1/coupon/session:started 帧缺 trace_id 时签发并随响应返回(签发不依赖
  写库成功);非 started 帧缺 trace_id 不签发、不写库(收尾没有 id 只能是异常调用,
  签发新 id 只会造出查不到发起信息的孤儿行)。新增 CouponSessionOut 响应模型——原
  dict[str,bool] 注解无法承载字符串 trace_id,FastAPI 响应校验会炸。
- 测试:更新 2 处旧断言(响应体多出 trace_id 字段),新增 compare 不带 id 签发用例 +
  coupon started 签发/回显、终尾缺 id 跳过写库 3 个用例。全量 30 passed + ruff clean。

配合 shaguabijia-app-android 同名分支 feat-unify-trace-id-backend-issued 的前端换源改动。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Reviewed-on: #210
2026-07-31 23:11:35 +08:00
exinglang 2a98e5147c fix(反馈): 优化图片处理与静态资源缓存 2026-07-31 22:31:45 +08:00
28 changed files with 916 additions and 82 deletions
@@ -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)
)
+34 -6
View File
@@ -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)),
+2 -2
View File
@@ -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,
+2 -1
View File
@@ -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 = []
# 全量环境
+2 -2
View File
@@ -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
+7 -1
View File
@@ -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
View File
@@ -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)")
+4 -2
View File
@@ -54,11 +54,13 @@ def _app_status(db_status: str) -> str:
def _record_out(fb) -> FeedbackRecordOut:
images = fb.images or []
return FeedbackRecordOut(
id=fb.id,
content=fb.content,
scene=getattr(fb, "scene", None),
images=fb.images or [],
images=images,
image_thumbnails=[media.feedback_thumbnail_url(url) for url in images],
status=_app_status(fb.status),
reject_reason=getattr(fb, "reject_reason", None),
reward_coins=getattr(fb, "reward_coins", None),
@@ -84,7 +86,7 @@ async def submit_feedback(
device_model: str = Form(default=""),
rom_name: str = Form(default=""),
android_version: str = Form(default=""),
images: list[UploadFile] = File(default=[]),
images: list[UploadFile] = File(default=[]), # noqa: B008 - FastAPI dependency declaration
) -> FeedbackOut:
content = content.strip()
contact = contact.strip()
+90 -2
View File
@@ -11,6 +11,7 @@
from __future__ import annotations
import json
import logging
import os
import secrets
import subprocess
@@ -18,8 +19,17 @@ from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from pathlib import Path
from typing import Any
from PIL import Image, ImageOps
from app.core.config import settings
logger = logging.getLogger("shagua.media")
_FEEDBACK_DIR = "feedback"
_FEEDBACK_THUMB_DIR = "feedback_thumbs"
_FEEDBACK_THUMB_MAX_PX = 256
_FEEDBACK_THUMB_QUALITY = 78
class MediaError(Exception):
"""上传文件不合法(类型/大小)。调用方转 400。"""
@@ -68,8 +78,86 @@ def save_avatar(user_id: int, data: bytes) -> str:
def save_feedback_image(user_id: int, data: bytes) -> str:
"""保存反馈截图,返回相对 URL(`/media/feedback/<file>`)"""
return _save_image("feedback", user_id, data)
"""保存反馈截图并预生成历史页缩略图,返回原图相对 URL"""
url = _save_image(_FEEDBACK_DIR, user_id, data)
# 缩略图失败不影响反馈受理;读取缩略图 URL 时会按需重试并回退原图。
ensure_feedback_thumbnail(url)
return url
def feedback_thumbnail_url(image_url: str) -> str:
"""把反馈原图 URL 映射成确定的缩略图 URL,不在 records 接口内做图片解码。
上传文件名由服务端生成且不会覆盖;旧数据在客户端真正请求可见图片时按需补图。
"""
paths = _feedback_thumbnail_paths(image_url)
return paths[2] if paths is not None else image_url
def _feedback_thumbnail_paths(image_url: str) -> tuple[Path, Path, str] | None:
prefix = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_DIR}/"
if not image_url.startswith(prefix):
return None
filename = image_url.removeprefix(prefix)
# 只接受当前目录下的单个文件名,避免数据库脏数据造成路径穿越。
if not filename or Path(filename).name != filename:
return None
source = _media_dir(_FEEDBACK_DIR) / filename
thumb_name = f"{Path(filename).stem}.jpg"
thumb = _media_dir(_FEEDBACK_THUMB_DIR) / thumb_name
thumb_url = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_THUMB_DIR}/{thumb_name}"
return source, thumb, thumb_url
def ensure_feedback_thumbnail(image_url: str) -> Path | None:
"""确保缩略图存在并返回文件;生成失败时回退原图,供动态缩略图路由使用。"""
paths = _feedback_thumbnail_paths(image_url)
if paths is None:
return None
source, thumb, _ = paths
if thumb.is_file():
return thumb
if not source.is_file():
return None
temp = thumb.with_name(f".{thumb.name}.{secrets.token_hex(4)}.tmp")
try:
with Image.open(source) as opened:
image = ImageOps.exif_transpose(opened)
image.thumbnail(
(_FEEDBACK_THUMB_MAX_PX, _FEEDBACK_THUMB_MAX_PX),
Image.Resampling.LANCZOS,
)
if image.mode != "RGB":
image = image.convert("RGB")
image.save(
temp,
format="JPEG",
quality=_FEEDBACK_THUMB_QUALITY,
optimize=True,
)
os.replace(temp, thumb)
return thumb
except (Image.DecompressionBombError, OSError, ValueError):
logger.warning("生成反馈缩略图失败: %s", source, exc_info=True)
return source
finally:
temp.unlink(missing_ok=True)
def feedback_thumbnail_file(filename: str) -> Path | None:
"""由缩略图文件名找到原反馈图并按需生成,非法/不存在返回 None。"""
if not filename or Path(filename).name != filename or Path(filename).suffix.lower() != ".jpg":
return None
stem = Path(filename).stem
for ext in (".jpg", ".png", ".webp"):
original = _media_dir(_FEEDBACK_DIR) / f"{stem}{ext}"
if original.is_file():
original_url = f"{settings.MEDIA_URL_PREFIX}/{_FEEDBACK_DIR}/{original.name}"
return ensure_feedback_thumbnail(original_url)
return None
def save_report_image(user_id: int, data: bytes) -> str:
+8 -4
View File
@@ -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))
+29
View File
@@ -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/startcoupon/session startedcompare.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]}"
+32 -2
View File
@@ -10,7 +10,7 @@ from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
@@ -44,6 +44,7 @@ from app.api.v1.tasks import router as tasks_router
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.api.v1.wxpay import router as wxpay_router
from app.core import media
from app.core.config import settings
from app.core.cps_reconcile_worker import (
start_cps_reconcile_worker,
@@ -82,6 +83,19 @@ setup_logging(debug=settings.APP_DEBUG)
logger = logging.getLogger("shagua.main")
class FeedbackMediaStaticFiles(StaticFiles):
"""反馈原图/缩略图文件名不可变,可长期缓存,避免列表反复回源。"""
async def get_response(self, path: str, scope):
response = await super().get_response(path, scope)
media_path = path.replace("\\", "/").lstrip("/")
if response.status_code == 200 and media_path.startswith(
("feedback/", "feedback_thumbs/")
):
response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
return response
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
# 提示而非强制建表:生产用 alembic upgrade head,本地 dev 也建议先跑一次 migration。
@@ -212,8 +226,24 @@ def download_apk() -> FileResponse:
)
@app.get(
f"{settings.MEDIA_URL_PREFIX}/feedback_thumbs/{{filename}}",
tags=["feedback"],
include_in_schema=False,
)
def feedback_thumbnail(filename: str) -> FileResponse:
"""旧反馈图按首次可见请求补缩略图;新图上传时已预生成。"""
path = media.feedback_thumbnail_file(filename)
if path is None:
raise HTTPException(status_code=404, detail="图片不存在")
return FileResponse(
path,
headers={"Cache-Control": "public, max-age=31536000, immutable"},
)
app.mount(
settings.MEDIA_URL_PREFIX,
StaticFiles(directory=str(_media_root)),
FeedbackMediaStaticFiles(directory=str(_media_root)),
name="media",
)
+1 -1
View File
@@ -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…";
# 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
+39 -10
View File
@@ -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
+21 -7
View File
@@ -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),
+12 -4
View File
@@ -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):
+18 -1
View File
@@ -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
+2
View File
@@ -33,6 +33,8 @@ class FeedbackRecordOut(BaseModel):
# 比价反馈的问题场景(找错商品/优惠不对…);普通反馈为 None
scene: str | None = None
images: list[str] = Field(default_factory=list)
# 与 images 下标一一对应;生成失败时该项回退原图 URL,兼容历史数据。
image_thumbnails: list[str] = Field(default_factory=list)
status: str
reject_reason: str | None = None
reward_coins: int | None = None
+3
View File
@@ -35,6 +35,9 @@ dependencies = [
# multipart form (FastAPI 表单上传依赖)
"python-multipart>=0.0.9",
# 用户反馈截图缩略图,避免 App 历史页为 48dp 小图下载数 MB 原图
"pillow>=11.0.0",
# admin 后台账号密码 hash(用户侧是手机号+验证码登录,不需要密码;admin 才用)
"bcrypt>=4.0.0",
+48 -11
View File
@@ -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()
+12
View File
@@ -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(
+29 -2
View File
@@ -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(
+131
View File
@@ -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 展示京东那条)"""
+34 -13
View File
@@ -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", "商品未找到"),
]
+12 -6
View File
@@ -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")
+65
View File
@@ -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
+77
View File
@@ -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)