Compare commits
34 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 9f804bf0cb | |||
| 63daeeaf9b | |||
| 523d970c45 | |||
| 44beb3b8de | |||
| 62b30342ed | |||
| 347c4c7de4 | |||
| e135ba9a84 | |||
| dd96fc2151 | |||
| fed3541a51 | |||
| 924e40a84e | |||
| 9598c7a1da | |||
| 0663ee5542 | |||
| 45a8e7b972 | |||
| 02d2e56ef6 | |||
| a7e8141497 | |||
| c930957e90 | |||
| 6c143dc9f2 | |||
| b6ece681f3 | |||
| 818ae1c9e1 | |||
| 4becde8d75 | |||
| af229e2a7b | |||
| 576b94b4bb | |||
| 46247fb3a9 | |||
| d0169ffb54 | |||
| bc321c1c64 | |||
| 7891984cd1 | |||
| 20cbc9e35e | |||
| 02d6300442 | |||
| 46ffa41931 | |||
| 01f97e72a4 | |||
| 5a66c302cb | |||
| 09b9381d03 | |||
| 7419f35f4b | |||
| 9de73152ec |
@@ -1,34 +0,0 @@
|
||||
"""savings_record.trace_id (下单归因到的比价 trace_id)
|
||||
|
||||
Revision ID: savings_record_trace_id
|
||||
Revises: comparison_updated_at
|
||||
Create Date: 2026-08-07 00:00:00.000000
|
||||
|
||||
「已下单」从店级改单次级:下单上报带上本次比价的 trace_id,落这一列,读取时按
|
||||
trace_id 精确对齐 comparison_record.trace_id —— 同一家店比价多次,只有真正下单的
|
||||
那一条标「已下单」。可空:demo 行 / 老客户端 / 历史订单没有 trace_id(→ 不进任何
|
||||
记录的「已下单」,不做回填)。见 repositories.comparison._ordered_trace_id_select。
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'savings_record_trace_id'
|
||||
down_revision: Union[str, Sequence[str], None] = 'comparison_updated_at'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table('savings_record', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('trace_id', sa.String(length=64), nullable=True))
|
||||
batch_op.create_index(batch_op.f('ix_savings_record_trace_id'), ['trace_id'], unique=False)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table('savings_record', schema=None) as batch_op:
|
||||
batch_op.drop_index(batch_op.f('ix_savings_record_trace_id'))
|
||||
batch_op.drop_column('trace_id')
|
||||
@@ -76,30 +76,30 @@ def _device_marketing_name(model: str | None) -> str | None:
|
||||
|
||||
|
||||
def _attach_comparison_order_status(db: Session, items: list[ComparisonRecord]) -> None:
|
||||
"""按 C 端既有口径给比价记录批量补充是否真实下单(按 trace_id 精确对齐真实下单上报)。"""
|
||||
"""按 C 端既有口径给比价记录批量补充是否真实下单。"""
|
||||
user_ids = {item.user_id for item in items if item.user_id is not None}
|
||||
trace_ids = {item.trace_id for item in items if item.trace_id}
|
||||
shop_names = {item.store_name for item in items if item.store_name}
|
||||
ordered_pairs: set[tuple[int, str]] = set()
|
||||
if user_ids and trace_ids:
|
||||
if user_ids and shop_names:
|
||||
rows = db.execute(
|
||||
select(SavingsRecord.user_id, SavingsRecord.trace_id)
|
||||
select(SavingsRecord.user_id, SavingsRecord.shop_name)
|
||||
.where(
|
||||
SavingsRecord.user_id.in_(user_ids),
|
||||
SavingsRecord.source == "compare",
|
||||
SavingsRecord.trace_id.in_(trace_ids),
|
||||
SavingsRecord.shop_name.in_(shop_names),
|
||||
)
|
||||
.distinct()
|
||||
).all()
|
||||
ordered_pairs = {
|
||||
(row.user_id, row.trace_id)
|
||||
(row.user_id, row.shop_name)
|
||||
for row in rows
|
||||
if row.trace_id is not None
|
||||
if row.shop_name is not None
|
||||
}
|
||||
for item in items:
|
||||
item.ordered = bool(
|
||||
item.user_id is not None
|
||||
and item.trace_id
|
||||
and (item.user_id, item.trace_id) in ordered_pairs
|
||||
and item.store_name
|
||||
and (item.user_id, item.store_name) in ordered_pairs
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -376,14 +376,15 @@ def dashboard_overview(
|
||||
.where(
|
||||
SavingsRecord.user_id == ComparisonRecord.user_id,
|
||||
SavingsRecord.source == "compare",
|
||||
SavingsRecord.trace_id.is_not(None),
|
||||
SavingsRecord.trace_id == ComparisonRecord.trace_id,
|
||||
SavingsRecord.shop_name.is_not(None),
|
||||
SavingsRecord.shop_name == ComparisonRecord.store_name,
|
||||
)
|
||||
.exists()
|
||||
)
|
||||
period_ordered_count = _count(
|
||||
ComparisonRecord,
|
||||
*period_comparison_conds,
|
||||
ComparisonRecord.store_name.is_not(None),
|
||||
ordered_exists,
|
||||
)
|
||||
|
||||
|
||||
@@ -127,16 +127,13 @@ def build_hits(
|
||||
hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
|
||||
else:
|
||||
# trace 判出卡点 → 上面带卡点报。其余一律回退耗时/步数兜底:可读但没判出卡点、
|
||||
# 读不到、无 trace,都过 total_ms/step 阈值——超长放弃照报,不再因「trace 可读但不
|
||||
# 原地卡」把超长放弃整条吞掉(线上 trace 几乎总可读,否则耗时阈值形同虚设)。
|
||||
# 读不到、无 trace,都过 total_ms/step 阈值——超长放弃照报(卡点列留空),不再因
|
||||
# 「trace 可读但不原地卡」把超长放弃整条吞掉(线上 trace 几乎总可读,否则耗时阈值形同虚设)。
|
||||
hit = classify_cancelled_fallback(
|
||||
rec,
|
||||
cancelled_ms_threshold=cancelled_ms_threshold,
|
||||
cancelled_step_threshold=cancelled_step_threshold,
|
||||
)
|
||||
# 没卡点但命中(超阈值放弃)→ 用末帧标「退出前在哪屏」(平台·环节·页面)
|
||||
if hit is not None and res is not None and res.last is not None:
|
||||
hit = _dc_replace(hit, stuck_point=res.last.label())
|
||||
if hit is not None:
|
||||
hits.append(hit)
|
||||
else:
|
||||
@@ -229,8 +226,6 @@ def _scan_and_alert() -> None:
|
||||
interval_min=interval_min,
|
||||
max_detail_per_type=settings.COMPARE_ALERT_MAX_DETAIL_PER_TYPE,
|
||||
max_total=settings.COMPARE_ALERT_MAX_TOTAL,
|
||||
cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
|
||||
cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
|
||||
)
|
||||
|
||||
try:
|
||||
|
||||
@@ -5,14 +5,9 @@
|
||||
`wallet.daily_auto_exchange`。
|
||||
|
||||
健壮性:
|
||||
- **持久化「当天已兑」标记**(app_config `auto_exchange.last_run_date`):当天已兑则整轮跳过,
|
||||
**标记跨进程重启不丢** → 同一北京日内多次启动/部署不再重复补扫。这是修 bug 的关键:原来用
|
||||
内存变量记「今天跑过没」,进程重启即归零,导致每次非 0 点部署都全量补扫一遍、把 0 点后才达标
|
||||
的用户在**非 0 点**兑成现金(用户反馈的「非 0 点也出现金币转现金记录」)。
|
||||
- **漏了 0 点即补**:标记 < 今天(如 0 点服务器宕机)时标记 != today → 照常补跑一轮(保留原
|
||||
timer 的 Persistent 语义)。这类补跑确实落非 0 点,但仅限「真漏了 0 点」的罕见场景,非常态。
|
||||
- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),做兜底
|
||||
防同机多进程 / 标记写入失败的竞态,不会重复兑。
|
||||
- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),
|
||||
故启动补跑 / 多次唤醒 / 进程重启都安全,不会重复兑。
|
||||
- **当天首跑即补**:进程起来时若当天还没兑过,立即兑一轮(等价原 timer 的 Persistent 补跑)。
|
||||
- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑(防跨进程并发导致 TOCTOU 双兑)。
|
||||
- **开关**:settings.AUTO_EXCHANGE_ENABLED=false 时不启动(与脚本/原 timer 同一开关)。
|
||||
"""
|
||||
@@ -32,14 +27,10 @@ from sqlalchemy.exc import SQLAlchemyError
|
||||
from app.core import rewards
|
||||
from app.core.config import settings
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.app_config import AppConfig
|
||||
from app.repositories import wallet as wallet_repo
|
||||
|
||||
logger = logging.getLogger("shagua.daily_exchange")
|
||||
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "daily_exchange.lock"
|
||||
# 「上次成功自动兑的北京日」标记,持久化在 app_config(仿 compare_alert 水位,不进 CONFIG_DEFS——
|
||||
# 它是 worker 内部运行状态,非运营可配项)。用它替代内存 last_run:重启不丢 → 同日不重复补扫。
|
||||
LAST_RUN_KEY = "auto_exchange.last_run_date"
|
||||
|
||||
|
||||
def _touch_lock() -> None:
|
||||
@@ -81,44 +72,9 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
|
||||
_LOCK_PATH.unlink()
|
||||
|
||||
|
||||
def _read_last_run(db) -> date | None:
|
||||
"""读持久化的「上次成功自动兑的北京日」标记(app_config,跨进程重启不丢)。脏值当作没跑过。"""
|
||||
row = db.get(AppConfig, LAST_RUN_KEY)
|
||||
if row is None or not row.value:
|
||||
return None
|
||||
try:
|
||||
return date.fromisoformat(row.value)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _write_last_run(db, day: date) -> None:
|
||||
"""落库「当天已兑」标记。用专用 key 直接写 AppConfig(仿 compare_alert 水位,不走 set_value)。"""
|
||||
iso = day.isoformat()
|
||||
row = db.get(AppConfig, LAST_RUN_KEY)
|
||||
if row is None:
|
||||
db.add(AppConfig(key=LAST_RUN_KEY, value=iso, updated_by_admin_id=None))
|
||||
else:
|
||||
row.value = iso
|
||||
db.commit()
|
||||
|
||||
|
||||
def _exchange_if_due(db, today: date) -> dict | None:
|
||||
"""当天未兑过(持久化标记 != today)才跑一轮并记标记;已兑过返回 None(整轮跳过)。
|
||||
|
||||
标记写在 daily_auto_exchange 之后:即便中途崩,标记仍是旧值 → 下次重跑,逐用户幂等会跳过已兑的、
|
||||
补完剩下的(安全)。真漏了 0 点(标记 < today)时标记 != today,仍会补跑,保留原「当天首跑即补」。
|
||||
"""
|
||||
if _read_last_run(db) == today:
|
||||
return None
|
||||
result = wallet_repo.daily_auto_exchange(db)
|
||||
_write_last_run(db, today)
|
||||
return result
|
||||
|
||||
|
||||
def _exchange_once(today: date) -> dict | None:
|
||||
def _exchange_once() -> dict:
|
||||
with SessionLocal() as db:
|
||||
return _exchange_if_due(db, today)
|
||||
return wallet_repo.daily_auto_exchange(db)
|
||||
|
||||
|
||||
async def _run_loop() -> None:
|
||||
@@ -133,15 +89,16 @@ async def _run_loop() -> None:
|
||||
|
||||
async def _run_locked_loop(interval: int) -> None:
|
||||
logger.info("daily auto-exchange worker started interval=%ss", interval)
|
||||
# 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。
|
||||
last_run: date | None = None
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
_touch_lock()
|
||||
today = rewards.cn_today()
|
||||
# 「今天是否已兑」以持久化标记为准(见 _exchange_if_due),不再用内存变量 →
|
||||
# 进程重启不会把当天当「没跑过」重复补扫;已兑当天返回 None(整轮跳过)。
|
||||
result = await asyncio.to_thread(_exchange_once, today)
|
||||
if result is not None:
|
||||
if last_run != today:
|
||||
result = await asyncio.to_thread(_exchange_once)
|
||||
last_run = today
|
||||
logger.info("daily auto-exchange done date=%s result=%s", today, result)
|
||||
except SQLAlchemyError:
|
||||
logger.exception("daily auto-exchange db error")
|
||||
|
||||
@@ -56,9 +56,6 @@ class SavingsRecord(Base):
|
||||
source_deeplink: Mapped[str | None] = mapped_column(String(512), nullable=True)
|
||||
# 客户端幂等键(UUID);demo 行为 NULL
|
||||
client_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
# 本次下单归因到的那次比价的 trace_id(客户端归因时从比价会话缓存带来;demo/老客户端/历史行为 NULL)。
|
||||
# 「已下单」按它精确对齐 comparison_record.trace_id —— 同店多次比价只标真正下单的那一条。
|
||||
trace_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
|
||||
device_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
|
||||
@@ -737,18 +737,18 @@ def harvest_abort(
|
||||
return rec
|
||||
|
||||
|
||||
def _ordered_trace_id_select(user_id: int):
|
||||
"""该用户「真实下单」(source='compare')覆盖到的比价 trace_id 的 select,给「已下单」筛选当子查询。
|
||||
def _ordered_shop_name_select(user_id: int):
|
||||
"""该用户「真实下单」(source='compare')覆盖到的店名 select,给「已下单」筛选当子查询。
|
||||
|
||||
口径与 [_ordered_trace_ids] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates
|
||||
口径与 [_ordered_shop_names] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates
|
||||
反查打标;这边是**分页之前**就要过滤,拿不到 candidates,只能整段下推成子查询。
|
||||
没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的 trace_id 可能上千,展开会撞 SQLite
|
||||
没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的店名可能上千,展开会撞 SQLite
|
||||
的绑定变量上限,而且又变回了那个「随下单量线性变慢」的老写法。
|
||||
"""
|
||||
return select(SavingsRecord.trace_id).where(
|
||||
return select(SavingsRecord.shop_name).where(
|
||||
SavingsRecord.user_id == user_id,
|
||||
SavingsRecord.source == "compare",
|
||||
SavingsRecord.trace_id.is_not(None),
|
||||
SavingsRecord.shop_name.is_not(None),
|
||||
)
|
||||
|
||||
|
||||
@@ -760,27 +760,27 @@ def _like_escape(kw: str) -> str:
|
||||
return kw.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
|
||||
|
||||
def _ordered_trace_ids(db: Session, user_id: int, candidates: set[str]) -> set[str]:
|
||||
"""[candidates] 里哪些 trace_id 被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。
|
||||
def _ordered_shop_names(db: Session, user_id: int, candidates: set[str]) -> set[str]:
|
||||
"""[candidates] 里哪些店名被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。
|
||||
|
||||
只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报带上本次比价的 trace_id →
|
||||
按 trace_id 精确对齐 comparison_record.trace_id:同一家店比价多次,只有真正下单的那一条会被
|
||||
标「已下单」。没带 trace_id 的下单(历史 / 老客户端)对齐不上任何记录 → 不进「已下单」。
|
||||
只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id,
|
||||
只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。
|
||||
语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。
|
||||
|
||||
⚠️ 只查**本页出现过的 trace_id**(candidates ≤ limit 条),不再把该用户全部下单 trace_id 捞回
|
||||
内存:老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回
|
||||
⚠️ 只查**本页出现过的店名**(candidates ≤ limit 条),不再把该用户全部下单店名捞回内存:
|
||||
老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回
|
||||
(避免 IN () 非法)。
|
||||
"""
|
||||
if not candidates:
|
||||
return set()
|
||||
rows = db.execute(
|
||||
select(SavingsRecord.trace_id).where(
|
||||
select(SavingsRecord.shop_name).where(
|
||||
SavingsRecord.user_id == user_id,
|
||||
SavingsRecord.source == "compare",
|
||||
SavingsRecord.trace_id.in_(candidates),
|
||||
SavingsRecord.shop_name.in_(candidates),
|
||||
).distinct()
|
||||
).scalars().all()
|
||||
return {t for t in rows if t}
|
||||
return {s for s in rows if s}
|
||||
|
||||
|
||||
def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]:
|
||||
@@ -832,7 +832,7 @@ def list_records(
|
||||
ordered: bool | None = None,
|
||||
keyword: str | None = None,
|
||||
) -> tuple[list[ComparisonRecord], int | None]:
|
||||
"""比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」标记(按 trace_id 精确对齐真实下单)+ 「看广告赚的金币」(瞬态,不写库)。"""
|
||||
"""比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。"""
|
||||
stmt = (
|
||||
select(ComparisonRecord)
|
||||
.where(ComparisonRecord.user_id == user_id)
|
||||
@@ -844,7 +844,7 @@ def list_records(
|
||||
# 分页之后一页里可能一条都不命中,列表看着就是空的/卡住的,得翻很多页才蹦出一条。
|
||||
if ordered:
|
||||
stmt = stmt.where(
|
||||
ComparisonRecord.trace_id.in_(_ordered_trace_id_select(user_id))
|
||||
ComparisonRecord.store_name.in_(_ordered_shop_name_select(user_id))
|
||||
)
|
||||
kw = (keyword or "").strip()
|
||||
if kw:
|
||||
@@ -865,16 +865,16 @@ def list_records(
|
||||
items = list(db.execute(stmt).scalars().all())
|
||||
next_cursor = items[-1].id if len(items) == limit else None
|
||||
|
||||
# 「已下单」标记:本页记录的 trace_id 若落在该用户真实下单(带 trace_id)的集合里即 True。
|
||||
# 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。
|
||||
# ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。
|
||||
page_traces = {it.trace_id for it in items}
|
||||
# ordered=True 时上面已按同一口径(_ordered_trace_id_select)筛过,本页必然全是已下单,
|
||||
# 省掉这次反查;其余情况按本页 trace_id 反查 savings。
|
||||
ordered_traces = page_traces if ordered else _ordered_trace_ids(db, user_id, page_traces)
|
||||
page_shops = {it.store_name for it in items if it.store_name}
|
||||
# ordered=True 时上面已按同一口径(_ordered_shop_name_select)筛过,本页必然全是已下单,
|
||||
# 省掉这次反查;其余情况照旧按本页店名反查 savings。
|
||||
ordered_shops = page_shops if ordered else _ordered_shop_names(db, user_id, page_shops)
|
||||
# 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。
|
||||
ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items])
|
||||
for it in items:
|
||||
it.ordered = it.trace_id in ordered_traces
|
||||
it.ordered = bool(it.store_name and it.store_name in ordered_shops)
|
||||
it.ad_coins_earned = ad_coins.get(it.trace_id, 0)
|
||||
|
||||
return items, next_cursor
|
||||
|
||||
@@ -177,7 +177,6 @@ def create_from_report(
|
||||
source_platform_name=req.source_platform_name,
|
||||
source_deeplink=req.source_deeplink,
|
||||
client_event_id=req.client_event_id,
|
||||
trace_id=req.trace_id,
|
||||
device_id=req.device_id,
|
||||
source="compare",
|
||||
# created_at 显式存 naive 北京 wall-clock(与 demo 行、聚合 _local_date 的 naive 分支一致)。
|
||||
|
||||
@@ -12,9 +12,6 @@ class OrderReportRequest(BaseModel):
|
||||
paid_amount_cents: int = Field(..., ge=0, description="实际支付金额(分)")
|
||||
device_id: str | None = Field(None, max_length=128)
|
||||
# ===== 比价时携带的记账信息(客户端从意图识别阶段缓存而来;旧版客户端可能不传,故全部可空)=====
|
||||
# 本次下单归因到的那次比价的 trace_id。服务端据此把「已下单」精确对齐到该条比价记录
|
||||
# (同店多次比价只标真正下单的那一条)。旧客户端不传 → None → 该单不进任何记录的「已下单」。
|
||||
trace_id: str | None = Field(None, max_length=64, description="归因到的比价 trace_id")
|
||||
shop_name: str | None = Field(None, max_length=128, description="门店名,如 肯德基宅急送(天北路店)")
|
||||
dishes: list[str] = Field(default_factory=list, description="菜品名列表")
|
||||
original_price_cents: int | None = Field(None, ge=0, description="源平台原价(分),省额=原价−实付")
|
||||
|
||||
@@ -64,31 +64,6 @@ def classify_cancelled_fallback(
|
||||
return None
|
||||
|
||||
|
||||
def _target_technical_failure_reason(
|
||||
rec: Any, biz_exclude_keywords: tuple[str, ...]
|
||||
) -> str | None:
|
||||
"""某目标平台「真技术失败」(pricebot 原始 status=='failed' 且 reason 非业务话术)的原因;无 → None。
|
||||
|
||||
记录级 fail_reason 是「展示口径」——一条比价里若某平台是干净业务结局(京东未找到菜),它会被派生
|
||||
成 headline,盖住另一平台的真技术崩溃(淘宝'比价过程出错')。这里扫 raw_payload.platform_results
|
||||
补判:任一目标平台 status='failed' 且 reason 不含业务词(pricebot 偶把打烊/不配送漏标成 failed,
|
||||
用 biz_exclude 过滤掉这些业务误标)→ 真技术崩溃。缺 raw_payload/platform_results / 结构异常 → None。
|
||||
"""
|
||||
raw = getattr(rec, "raw_payload", None)
|
||||
pr = raw.get("platform_results") if isinstance(raw, dict) else None
|
||||
if not isinstance(pr, dict):
|
||||
return None
|
||||
for v in pr.values():
|
||||
if not isinstance(v, dict) or v.get("is_source"):
|
||||
continue
|
||||
if v.get("status") != "failed":
|
||||
continue
|
||||
reason = (v.get("reason") or "").strip()
|
||||
if not any(w in reason for w in biz_exclude_keywords):
|
||||
return reason or "比价过程出错"
|
||||
return None
|
||||
|
||||
|
||||
def classify_record(
|
||||
rec: Any,
|
||||
*,
|
||||
@@ -112,11 +87,7 @@ def classify_record(
|
||||
return make_hit(rec, "T6", f"识别失败·{fail_reason[:80]}")
|
||||
if any(w in fail_reason for w in timeout_keywords):
|
||||
return make_hit(rec, "T2", fail_reason[:80])
|
||||
# fail_reason 是干净业务 headline,但可能盖住某目标平台的真技术失败(比价过程出错)→ 补判 T1。
|
||||
tech = _target_technical_failure_reason(rec, biz_exclude_keywords)
|
||||
if tech:
|
||||
return make_hit(rec, "T1", f"技术失败·{tech[:80]}")
|
||||
return None # 纯业务失败,不报
|
||||
return None
|
||||
if status == "cancelled":
|
||||
return classify_cancelled_fallback(
|
||||
rec,
|
||||
|
||||
@@ -180,26 +180,12 @@ _TABLE_COLUMNS = [
|
||||
{"name": "phone", "display_name": "手机号", "data_type": "text"},
|
||||
{"name": "cost", "display_name": "用时", "data_type": "text"},
|
||||
{"name": "reason", "display_name": "失败原因", "data_type": "text"},
|
||||
{"name": "stuck", "display_name": "末帧", "data_type": "text"},
|
||||
{"name": "stuck", "display_name": "卡点", "data_type": "text"},
|
||||
{"name": "ver", "display_name": "版本", "data_type": "text"},
|
||||
{"name": "trace", "display_name": "trace", "data_type": "lark_md"},
|
||||
]
|
||||
|
||||
|
||||
def _type_criteria(t: str, cancelled_ms_threshold: int, cancelled_step_threshold: int) -> str:
|
||||
"""各触发类型的「判据」文案,展示在卡片摘要里让收报警的人一眼知道为什么报。
|
||||
T5 带当前配置阈值(耗时 ms→s、步数);T1 覆盖两条来源(整场系统错 + 混合单里任一平台 status=failed)。"""
|
||||
if t == "T5":
|
||||
return f"耗时>{round(cancelled_ms_threshold / 1000)}s 或 步数>{cancelled_step_threshold}"
|
||||
if t == "T1":
|
||||
return "无业务原因的系统错 或 任一平台 status=failed"
|
||||
if t == "T2":
|
||||
return "原因含 超时/启动/加载"
|
||||
if t == "T6":
|
||||
return "原因含 未识别"
|
||||
return ""
|
||||
|
||||
|
||||
def format_alert_card(
|
||||
hits: list[AlertHit],
|
||||
*,
|
||||
@@ -208,8 +194,6 @@ def format_alert_card(
|
||||
interval_min: int,
|
||||
max_detail_per_type: int,
|
||||
max_total: int,
|
||||
cancelled_ms_threshold: int = 90000,
|
||||
cancelled_step_threshold: int = 30,
|
||||
) -> dict:
|
||||
"""返回飞书 schema 2.0 卡片 dict(配合 send_feishu_card 发送)。
|
||||
|
||||
@@ -241,15 +225,6 @@ def format_alert_card(
|
||||
f"**合计 {total} 条**:" + " | ".join(count_parts)
|
||||
)
|
||||
|
||||
# ---------- 判据说明(本期出现的类型各给一行判据,T5 带当前配置阈值)----------
|
||||
criteria_parts = [
|
||||
f"{ALERT_TYPE_LABELS[t]}={_type_criteria(t, cancelled_ms_threshold, cancelled_step_threshold)}"
|
||||
for t in _TYPE_ORDER
|
||||
if grouped_count.get(t)
|
||||
]
|
||||
if criteria_parts:
|
||||
md_content += "\n判据:" + " | ".join(criteria_parts)
|
||||
|
||||
# ---------- 截断:超 max_total 只出摘要 ----------
|
||||
if total > max_total:
|
||||
md_content += f"\n超 {max_total} 条仅列计数,明细见分析库 comparison_record"
|
||||
|
||||
@@ -46,22 +46,17 @@ class StuckPoint:
|
||||
pipeline_step: str
|
||||
frames: int # 末段连续困住的帧数(上限 max_tail)
|
||||
stuck_ms: int | None = None # 末段连续卡住的时长(ms);无 timestamp 时 None
|
||||
detected_page: str | None = None # 末帧所在页面(pricebot detected_page 原值);无则 None
|
||||
|
||||
def label(self) -> str:
|
||||
p = PLATFORM_LABELS.get(self.platform, self.platform)
|
||||
s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
|
||||
base = f"{p}·{s}"
|
||||
if self.detected_page:
|
||||
base += f"·{self.detected_page}" # 页面暂用 pricebot 原值(英文),无中文映射
|
||||
return base
|
||||
return f"{p}·{s}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StuckResult:
|
||||
readable: bool # trace 是否读到(区分「读不到」与「读到但没卡」)
|
||||
points: list[StuckPoint] # 卡死平台列表;readable=True 且空 = 确认没卡死
|
||||
last: StuckPoint | None = None # 末帧(帧数最多平台的末帧,带 detected_page);读不到 → None
|
||||
|
||||
|
||||
def dir_name_from_trace_url(trace_url: str | None) -> str | None:
|
||||
@@ -113,7 +108,7 @@ def _platform_stuck(
|
||||
stuck_ms = round((t_last - t_first).total_seconds() * 1000) if t_last and t_first else None
|
||||
if stuck_ms is not None and stuck_ms < 0:
|
||||
stuck_ms = None # 时钟不单调(帧 timestamp 回退)→ 降级为不显示时长
|
||||
return StuckPoint(platform, last_ps, count, stuck_ms, detected_page=last_pg)
|
||||
return StuckPoint(platform, last_ps, count, stuck_ms)
|
||||
|
||||
|
||||
def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
|
||||
@@ -123,8 +118,6 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
|
||||
return StuckResult(readable=False, points=[])
|
||||
points: list[StuckPoint] = []
|
||||
any_frames = False
|
||||
last: StuckPoint | None = None
|
||||
best_n = -1
|
||||
for pdir in sorted(trace_dir.iterdir()):
|
||||
if not pdir.is_dir():
|
||||
continue
|
||||
@@ -135,15 +128,9 @@ def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> Stuc
|
||||
sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
|
||||
if sp is not None:
|
||||
points.append(sp)
|
||||
# 末帧:取帧数最多平台的末帧(与 last_step 同口径),带 detected_page
|
||||
if len(step_files) > best_n:
|
||||
ps, pg, _ = _read_head(step_files[-1])
|
||||
if ps is not None:
|
||||
best_n = len(step_files)
|
||||
last = StuckPoint(pdir.name, ps, len(step_files), detected_page=pg)
|
||||
if not any_frames:
|
||||
return StuckResult(readable=False, points=[])
|
||||
return StuckResult(readable=True, points=points, last=last)
|
||||
return StuckResult(readable=True, points=points)
|
||||
except OSError:
|
||||
return StuckResult(readable=False, points=[])
|
||||
|
||||
@@ -164,9 +151,9 @@ def last_step(trace_dir: Path) -> StuckPoint | None:
|
||||
if best is None:
|
||||
return None
|
||||
_, platform, step_files = best
|
||||
ps, pg, _ts = _read_head(step_files[-1])
|
||||
ps, _pg, _ts = _read_head(step_files[-1])
|
||||
if ps is None:
|
||||
return None
|
||||
return StuckPoint(platform, ps, len(step_files), detected_page=pg) # stuck_ms=None(failed 不算时长)
|
||||
return StuckPoint(platform, ps, len(step_files)) # stuck_ms=None(failed 不算时长)
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
@@ -5,13 +5,9 @@
|
||||
|
||||
## 现状(2026-06 起):已改为 App 进程内任务,不再需要 systemd timer
|
||||
「0 点自动兑换」现由 **App 进程内后台任务** `app.core.daily_exchange_worker` 负责:App 一起来就
|
||||
每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥)。**无需再装 / 启用
|
||||
每 10 分钟检查、跨过北京 0 点自动跑一轮(逐用户幂等、文件锁互斥、重启补跑当天遗漏)。**无需再装 / 启用
|
||||
`daily-exchange.timer`**。
|
||||
|
||||
- **「当天已兑」标记持久化在 app_config**(key=`auto_exchange.last_run_date`):当天兑过后,同一北京日
|
||||
内进程重启 / 部署**不再重复补扫**——避免把 0 点后才达标的用户在非 0 点兑现金。只有真漏了 0 点
|
||||
(标记 < 今天,如 0 点服务器宕机)才会在重启后补跑一轮。
|
||||
|
||||
- 开关仍是 `.env` 的 `AUTO_EXCHANGE_ENABLED`(false → worker 不启动);间隔由 `AUTO_EXCHANGE_CHECK_INTERVAL_SEC`(默认 600s=10min)控。
|
||||
- 下方 systemd timer / 脚本属**遗留 + 手动应急**:逻辑同一套且幂等,可手动 `--once` 补跑;
|
||||
但**不要再 `enable` timer 与进程内 worker 并存**(虽幂等不会双兑,纯属多余)。
|
||||
|
||||
@@ -906,7 +906,6 @@ def test_comparison_records_show_real_order_status(
|
||||
order_amount_cents=1800,
|
||||
saved_amount_cents=300,
|
||||
shop_name="真实下单店",
|
||||
trace_id="comparison-ordered-shop",
|
||||
source="compare",
|
||||
client_event_id="admin-comparison-real-order",
|
||||
),
|
||||
|
||||
@@ -282,8 +282,8 @@ def test_card_has_table_seven_columns():
|
||||
cols = table["columns"]
|
||||
assert len(cols) == 7
|
||||
display_names = [c["display_name"] for c in cols]
|
||||
assert display_names == ["时间", "手机号", "用时", "失败原因", "末帧", "版本", "trace"]
|
||||
# 「末帧」列在「失败原因」后、「版本」前
|
||||
assert display_names == ["时间", "手机号", "用时", "失败原因", "卡点", "版本", "trace"]
|
||||
# 「卡点」列在「失败原因」后、「版本」前
|
||||
names = [c["name"] for c in cols]
|
||||
reason_idx = names.index("reason")
|
||||
stuck_idx = names.index("stuck")
|
||||
@@ -570,29 +570,3 @@ def test_card_stuck_point_shown_in_row():
|
||||
)
|
||||
row_without = card_without["body"]["elements"][1]["rows"][0]
|
||||
assert row_without["stuck"] == "-"
|
||||
|
||||
|
||||
def test_card_shows_criteria_legend_with_config_thresholds():
|
||||
"""卡片摘要含「判据」说明:深度放弃带当前配置阈值、技术失败说明任一平台 status=failed。"""
|
||||
hits = _card_hits(1, "T5", total_ms=95000) + _card_hits(1, "T1")
|
||||
card = format_alert_card(
|
||||
hits, window_label="w", phone_map={}, interval_min=15,
|
||||
max_detail_per_type=20, max_total=50,
|
||||
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
|
||||
)
|
||||
md = card["body"]["elements"][0]["content"]
|
||||
assert "判据" in md
|
||||
assert "耗时>90s 或 步数>30" in md # 深度放弃判据带配置值
|
||||
assert "任一平台 status=failed" in md # 技术失败判据
|
||||
|
||||
|
||||
def test_card_criteria_reflects_custom_thresholds():
|
||||
"""判据里的阈值随配置变化(不是写死 90/30)。"""
|
||||
card = format_alert_card(
|
||||
_card_hits(1, "T5", total_ms=200000),
|
||||
window_label="w", phone_map={}, interval_min=15,
|
||||
max_detail_per_type=20, max_total=50,
|
||||
cancelled_ms_threshold=180000, cancelled_step_threshold=45,
|
||||
)
|
||||
md = card["body"]["elements"][0]["content"]
|
||||
assert "耗时>180s 或 步数>45" in md
|
||||
|
||||
@@ -59,34 +59,3 @@ def test_reason_texts():
|
||||
assert t1_empty.reason == "技术失败·比价过程出错"
|
||||
t5 = classify_record(_rec(status="cancelled", total_ms=98000, step_count=26), **KW)
|
||||
assert t5.reason == "深度放弃"
|
||||
|
||||
|
||||
# ---- failed 混合单:业务 headline 盖住某平台真技术失败(线上 trace 20260806_144544)----
|
||||
|
||||
def test_failed_business_headline_masks_target_technical_failure_reports_t1():
|
||||
# fail_reason 被派生成京东业务原因(未找到),却盖住淘宝 status=failed 的真技术崩溃 → 补判 T1。
|
||||
rec = _rec(
|
||||
status="failed", fail_reason="京东此店内未找到这些菜品",
|
||||
information="比价过程出错,请稍后重试",
|
||||
raw_payload={"platform_results": {
|
||||
"meituan": {"status": "source", "is_source": True},
|
||||
"jd_waimai": {"status": "items_not_found", "reason": "京东此店内未找到这些菜品", "is_source": False},
|
||||
"taobao_flash": {"status": "failed", "reason": "比价过程出错,请稍后重试", "is_source": False},
|
||||
}},
|
||||
)
|
||||
hit = classify_record(rec, **KW)
|
||||
assert hit is not None
|
||||
assert hit.alert_type == "T1"
|
||||
|
||||
|
||||
def test_failed_target_failed_but_business_reason_no_false_positive():
|
||||
# pricebot 把打烊漏标成 status=failed,但 reason 是业务话术 → 不算技术崩溃,不报(不误报)。
|
||||
rec = _rec(
|
||||
status="failed", fail_reason="京东此店内未找到这些菜品",
|
||||
information="比价过程出错,请稍后重试",
|
||||
raw_payload={"platform_results": {
|
||||
"jd_waimai": {"status": "items_not_found", "reason": "京东此店内未找到这些菜品", "is_source": False},
|
||||
"taobao_flash": {"status": "failed", "reason": "门店已打烊,无法比价", "is_source": False},
|
||||
}},
|
||||
)
|
||||
assert classify_record(rec, **KW) is None
|
||||
|
||||
@@ -46,11 +46,10 @@ def test_cancelled_stuck_reports_via_trace(tmp_path):
|
||||
assert hits[0].reason == "深度放弃"
|
||||
assert "美团·加菜" in hits[0].stuck_point
|
||||
assert "帧" in hits[0].stuck_point
|
||||
assert "meal_detail_popup" in hits[0].stuck_point # 卡点带末帧页面
|
||||
|
||||
|
||||
def test_cancelled_readable_not_stuck_long_duration_reports(tmp_path):
|
||||
# trace 可读但没判出原地卡点:仍过耗时/步数阈值兜底,超长(>90s)照报 T5;没卡点也用末帧标"退出前在哪屏"。
|
||||
# trace 可读但没判出原地卡点:仍过耗时/步数阈值兜底,超长(>90s)照报 T5,卡点列留空。
|
||||
# (线上 trace 几乎总可读,若不回退则 total_ms 阈值形同虚设、超长放弃永不报——见 compare-fail-alert 排查。)
|
||||
p = tmp_path / "20260804_y" / "eleme"
|
||||
_frame(p, 0, "set_address", "home")
|
||||
@@ -62,7 +61,7 @@ def test_cancelled_readable_not_stuck_long_duration_reports(tmp_path):
|
||||
assert len(hits) == 1
|
||||
assert hits[0].alert_type == "T5"
|
||||
assert hits[0].reason == "深度放弃"
|
||||
assert hits[0].stuck_point == "饿了么·进店·store" # 无卡点 → 用末帧(平台·环节·页面)标退出前在哪屏
|
||||
assert hits[0].stuck_point is None # 没卡点 → 卡片卡点列显 "-"
|
||||
|
||||
|
||||
def test_cancelled_readable_not_stuck_short_no_report(tmp_path):
|
||||
|
||||
+31
-122
@@ -223,10 +223,10 @@ def test_stats_compare_count_and_saved(client) -> None:
|
||||
|
||||
|
||||
def test_records_ordered_flag(client) -> None:
|
||||
"""「已下单」按 trace_id 标记:下单上报带上本次比价的 trace_id,精确命中该条记录才 True。
|
||||
"""「已下单」店级标记:店名命中该用户 source='compare' 的下单记录才 True。
|
||||
|
||||
覆盖 list_records 只按**本页 trace_id**反查 savings 的写法(不把该用户全部下单 trace_id 捞回
|
||||
内存再取交集)——两种写法结果必须一致,故这里逐条断言。
|
||||
覆盖 list_records 只按**本页店名**反查 savings 的写法(原来是把该用户全部下单店名捞回内存
|
||||
再取交集,随下单量线性变慢)——两种写法结果必须一致,故这里按店名逐条断言。
|
||||
"""
|
||||
token = _login(client, "13800002010")
|
||||
|
||||
@@ -236,15 +236,14 @@ def test_records_ordered_flag(client) -> None:
|
||||
other["store_name"] = "没下过单的店"
|
||||
client.post("/api/v1/compare/record", json=other, headers=_auth(token))
|
||||
|
||||
# 下单前:两条都不该带「已下单」。按 trace_id 断言本测试自己的两条(不受同库其它用例数据干扰)。
|
||||
flags = {
|
||||
it["trace_id"]: it["ordered"]
|
||||
for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
# 下单前:两条都不该带「已下单」
|
||||
items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
assert {it["store_name"]: it["ordered"] for it in items} == {
|
||||
"海底捞(朝阳店)": False,
|
||||
"没下过单的店": False,
|
||||
}
|
||||
assert flags["ord-1"] is False
|
||||
assert flags["ord-2"] is False
|
||||
|
||||
# 对海底捞那次比价真实下单一笔:上报带上该次比价的 trace_id(order/report 写 source='compare')
|
||||
# 对海底捞真实下单一笔(order/report 写 source='compare' 的 savings_record)
|
||||
r = client.post(
|
||||
"/api/v1/order/report",
|
||||
json={
|
||||
@@ -256,28 +255,23 @@ def test_records_ordered_flag(client) -> None:
|
||||
"paid_amount_cents": 12350,
|
||||
"shop_name": "海底捞(朝阳店)",
|
||||
"original_price_cents": 12850,
|
||||
"trace_id": "ord-1",
|
||||
},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
# 下单后:只有命中 trace_id(ord-1)那条翻成 True,同店的另一次比价(ord-2)不受影响
|
||||
flags = {
|
||||
it["trace_id"]: it["ordered"]
|
||||
for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
# 下单后:只有同店名那条翻成 True,另一条不受影响
|
||||
items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
assert {it["store_name"]: it["ordered"] for it in items} == {
|
||||
"海底捞(朝阳店)": True,
|
||||
"没下过单的店": False,
|
||||
}
|
||||
assert flags["ord-1"] is True
|
||||
assert flags["ord-2"] is False
|
||||
|
||||
# 别人的下单不该影响本人标记(_ordered_trace_ids 按 user_id 过滤)
|
||||
# 别人的下单不该影响本人标记(_ordered_shop_names 按 user_id 过滤)
|
||||
token_b = _login(client, "13800002011")
|
||||
client.post("/api/v1/compare/record", json=_food_payload("ord-b"), headers=_auth(token_b))
|
||||
flags_b = {
|
||||
it["trace_id"]: it["ordered"]
|
||||
for it in client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"]
|
||||
}
|
||||
assert flags_b["ord-b"] is False
|
||||
items_b = client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"]
|
||||
assert [it["ordered"] for it in items_b] == [False]
|
||||
|
||||
|
||||
def test_records_list_omits_raw_payload(client) -> None:
|
||||
@@ -322,24 +316,20 @@ def test_records_ordered_filter(client) -> None:
|
||||
q["store_name"] = "没下过单的店"
|
||||
client.post("/api/v1/compare/record", json=q, headers=_auth(token))
|
||||
|
||||
# 对这 3 条「下过单的店」比价分别真实下单(各带自己的 trace_id);2 条「没下过单的店」不下单
|
||||
for i in range(3):
|
||||
r = client.post(
|
||||
"/api/v1/order/report",
|
||||
json={
|
||||
"client_event_id": f"evt-ordered-filter-{i}",
|
||||
"platform": "美团",
|
||||
"platform_package": "com.sankuai.meituan",
|
||||
"pay_channel": "wechat",
|
||||
"compared_price_cents": 12350,
|
||||
"paid_amount_cents": 12350,
|
||||
"shop_name": "下过单的店",
|
||||
"original_price_cents": 12850,
|
||||
"trace_id": f"of-ordered-{i}",
|
||||
},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
client.post(
|
||||
"/api/v1/order/report",
|
||||
json={
|
||||
"client_event_id": "evt-ordered-filter",
|
||||
"platform": "美团",
|
||||
"platform_package": "com.sankuai.meituan",
|
||||
"pay_channel": "wechat",
|
||||
"compared_price_cents": 12350,
|
||||
"paid_amount_cents": 12350,
|
||||
"shop_name": "下过单的店",
|
||||
"original_price_cents": 12850,
|
||||
},
|
||||
headers=_auth(token),
|
||||
)
|
||||
|
||||
# 不传 ordered:5 条全出(「全部记录」tab 口径不变)
|
||||
assert len(client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]) == 5
|
||||
@@ -449,84 +439,3 @@ def test_trace_id_required(client) -> None:
|
||||
token = _login(client, "13800002009")
|
||||
r = client.post("/api/v1/compare/record", json={"business_type": "food"}, headers=_auth(token))
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_records_ordered_by_trace_id(client) -> None:
|
||||
"""「已下单」精确到单次比价:同一家店比价多次,只有真实下单的那一次(下单上报带 trace_id)才算已下单。
|
||||
|
||||
下单上报带上本次比价的 trace_id,服务端按 trace_id 对齐 —— 同店其它比价(哪怕成功)不再被一并
|
||||
标「已下单」。这是从"店级"改"单次级"的核心契约。
|
||||
"""
|
||||
token = _login(client, "13800002016")
|
||||
|
||||
# 同一家店(默认 payload 店名都是「海底捞(朝阳店)」)比价两次,不同 trace_id
|
||||
client.post("/api/v1/compare/record", json=_food_payload("ord-trace-a"), headers=_auth(token))
|
||||
client.post("/api/v1/compare/record", json=_food_payload("ord-trace-b"), headers=_auth(token))
|
||||
|
||||
# 只对其中一次(trace-a)真实下单,上报带上该次比价的 trace_id
|
||||
r = client.post(
|
||||
"/api/v1/order/report",
|
||||
json={
|
||||
"client_event_id": "evt-trace-a",
|
||||
"platform": "美团",
|
||||
"platform_package": "com.sankuai.meituan",
|
||||
"pay_channel": "wechat",
|
||||
"compared_price_cents": 12350,
|
||||
"paid_amount_cents": 12350,
|
||||
"shop_name": "海底捞(朝阳店)",
|
||||
"original_price_cents": 12850,
|
||||
"trace_id": "ord-trace-a",
|
||||
},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
# ordered=true 只出下单的那条(trace-a);同店没下单的 trace-b 不进来
|
||||
page = client.get("/api/v1/compare/records?ordered=true", headers=_auth(token)).json()
|
||||
assert [it["trace_id"] for it in page["items"]] == ["ord-trace-a"]
|
||||
|
||||
# 全部记录里,只有 trace-a 带 ordered=true,同店的 trace-b 仍是 False
|
||||
items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
assert {it["trace_id"]: it["ordered"] for it in items} == {
|
||||
"ord-trace-a": True,
|
||||
"ord-trace-b": False,
|
||||
}
|
||||
|
||||
|
||||
def test_records_ordered_ignores_order_without_trace_id(client) -> None:
|
||||
"""无 trace_id 的下单(老客户端/历史)即使店名相同也不进「已下单」—— 锁定「无店名回退」契约。
|
||||
|
||||
下单上报不带 trace_id → savings_record.trace_id 为空 → 对齐不上任何比价记录。老逻辑靠店名
|
||||
会把同店比价一并标已下单,这条正是要防它被改回去。
|
||||
"""
|
||||
token = _login(client, "13800002017")
|
||||
|
||||
# 一条比价记录(默认店名「海底捞(朝阳店)」)
|
||||
client.post("/api/v1/compare/record", json=_food_payload("no-trace-order"), headers=_auth(token))
|
||||
|
||||
# 同店名、但**不带 trace_id** 的下单(模拟老客户端)
|
||||
r = client.post(
|
||||
"/api/v1/order/report",
|
||||
json={
|
||||
"client_event_id": "evt-no-trace",
|
||||
"platform": "美团",
|
||||
"platform_package": "com.sankuai.meituan",
|
||||
"pay_channel": "wechat",
|
||||
"compared_price_cents": 12350,
|
||||
"paid_amount_cents": 12350,
|
||||
"shop_name": "海底捞(朝阳店)",
|
||||
"original_price_cents": 12850,
|
||||
},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
# 店名虽同,但订单没 trace_id → 该记录不该被标「已下单」,也不进 ordered=true(无店名回退)
|
||||
flags = {
|
||||
it["trace_id"]: it["ordered"]
|
||||
for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]
|
||||
}
|
||||
assert flags["no-trace-order"] is False
|
||||
assert client.get(
|
||||
"/api/v1/compare/records?ordered=true", headers=_auth(token)
|
||||
).json()["items"] == []
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
"""0 点自动兑金币 worker:持久化「当天已兑」标记,防同日重启重复补扫。
|
||||
|
||||
背景(路径 B bug):worker 原用内存变量 last_run 记「今天跑过没」,进程重启即归零 → 每次
|
||||
非 0 点部署/重启都会全量补扫一遍,把 0 点后才达标的用户在**非 0 点**兑成现金(用户反馈的
|
||||
「非 0 点也出现金币转现金记录」)。修复:标记持久化到 app_config,跨重启不丢。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import delete, select, update
|
||||
|
||||
from app.core import daily_exchange_worker as w
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.app_config import AppConfig
|
||||
from app.models.user import User
|
||||
from app.models.wallet import CashTransaction, CoinAccount
|
||||
from app.repositories import wallet as wallet_repo
|
||||
|
||||
_PHONE_SEQ = [0]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_exchange_state():
|
||||
"""daily_auto_exchange 全表扫描 + app_config 标记会跨用例泄漏(SQLite 测试库 session 级共享,
|
||||
commit 后 rollback 是 no-op),故每个用例先清零所有余额 + 清掉自动兑标记,保证干净起步。"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
db.execute(update(CoinAccount).values(
|
||||
coin_balance=0, cash_balance_cents=0, invite_cash_balance_cents=0))
|
||||
db.execute(delete(AppConfig).where(AppConfig.key == w.LAST_RUN_KEY))
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
yield
|
||||
|
||||
|
||||
def _new_user(db, *, coin: int) -> int:
|
||||
"""建一个 User + 指定金币余额的 CoinAccount,返回 user_id。"""
|
||||
_PHONE_SEQ[0] += 1
|
||||
# 198 段避免撞其他测试文件的固定手机号 / username UNIQUE
|
||||
u = User(phone=f"198{_PHONE_SEQ[0]:08d}", username=f"ax{_PHONE_SEQ[0]}", status="active")
|
||||
db.add(u)
|
||||
db.flush()
|
||||
acc = wallet_repo.get_or_create_account(db, u.id, commit=False)
|
||||
acc.coin_balance = coin
|
||||
acc.total_coin_earned = coin
|
||||
db.flush()
|
||||
return u.id
|
||||
|
||||
|
||||
def test_first_run_converts_and_records_marker() -> None:
|
||||
"""首轮(标记为空)→ 到分全额兑 + 落库当天标记。"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
uid = _new_user(db, coin=300)
|
||||
db.commit()
|
||||
|
||||
today = date(2026, 8, 6)
|
||||
result = w._exchange_if_due(db, today)
|
||||
|
||||
assert result is not None and result["converted"] >= 1
|
||||
acc = db.get(CoinAccount, uid)
|
||||
assert acc.coin_balance == 0 and acc.cash_balance_cents > 0 # 300 为整分,无零头
|
||||
assert w._read_last_run(db) == today # 标记落库
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_same_day_restart_does_not_resweep() -> None:
|
||||
"""路径 B 回归:0 点兑过后,同日进程重启(下午部署)不得再补扫当天新达标用户。"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
today = date(2026, 8, 6)
|
||||
first = _new_user(db, coin=200)
|
||||
db.commit()
|
||||
assert w._exchange_if_due(db, today) is not None
|
||||
assert db.get(CoinAccount, first).coin_balance == 0 # 首轮兑掉
|
||||
|
||||
# 0 点后才达标的用户(白天攒够;或 0 点是零头、白天赚够)
|
||||
late = _new_user(db, coin=500)
|
||||
db.commit()
|
||||
|
||||
# 模拟同一北京日内进程重启 → 持久化标记 == today → 整轮跳过,不碰 late
|
||||
assert w._exchange_if_due(db, today) is None
|
||||
acc = db.get(CoinAccount, late)
|
||||
assert acc.coin_balance == 500 and acc.cash_balance_cents == 0
|
||||
assert db.execute(select(CashTransaction).where(
|
||||
CashTransaction.user_id == late,
|
||||
CashTransaction.biz_type == "exchange_in",
|
||||
)).first() is None # late 无任何兑现金流水
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_new_day_runs_again() -> None:
|
||||
"""跨到北京新的一天(或真漏了 0 点,标记 < today)→ 照常补跑。"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
w._write_last_run(db, date(2026, 8, 6)) # 昨天已兑
|
||||
late = _new_user(db, coin=500)
|
||||
db.commit()
|
||||
|
||||
result = w._exchange_if_due(db, date(2026, 8, 7))
|
||||
assert result is not None and result["converted"] >= 1
|
||||
assert db.get(CoinAccount, late).coin_balance == 0
|
||||
assert w._read_last_run(db) == date(2026, 8, 7)
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
@@ -29,7 +29,7 @@ def test_stuck_when_tail_repeats_same_step(tmp_path):
|
||||
_frame(pdir, i, "add_one_dish", "meal_detail_popup")
|
||||
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
|
||||
assert res.readable is True
|
||||
assert res.points == [StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")]
|
||||
assert res.points == [StuckPoint("meituan", "add_one_dish", 20)]
|
||||
|
||||
|
||||
def test_not_stuck_when_progressing(tmp_path):
|
||||
@@ -81,7 +81,7 @@ def test_per_platform_one_stuck_one_normal(tmp_path):
|
||||
_frame(e, i, "enter_store", "store")
|
||||
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
|
||||
assert res.readable is True
|
||||
assert res.points == [StuckPoint("meituan", "add_one_dish", 18, detected_page="meal_detail_popup")]
|
||||
assert res.points == [StuckPoint("meituan", "add_one_dish", 18)]
|
||||
|
||||
|
||||
def test_last_step_returns_busiest_platform_last_env(tmp_path):
|
||||
@@ -92,27 +92,7 @@ def test_last_step_returns_busiest_platform_last_env(tmp_path):
|
||||
for i in range(3):
|
||||
_frame(e, i, "enter_store", "store")
|
||||
sp = last_step(tmp_path)
|
||||
assert sp == StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup")
|
||||
|
||||
|
||||
def test_stuck_point_label_includes_page():
|
||||
# label 带页面(平台/环节中文 + 页面原始英文);无页面时只到环节
|
||||
assert StuckPoint("meituan", "add_one_dish", 20, detected_page="meal_detail_popup").label() \
|
||||
== "美团·加菜·meal_detail_popup"
|
||||
assert StuckPoint("meituan", "add_one_dish", 20).label() == "美团·加菜"
|
||||
|
||||
|
||||
def test_read_stuck_points_returns_last_frame(tmp_path):
|
||||
# res.last = 帧数最多平台的末帧(带 detected_page),供"退出前在哪屏"用
|
||||
m = tmp_path / "meituan"
|
||||
for i in range(6):
|
||||
_frame(m, i, "enter_store", "store")
|
||||
_frame(m, 6, "add_one_dish", "menu") # 末帧换到 menu
|
||||
e = tmp_path / "eleme"
|
||||
_frame(e, 0, "set_address", "home")
|
||||
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
|
||||
assert res.last == StuckPoint("meituan", "add_one_dish", 7, detected_page="menu")
|
||||
assert res.points == [] # 没卡死,但末帧照样有
|
||||
assert sp == StuckPoint("meituan", "add_one_dish", 20)
|
||||
|
||||
|
||||
def test_dir_name_from_trace_url():
|
||||
|
||||
Reference in New Issue
Block a user