feat(admin): 比价记录展示口径共享模块(外部缺失判为成功) #219

Merged
guke merged 32 commits from feat-compare-fail-alert into main 2026-08-05 17:30:35 +08:00
22 changed files with 3363 additions and 0 deletions
+50
View File
@@ -0,0 +1,50 @@
"""comparison_record 加 updated_at 列(报警水位)+ 回填现有行 + 索引
新增 updated_at:server_default + onupdate = func.now()(DB 时钟)。比价失败报警 worker 用它做
单调水位(WHERE updated_at > watermark)。加列后回填现有行 = created_at,避免冷启动 max(updated_at)
为 NULL;再置 NOT NULL + 建索引 ix_comparison_updated(水位查询按它)。batch 模式兼容 SQLite。
Revision ID: comparison_updated_at
Revises: deepseek_v4_flash_price
Create Date: 2026-08-04 00:00:00.000000
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "comparison_updated_at"
down_revision: str | Sequence[str] | None = "deepseek_v4_flash_price"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# 1) 先加可空列(不带 default,避免各库对 add-column-with-default 的差异)
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.add_column(sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True))
# 2) 回填现有行 = created_at(全新环境表为空,回填 no-op)
op.get_bind().execute(
sa.text(
"UPDATE comparison_record SET updated_at = created_at WHERE updated_at IS NULL"
)
)
# 3) 置 NOT NULL + server_default + 建索引
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.alter_column(
"updated_at",
existing_type=sa.DateTime(timezone=True),
nullable=False,
server_default=sa.text("CURRENT_TIMESTAMP"),
)
batch_op.create_index("ix_comparison_updated", ["updated_at"], unique=False)
def downgrade() -> None:
with op.batch_alter_table("comparison_record", schema=None) as batch_op:
batch_op.drop_index("ix_comparison_updated")
batch_op.drop_column("updated_at")
+312
View File
@@ -0,0 +1,312 @@
"""比价失败报警后台任务:周期扫 comparison_record 新落定记录 → 规则命中 → 飞书汇总。
结构仿 heartbeat_monitor_worker(单实例文件锁 + asyncio 轮询 + 优雅退出);发送与 DB 全同步,
放 asyncio.to_thread。水位存 app_config(key=compare_alert.last_watermark,值=上次处理的最大
updated_at ISO 串),查询用 updated_at 自身比较、规避时区。见 spec 第 5/6 节。
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from dataclasses import replace as _dc_replace
from datetime import datetime
from pathlib import Path
from sqlalchemy import func, select
from sqlalchemy.exc import SQLAlchemyError
from app.core.config import settings
from app.core.rewards import CN_TZ
from app.db.session import SessionLocal
from app.integrations import feishu_notifier
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.user import User
from app.services import trace_stuck
from app.services.compare_alert import (
AlertHit,
classify_cancelled_fallback,
classify_record,
make_hit,
)
from app.services.compare_alert_format import format_alert_card
logger = logging.getLogger("shagua.compare_alert")
WATERMARK_KEY = "compare_alert.last_watermark"
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "compare_alert.lock"
def _read_watermark(db) -> datetime | None:
row = db.get(AppConfig, WATERMARK_KEY)
if row is None or not row.value:
return None
try:
return datetime.fromisoformat(row.value)
except (ValueError, TypeError):
return None
def _write_watermark(db, value: datetime) -> None:
iso = value.isoformat()
row = db.get(AppConfig, WATERMARK_KEY)
if row is None:
db.add(AppConfig(key=WATERMARK_KEY, value=iso, updated_by_admin_id=None))
else:
row.value = iso
db.commit()
def _send_card(card: dict) -> None:
"""发飞书交互卡片(webhook 为空则只打日志、不外发)。失败抛 FeishuNotifyError 由调用方处理。"""
webhook = settings.COMPARE_ALERT_FEISHU_WEBHOOK
if not webhook:
title = card.get("header", {}).get("title", {}).get("content", "")
logger.info("[compare-alert] webhook 未配置,仅打印: title=%s", title)
return
feishu_notifier.send_feishu_card(
webhook, card, timeout=settings.COMPARE_ALERT_FEISHU_TIMEOUT_SEC
)
def _trace_dir(base: Path, trace_url: str | None) -> Path | None:
"""URL → trace 目录 Path;trace_url 缺失/无法解析 → None。"""
name = trace_stuck.dir_name_from_trace_url(trace_url)
if not name:
return None
return base / name
def _fmt_stuck(sp: trace_stuck.StuckPoint) -> str:
"""StuckPoint → 「平台·环节 110帧/32s」;stuck_ms 为 None 时省略时长。"""
s = f"{sp.label()} {sp.frames}"
if sp.stuck_ms is not None:
s += f"/{round(sp.stuck_ms / 1000)}s"
return s
def build_hits(
records: list,
*,
work_log_dir: str,
stuck_threshold: int,
max_tail: int,
max_trace_reads: int,
cancelled_ms_threshold: int,
cancelled_step_threshold: int,
timeout_keywords: tuple[str, ...],
unrecognized_keywords: tuple[str, ...],
biz_exclude_keywords: tuple[str, ...],
) -> list[AlertHit]:
"""编排:cancelled 走 trace 优先(读到确认没卡则不报,读不到回退保底);failed 附卡点。
trace 读取限量 max_trace_reads 次/轮;任何 trace 异常都在 trace_stuck 内部降级为
「读不到」,cancelled 因而回退保底、failed 不附卡点,绝不影响报警发送。
"""
base = Path(work_log_dir) if work_log_dir else None
reads = 0
hits: list = []
for rec in records:
if rec.status == "cancelled":
res = None
if base is not None and reads < max_trace_reads:
td = _trace_dir(base, rec.trace_url)
if td is not None:
res = trace_stuck.read_stuck_points(
td, threshold=stuck_threshold, max_tail=max_tail
)
reads += 1
if res is not None and res.readable:
if res.points:
stuck = "".join(_fmt_stuck(sp) for sp in res.points)
hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
else:
hit = None # 读到且确认没卡 → 不报
else:
hit = classify_cancelled_fallback(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
)
if hit is not None:
hits.append(hit)
else:
hit = classify_record(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
timeout_keywords=timeout_keywords,
unrecognized_keywords=unrecognized_keywords,
biz_exclude_keywords=biz_exclude_keywords,
)
if (
hit is not None
and hit.alert_type in ("T1", "T2", "T6")
and base is not None
and reads < max_trace_reads
):
td = _trace_dir(base, rec.trace_url)
if td is not None:
sp = trace_stuck.last_step(td)
reads += 1
if sp is not None:
# failed 是「末帧停在哪」:last_step 返回的 frames 是该平台总帧数(非"卡住"帧数)、
# stuck_ms=None,带上帧数/时长会误导,故只用 label() 显示环节
hit = _dc_replace(hit, stuck_point=sp.label())
if hit is not None:
hits.append(hit)
return hits
def _scan_and_alert() -> None:
"""一轮:读水位 → 查有更新记录 → 规则 → 有命中发飞书 → 成功推进水位。同步,放 to_thread 调。"""
with SessionLocal() as db:
watermark = _read_watermark(db)
if watermark is None:
# 冷启动:水位 = 当前 max(updated_at),不回溯历史失败。空表则本轮不建水位、下轮再说——
# 不用 datetime.now():那是本地时钟,与 SQLite 的 updated_at(UTC CURRENT_TIMESTAMP)
# 不同源、会差 8h,导致新记录永远追不上水位。只用 DB 产出的 updated_at 值。
max_updated = db.scalar(select(func.max(ComparisonRecord.updated_at)))
if max_updated is None:
logger.info("[compare-alert] 冷启动:表空,待有记录后再建水位")
return
_write_watermark(db, max_updated)
logger.info("[compare-alert] 冷启动,水位初始化=%s", max_updated)
return
records = list(
db.scalars(
select(ComparisonRecord)
.where(ComparisonRecord.updated_at > watermark)
.order_by(ComparisonRecord.updated_at.asc())
)
)
if not records:
return
batch_max = max(r.updated_at for r in records)
hits = build_hits(
records,
work_log_dir=settings.COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR,
stuck_threshold=settings.COMPARE_ALERT_STUCK_FRAME_THRESHOLD,
max_tail=settings.COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES,
max_trace_reads=settings.COMPARE_ALERT_TRACE_MAX_RECORDS,
cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
timeout_keywords=settings.compare_alert_timeout_keywords,
unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
)
if hits or settings.COMPARE_ALERT_SEND_EMPTY:
label = datetime.now(CN_TZ).strftime("%Y-%m-%d %H:%M")
interval_min = max(1, settings.COMPARE_ALERT_SCAN_INTERVAL_SEC // 60)
if hits:
# join User 取手机号
uids = {h.user_id for h in hits if h.user_id is not None}
if uids:
users = db.scalars(select(User).where(User.id.in_(uids)))
phone_map = {u.id: u.phone for u in users}
else:
phone_map = {}
else:
phone_map = {}
card = format_alert_card(
hits,
window_label=label,
phone_map=phone_map,
interval_min=interval_min,
max_detail_per_type=settings.COMPARE_ALERT_MAX_DETAIL_PER_TYPE,
max_total=settings.COMPARE_ALERT_MAX_TOTAL,
)
try:
_send_card(card)
except feishu_notifier.FeishuNotifyError:
logger.warning("[compare-alert] 发送失败,水位不推进、下轮补发", exc_info=True)
return # 不推进水位
_write_watermark(db, batch_max)
if hits:
logger.info("[compare-alert] 本轮命中 %d 条,水位推进到 %s", len(hits), batch_max)
# ---- 单实例锁 + 轮询循环(结构同 heartbeat_monitor_worker)---------------------
def _touch_lock() -> None:
with contextlib.suppress(FileNotFoundError):
os.utime(_LOCK_PATH, None)
@contextlib.contextmanager
def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
fd: int | None = None
try:
try:
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
except FileExistsError:
try:
age = time.time() - _LOCK_PATH.stat().st_mtime
except FileNotFoundError:
age = stale_after_sec + 1
if age > stale_after_sec:
with contextlib.suppress(FileNotFoundError):
_LOCK_PATH.unlink()
try:
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
except FileExistsError:
fd = None
if fd is None:
yield False
return
os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
yield True
finally:
if fd is not None:
os.close(fd)
with contextlib.suppress(FileNotFoundError):
_LOCK_PATH.unlink()
async def _run_loop() -> None:
interval = max(60, int(settings.COMPARE_ALERT_SCAN_INTERVAL_SEC))
lock_stale_after = max(interval * 3, 600)
with _single_instance_lock(lock_stale_after) as lock_acquired:
if not lock_acquired:
logger.warning("compare-alert skipped: another worker owns lock")
return
logger.info("compare-alert worker started interval=%ss", interval)
try:
while True:
try:
_touch_lock()
await asyncio.to_thread(_scan_and_alert)
except SQLAlchemyError:
logger.exception("compare-alert db error")
except Exception: # noqa: BLE001 - 后台任务不因单次异常退出
logger.exception("compare-alert unexpected error")
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("compare-alert worker stopped")
raise
def start_compare_alert_worker() -> asyncio.Task | None:
if not settings.COMPARE_ALERT_ENABLED:
logger.info("compare-alert worker disabled")
return None
return asyncio.create_task(_run_loop(), name="compare-alert-worker")
async def stop_compare_alert_worker(task: asyncio.Task | None) -> None:
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
+36
View File
@@ -132,6 +132,26 @@ class Settings(BaseSettings):
HEARTBEAT_TIMEOUT_MINUTES: int = 60 # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀)
HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期
# ===== 比价失败报警(常驻 worker 周期扫 comparison_record → 飞书汇总)=====
COMPARE_ALERT_ENABLED: bool = False # 总开关(默认关;启用用 .env COMPARE_ALERT_ENABLED=true 覆盖,别改这默认值);关时 worker 不启动
COMPARE_ALERT_SCAN_INTERVAL_SEC: int = 900 # 扫描间隔(默认 15min,可配 1800=30min)
COMPARE_ALERT_FEISHU_WEBHOOK: str = "" # 群机器人 webhook(敏感,放 .env 别硬编码进代码);空则 worker 仅打日志不外发
COMPARE_ALERT_FEISHU_TIMEOUT_SEC: float = 10.0 # 飞书 POST 读/连超时
COMPARE_ALERT_CANCELLED_MS_THRESHOLD: int = 90000 # T5 耗时阈值(ms)
COMPARE_ALERT_CANCELLED_STEP_THRESHOLD: int = 30 # T5 步数阈值
COMPARE_ALERT_TIMEOUT_KEYWORDS: str = "超时,启动,加载" # T2 关键词(逗号分隔)
COMPARE_ALERT_UNRECOGNIZED_KEYWORDS: str = "未识别" # T6 关键词
COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS: str = "未找到,打烊,起送,门店,店内,不配送,这些菜,未入驻,休息" # T1 information 业务词排除
COMPARE_ALERT_MAX_DETAIL_PER_TYPE: int = 20 # 单类型明细截断
COMPARE_ALERT_MAX_TOTAL: int = 50 # 本期总命中截断(超则只给计数)
COMPARE_ALERT_SEND_EMPTY: bool = False # 无命中是否发「本期无异常」简讯
# ===== 卡死定位(读 pricebot trace 末段判原地打转)=====
COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR: str = "" # pricebot work_logs 绝对路径(敏感,放 .env);空=跳过 trace、cancelled 全走保底
COMPARE_ALERT_STUCK_FRAME_THRESHOLD: int = 15 # 末段连续同环节达此帧数判卡死
COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES: int = 40 # 每平台最多往前读多少帧
COMPARE_ALERT_TRACE_MAX_RECORDS: int = 30 # 每轮最多对多少条命中记录读 trace(限量)
# ===== 短信 =====
SMS_MOCK: bool = True
SMS_CODE_TTL_SEC: int = 300
@@ -217,6 +237,22 @@ class Settings(BaseSettings):
phones.add(self.test_account_phone)
return frozenset(phones)
def _csv(self, raw: str) -> tuple[str, ...]:
"""逗号分隔字符串 → 去空白非空元组(报警关键词解析共用)。"""
return tuple(w.strip() for w in raw.split(",") if w.strip())
@property
def compare_alert_timeout_keywords(self) -> tuple[str, ...]:
return self._csv(self.COMPARE_ALERT_TIMEOUT_KEYWORDS)
@property
def compare_alert_unrecognized_keywords(self) -> tuple[str, ...]:
return self._csv(self.COMPARE_ALERT_UNRECOGNIZED_KEYWORDS)
@property
def compare_alert_biz_exclude_keywords(self) -> tuple[str, ...]:
return self._csv(self.COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS)
# ===== 美团联盟 CPS =====
# 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
MT_CPS_APP_KEY: str = ""
+48
View File
@@ -0,0 +1,48 @@
"""飞书群自定义机器人发送(text / post 消息)。
自定义机器人「关键词」验证:消息 content.text 必须含机器人配置的关键词,否则飞书返回 code!=0
(如 19024 Key Words Not Found)——本项目消息由 compare_alert_format 生成,标题已含「比价失败报警」。
不需要签名(sign)/IP 白名单。文档:https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot
"""
from __future__ import annotations
import httpx
class FeishuNotifyError(Exception):
"""飞书发送失败(网络错误 / 非 2xx / 业务 code!=0,含关键词不匹配)。"""
def _post_feishu(webhook_url: str, payload: dict, timeout: float) -> None:
"""内部 helper:POST payload 到飞书 webhook 并校验响应。失败抛 FeishuNotifyError。"""
try:
resp = httpx.post(webhook_url, json=payload, timeout=timeout)
except httpx.HTTPError as e:
raise FeishuNotifyError(f"feishu request failed: {e}") from e
if resp.status_code >= 300:
raise FeishuNotifyError(f"feishu http {resp.status_code}: {resp.text[:200]}")
try:
data = resp.json()
except ValueError as e:
raise FeishuNotifyError(f"feishu bad json: {resp.text[:200]}") from e
code = data.get("code", data.get("StatusCode", 0))
if code not in (0, None):
raise FeishuNotifyError(f"feishu code={code} msg={data.get('msg') or data.get('StatusMessage')}")
def send_feishu_text(webhook_url: str, text: str, *, timeout: float = 10.0) -> None:
"""POST 一条 text 消息到飞书群机器人 webhook。失败(网络/HTTP/业务 code)抛 FeishuNotifyError。"""
payload = {"msg_type": "text", "content": {"text": text}}
_post_feishu(webhook_url, payload, timeout)
def send_feishu_post(webhook_url: str, title: str, content: list, *, timeout: float = 10.0) -> None:
"""发飞书富文本(post)。content 是段落数组,每段是元素数组[{tag:text/a,...}]。失败抛 FeishuNotifyError。"""
payload = {"msg_type": "post", "content": {"post": {"zh_cn": {"title": title, "content": content}}}}
_post_feishu(webhook_url, payload, timeout)
def send_feishu_card(webhook_url: str, card: dict, *, timeout: float = 10.0) -> None:
"""发飞书交互卡片(interactive)。card 为 schema 2.0 卡片 dict。失败抛 FeishuNotifyError。"""
payload = {"msg_type": "interactive", "card": card}
_post_feishu(webhook_url, payload, timeout)
+6
View File
@@ -45,6 +45,10 @@ 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.compare_alert_worker import (
start_compare_alert_worker,
stop_compare_alert_worker,
)
from app.core.config import settings
from app.core.cps_reconcile_worker import (
start_cps_reconcile_worker,
@@ -123,6 +127,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
observe_task = start_observe_worker()
inactivity_task = start_inactivity_reset_worker()
llm_cost_backfill_task = start_llm_cost_backfill_worker()
compare_alert_task = start_compare_alert_worker()
try:
yield
finally:
@@ -133,6 +138,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
await stop_observe_worker(observe_task)
await stop_inactivity_reset_worker(inactivity_task)
await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
await stop_compare_alert_worker(compare_alert_task)
await aclose_pricebot_client()
mt_meituan.close_client()
logger.info("shutting down")
+12
View File
@@ -49,6 +49,9 @@ class ComparisonRecord(Base):
# 单列 user_id 索引只能过滤,排序仍要把该用户全部记录取出来排一遍;这条复合索引的**反向扫**
# 恰好等于 (created_at DESC, id DESC),PG 直接取前 n 条、免排序。列序不能动。
Index("ix_comparison_user_created", "user_id", "created_at", "id"),
# 比价报警 worker 的水位查询 WHERE updated_at > watermark 走它。显式命名(不用列上
# index=True 的自动名 ix_comparison_record_updated_at)以与迁移 create_index 同名、免 autogenerate 漂移。
Index("ix_comparison_updated", "updated_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -160,6 +163,15 @@ class ComparisonRecord(Base):
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
# 记录任一次更新(建 running 行 / done / abort 落终态)的 DB 时钟时间。比价报警 worker 的
# 水位列:按 updated_at 单调推进扫描,任何记录落定/更新都刷新它 > 水位、必被下轮扫到,
# 根治 created_at 水位漏掉「慢失败」(落定延迟 p99 达 7-10min)。onupdate 在 ORM UPDATE 时自动刷新。
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
def __repr__(self) -> str: # pragma: no cover
return (
+98
View File
@@ -0,0 +1,98 @@
"""比价失败报警规则:一条记录 → 命中的 AlertHit(或 None)。
纯函数不碰 DB(阈值/关键词由调用方从 config 传入),便于单测与调阈值判定顺序保证四类互斥:
failed fail_reason ( information 非业务)=T1 / 含未识别=T6 / 含超时词=T2 / 其余业务不报;
cancelled 且耗时或步数超阈值=T5(深度放弃);success/running 不报
口径依据见 docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md 3
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Any
ALERT_TYPE_LABELS: dict[str, str] = {
"T1": "系统技术失败",
"T6": "商品识别失败",
"T2": "启动/超时失败",
"T5": "深度放弃(cancelled)",
}
@dataclass(frozen=True)
class AlertHit:
trace_id: str
alert_type: str
reason: str
app_version: str | None
created_at: datetime | None
trace_url: str | None
user_id: int | None
total_ms: int | None = None
step_count: int | None = None
stuck_point: str | None = None
def make_hit(rec: Any, alert_type: str, reason: str) -> AlertHit:
return AlertHit(
trace_id=rec.trace_id,
alert_type=alert_type,
reason=reason,
app_version=getattr(rec, "app_version", None),
created_at=getattr(rec, "created_at", None),
trace_url=getattr(rec, "trace_url", None),
user_id=getattr(rec, "user_id", None),
total_ms=getattr(rec, "total_ms", None),
step_count=getattr(rec, "step_count", None),
)
def classify_cancelled_fallback(
rec: Any,
*,
cancelled_ms_threshold: int,
cancelled_step_threshold: int,
) -> AlertHit | None:
"""cancelled 保底判定(读不到 trace 时用):超耗时或步数阈值 → T5 深度放弃,否则 None。纯函数。"""
ms = rec.total_ms
step = rec.step_count
deep = (ms is not None and ms > cancelled_ms_threshold) or (
step is not None and step > cancelled_step_threshold
)
if deep:
return make_hit(rec, "T5", "深度放弃")
return None
def classify_record(
rec: Any,
*,
cancelled_ms_threshold: int,
cancelled_step_threshold: int,
timeout_keywords: tuple[str, ...],
unrecognized_keywords: tuple[str, ...],
biz_exclude_keywords: tuple[str, ...],
) -> AlertHit | None:
"""判定单条记录是否触发报警。rec 需有 status/fail_reason/information/total_ms/step_count/
trace_id/app_version 属性(ComparisonRecord 或等价对象)"""
status = rec.status
if status == "failed":
fail_reason = rec.fail_reason
if fail_reason is None:
info = (rec.information or "").strip()
if info and any(w in info for w in biz_exclude_keywords):
return None
return make_hit(rec, "T1", f"技术失败·{info[:80] or '比价过程出错'}")
if any(w in fail_reason for w in unrecognized_keywords):
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])
return None
if status == "cancelled":
return classify_cancelled_fallback(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
)
return None
+243
View File
@@ -0,0 +1,243 @@
"""AlertHit[] → 飞书群机器人消息。
提供三个格式化函数:
- format_alert_message: 纯文本(保留,已有集成测试依赖)
- format_alert_post: 富文本 post(行式明细:时间手机版本原因trace 超链接)
- format_alert_card: schema 2.0 卡片 + table 组件(正式发送格式)
按触发类型分组,每类给计数 + 明细(trace/版本/原因)两级截断防报警风暴:单类型超
max_detail_per_type 只列前 N + 另有 M ;本期总量超 max_total 只给各类型计数提示去分析库查
标题含关键词比价失败报警飞书自定义机器人用关键词验证,消息必须含它,否则被拒收
"""
from __future__ import annotations
from app.services.compare_alert import ALERT_TYPE_LABELS, AlertHit
ALERT_KEYWORD = "比价失败报警"
_TYPE_ORDER = ("T1", "T6", "T2", "T5")
def _detail_line(h: AlertHit) -> str:
ver = h.app_version or "?"
return f" - trace {h.trace_id} | {ver} | {h.reason}"
def format_alert_message(
hits: list[AlertHit],
*,
window_label: str,
max_detail_per_type: int,
max_total: int,
) -> str:
total = len(hits)
header = f"🚨 {ALERT_KEYWORD} · {window_label} · 本期触发 {total}"
grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
for h in hits:
grouped.setdefault(h.alert_type, []).append(h)
lines = [header]
counts_only = total > max_total
for t in _TYPE_ORDER:
bucket = grouped.get(t) or []
if not bucket:
continue
lines.append(f"{ALERT_TYPE_LABELS[t]} {len(bucket)}")
if counts_only:
continue
shown = bucket[:max_detail_per_type]
lines.extend(_detail_line(h) for h in shown)
if len(bucket) > max_detail_per_type:
lines.append(f" …另有 {len(bucket) - max_detail_per_type}")
if counts_only:
lines.append(f"(本期命中超 {max_total} 条,仅列计数,明细见分析库 comparison_record)")
return "\n".join(lines)
def format_alert_post(
hits: list[AlertHit],
*,
window_label: str,
phone_map: dict[int, str],
max_detail_per_type: int,
max_total: int,
) -> tuple[str, list]:
"""行式富文本:返回 (title, content)。title 含 ALERT_KEYWORD(飞书关键词验证)。
content: 摘要段(各类型计数) + 表头段 + 明细行(每条时间手机版本原因+ trace 超链接 a 元素)
phone_map: {user_id: phone};明细手机号取 phone_map.get(hit.user_id) or "-"
截断规则:总命中 > max_total 只出摘要+各类型计数(不列明细);
否则明细最多列前 max_detail_per_type ,超出加另有 N
"""
total = len(hits)
title = f"🚨 {ALERT_KEYWORD} · {window_label}"
grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
for h in hits:
grouped.setdefault(h.alert_type, []).append(h)
# 摘要段:各类型计数
summary_parts = [f"{ALERT_TYPE_LABELS[t]} {len(grouped[t])}" for t in _TYPE_ORDER if grouped.get(t)]
summary_text = f"合计 {total} 条:" + " ".join(summary_parts)
content: list[list[dict]] = [
[{"tag": "text", "text": summary_text}],
]
counts_only = total > max_total
if counts_only:
content.append([{"tag": "text", "text": f"(本期命中超 {max_total} 条,仅列计数,明细见分析库 comparison_record)"}])
return title, content
# 表头段
content.append([{"tag": "text", "text": "时间 | 手机号 | 版本 | 失败原因 | trace"}])
# 明细行(按类型顺序展开,每条一段)
shown_count = 0
for t in _TYPE_ORDER:
bucket = grouped.get(t) or []
if not bucket:
continue
for h in bucket[:max_detail_per_type]:
time_str = h.created_at.strftime("%m-%d %H:%M") if h.created_at else "-"
phone = phone_map.get(h.user_id) if h.user_id is not None else None
phone = phone or "-"
ver = h.app_version or "-"
row: list[dict] = [{"tag": "text", "text": f"{time_str} {phone} {ver} {h.reason} "}]
if h.trace_url:
row.append({"tag": "a", "text": "trace", "href": h.trace_url})
else:
row.append({"tag": "text", "text": h.trace_id[:16]})
content.append(row)
shown_count += 1
if len(bucket) > max_detail_per_type:
content.append([{"tag": "text", "text": f"…另有 {len(bucket) - max_detail_per_type}"}])
return title, content
# ---- format_alert_card (schema 2.0 卡片 + table 组件) ----
def _cost_cell(total_ms: int | None, step_count: int | None) -> str:
"""组合「用时」列值。有 ms → '{N}s',有 step_count → '{M}',两者用 ' / ' 连;都无 → '-'"""
parts = []
if total_ms is not None:
parts.append(f"{round(total_ms / 1000)}s")
if step_count is not None:
parts.append(f"{step_count}")
return " / ".join(parts) or "-"
def _build_card(title_text: str, elements: list[dict]) -> dict:
"""组装 schema 2.0 红色 header 卡片;三条路径只需决定 elements。"""
return {
"schema": "2.0",
"header": {
"title": {"tag": "plain_text", "content": title_text},
"template": "red",
},
"body": {"elements": elements},
}
def _build_table_rows(
hits: list[AlertHit],
*,
phone_map: dict[int, str],
max_detail_per_type: int,
) -> list[dict]:
"""按 _TYPE_ORDER 顺序展开,每类型最多 max_detail_per_type 条。"""
grouped: dict[str, list[AlertHit]] = {t: [] for t in _TYPE_ORDER}
for h in hits:
if h.alert_type in grouped:
grouped[h.alert_type].append(h)
# 非 _TYPE_ORDER 类型静默跳过(与既有行为一致)
rows = []
for t in _TYPE_ORDER:
bucket = grouped.get(t) or []
for h in bucket[:max_detail_per_type]:
time_str = h.created_at.strftime("%m-%d %H:%M") if h.created_at else "-"
phone = (phone_map.get(h.user_id) if h.user_id is not None else None) or "-"
trace = (
f"[链接]({h.trace_url})" if h.trace_url
else (h.trace_id or "")[:12]
)
rows.append({
"time": time_str,
"phone": phone,
"cost": _cost_cell(h.total_ms, h.step_count),
"reason": h.reason,
"stuck": h.stuck_point or "-",
"ver": h.app_version or "-",
"trace": trace,
})
return rows
_TABLE_COLUMNS = [
{"name": "time", "display_name": "时间", "data_type": "text"},
{"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": "ver", "display_name": "版本", "data_type": "text"},
{"name": "trace", "display_name": "trace", "data_type": "lark_md"},
]
def format_alert_card(
hits: list[AlertHit],
*,
window_label: str,
phone_map: dict[int, str],
interval_min: int,
max_detail_per_type: int,
max_total: int,
) -> dict:
"""返回飞书 schema 2.0 卡片 dict(配合 send_feishu_card 发送)。
- header: template=redtitle ALERT_KEYWORD飞书关键词验证必须
- body 第一个元素: markdown 摘要数据范围 + 合计 + 各类型计数
- hits: 只有摘要本期无异常
- total > max_total: 只有摘要提示去 comparison_record 不加 table
- 否则: 第二个元素为 table列序 time/phone/cost/reason/stuck/ver/trace
"""
total = len(hits)
title_text = f"🚨 {ALERT_KEYWORD} · {window_label}"
# ---------- 空 hits ----------
if total == 0:
md_content = f"数据范围:近 {interval_min} 分钟\n本期无异常"
return _build_card(title_text, [{"tag": "markdown", "content": md_content}])
# ---------- 摘要 ----------
grouped_count: dict[str, int] = {}
for h in hits:
grouped_count[h.alert_type] = grouped_count.get(h.alert_type, 0) + 1
count_parts = [
f"{ALERT_TYPE_LABELS[t]} {grouped_count[t]}"
for t in _TYPE_ORDER
if grouped_count.get(t)
]
md_content = (
f"数据范围:近 {interval_min} 分钟\n"
f"**合计 {total} 条**" + " ".join(count_parts)
)
# ---------- 截断:超 max_total 只出摘要 ----------
if total > max_total:
md_content += f"\n{max_total} 条仅列计数,明细见分析库 comparison_record"
return _build_card(title_text, [{"tag": "markdown", "content": md_content}])
# ---------- 常规:摘要 + table ----------
rows = _build_table_rows(hits, phone_map=phone_map, max_detail_per_type=max_detail_per_type)
table_element = {
"tag": "table",
"page_size": 10,
"row_height": "low",
"header_style": {"background_style": "grey", "bold": True},
"columns": _TABLE_COLUMNS,
"rows": rows,
}
return _build_card(title_text, [{"tag": "markdown", "content": md_content}, table_element])
+159
View File
@@ -0,0 +1,159 @@
"""比价卡死定位:读 pricebot trace 末段,判某平台是否原地打转(卡死)。
同机直读 {WORK_LOG_DIR}/{dir_name}/{platform}/step_*.json,只取头部字段
(pipeline_step/detected_page),不解析后面的无障碍树(windows,占单帧 99% 体积)
判据与降级见 docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
# pipeline_step / 平台名 → 中文(映射不到原样显示英文,不阻断)。按 pricebot 实际枚举补全。
PIPELINE_STEP_LABELS: dict[str, str] = {
"set_address": "定位",
"enter_store": "进店",
"add_one_dish": "加菜",
"match_dish": "找菜",
"checkout": "结算",
}
PLATFORM_LABELS: dict[str, str] = {
"meituan": "美团",
"eleme": "饿了么",
"jd_waimai": "京东外卖",
}
_PIPE_RE = re.compile(r'"pipeline_step":\s*"([^"]*)"')
_PAGE_RE = re.compile(r'"detected_page":\s*"([^"]*)"')
_TS_RE = re.compile(r'"timestamp":\s*"([^"]*)"')
_STEP_NUM_RE = re.compile(r"step_(\d+)")
def _parse_ts(s: str | None) -> datetime | None:
if not s:
return None
try:
return datetime.fromisoformat(s)
except (ValueError, TypeError):
return None
@dataclass(frozen=True)
class StuckPoint:
platform: str
pipeline_step: str
frames: int # 末段连续困住的帧数(上限 max_tail)
stuck_ms: int | None = None # 末段连续卡住的时长(ms);无 timestamp 时 None
def label(self) -> str:
p = PLATFORM_LABELS.get(self.platform, self.platform)
s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
return f"{p}·{s}"
@dataclass(frozen=True)
class StuckResult:
readable: bool # trace 是否读到(区分「读不到」与「读到但没卡」)
points: list[StuckPoint] # 卡死平台列表;readable=True 且空 = 确认没卡死
def dir_name_from_trace_url(trace_url: str | None) -> str | None:
""".../traces/{dir_name}/ → dir_name;空/异常 → None。"""
if not trace_url:
return None
name = trace_url.rstrip("/").rsplit("/", 1)[-1]
return name or None
def _step_num(path: Path) -> int:
m = _STEP_NUM_RE.search(path.name)
return int(m.group(1)) if m else -1
def _read_head(path: Path, nbytes: int = 4096) -> tuple[str | None, str | None, str | None]:
"""只读文件头部,抠 (pipeline_step, detected_page, timestamp)。它们在 json 最前面。"""
try:
with open(path, encoding="utf-8", errors="replace") as f:
head = f.read(nbytes)
except OSError:
return None, None, None
ps = _PIPE_RE.search(head)
pg = _PAGE_RE.search(head)
ts = _TS_RE.search(head)
return (ps.group(1) if ps else None, pg.group(1) if pg else None, ts.group(1) if ts else None)
def _platform_stuck(
platform: str, step_files: list[Path], threshold: int, max_tail: int
) -> StuckPoint | None:
"""末帧往前数连续同 (pipeline_step, detected_page) 的帧数 ≥threshold → 卡死。"""
tail = step_files[-max_tail:]
heads = [_read_head(p) for p in tail] # [(ps, pg, ts), ...]
last_ps, last_pg, _ = heads[-1]
if last_ps is None:
return None
seg_ts: list[str | None] = [] # 连续段的 timestamp(逆序:末帧在前)
for ps, pg, ts in reversed(heads):
if ps == last_ps and pg == last_pg:
seg_ts.append(ts)
else:
break
count = len(seg_ts)
if count < threshold:
return None
# seg_ts[0]=末帧, seg_ts[-1]=段首帧;两端都能解析才算时长
t_last, t_first = _parse_ts(seg_ts[0]), _parse_ts(seg_ts[-1])
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)
def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
"""逐平台判卡死。目录不存在/异常/无平台帧 → StuckResult(readable=False, points=[])。"""
try:
if not trace_dir.is_dir():
return StuckResult(readable=False, points=[])
points: list[StuckPoint] = []
any_frames = False
for pdir in sorted(trace_dir.iterdir()):
if not pdir.is_dir():
continue
step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
if not step_files:
continue
any_frames = True
sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
if sp is not None:
points.append(sp)
if not any_frames:
return StuckResult(readable=False, points=[])
return StuckResult(readable=True, points=points)
except OSError:
return StuckResult(readable=False, points=[])
def last_step(trace_dir: Path) -> StuckPoint | None:
"""failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。"""
try:
if not trace_dir.is_dir():
return None
best: tuple[int, str, list[Path]] | None = None
for pdir in sorted(trace_dir.iterdir()):
if not pdir.is_dir():
continue
step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
# 平局(同帧数)时取字典序第一个平台(sorted 保证稳定)
if step_files and (best is None or len(step_files) > best[0]):
best = (len(step_files), pdir.name, step_files)
if best is None:
return None
_, platform, step_files = best
ps, _pg, _ts = _read_head(step_files[-1])
if ps is None:
return None
return StuckPoint(platform, ps, len(step_files)) # stuck_ms=None(failed 不算时长)
except OSError:
return None
@@ -0,0 +1,735 @@
# 比价卡死定位报警增强 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 让 cancelled 报警用 trace 末段「原地打转」判卡死并定位卡在哪个环节,读不到 trace 回退耗时/帧数保底;failed 类附卡点。
**Architecture:** 判定保持纯函数(`compare_alert.py`),trace 读取单独成层(`trace_stuck.py`,同机直读 pricebot work_logs、只读帧头部字段),worker 编排(cancelled trace 优先 + 保底、failed 附卡点)。卡点拼进 `AlertHit.reason`,复用现有 `format_alert_post` 展示,不依赖卡片 table 固化。
**Tech Stack:** Python 3.11+、FastAPI、SQLAlchemy、pytest。无新依赖(仅标准库 `re`/`json`/`pathlib`/`dataclasses`)。
参考 spec`docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md`
---
## File Structure
- **Create** `app/services/trace_stuck.py` — 卡死判据 + 薄 IO。`StuckPoint`/`StuckResult``read_stuck_points``last_step``dir_name_from_trace_url`、文案表。
- **Create** `tests/test_trace_stuck.py` — trace_stuck 单测。
- **Modify** `app/core/config.py` — 加 4 个配置项。
- **Modify** `app/services/compare_alert.py``_hit` 改公开 `make_hit`;新增纯函数 `classify_cancelled_fallback``classify_record` 的 cancelled 分支改调它(行为不变)。
- **Create** `tests/test_compare_alert_fallback.py``classify_cancelled_fallback`/`make_hit` 单测。
- **Modify** `app/core/compare_alert_worker.py` — 新增 `build_hits`/`_trace_dir` 编排;`_scan_and_alert``build_hits` 替换 `classify_batch`
- **Create** `tests/test_compare_alert_stuck_worker.py``build_hits` 集成测。
---
## Task 1: 配置项
**Files:**
- Modify: `app/core/config.py:147`(在 `COMPARE_ALERT_SEND_EMPTY` 行后追加)
- [ ] **Step 1: 加 4 个配置字段**
`app/core/config.py` 第 147 行 `COMPARE_ALERT_SEND_EMPTY: bool = False ...` 之后,紧接着追加:
```python
# ===== 卡死定位(读 pricebot trace 末段判原地打转)=====
COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR: str = "" # pricebot work_logs 绝对路径(敏感,放 .env);空=跳过 trace、cancelled 全走保底
COMPARE_ALERT_STUCK_FRAME_THRESHOLD: int = 15 # 末段连续同环节达此帧数判卡死
COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES: int = 40 # 每平台最多往前读多少帧
COMPARE_ALERT_TRACE_MAX_RECORDS: int = 30 # 每轮最多对多少条命中记录读 trace(限量)
```
- [ ] **Step 2: 跑现有测试确认不破**
Run: `pytest tests/ -q -k "config or defaults"`
Expected: PASS(新增字段都有默认值,不影响 `test_defaults`
- [ ] **Step 3: Commit**
```bash
git add app/core/config.py
git commit -m "feat(compare-alert): 卡死定位 4 个配置项
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 2: trace_stuck 模块
**Files:**
- Create: `app/services/trace_stuck.py`
- Test: `tests/test_trace_stuck.py`
- [ ] **Step 1: 写失败测试**
创建 `tests/test_trace_stuck.py`
```python
"""trace_stuck 单测:用 tmp 造 step_*.json(只含头部字段)验证卡死判据。"""
import json
from pathlib import Path
from app.services.trace_stuck import (
StuckPoint,
dir_name_from_trace_url,
last_step,
read_stuck_points,
)
def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
"""造一帧 step json:头部放 pipeline_step/detected_page,尾部塞大 windows 模拟真实。"""
pdir.mkdir(parents=True, exist_ok=True)
body = {
"trace_id": "t", "step": idx, "platform": pdir.name,
"pipeline_step": step, "detected_page": page,
"windows": [{"nodes": ["x" * 200]}],
}
(pdir / f"step_{idx:03d}.json").write_text(
json.dumps(body, ensure_ascii=False), encoding="utf-8"
)
def test_stuck_when_tail_repeats_same_step(tmp_path):
pdir = tmp_path / "meituan"
for i in range(20):
_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)]
def test_not_stuck_when_progressing(tmp_path):
pdir = tmp_path / "eleme"
_frame(pdir, 0, "set_address", "home")
for i in range(1, 8):
_frame(pdir, i, "enter_store", "store")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.readable is True
assert res.points == []
def test_adding_many_dishes_not_stuck_when_page_changes(tmp_path):
# add_one_dish 重复但 detected_page 在跳(换菜)=推进,不判卡死
pdir = tmp_path / "meituan"
for i in range(20):
_frame(pdir, i, "add_one_dish", "menu" if i % 2 == 0 else "dish_popup")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.points == []
def test_below_threshold_not_stuck(tmp_path):
pdir = tmp_path / "meituan"
for i in range(10): # < 15
_frame(pdir, i, "add_one_dish", "meal_detail_popup")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.points == []
def test_missing_dir_not_readable(tmp_path):
res = read_stuck_points(tmp_path / "nope", threshold=15, max_tail=40)
assert res.readable is False
assert res.points == []
def test_empty_dir_no_platform_frames_not_readable(tmp_path):
(tmp_path / "emptysub").mkdir()
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.readable is False
def test_per_platform_one_stuck_one_normal(tmp_path):
m = tmp_path / "meituan"
for i in range(18):
_frame(m, i, "add_one_dish", "meal_detail_popup")
e = tmp_path / "eleme"
_frame(e, 0, "set_address", "home")
for i in range(1, 6):
_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)]
def test_last_step_returns_busiest_platform_last_env(tmp_path):
m = tmp_path / "meituan"
for i in range(20):
_frame(m, i, "add_one_dish", "meal_detail_popup")
e = tmp_path / "eleme"
for i in range(3):
_frame(e, i, "enter_store", "store")
sp = last_step(tmp_path)
assert sp == StuckPoint("meituan", "add_one_dish", 20)
def test_dir_name_from_trace_url():
assert dir_name_from_trace_url("https://x/traces/20260804_1_abc/") == "20260804_1_abc"
assert dir_name_from_trace_url("https://x/traces/20260804_1_abc") == "20260804_1_abc"
assert dir_name_from_trace_url("") is None
assert dir_name_from_trace_url(None) is None
```
- [ ] **Step 2: 跑测试确认失败**
Run: `pytest tests/test_trace_stuck.py -q`
Expected: FAIL`ModuleNotFoundError: app.services.trace_stuck`
- [ ] **Step 3: 实现 trace_stuck.py**
创建 `app/services/trace_stuck.py`
```python
"""比价卡死定位:读 pricebot trace 末段,判某平台是否原地打转(卡死)。
同机直读 {WORK_LOG_DIR}/{dir_name}/{platform}/step_*.json,只取头部字段
(pipeline_step/detected_page),不解析后面的无障碍树(windows,占单帧 99% 体积)。
判据与降级见 docs/superpowers/specs/2026-08-05-compare-stuck-detection-design.md。
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
# pipeline_step / 平台名 → 中文(映射不到原样显示英文,不阻断)。按 pricebot 实际枚举补全。
PIPELINE_STEP_LABELS: dict[str, str] = {
"set_address": "定位",
"enter_store": "进店",
"add_one_dish": "加菜",
"match_dish": "找菜",
"checkout": "结算",
}
PLATFORM_LABELS: dict[str, str] = {
"meituan": "美团",
"eleme": "饿了么",
"jd_waimai": "京东外卖",
}
_PIPE_RE = re.compile(r'"pipeline_step":\s*"([^"]*)"')
_PAGE_RE = re.compile(r'"detected_page":\s*"([^"]*)"')
_STEP_NUM_RE = re.compile(r"step_(\d+)")
@dataclass(frozen=True)
class StuckPoint:
platform: str
pipeline_step: str
frames: int # 末段连续困住的帧数(上限 max_tail)
def label(self) -> str:
p = PLATFORM_LABELS.get(self.platform, self.platform)
s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
return f"{p}·{s}"
@dataclass(frozen=True)
class StuckResult:
readable: bool # trace 是否读到(区分「读不到」与「读到但没卡」)
points: list[StuckPoint] # 卡死平台列表;readable=True 且空 = 确认没卡死
def dir_name_from_trace_url(trace_url: str | None) -> str | None:
""".../traces/{dir_name}/ → dir_name;空/异常 → None。"""
if not trace_url:
return None
name = trace_url.rstrip("/").rsplit("/", 1)[-1]
return name or None
def _step_num(path: Path) -> int:
m = _STEP_NUM_RE.search(path.name)
return int(m.group(1)) if m else -1
def _read_head(path: Path, nbytes: int = 4096) -> tuple[str | None, str | None]:
"""只读文件头部,抠 (pipeline_step, detected_page)。它们在 json 最前面。"""
with open(path, "r", encoding="utf-8") as f:
head = f.read(nbytes)
ps = _PIPE_RE.search(head)
pg = _PAGE_RE.search(head)
return (ps.group(1) if ps else None, pg.group(1) if pg else None)
def _platform_stuck(
platform: str, step_files: list[Path], threshold: int, max_tail: int
) -> StuckPoint | None:
"""末帧往前数连续同 (pipeline_step, detected_page) 的帧数 ≥threshold → 卡死。"""
tail = step_files[-max_tail:]
heads = [_read_head(p) for p in tail]
last_ps, last_pg = heads[-1]
if last_ps is None:
return None
count = 0
for ps, pg in reversed(heads):
if ps == last_ps and pg == last_pg:
count += 1
else:
break
if count >= threshold:
return StuckPoint(platform, last_ps, count)
return None
def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
"""逐平台判卡死。目录不存在/异常/无平台帧 → StuckResult(readable=False, points=[])。"""
try:
if not trace_dir.is_dir():
return StuckResult(readable=False, points=[])
points: list[StuckPoint] = []
any_frames = False
for pdir in sorted(trace_dir.iterdir()):
if not pdir.is_dir():
continue
step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
if not step_files:
continue
any_frames = True
sp = _platform_stuck(pdir.name, step_files, threshold, max_tail)
if sp is not None:
points.append(sp)
if not any_frames:
return StuckResult(readable=False, points=[])
return StuckResult(readable=True, points=points)
except OSError:
return StuckResult(readable=False, points=[])
def last_step(trace_dir: Path) -> StuckPoint | None:
"""failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。"""
try:
if not trace_dir.is_dir():
return None
best: tuple[int, str, list[Path]] | None = None
for pdir in sorted(trace_dir.iterdir()):
if not pdir.is_dir():
continue
step_files = sorted(pdir.glob("step_*.json"), key=_step_num)
if step_files and (best is None or len(step_files) > best[0]):
best = (len(step_files), pdir.name, step_files)
if best is None:
return None
_, platform, step_files = best
ps, _pg = _read_head(step_files[-1])
if ps is None:
return None
return StuckPoint(platform, ps, len(step_files))
except OSError:
return None
```
- [ ] **Step 4: 跑测试确认通过**
Run: `pytest tests/test_trace_stuck.py -q`
Expected: PASS9 passed
- [ ] **Step 5: Commit**
```bash
git add app/services/trace_stuck.py tests/test_trace_stuck.py
git commit -m "feat(compare-alert): trace_stuck 卡死判据(末段原地打转)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 3: compare_alert 抽出 fallback + 公开 make_hit
**Files:**
- Modify: `app/services/compare_alert.py``_hit``make_hit`;新增 `classify_cancelled_fallback`cancelled 分支改调它)
- Test: `tests/test_compare_alert_fallback.py`
- [ ] **Step 1: 写失败测试**
创建 `tests/test_compare_alert_fallback.py`
```python
"""classify_cancelled_fallback / make_hit 单测。"""
from app.services.compare_alert import classify_cancelled_fallback, make_hit
class _Rec:
def __init__(self, **kw):
self.trace_id = kw.get("trace_id", "t")
self.status = kw.get("status", "cancelled")
self.total_ms = kw.get("total_ms")
self.step_count = kw.get("step_count")
self.fail_reason = kw.get("fail_reason")
self.information = kw.get("information")
self.app_version = kw.get("app_version")
self.created_at = kw.get("created_at")
self.trace_url = kw.get("trace_url")
self.user_id = kw.get("user_id")
def test_fallback_deep_by_ms():
hit = classify_cancelled_fallback(
_Rec(total_ms=95000, step_count=5),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is not None and hit.alert_type == "T5" and "深度放弃" in hit.reason
def test_fallback_deep_by_step():
hit = classify_cancelled_fallback(
_Rec(total_ms=1000, step_count=35),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is not None and hit.alert_type == "T5"
def test_fallback_shallow_none():
hit = classify_cancelled_fallback(
_Rec(total_ms=5000, step_count=3),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is None
def test_make_hit_carries_fields():
hit = make_hit(_Rec(trace_id="tx", app_version="0.6.0"), "T5", "卡在 美团·加菜")
assert hit.trace_id == "tx"
assert hit.reason == "卡在 美团·加菜"
assert hit.app_version == "0.6.0"
```
- [ ] **Step 2: 跑测试确认失败**
Run: `pytest tests/test_compare_alert_fallback.py -q`
Expected: FAIL`ImportError: cannot import name 'classify_cancelled_fallback'`
- [ ] **Step 3: 改 compare_alert.py**
`app/services/compare_alert.py`
(a) 把 `def _hit(` 改名为 `def make_hit(`(第 33 行),并把 `classify_record` 内 4 处 `_hit(` 调用改成 `make_hit(`(原 T1/T6/T2/T5 分支)。
(b) 在 `make_hit` 之后、`classify_record` 之前,新增:
```python
def classify_cancelled_fallback(
rec: Any,
*,
cancelled_ms_threshold: int,
cancelled_step_threshold: int,
) -> AlertHit | None:
"""cancelled 保底判定(读不到 trace 时用):超耗时或步数阈值 → T5 深度放弃,否则 None。纯函数。"""
ms = rec.total_ms
step = rec.step_count
deep = (ms is not None and ms > cancelled_ms_threshold) or (
step is not None and step > cancelled_step_threshold
)
if deep:
return make_hit(
rec, "T5",
f"深度放弃·等待 {round((ms or 0) / 1000)}s / {step or 0} 步后退出",
)
return None
```
(c) 把 `classify_record` 里的 cancelled 分支(原 `if status == "cancelled":` 那整段)替换为:
```python
if status == "cancelled":
return classify_cancelled_fallback(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
)
return None
```
`classify_record` 行为不变,只是把 cancelled 逻辑抽到 `classify_cancelled_fallback`。)
- [ ] **Step 4: 跑测试确认通过(含现有 rules 测试不回归)**
Run: `pytest tests/test_compare_alert_fallback.py tests/test_compare_alert_rules.py -q`
Expected: PASS(新测试 4 passed,现有 rules 测试仍全 PASS
- [ ] **Step 5: Commit**
```bash
git add app/services/compare_alert.py tests/test_compare_alert_fallback.py
git commit -m "feat(compare-alert): 抽出 classify_cancelled_fallback + 公开 make_hit
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 4: worker 编排 build_hits
**Files:**
- Modify: `app/core/compare_alert_worker.py`(加 imports、`_trace_dir``build_hits``_scan_and_alert` 改用 `build_hits`
- Test: `tests/test_compare_alert_stuck_worker.py`
- [ ] **Step 1: 写失败测试**
创建 `tests/test_compare_alert_stuck_worker.py`
```python
"""build_hits 集成测:cancelled trace 优先/保底切换、failed 附卡点、限量。"""
import json
from pathlib import Path
from app.core.compare_alert_worker import build_hits
class _Rec:
def __init__(self, **kw):
self.trace_id = kw.get("trace_id", "t")
self.status = kw.get("status", "cancelled")
self.total_ms = kw.get("total_ms")
self.step_count = kw.get("step_count")
self.fail_reason = kw.get("fail_reason")
self.information = kw.get("information")
self.app_version = kw.get("app_version")
self.created_at = kw.get("created_at")
self.trace_url = kw.get("trace_url")
self.user_id = kw.get("user_id")
def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
pdir.mkdir(parents=True, exist_ok=True)
body = {"pipeline_step": step, "detected_page": page, "windows": [{"n": ["x" * 200]}]}
(pdir / f"step_{idx:03d}.json").write_text(
json.dumps(body, ensure_ascii=False), encoding="utf-8"
)
_KW = dict(
stuck_threshold=15, max_tail=40, max_trace_reads=30,
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
timeout_keywords=("超时",), unrecognized_keywords=("未识别",), biz_exclude_keywords=(),
)
def test_cancelled_stuck_reports_via_trace(tmp_path):
for i in range(18):
_frame(tmp_path / "20260804_x" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_x/",
total_ms=5000, step_count=3) # 保底不会中,靠 trace 判卡死
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T5"
assert "卡在" in hits[0].reason and "美团·加菜" in hits[0].reason
def test_cancelled_readable_not_stuck_no_report(tmp_path):
# trace 确认没卡(在推进);即便 total_ms/step 超阈值也不报(信 trace,不回退保底)
p = tmp_path / "20260804_y" / "eleme"
_frame(p, 0, "set_address", "home")
for i in range(1, 6):
_frame(p, i, "enter_store", "store")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_y/",
total_ms=95000, step_count=40)
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert hits == []
def test_cancelled_unreadable_falls_back(tmp_path):
rec = _Rec(status="cancelled", trace_url="https://x/traces/nope/",
total_ms=95000, step_count=3)
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T5" and "深度放弃" in hits[0].reason
def test_no_work_log_dir_uses_fallback(tmp_path):
rec = _Rec(status="cancelled", trace_url="https://x/traces/y/",
total_ms=95000, step_count=3)
hits = build_hits([rec], work_log_dir="", **_KW)
assert len(hits) == 1 and "深度放弃" in hits[0].reason
def test_failed_gets_stuck_point_appended(tmp_path):
for i in range(20):
_frame(tmp_path / "20260804_f" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="failed", fail_reason="启动超时",
trace_url="https://x/traces/20260804_f/")
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T2"
assert "卡在 美团·加菜" in hits[0].reason
def test_max_trace_reads_zero_skips_trace(tmp_path):
for i in range(18):
_frame(tmp_path / "20260804_z" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_z/",
total_ms=95000, step_count=3)
kw = {**_KW, "max_trace_reads": 0}
hits = build_hits([rec], work_log_dir=str(tmp_path), **kw)
# 没读 trace → 回退保底 → deep(95s) → 深度放弃
assert len(hits) == 1 and "深度放弃" in hits[0].reason
```
- [ ] **Step 2: 跑测试确认失败**
Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
Expected: FAIL`ImportError: cannot import name 'build_hits'`
- [ ] **Step 3: 改 compare_alert_worker.py**
(a) 顶部 imports 段,把
```python
from app.services.compare_alert import classify_batch
from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_post
```
改为
```python
from dataclasses import replace as _dc_replace
from app.services import trace_stuck
from app.services.compare_alert import (
classify_cancelled_fallback,
classify_record,
make_hit,
)
from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_post
```
(b) 在 `_scan_and_alert` 之前新增两个函数:
```python
def _trace_dir(base: Path, trace_url: str | None) -> Path | None:
name = trace_stuck.dir_name_from_trace_url(trace_url)
if not name:
return None
return base / name
def build_hits(
records: list,
*,
work_log_dir: str,
stuck_threshold: int,
max_tail: int,
max_trace_reads: int,
cancelled_ms_threshold: int,
cancelled_step_threshold: int,
timeout_keywords: tuple[str, ...],
unrecognized_keywords: tuple[str, ...],
biz_exclude_keywords: tuple[str, ...],
) -> list:
"""编排:cancelled 走 trace 优先(读到确认没卡则不报,读不到回退保底);failed 附卡点。
trace 读取限量 max_trace_reads 次/轮;任何 trace 异常都在 trace_stuck 内部降级为
「读不到」,cancelled 因而回退保底、failed 不附卡点,绝不影响报警发送。
"""
base = Path(work_log_dir) if work_log_dir else None
reads = 0
hits: list = []
for rec in records:
if rec.status == "cancelled":
res = None
if base is not None and reads < max_trace_reads:
td = _trace_dir(base, rec.trace_url)
if td is not None:
res = trace_stuck.read_stuck_points(
td, threshold=stuck_threshold, max_tail=max_tail
)
reads += 1
if res is not None and res.readable:
if res.points:
reason = "卡在 " + "、".join(sp.label() for sp in res.points)
hit = make_hit(rec, "T5", reason)
else:
hit = None # 读到且确认没卡 → 不报
else:
hit = classify_cancelled_fallback(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
)
if hit is not None:
hits.append(hit)
else:
hit = classify_record(
rec,
cancelled_ms_threshold=cancelled_ms_threshold,
cancelled_step_threshold=cancelled_step_threshold,
timeout_keywords=timeout_keywords,
unrecognized_keywords=unrecognized_keywords,
biz_exclude_keywords=biz_exclude_keywords,
)
if (
hit is not None
and hit.alert_type in ("T1", "T2", "T6")
and base is not None
and reads < max_trace_reads
):
td = _trace_dir(base, rec.trace_url)
if td is not None:
sp = trace_stuck.last_step(td)
reads += 1
if sp is not None:
hit = _dc_replace(hit, reason=f"{hit.reason}|卡在 {sp.label()}")
if hit is not None:
hits.append(hit)
return hits
```
(c) 在 `_scan_and_alert` 里,把
```python
hits = classify_batch(
records,
cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
timeout_keywords=settings.compare_alert_timeout_keywords,
unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
)
```
替换为
```python
hits = build_hits(
records,
work_log_dir=settings.COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR,
stuck_threshold=settings.COMPARE_ALERT_STUCK_FRAME_THRESHOLD,
max_tail=settings.COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES,
max_trace_reads=settings.COMPARE_ALERT_TRACE_MAX_RECORDS,
cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
timeout_keywords=settings.compare_alert_timeout_keywords,
unrecognized_keywords=settings.compare_alert_unrecognized_keywords,
biz_exclude_keywords=settings.compare_alert_biz_exclude_keywords,
)
```
- [ ] **Step 4: 跑测试确认通过**
Run: `pytest tests/test_compare_alert_stuck_worker.py -q`
Expected: PASS6 passed
- [ ] **Step 5: 跑报警相关全量测试确认不回归**
Run: `pytest tests/ -q -k "compare_alert or trace_stuck"`
Expected: PASS(全绿)
- [ ] **Step 6: Commit**
```bash
git add app/core/compare_alert_worker.py tests/test_compare_alert_stuck_worker.py
git commit -m "feat(compare-alert): worker 编排 build_hits(cancelled trace 优先+保底、failed 附卡点)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## 收尾
- [ ] **全量测试**`pytest -q`(对齐 preexisting 失败基线,不新增失败)
- [ ] **lint**`ruff check app/services/trace_stuck.py app/core/compare_alert_worker.py app/services/compare_alert.py`
- [ ] **本地联调(可选)**:把 `.env``COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR` 指向本地 `E:\project\pricebot-backend\data\work_logs`,用真实 cancelled trace 目录验证卡点文案。
- [ ] **清理临时脚本**`git rm --cached` 无关,直接删 `scripts/_probe_trace_timing.py`(若确认不再用,另行确认 `scripts/_test_alert_card.py`)。
## 不在本 plan(后续单独排)
- 卡片 schema 2.0 table 组件固化(`format_alert_card` + `send_feishu_card`,当前仍在 `scripts/_test_alert_card.py`)——卡点已随 `reason` 在现有 `format_alert_post` 展示,不阻塞本功能。
- pricebot 侧改动(本方案零改 pricebot)。
- `PIPELINE_STEP_LABELS` 全枚举补全(映射不到原样英文,可随线上观察增量补)。
@@ -0,0 +1,197 @@
# 比价失败报警机制 · 设计文档
- **日期**2026-08-04
- **状态**:设计待评审(v2,含数据复审修订)
- **范围**app-server`shaguabijia-app-server`
- **数据源**`comparison_record` 单表(线上快照已导入本地 `cr_analysis` 分析库,3867 行,覆盖 2026-06-09 ~ 08-04
## 复审修订记录(2026-08-04, v2
结合 `cr_analysis` 实测数据复审后的改动:
1. **[必修] 水位改用 `updated_at`**:原 `created_at` 水位会系统性漏报慢失败(落定延迟 p99 达 7.5–10min)。改为给 `comparison_record``updated_at` 列、水位按 `updated_at` 单调推进(§5/§6)。
2. **新增规则 T6「商品识别失败」**:「未识别到商品」96 条纳入报警,作为识别能力信号(§3)。
3. **T1 加业务词排除**:清掉 `fail_reason IS NULL` 里 7 条 `information` 实为业务的误报(§3)。
4. 补充:NULL 语义、时区口径、`business_type` 复核、单窗口截断阈值(§7/§9)。
5. 数据证伪、未采纳的改动:T2 关键词已完备(35 条技术词全被「超时/启动/加载」覆盖),不扩。
---
## 1. 背景与目标
比价(外卖 `business_type=food`)由客户端无障碍自动化 + pricebot 多平台 LLM 驱动,链路长、失败形态多。目前**没有任何主动发现失败的手段**——只能人工翻库或等用户反馈。
**目标**:新增一个**近实时、记录级**的比价失败报警机制。每隔 15–30 分钟扫描新落定的比价记录,逐条按预定规则判定「是否属于需要关注的失败」,把命中的记录**汇总成一条飞书消息**发到告警群,并**报出每条触发的原因**。
**关键设计取向**(均由数据分析与评审确认):
- **粒度是「记录」不是「失败率」**:逐条判定,不算比率、不设样本量门槛、不做基线对比。日均比价量小(完成约 45 条/天),比率方案在小窗口会剧烈抖动;记录级方案规避了这个问题。
- **只报「技术性失败」「识别失败」与「深度放弃」**,不报正常业务结局。
- **有触发才发,无触发静默**:不刷屏。
## 2. 数据分析依据(基线)
全量 3867 条记录级 `status` 分布(详见附录 A):
| status | 数量 | 占比 | 说明 |
|---|---|---|---|
| success | 1501 | 38.8% | 成功(含 `below_minimum` 未满起送,被归一为 success) |
| cancelled | 1394 | 36.0% | 用户中途退出 |
| failed | 957 | 24.7% | 失败(T1 技术 386 + T2 超时 35 + T6 识别 96 + 业务 440 |
| running | 15 | 0.4% | 悬挂未收尾 |
支撑规则设计的关键事实:
- **`failed` 是混合桶**`fail_reason IS NULL` 的 393 条是纯系统技术失败(其中「比价过程出错,请稍后重试」占 294,是 `_GENERIC_INFO` 兜底黑话);`fail_reason` 非空的多为业务结局,但夹杂「启动淘宝超时」等技术问题 35 条、「未识别到商品」96 条。
- **业务失败不该报**:打烊、无此店、无此菜、未起送、单点不配送是正常结局。
- **落定延迟很长**`total_ms`(≈ 记录从建行到落定的时长)p99 = failed 449s、cancelled 602smax 16min。**这是水位必须用 `updated_at` 而非 `created_at` 的直接依据**。
- **cancelled 缺退出上下文**:99.3% 终止原因就一句「用户终止比价」,且 100% 没有 `platforms`/结果数据(都在 `running` 阶段被中止)。唯一可用信号是退出时机(`total_ms`/`step_count`)。cancelled 的 `step_count` 中位 5、p90 31`total_ms` 中位 24s、p90 124s。参照系:一次成功比价中位 113s / 38 步。
## 3. 报警规则(v1
worker 每轮查询「上次水位之后有更新」的记录,对每条按下表判定;命中任一即计入本期汇总。四个规则互斥(一条记录最多归一类)。
| 类型 | 判定条件(SQL 语义) | 触发原因文案 | 历史量(2月) |
|---|---|---|---|
| **T1 系统技术失败** | `status='failed' AND fail_reason IS NULL AND (information IS NULL OR information !~ 业务词)` | `技术失败·{information 去空白截断; 空则"比价过程出错"}` | ≈386 |
| **T6 商品识别失败** | `status='failed' AND fail_reason ~ '未识别'` | `识别失败·未识别到商品` | ≈96 |
| **T2 超时/启动失败** | `status='failed' AND fail_reason IS NOT NULL AND fail_reason ~ 超时关键词` | `{fail_reason}`(如「启动淘宝超时」) | ≈35 |
| **T5 cancelled 深度放弃** | `status='cancelled' AND (total_ms > 90000 OR step_count > 30)` | `深度放弃·等待 {total_ms/1000 取整}s / {step_count} 步后退出` | ≈250 |
**判定顺序(保证互斥)**
1. `status='failed'`
- `fail_reason IS NULL` → 若 `information` 命中**业务词**`未找到|打烊|起送|门店|店内|不配送|这些菜|未入驻|休息`)则**不报**(业务失败漏派生 fail_reason,约 7 条);否则 **T1**
- `fail_reason` 含「未识别」→ **T6**
- `fail_reason` 含超时关键词(`超时|启动|加载`)→ **T2**
- 其余(干净业务原因)→ **不报**
2. `status='cancelled'` 且(`total_ms>90000``step_count>30`)→ **T5**;否则不报。
3. `status IN ('success','running')` → 不报。
**阈值/关键词(可配初值)**:T5 的 `90000ms`/`30步` 取自 cancelled 分布约 p90(评审选定「B 中档」)。超时关键词 `超时,启动,加载`、识别关键词 `未识别`、业务排除词均可配。
每条命中记录在汇总里附带:`trace_id``app_version``business_type`、触发原因文案、`created_at`
## 4. 非目标与暂缓项
| 项 | 处理 | 原因 |
|---|---|---|
| 业务失败(打烊/无店/无菜/未起送/单点不配送) | **不报** | 正常业务结局 |
| success、早退 cancelled(≤90s 且 ≤30 步) | **不报** | 无报警价值 |
| **T3 running 悬挂** | **本期暂缓** | 评审决定先不报;但见下方 🔴 |
| **T4 单平台适配失效**`platforms[].status='failed'`) | 暂不纳入(未来增强) | 量大、与整体失败重叠、噪音高 |
| 失败率 / cancelled 率等**比率型**指标 | 不做 | 本设计是记录级 |
> 🔴 **待独立排查的回归线索**(非本报警范围,留档):`running` 悬挂 15 条**全部集中在 2026-07-28 之后**,此前两个月几乎为 0。强烈提示某次发版后 `harvest_done`/`harvest_abort` 收尾链路(`app/repositories/comparison.py`)回归,建议单独开 issue。排查确认后可在 v2 把 T3 作为独立高频告警加回。
## 5. 架构与组件
沿用项目现有**常驻 asyncio worker** 范式(与 `heartbeat_monitor_worker` 等一致)。各组件单一职责、可独立测试:
| 组件 | 路径 | 职责 |
|---|---|---|
| **数据模型改动** | `app/models/comparison.py` + alembic 迁移 | `comparison_record` 新增 `updated_at``server_default=func.now()`, `onupdate=func.now()`+ 索引 `ix_comparison_updated`。为水位提供单调递增的落定时间。 |
| **规则模块** | `app/services/compare_alert.py` | 纯函数:输入一批 ORM 记录 → 输出 `[(记录, 触发类型, 原因文案)]`。判定逻辑与阈值全在此,无 I/O,易测易调。 |
| **扫描 worker** | `app/core/compare_alert_worker.py` | 仿 `heartbeat_monitor_worker`:单实例文件锁 + `asyncio` 轮询 + 优雅退出。每轮:读水位 → 查有更新记录 → 调规则 → 有命中则格式化并发飞书 → 推进水位。 |
| **飞书通知器** | `app/integrations/feishu_notifier.py` | 实现群机器人 webhook 发送。发送失败抛异常由 worker 处理。 |
| **水位存储** | 复用 `app_config` 表 | key=`compare_alert.last_watermark`value=上次处理的最大 `updated_at`。 |
| **启停挂载** | `app/main.py` lifespan | `start_compare_alert_worker()` / `stop_compare_alert_worker()`,与现有 worker 同处注册。 |
> **`onupdate` 生效前提**:现有 `harvest_done`/`harvest_abort`/`upsert_record` 均走 ORM `setattr`+`commit` 更新,`onupdate=func.now()` 会自动刷新 `updated_at`,无需改写路径。
## 6. 数据流与水位管理(updated_at 方案)
```
每 interval 秒:
读 app_config['compare_alert.last_watermark'] → watermark
(空 → 冷启动:watermark = 当前 max(updated_at),只报之后新落定的,不回溯历史)
查 comparison_record
WHERE updated_at > watermark
ORDER BY updated_at ASC
逐条套 T1/T6/T2/T5 规则 → 命中集合(按类型分组)
若命中集合非空:
格式化飞书消息 → feishu_notifier.send()
成功 → 水位 = 本批 max(updated_at)
失败 → 不更新水位(log),下一轮重扫补发
若命中集合为空:
水位 = 本批 max(updated_at)(无记录则不动;可选 SEND_EMPTY 发简讯)
```
- **零漏报**:任何记录落定/更新时 `updated_at` 刷新为当前 DB 时钟 > 水位,必被下一轮扫到——无论 `created_at` 多早、落定多慢(根治了 `created_at` 水位漏掉慢失败的问题)。
- **规避时区**:水位存的是 DB 产出的 `updated_at` 值,查询用 `updated_at` 自身比较,**不依赖 worker 本地时钟与 DB 时钟对齐**(`created_at` 存 naive 北京、`func.now()` 为 DB 时钟,二者口径不同,但本方案只用 `updated_at` 自比较,不受影响)。
- **发送失败不推进水位**:保证不漏;恢复后一次补发。
- **一条记录可能被扫多次**running 更新→落定更新,`updated_at` 变两次):但只有落定后 `status` 才命中规则,running 阶段扫到不命中,无副作用;不会重复报。
## 7. 飞书消息格式
群机器人消息(文本或富文本 `post`),按类型分组:
```
🚨 比价失败报警 · 2026-08-04 08:0008:30 · 本期触发 7 条
• 系统技术失败 3 条
- trace abc123 | v0.3.4 | 比价过程出错
• 商品识别失败 2 条
- trace abc200 | v0.5.1 | 未识别到商品
• 启动/超时失败 1 条
- trace def456 | v0.3.4 | 启动淘宝超时
• 深度放弃(cancelled) 1 条
- trace ghi789 | v0.6.3 | 等待 98s / 26 步后退出
```
- **截断阈值**:单类型明细超 `MAX_DETAIL_PER_TYPE`(默认 20)条时,只列前 20 条 + 「另有 N 条」;本期总命中超 `MAX_TOTAL`(默认 50)条时降级为只给各类型计数,提示去分析库查(防报警风暴,如 07-14 那种高失败日)。
## 8. 配置项
`app/core/config.py``pydantic-settings`):
| 配置 | 默认 | 说明 |
|---|---|---|
| `COMPARE_ALERT_ENABLED` | `False` | 总开关;关时 worker 不启动 |
| `COMPARE_ALERT_SCAN_INTERVAL_SEC` | `1800` | 扫描间隔,可配 90015min |
| `COMPARE_ALERT_FEISHU_WEBHOOK` | `""` | 群机器人 webhook;空则 worker 仅打日志不外发 |
| `COMPARE_ALERT_CANCELLED_MS_THRESHOLD` | `90000` | T5 耗时阈值(ms |
| `COMPARE_ALERT_CANCELLED_STEP_THRESHOLD` | `30` | T5 步数阈值 |
| `COMPARE_ALERT_TIMEOUT_KEYWORDS` | `"超时,启动,加载"` | T2 关键词 |
| `COMPARE_ALERT_UNRECOGNIZED_KEYWORDS` | `"未识别"` | T6 关键词 |
| `COMPARE_ALERT_BIZ_EXCLUDE_KEYWORDS` | `"未找到,打烊,起送,门店,店内,不配送,这些菜,未入驻,休息"` | T1 的 information 业务词排除 |
| `COMPARE_ALERT_MAX_DETAIL_PER_TYPE` | `20` | 单类型明细截断 |
| `COMPARE_ALERT_MAX_TOTAL` | `50` | 本期总命中截断(超则只给计数) |
| `COMPARE_ALERT_SEND_EMPTY` | `False` | 无命中是否发「本期无异常」简讯 |
> **`business_type` 复核**:当前数据全为 `food`,规则未按 `business_type` 限定。接入 `ecom`/`coupon` 时需复核各规则(尤其 T2/T6 关键词与 T5 阈值是否仍适用)。
## 9. 错误处理与边界
- **worker 单轮异常吞掉不退出**`except Exception: logger.exception`),仿 heartbeat。
- **DB / 飞书发送异常**:log,本轮不推进水位,下轮重试补发。
- **单实例锁**:文件锁 `data/compare_alert.lock`O_CREAT|O_EXCL + stale 检测)。
- **NULL 语义**T5 中 `total_ms`/`step_count` 为 NULL 的 cancelledSQL 比较 `NULL>90000` 为 false → 不命中(无数据不报,符合预期)。T1 中 `information IS NULL` 时业务词排除不触发(视为非业务)→ 仍属 T1,原因文案兜底「比价过程出错」。
- **冷启动不回溯历史**:首次启动水位=当前 `max(updated_at)`,避免把历史失败一次性全报。
## 10. 测试策略
仿现有 `tests/` 风格(`TestClient` + monkeypatch 外部依赖,SQLite 临时库):
- `test_compare_alert_rules.py`:喂各类记录(T1/T6/T2/T5 命中样本 + 业务失败/success/早退 cancelled/running 反例 + T1 业务词误入反例),断言分类与原因文案;覆盖阈值边界(`total_ms=90000` 不命中、`90001` 命中)与 NULL 语义。
- `test_compare_alert_worker.py`monkeypatch notifier 与 `SessionLocal`,验证 `updated_at` 水位推进、发送失败不推进、冷启动=max、命中汇总、截断逻辑。
- `test_feishu_notifier.py`monkeypatch HTTP,验证消息体格式与发送失败抛异常。
- 迁移测试:`updated_at` 列 + 索引存在,`onupdate` 在 ORM 更新时刷新。
## 11. 未来增强
1. **T3 running 悬挂告警**:待第 4 节 🔴 回归排查后,作为独立高频告警加回。
2. **T4 单平台适配失效**`platforms[].status='failed'` 逐平台维度。
3. **cancelled 退出上下文埋点**:客户端终止时上报退出阶段、已比出平台数、是否已看到中间结果——让 cancelled 从「只有时机」升级为「可归因」。
4. **分维度统计**:汇总附带按 `app_version`/`source_platform` 的命中分布,辅助定位回归版本/平台。
5. **趋势型报警**:记录级之上叠加比率/环比(需另设样本量保护)。
## 附录 A:分析数据来源与复现
- **来源**:线上 PostgreSQL 16 `pg_dump` 单表 `comparison_record`plain SQL187MB)。
- **本地环境**Docker 容器 `shaguabijia-pg`postgres:16-alpine),独立分析库 `cr_analysis`(用户 `shaguabijia_app`)。导入:`docker cp` dump 进容器后 `psql -f`(末尾外键引用 `public.user` 报错属预期,单表 dump 无 user 表,不影响数据与索引)。
- **样本**3867 行,2026-06-09 ~ 08-04。
- **关键分布**(供实现期回归对照):
| 指标 | success | failed | cancelled |
|---|---|---|---|
| 数量 | 1501 | 957 | 1394 |
| step_count 中位 / p90 | 38 / 61 | 22 / 47 | 5 / 31 |
| total_ms 中位 / p90 / p99 | 113s / 200s / 391s | 77s / 168s / 449s | 24s / 124s / 602s |
- **failed 细分**(合计 957):T1 系统技术 386(`fail_reason IS NULL` 393 − 业务误入 7)、T2 超时/启动 35、T6 识别失败 96、业务失败 440(含误入的 7 条)。
@@ -0,0 +1,163 @@
# 比价「卡死定位」报警增强设计
- 日期:2026-08-05
- 分支:feat-compare-fail-alert(延续一期)
- 关联:`docs/superpowers/specs/2026-08-04-compare-fail-alert-design.md`(一期报警)
## 1. 背景与问题
一期报警对 **cancelled(中途退出)** 的判据(`app/services/compare_alert.py``classify_record`):
```
cancelled 且 (total_ms > 90s 或 step_count > 30步) → T5 深度放弃
```
这个判据量的是「投入多少」,不是「卡没卡」,两头都错:
- **漏报**:一进平台就卡在登录墙 / 加载失败,5 秒 2 步就退 → 判「不深度」→ 不报。但这是真卡死。
- **误报**:用户正常挑了 100 秒、点了 40 步,比完价不满意退了 → `>90s` → 报「深度放弃」。但根本没卡。
根因:`total_ms`/`step_count` 是**整场**的量,把「卡在一步反复失败」和「正常深度使用」混为一谈。
### 1.1 数据佐证(真实 trace
- **卡死例**`20260804_114703` meituan):`pipeline_step``set_address`(step 0-1) → `enter_store`(3-8) → **`add_one_dish`(9 一路到 120+,110+ 帧全困在这一个环节)**。且 `timing.json` 里根本没有 meituan——`step_profiler` 只在平台 `is_done` 时落 timing,卡死平台永不 done。
- **正常例**`20260803_165239` eleme):`set_address`(0) → `enter_store`(1-7) → done,每个环节 ≤7 帧就推进走了。
**卡死的结构特征**:某 `pipeline_step` 连续几十上百帧不变(原地打转);正常则是逐环节推进、单环节 ≤7 帧。两者空档极大(7 vs 110+),可用一个帧数阈值干净区分,且**不需要大量数据归纳环节语义**。
## 2. 目标
- cancelled 判据:从「整场耗时/帧数阈值」→「trace 末段原地打转」,抓真卡死(含短时卡死)、不误报正常深度使用。
- **判定与展示一体**:直接报「卡在 平台·环节」。
- failed 类(T1/T2/T6):判定不变,best-effort 补卡点定位。
- 稳:读不到 trace 回退原耗时/帧数保底,**绝不阻断报警发送**。
## 3. 取数:同机直读(不改 pricebot)
app-server 与 pricebot **同机**。trace 落盘在 `{WORK_LOG_DIR}/{dir_name}/`
- **dir_name 从 `comparison_record.trace_url` 尾段抠**`trace_url = {base}/traces/{dir_name}/`,尾段就是磁盘目录名,新老格式都对得上,规避从 `trace_id` 反推老格式「首帧时刻」的难题。
- 只读末段帧的**头部字段**`pipeline_step` / `detected_page`),不解析后面的无障碍树(`windows`,占单帧 99% 体积)。
- 不改 pricebot、不需要 `INTERNAL_API_SECRET`、不走网络。
> 备选途径(已否决):pricebot 加内部接口(要改两仓 + secret)、公网 `trace_url` GET timing.json(本地 SSL 大面积超时 + timing.json 缺卡死平台)。同机直读最优。
## 4. 架构分层
保持判定纯函数、IO 单独成层:
| 模块 | 职责 | 性质 |
|---|---|---|
| `services/compare_alert.py`(微调) | failed 判定不变;cancelled 只保留**回退保底**判定(`>90s`/`>30步` | 纯函数 |
| `services/trace_stuck.py`(新) | 给定 trace 目录 → 逐平台读末段 → 判「原地打转」→ 返回卡点列表 | 薄 IO + 纯逻辑 |
| `core/compare_alert_worker.py`(编排) | 先跑纯 `classify_batch` 出候选,再对候选调 `trace_stuck` 增强 | 编排 |
## 5. trace_stuck 模块
### 5.1 卡死判据(逐平台)
对某平台的 `step_*.json` 序列,从**末帧往前**数,连续 `(pipeline_step, detected_page)` 都相同的帧数 ≥ N → 判该平台卡死,卡点 = 该 `pipeline_step`
- `N = COMPARE_ALERT_STUCK_FRAME_THRESHOLD`(默认 **15**;正常环节 ≤7 帧、卡死 110+ 帧,空档极大)。
- **「无推进」= `(pipeline_step, detected_page)` 双不变**(页面没跳转、环节没变)。这样能区分:
- 「加多菜」:`pipeline_step` 相同但 `detected_page` 在跳(换菜/回菜单)= 推进 → 不判卡死;
- 「卡在一步」:两者都不变 = 原地打转 → 卡死。
- 从末帧往前最多读 `COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES`(默认 **40**)帧,够判 ≥N 即停,防超长 trace 全读。
### 5.2 逐平台聚合(B 方案:不漏)
一条 trace **逐平台**判,所有卡死平台都收集——不只「帧数最多」的那个。因为「帧数最多」会在**卡死平台帧数不是最多**时漏报(如另一平台正常加了 8 道菜跑了 30 帧、卡死平台一进就卡登录 5 帧退),而那恰是短时卡死。多个卡死平台都列进 reason。
### 5.3 接口
```python
@dataclass(frozen=True)
class StuckPoint:
platform: str
pipeline_step: str
frames: int # 末段连续困住的帧数
@dataclass(frozen=True)
class StuckResult:
readable: bool # trace 是否读到(区分「读不到」与「读到但没卡」)
points: list[StuckPoint] # 卡死平台列表;readable=True 且空 = 确认没卡死
def read_stuck_points(trace_dir: Path, *, threshold: int, max_tail: int) -> StuckResult:
"""逐平台判卡死。目录不存在/异常/无平台帧 → StuckResult(readable=False, points=[])。"""
def last_step(trace_dir: Path) -> StuckPoint | None:
"""failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。"""
```
## 6. 判定流
### 6.1 cancelledtrace 优先 → 保底)
```
worker 对 cancelled 候选:
res = read_stuck_points(dir)
if not res.readable: # 读不到 trace(目录被清/生产一时读不到)→ 回退保底
>90s或>30步 → T5「深度放弃·等待Xs/Y步」; 否则不报
elif res.points: # 读到且有卡死平台 → 报卡死
报 T5, reason = "卡在 " + "、".join(f"{平台}·{环节}" for res.points)
else: # 读到且没卡死(末段在推进 = 正常深度使用后退出)→ 不报
不报
```
### 6.2 failedT1/T2/T6,判定不变 + 附卡点)
```
worker 对 failed 命中:
sp = last_step(dir) # 读不到 → None
if sp: reason += f"|卡在 {平台}·{环节}"
```
`failed` 只取「末帧停在哪」,不要求原地打转(它已失败、末帧即失败点)。intent 阶段就失败(无平台目录,典型 T6)→ 不附,reason 原样。
## 7. 卡点文案映射
`PIPELINE_STEP_LABELS`(小映射表,映射不到原样显示英文、不阻断):
| pipeline_step | 中文 |
|---|---|
| `set_address` | 定位 |
| `enter_store` | 进店 |
| `add_one_dish` | 加菜 |
| …(实现时按 pricebot 实际枚举补全) | |
平台名同样映射(`meituan`→美团、`eleme`→饿了么、`jd_waimai`→京东外卖)。
## 8. 配置(`app/core/config.py`;路径敏感项放 `.env`
| 配置 | 默认 | 说明 |
|---|---|---|
| `COMPARE_ALERT_PRICEBOT_WORK_LOG_DIR` | `""` | pricebot work_logs 绝对路径;**空 = 跳过 trace、全走保底**(行为等同一期) |
| `COMPARE_ALERT_STUCK_FRAME_THRESHOLD` | `15` | N:末段连续同环节达此帧数判卡死 |
| `COMPARE_ALERT_TRACE_MAX_TAIL_FRAMES` | `40` | 每平台最多往前读多少帧 |
| `COMPARE_ALERT_TRACE_MAX_RECORDS` | `30` | 每轮最多对多少条命中记录读 trace(限量) |
## 9. 展示
卡片「失败原因」列下附一行卡点小字,**不新增列**。cancelled 卡死时卡点即 reason 本身;failed 的卡点附在原因后。
> 依赖:本期展示复用一期卡片的「失败原因」列。若一期卡片(schema 2.0 table 组件,当前仍在临时脚本 `scripts/_test_alert_card.py`)尚未固化为正式 `format_alert_card` + `send_feishu_card`,本期实现时一并固化。
## 10. 降级与成本
- 只对命中记录读、限量 `MAX_RECORDS`、每平台只读末段头部字段、单文件读加超时。
- **任何异常降级**cancelled 回退保底、failed 不附卡点,绝不阻断报警。
- `work_log_dir` 未配 → 整个 trace 增强跳过,行为等同一期(纯保底)。
## 11. 测试
- **trace_stuck 单测**:卡死正例(meituan 目录 → 判出 `add_one_dish`)、正常负例(eleme → 不判卡死)、加多菜不误判(`detected_page` 在变)、末段不足 N 帧、读不到目录降级。
- **worker 集成**:trace 优先命中 vs 读不到回退保底切换;failed 附卡点;限量 `MAX_RECORDS` 生效。
- fixture 用 tmp 造 `step_*.json`**只含头部字段**trace_id/step/platform/pipeline_step/detected_page)即可,不需无障碍树。
## 12. 不做(YAGNI
- 不改 pricebot(不加内部接口)。
- 不落库(不加 `comparison_record` 列、不做迁移)。
- 不做每帧耗时(`timing.json` 缺卡死平台,且报警用不上逐帧耗时)。
- 不做环节黑白名单 / 语义分类(结构判据已够,且需大数据)。
+1
View File
@@ -31,6 +31,7 @@ os.environ.setdefault("WXPAY_MCH_ID", "test-mch")
os.environ.setdefault("WXPAY_MCH_SERIAL_NO", "test-serial")
os.environ.setdefault("WXPAY_PUBLIC_KEY_ID", "test-pubkey-id")
os.environ.setdefault("RATE_LIMIT_ENABLED", "false") # 限流内存计数会跨用例累加,测试关掉
os.environ.setdefault("COMPARE_ALERT_ENABLED", "false") # 报警 worker 测试不启动(避免 .env 的 true 干扰 test_defaults)
# 穿山甲发奖回调:测试里开启 + 给个 mock 验签密钥,test 内自签自验闭环
os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")
+17
View File
@@ -0,0 +1,17 @@
"""比价报警配置默认值与关键词解析。"""
from __future__ import annotations
from app.core.config import settings
def test_defaults() -> None:
assert settings.COMPARE_ALERT_ENABLED is False
assert settings.COMPARE_ALERT_SCAN_INTERVAL_SEC == 900
assert settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD == 90000
assert settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD == 30
def test_keyword_parsing() -> None:
assert settings.compare_alert_timeout_keywords == ("超时", "启动", "加载")
assert settings.compare_alert_unrecognized_keywords == ("未识别",)
assert "打烊" in settings.compare_alert_biz_exclude_keywords
+62
View File
@@ -0,0 +1,62 @@
"""classify_cancelled_fallback / make_hit 单测。"""
from app.services.compare_alert import classify_cancelled_fallback, make_hit
class _Rec:
def __init__(self, **kw):
self.trace_id = kw.get("trace_id", "t")
self.status = kw.get("status", "cancelled")
self.total_ms = kw.get("total_ms")
self.step_count = kw.get("step_count")
self.fail_reason = kw.get("fail_reason")
self.information = kw.get("information")
self.app_version = kw.get("app_version")
self.created_at = kw.get("created_at")
self.trace_url = kw.get("trace_url")
self.user_id = kw.get("user_id")
def test_fallback_deep_by_ms():
hit = classify_cancelled_fallback(
_Rec(total_ms=95000, step_count=5),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is not None and hit.alert_type == "T5" and "深度放弃" in hit.reason
def test_fallback_deep_by_step():
hit = classify_cancelled_fallback(
_Rec(total_ms=1000, step_count=35),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is not None and hit.alert_type == "T5"
def test_fallback_shallow_none():
hit = classify_cancelled_fallback(
_Rec(total_ms=5000, step_count=3),
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
)
assert hit is None
def test_make_hit_carries_fields():
hit = make_hit(_Rec(trace_id="tx", app_version="0.6.0"), "T5", "卡在 美团·加菜")
assert hit.trace_id == "tx"
assert hit.reason == "卡在 美团·加菜"
assert hit.app_version == "0.6.0"
def test_make_hit_carries_total_ms_and_step():
hit = make_hit(_Rec(trace_id="tx", total_ms=602000, step_count=157), "T5", "深度放弃")
assert hit.total_ms == 602000
assert hit.step_count == 157
def test_make_hit_total_ms_step_default_none():
# rec 没有这两个属性时安全降级为 None(不报错)
class _Bare:
trace_id = "b"; status = "cancelled"; reason = None
app_version = None; created_at = None; trace_url = None; user_id = None
hit = make_hit(_Bare(), "T1", "技术失败")
assert hit.total_ms is None and hit.step_count is None
+572
View File
@@ -0,0 +1,572 @@
"""AlertHit[] → 飞书消息:分组 / 截断 / 含关键词(text + post 两种格式)。"""
from __future__ import annotations
from datetime import datetime
from app.services.compare_alert import AlertHit
from app.services.compare_alert_format import ALERT_KEYWORD, format_alert_message, format_alert_post
def _hits(n, alert_type="T1"):
return [
AlertHit(
trace_id=f"t{i}",
alert_type=alert_type,
reason="比价过程出错",
app_version="v0.3.4",
created_at=None,
trace_url=None,
user_id=None,
)
for i in range(n)
]
# ---- format_alert_message(纯文本,保留原有测试) ----
def test_contains_keyword_and_count():
msg = format_alert_message(
_hits(2), window_label="2026-08-04 08:0008:30",
max_detail_per_type=20, max_total=50,
)
assert ALERT_KEYWORD in msg
assert "本期触发 2 条" in msg
assert "系统技术失败 2 条" in msg
assert "t0" in msg and "v0.3.4" in msg
def test_group_by_type():
hits = _hits(1, "T1") + _hits(1, "T6") + _hits(1, "T5")
msg = format_alert_message(hits, window_label="w", max_detail_per_type=20, max_total=50)
assert "系统技术失败 1 条" in msg
assert "商品识别失败 1 条" in msg
assert "深度放弃(cancelled) 1 条" in msg
def test_per_type_truncation():
msg = format_alert_message(_hits(25), window_label="w", max_detail_per_type=20, max_total=50)
assert msg.count("t0") == 1
assert "另有 5 条" in msg
def test_total_truncation_counts_only():
msg = format_alert_message(_hits(60), window_label="w", max_detail_per_type=20, max_total=50)
assert "系统技术失败 60 条" in msg
assert "t0" not in msg
assert "分析库" in msg
# ---- format_alert_post(富文本 post) ----
def _hits_with_meta(n, alert_type="T1", *, user_id=None, trace_url=None, created_at=None):
return [
AlertHit(
trace_id=f"tr{i}",
alert_type=alert_type,
reason="比价过程出错",
app_version="v1.2.3",
created_at=created_at or datetime(2026, 8, 4, 10, 30),
trace_url=trace_url,
user_id=user_id,
)
for i in range(n)
]
def test_post_title_contains_keyword():
hits = _hits_with_meta(1)
title, content = format_alert_post(
hits, window_label="2026-08-04 10:00", phone_map={}, max_detail_per_type=20, max_total=50,
)
assert ALERT_KEYWORD in title
def test_post_content_is_list():
hits = _hits_with_meta(2)
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
assert isinstance(content, list)
assert len(content) >= 1
# 每个段落是 list[dict]
for para in content:
assert isinstance(para, list)
for elem in para:
assert "tag" in elem
def test_post_summary_count():
hits = _hits_with_meta(3, "T1") + _hits_with_meta(2, "T6")
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
# 摘要第一段含合计数和类型计数
first_para_text = "".join(e.get("text", "") for e in content[0])
assert "合计 5 条" in first_para_text
assert "系统技术失败 3" in first_para_text
assert "商品识别失败 2" in first_para_text
def test_post_phone_map_applied():
hits = _hits_with_meta(1, user_id=42)
title, content = format_alert_post(
hits, window_label="w", phone_map={42: "13800138000"}, max_detail_per_type=20, max_total=50,
)
# 找明细行(非摘要非表头)中含手机号
all_text = " ".join(
e.get("text", "") for para in content for e in para
)
assert "13800138000" in all_text
def test_post_no_user_id_shows_dash():
hits = _hits_with_meta(1, user_id=None)
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
all_text = " ".join(e.get("text", "") for para in content for e in para)
assert " - " in all_text
def test_post_trace_url_becomes_a_element():
hits = _hits_with_meta(1, trace_url="https://trace.example.com/tr0")
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
# 找 tag=a 的元素
a_elements = [e for para in content for e in para if e.get("tag") == "a"]
assert len(a_elements) == 1
assert a_elements[0]["href"] == "https://trace.example.com/tr0"
assert a_elements[0]["text"] == "trace"
def test_post_no_trace_url_shows_trace_id_prefix():
hits = _hits_with_meta(1, trace_url=None)
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
# 无 trace_url 时:tag=text, text=trace_id[:16]
detail_texts = [e.get("text", "") for para in content for e in para if e.get("tag") == "text"]
# trace_id 是 tr0,截 16 位
assert any("tr0" in t for t in detail_texts)
def test_post_total_truncation_no_detail():
hits = _hits_with_meta(60)
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
# 超 max_total:只有摘要+截断提示,无表头、无明细
all_text = " ".join(e.get("text", "") for para in content for e in para)
assert "合计 60 条" in all_text
assert "时间 手机号" not in all_text
assert "tr0" not in all_text
assert "分析库" in all_text
def test_post_per_type_truncation():
hits = _hits_with_meta(25)
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
all_text = " ".join(e.get("text", "") for para in content for e in para)
assert "另有 5 条" in all_text
def test_post_created_at_formatting():
hits = [
AlertHit(
trace_id="tx1",
alert_type="T1",
reason="测试",
app_version="v2.0",
created_at=datetime(2026, 8, 4, 10, 30),
trace_url=None,
user_id=None,
)
]
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
all_text = " ".join(e.get("text", "") for para in content for e in para)
assert "08-04 10:30" in all_text
def test_post_no_created_at_shows_dash():
hits = [
AlertHit(
trace_id="tx2",
alert_type="T1",
reason="测试",
app_version=None,
created_at=None,
trace_url=None,
user_id=None,
)
]
title, content = format_alert_post(
hits, window_label="w", phone_map={}, max_detail_per_type=20, max_total=50,
)
all_text = " ".join(e.get("text", "") for para in content for e in para)
# 无 created_at 时间显示 "-"
assert "- " in all_text
# ---- format_alert_card(schema 2.0 卡片) ----
from app.services.compare_alert_format import format_alert_card # noqa: E402
def _card_hits(n, alert_type="T1", *, total_ms=None, step_count=None, user_id=None,
trace_url=None, created_at=None):
return [
AlertHit(
trace_id=f"card{i}",
alert_type=alert_type,
reason="比价过程出错",
app_version="v1.5.0",
created_at=created_at or datetime(2026, 8, 4, 10, 30),
trace_url=trace_url,
user_id=user_id,
total_ms=total_ms,
step_count=step_count,
)
for i in range(n)
]
def test_card_schema_and_header():
card = format_alert_card(
_card_hits(1),
window_label="2026-08-04 10:00",
phone_map={},
interval_min=15,
max_detail_per_type=20,
max_total=50,
)
assert card["schema"] == "2.0"
assert card["header"]["template"] == "red"
assert ALERT_KEYWORD in card["header"]["title"]["content"]
def test_card_body_markdown_summary():
card = format_alert_card(
_card_hits(3),
window_label="2026-08-04 10:00",
phone_map={},
interval_min=15,
max_detail_per_type=20,
max_total=50,
)
elements = card["body"]["elements"]
md = elements[0]
assert md["tag"] == "markdown"
assert "近 15 分钟" in md["content"]
assert "合计 3 条" in md["content"]
def test_card_has_table_seven_columns():
card = format_alert_card(
_card_hits(2),
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
elements = card["body"]["elements"]
# 第二个元素是 table
assert len(elements) >= 2
table = elements[1]
assert table["tag"] == "table"
cols = table["columns"]
assert len(cols) == 7
display_names = [c["display_name"] for c in cols]
assert display_names == ["时间", "手机号", "用时", "失败原因", "卡点", "版本", "trace"]
# 「卡点」列在「失败原因」后、「版本」前
names = [c["name"] for c in cols]
reason_idx = names.index("reason")
stuck_idx = names.index("stuck")
ver_idx = names.index("ver")
assert reason_idx < stuck_idx < ver_idx
# trace 列用 lark_md
trace_col = next(c for c in cols if c["name"] == "trace")
assert trace_col["data_type"] == "lark_md"
# 其余列 data_type 均为 text
for c in cols:
if c["name"] != "trace":
assert c["data_type"] == "text"
def test_card_cost_cell_format():
"""cost 格式: total_ms 和 step_count 均有值时 '{Ns} / {M步}'"""
hits = [
AlertHit(
trace_id="t1",
alert_type="T1",
reason="r",
app_version="v1",
created_at=datetime(2026, 8, 4, 10, 0),
trace_url=None,
user_id=None,
total_ms=602_000,
step_count=157,
)
]
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["cost"] == "602s / 157步"
def test_card_cost_only_ms():
hits = [
AlertHit(
trace_id="t2",
alert_type="T1",
reason="r",
app_version="v1",
created_at=None,
trace_url=None,
user_id=None,
total_ms=30_000,
step_count=None,
)
]
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["cost"] == "30s"
def test_card_cost_only_steps():
hits = [
AlertHit(
trace_id="t3",
alert_type="T1",
reason="r",
app_version="v1",
created_at=None,
trace_url=None,
user_id=None,
total_ms=None,
step_count=42,
)
]
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["cost"] == "42步"
def test_card_cost_none_when_both_missing():
hits = _card_hits(1, total_ms=None, step_count=None)
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["cost"] == "-"
def test_card_total_truncation_no_table():
"""超 max_total 时只有 markdown 摘要,没有 table 元素"""
card = format_alert_card(
_card_hits(60),
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
elements = card["body"]["elements"]
assert len(elements) == 1
assert elements[0]["tag"] == "markdown"
assert "comparison_record" in elements[0]["content"]
def test_card_empty_hits():
"""空 hits 返回含「本期无异常」的卡片(无 table)"""
card = format_alert_card(
[],
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
assert card["schema"] == "2.0"
elements = card["body"]["elements"]
assert len(elements) == 1
assert elements[0]["tag"] == "markdown"
assert "本期无异常" in elements[0]["content"]
def test_card_rows_count_respects_per_type_limit():
"""每类型最多 max_detail_per_type 条"""
hits = _card_hits(25, "T1")
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
assert len(table["rows"]) == 20
def test_card_trace_url_becomes_markdown_link():
hits = _card_hits(1, trace_url="https://trace.example.com/t0")
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["trace"] == "[链接](https://trace.example.com/t0)"
def test_card_no_trace_url_shows_trace_id_prefix():
hits = _card_hits(1, trace_url=None)
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
# trace_id = "card0"[:12]
assert row["trace"] == "card0"
def test_card_phone_from_map():
hits = _card_hits(1, user_id=7)
card = format_alert_card(
hits,
window_label="w",
phone_map={7: "13912345678"},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["phone"] == "13912345678"
def test_card_version_not_truncated():
"""版本完整显示,不缩写"""
hits = [
AlertHit(
trace_id="tv1",
alert_type="T1",
reason="r",
app_version="v2.15.3-release",
created_at=None,
trace_url=None,
user_id=None,
)
]
card = format_alert_card(
hits,
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
row = table["rows"][0]
assert row["ver"] == "v2.15.3-release"
def test_card_table_has_page_size_and_header_style():
card = format_alert_card(
_card_hits(1),
window_label="w",
phone_map={},
interval_min=10,
max_detail_per_type=20,
max_total=50,
)
table = card["body"]["elements"][1]
assert "page_size" in table
assert "header_style" in table
assert table["header_style"].get("background_style") == "grey"
assert table["header_style"].get("bold") is True
def test_card_stuck_point_shown_in_row():
"""stuck_point 有值时行中 stuck 列正确显示;无值时显示「-」。"""
import dataclasses
from app.services.compare_alert import make_hit
class _Rec:
trace_id = "sp1"
status = "cancelled"
total_ms = 90000
step_count = 20
fail_reason = None
information = None
app_version = "v1.0"
created_at = datetime(2026, 8, 5, 10, 0)
trace_url = None
user_id = None
hit_with = dataclasses.replace(
make_hit(_Rec(), "T5", "深度放弃"),
stuck_point="美团·加菜 110帧/32s",
)
hit_without = make_hit(_Rec(), "T5", "深度放弃") # stuck_point=None
card_with = format_alert_card(
[hit_with],
window_label="w", phone_map={}, interval_min=5,
max_detail_per_type=20, max_total=50,
)
row_with = card_with["body"]["elements"][1]["rows"][0]
assert row_with["stuck"] == "美团·加菜 110帧/32s"
card_without = format_alert_card(
[hit_without],
window_label="w", phone_map={}, interval_min=5,
max_detail_per_type=20, max_total=50,
)
row_without = card_without["body"]["elements"][1]["rows"][0]
assert row_without["stuck"] == "-"
+46
View File
@@ -0,0 +1,46 @@
"""updated_at 列存在、回填 = created_at、onupdate 在 ORM 更新时刷新。"""
from __future__ import annotations
import time
from sqlalchemy import inspect
from app.db.session import SessionLocal, engine
from app.models.comparison import ComparisonRecord
def test_updated_at_column_and_index_exist() -> None:
insp = inspect(engine)
cols = {c["name"] for c in insp.get_columns("comparison_record")}
assert "updated_at" in cols
idx_names = {i["name"] for i in insp.get_indexes("comparison_record")}
# 迁移环境手动建索引名为 ix_comparison_updated;
# 测试环境 create_all() 按 SQLAlchemy 命名约定生成 ix_comparison_record_updated_at。
# 两种环境都验通过即可。
assert (
"ix_comparison_updated" in idx_names
or "ix_comparison_record_updated_at" in idx_names
), f"updated_at index not found; available indexes: {idx_names}"
def test_onupdate_refreshes_updated_at() -> None:
db = SessionLocal()
try:
rec = ComparisonRecord(trace_id="alert-updated-at-onupdate", status="running")
db.add(rec)
db.commit()
db.refresh(rec)
first = rec.updated_at
assert first is not None
time.sleep(1.1) # SQLite CURRENT_TIMESTAMP 秒级精度,睡过 1 秒才看得出变化
rec.status = "failed"
db.commit()
db.refresh(rec)
assert rec.updated_at > first
finally:
db.rollback()
db.query(ComparisonRecord).filter(
ComparisonRecord.trace_id == "alert-updated-at-onupdate"
).delete()
db.commit()
db.close()
+61
View File
@@ -0,0 +1,61 @@
"""比价报警规则分类:记录 → AlertHit | None(纯函数,不碰 DB)。"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from app.services.compare_alert import classify_record
KW = dict(
cancelled_ms_threshold=90000,
cancelled_step_threshold=30,
timeout_keywords=("超时", "启动", "加载"),
unrecognized_keywords=("未识别",),
biz_exclude_keywords=("未找到", "打烊", "起送", "门店", "店内", "不配送", "这些菜", "未入驻", "休息"),
)
def _rec(**kw):
base = dict(
status="failed", fail_reason=None, information=None,
total_ms=None, step_count=None, trace_id="t", app_version=None, business_type="food",
)
base.update(kw)
return SimpleNamespace(**base)
@pytest.mark.parametrize(
("rec", "expected_type"),
[
(_rec(status="failed", fail_reason=None, information="比价过程出错,请稍后重试"), "T1"),
(_rec(status="failed", fail_reason=None, information=None), "T1"),
(_rec(status="failed", fail_reason=None, information="美团外卖门店已打烊,无法比价"), None),
(_rec(status="failed", fail_reason="未识别到商品"), "T6"),
(_rec(status="failed", fail_reason="启动淘宝超时, 请稍后重试"), "T2"),
(_rec(status="failed", fail_reason="淘宝闪购店内未找到这些菜品"), None),
(_rec(status="cancelled", total_ms=98000, step_count=10), "T5"),
(_rec(status="cancelled", total_ms=20000, step_count=31), "T5"),
(_rec(status="cancelled", total_ms=20000, step_count=5), None),
(_rec(status="cancelled", total_ms=None, step_count=None), None),
(_rec(status="success"), None),
(_rec(status="running"), None),
],
)
def test_classify_record_type(rec, expected_type):
hit = classify_record(rec, **KW)
assert (hit.alert_type if hit else None) == expected_type
def test_t5_boundary_exclusive():
assert classify_record(_rec(status="cancelled", total_ms=90000, step_count=30), **KW) is None
assert classify_record(_rec(status="cancelled", total_ms=90001, step_count=30), **KW).alert_type == "T5"
def test_reason_texts():
t1 = classify_record(_rec(status="failed", fail_reason=None, information=" 比价过程出错 "), **KW)
assert t1.reason == "技术失败·比价过程出错"
t1_empty = classify_record(_rec(status="failed", fail_reason=None, information=None), **KW)
assert t1_empty.reason == "技术失败·比价过程出错"
t5 = classify_record(_rec(status="cancelled", total_ms=98000, step_count=26), **KW)
assert t5.reason == "深度放弃"
+129
View File
@@ -0,0 +1,129 @@
"""build_hits 集成测:cancelled trace 优先/保底切换、failed 附卡点、限量。"""
import json
from pathlib import Path
from app.core.compare_alert_worker import _fmt_stuck, build_hits
from app.services.trace_stuck import StuckPoint
class _Rec:
def __init__(self, **kw):
self.trace_id = kw.get("trace_id", "t")
self.status = kw.get("status", "cancelled")
self.total_ms = kw.get("total_ms")
self.step_count = kw.get("step_count")
self.fail_reason = kw.get("fail_reason")
self.information = kw.get("information")
self.app_version = kw.get("app_version")
self.created_at = kw.get("created_at")
self.trace_url = kw.get("trace_url")
self.user_id = kw.get("user_id")
def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
pdir.mkdir(parents=True, exist_ok=True)
body = {"pipeline_step": step, "detected_page": page, "windows": [{"n": ["x" * 200]}]}
(pdir / f"step_{idx:03d}.json").write_text(
json.dumps(body, ensure_ascii=False), encoding="utf-8"
)
_KW = dict(
stuck_threshold=15, max_tail=40, max_trace_reads=30,
cancelled_ms_threshold=90000, cancelled_step_threshold=30,
timeout_keywords=("超时",), unrecognized_keywords=("未识别",), biz_exclude_keywords=(),
)
def test_cancelled_stuck_reports_via_trace(tmp_path):
for i in range(18):
_frame(tmp_path / "20260804_x" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_x/",
total_ms=5000, step_count=3) # 保底不会中,靠 trace 判卡死
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T5"
assert hits[0].reason == "深度放弃"
assert "美团·加菜" in hits[0].stuck_point
assert "" in hits[0].stuck_point
def test_cancelled_readable_not_stuck_no_report(tmp_path):
# trace 确认没卡(在推进);即便 total_ms/step 超阈值也不报(信 trace,不回退保底)
p = tmp_path / "20260804_y" / "eleme"
_frame(p, 0, "set_address", "home")
for i in range(1, 6):
_frame(p, i, "enter_store", "store")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_y/",
total_ms=95000, step_count=40)
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert hits == []
def test_cancelled_unreadable_falls_back(tmp_path):
rec = _Rec(status="cancelled", trace_url="https://x/traces/nope/",
total_ms=95000, step_count=3)
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T5" and "深度放弃" in hits[0].reason
def test_no_work_log_dir_uses_fallback(tmp_path):
rec = _Rec(status="cancelled", trace_url="https://x/traces/y/",
total_ms=95000, step_count=3)
hits = build_hits([rec], work_log_dir="", **_KW)
assert len(hits) == 1 and "深度放弃" in hits[0].reason
def test_failed_gets_stuck_point_appended(tmp_path):
for i in range(20):
_frame(tmp_path / "20260804_f" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="failed", fail_reason="启动超时",
trace_url="https://x/traces/20260804_f/")
hits = build_hits([rec], work_log_dir=str(tmp_path), **_KW)
assert len(hits) == 1
assert hits[0].alert_type == "T2"
assert "美团·加菜" in hits[0].stuck_point
assert "卡在" not in hits[0].reason
def test_max_trace_reads_zero_skips_trace(tmp_path):
for i in range(18):
_frame(tmp_path / "20260804_z" / "meituan", i, "add_one_dish", "meal_detail_popup")
rec = _Rec(status="cancelled", trace_url="https://x/traces/20260804_z/",
total_ms=95000, step_count=3)
kw = {**_KW, "max_trace_reads": 0}
hits = build_hits([rec], work_log_dir=str(tmp_path), **kw)
# 没读 trace → 回退保底 → deep(95s) → 深度放弃
assert len(hits) == 1 and "深度放弃" in hits[0].reason
def test_shared_reads_budget_across_branches(tmp_path):
# cancelled 和 failed 共用 max_trace_reads 预算;预算=1 时 cancelled 先消耗,failed 拿不到卡点
for i in range(18):
_frame(tmp_path / "20260804_c" / "meituan", i, "add_one_dish", "meal_detail_popup")
for i in range(20):
_frame(tmp_path / "20260804_d" / "meituan", i, "add_one_dish", "meal_detail_popup")
cancelled = _Rec(status="cancelled", trace_url="https://x/traces/20260804_c/",
total_ms=5000, step_count=3)
failed = _Rec(status="failed", fail_reason="启动超时",
trace_url="https://x/traces/20260804_d/")
kw = {**_KW, "max_trace_reads": 1}
hits = build_hits([cancelled, failed], work_log_dir=str(tmp_path), **kw)
assert len(hits) == 2
# cancelled 消耗了唯一预算 → 报卡死,卡点在 stuck_point
assert hits[0].alert_type == "T5" and hits[0].stuck_point is not None
# failed 超预算 → 仍报 T2,但 stuck_point 为 None
assert hits[1].alert_type == "T2" and hits[1].stuck_point is None
# ---- _fmt_stuck 单测 ----
def test_fmt_stuck_with_ms():
sp = StuckPoint(platform="meituan", pipeline_step="add_one_dish", frames=110, stuck_ms=32000)
assert _fmt_stuck(sp) == "美团·加菜 110帧/32s"
def test_fmt_stuck_without_ms():
sp = StuckPoint(platform="meituan", pipeline_step="add_one_dish", frames=110, stuck_ms=None)
assert _fmt_stuck(sp) == "美团·加菜 110帧"
+93
View File
@@ -0,0 +1,93 @@
"""扫描 worker 的同步核心 _scan_and_alert:水位读写 / 冷启动 / 发送成功推进、失败不推进。
monkeypatch 掉真正的飞书发送;用真实 SQLite(SessionLocal)插入几条记录
"""
from __future__ import annotations
import time
import pytest
from app.core import compare_alert_worker as w
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
WM_KEY = w.WATERMARK_KEY
@pytest.fixture()
def clean_db():
db = SessionLocal()
db.query(ComparisonRecord).delete()
db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
db.commit()
yield db
db.query(ComparisonRecord).delete()
db.query(AppConfig).filter(AppConfig.key == WM_KEY).delete()
db.commit()
db.close()
def _add(db, trace, status, **kw):
rec = ComparisonRecord(trace_id=trace, status=status, **kw)
db.add(rec)
db.commit()
db.refresh(rec)
return rec
def _card_title(card: dict) -> str:
return card.get("header", {}).get("title", {}).get("content", "")
def _card_md_content(card: dict) -> str:
elements = card.get("body", {}).get("elements", [])
return elements[0].get("content", "") if elements else ""
def test_cold_start_sets_watermark_no_alert(clean_db, monkeypatch):
# 冷启动:表非空 → 水位=当前 max(updated_at),不回溯已有历史失败、不报
_add(clean_db, "old-fail", "failed", fail_reason=None, information="比价过程出错")
sent = []
monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
w._scan_and_alert()
assert sent == [] # 冷启动不报历史
row = clean_db.get(AppConfig, WM_KEY)
assert row is not None # 水位已初始化
def test_alerts_on_new_failed_and_advances(clean_db, monkeypatch):
# seed 一条 + 冷启动建水位;sleep 1.1s 拉开时间(SQLite 秒级精度,否则新记录同秒、追不上水位)
_add(clean_db, "seed", "running")
monkeypatch.setattr(w, "_send_card", lambda card: None)
w._scan_and_alert() # 冷启动,水位=seed.updated_at
time.sleep(1.1)
sent = []
monkeypatch.setattr(w, "_send_card", lambda card: sent.append(card))
_add(clean_db, "new-fail", "failed", fail_reason=None, information="比价过程出错")
w._scan_and_alert()
assert len(sent) == 1
card = sent[0]
# title 含关键词
assert "比价失败报警" in _card_title(card)
# markdown 摘要含"系统技术失败"
assert "系统技术失败 1" in _card_md_content(card)
def test_send_failure_does_not_advance_watermark(clean_db, monkeypatch):
_add(clean_db, "seed2", "running")
monkeypatch.setattr(w, "_send_card", lambda card: None)
w._scan_and_alert() # 冷启动建水位
wm_before = clean_db.get(AppConfig, WM_KEY).value
time.sleep(1.1)
_add(clean_db, "fail-send", "failed", fail_reason=None, information="比价过程出错")
def boom(card):
raise w.feishu_notifier.FeishuNotifyError("down")
monkeypatch.setattr(w, "_send_card", boom)
w._scan_and_alert() # 发送失败
clean_db.expire_all()
wm_after = clean_db.get(AppConfig, WM_KEY).value
assert wm_after == wm_before # 未推进,下轮补发
+169
View File
@@ -0,0 +1,169 @@
"""飞书群机器人发送:text/post 消息体格式 / 关键词失败 / 网络失败,均 monkeypatch httpx 不真发。"""
from __future__ import annotations
import httpx
import pytest
from app.integrations import feishu_notifier
# ---- send_feishu_text ----
def test_send_posts_text_payload(monkeypatch):
captured = {}
def fake_post(url, json, timeout):
captured["url"] = url
captured["json"] = json
return httpx.Response(200, json={"code": 0, "msg": "success"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
feishu_notifier.send_feishu_text("https://open.feishu.cn/hook/xxx", "比价失败报警 · test", timeout=5.0)
assert captured["url"] == "https://open.feishu.cn/hook/xxx"
assert captured["json"] == {"msg_type": "text", "content": {"text": "比价失败报警 · test"}}
def test_send_raises_on_keyword_rejection(monkeypatch):
def fake_post(url, json, timeout):
return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_text("https://hook", "无关键词文本", timeout=5.0)
def test_send_raises_on_http_error(monkeypatch):
def fake_post(url, json, timeout):
return httpx.Response(500, text="boom")
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_text("https://hook", "比价失败报警", timeout=5.0)
# ---- send_feishu_post ----
def test_send_post_payload_structure(monkeypatch):
"""send_feishu_post 发出 msg_type=post 的 payload,结构符合飞书 post 格式。"""
captured = {}
def fake_post(url, json, timeout):
captured["url"] = url
captured["json"] = json
return httpx.Response(200, json={"code": 0, "msg": "success"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
content = [[{"tag": "text", "text": "合计 3 条"}]]
feishu_notifier.send_feishu_post(
"https://open.feishu.cn/hook/yyy",
"🚨 比价失败报警 · 2026-08-04",
content,
timeout=5.0,
)
assert captured["url"] == "https://open.feishu.cn/hook/yyy"
payload = captured["json"]
assert payload["msg_type"] == "post"
zh_cn = payload["content"]["post"]["zh_cn"]
assert zh_cn["title"] == "🚨 比价失败报警 · 2026-08-04"
assert zh_cn["content"] == content
def test_send_post_raises_on_code_nonzero(monkeypatch):
"""send_feishu_post code!=0 时抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_post(
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
)
def test_send_post_raises_on_http_error(monkeypatch):
"""send_feishu_post 非 2xx 抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
return httpx.Response(500, text="internal server error")
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_post(
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
)
def test_send_post_raises_on_network_error(monkeypatch):
"""send_feishu_post 网络异常抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
raise httpx.ConnectError("connection refused")
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_post(
"https://hook", "🚨 比价失败报警", [[{"tag": "text", "text": "test"}]], timeout=5.0
)
# ---- send_feishu_card ----
def test_send_card_payload_structure(monkeypatch):
"""send_feishu_card 发出 msg_type=interactive + card 字段的 payload。"""
captured = {}
def fake_post(url, json, timeout):
captured["url"] = url
captured["json"] = json
return httpx.Response(200, json={"code": 0, "msg": "success"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
card = {
"schema": "2.0",
"header": {"title": {"tag": "plain_text", "content": "🚨 比价失败报警 · test"}, "template": "red"},
"body": {"elements": [{"tag": "markdown", "content": "近 15 分钟"}]},
}
feishu_notifier.send_feishu_card("https://open.feishu.cn/hook/zzz", card, timeout=5.0)
assert captured["url"] == "https://open.feishu.cn/hook/zzz"
assert captured["json"]["msg_type"] == "interactive"
assert captured["json"]["card"] is card
def test_send_card_uses_default_timeout(monkeypatch):
"""send_feishu_card 默认 timeout=10.0。"""
captured = {}
def fake_post(url, json, timeout):
captured["timeout"] = timeout
return httpx.Response(200, json={"code": 0, "msg": "success"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
assert captured["timeout"] == 10.0
def test_send_card_raises_on_code_nonzero(monkeypatch):
"""send_feishu_card code!=0 时抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
return httpx.Response(200, json={"code": 19024, "msg": "Key Words Not Found"})
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
def test_send_card_raises_on_http_error(monkeypatch):
"""send_feishu_card 非 2xx 抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
return httpx.Response(500, text="boom")
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
def test_send_card_raises_on_network_error(monkeypatch):
"""send_feishu_card 网络异常抛 FeishuNotifyError。"""
def fake_post(url, json, timeout):
raise httpx.ConnectError("connection refused")
monkeypatch.setattr(feishu_notifier.httpx, "post", fake_post)
with pytest.raises(feishu_notifier.FeishuNotifyError):
feishu_notifier.send_feishu_card("https://hook", {"schema": "2.0"})
+154
View File
@@ -0,0 +1,154 @@
"""trace_stuck 单测:用 tmp 造 step_*.json(只含头部字段)验证卡死判据。"""
import json
from pathlib import Path
from app.services.trace_stuck import (
StuckPoint,
dir_name_from_trace_url,
last_step,
read_stuck_points,
)
def _frame(pdir: Path, idx: int, step: str, page: str) -> None:
"""造一帧 step json:头部放 pipeline_step/detected_page,尾部塞大 windows 模拟真实。"""
pdir.mkdir(parents=True, exist_ok=True)
body = {
"trace_id": "t", "step": idx, "platform": pdir.name,
"pipeline_step": step, "detected_page": page,
"windows": [{"nodes": ["x" * 200]}],
}
(pdir / f"step_{idx:03d}.json").write_text(
json.dumps(body, ensure_ascii=False), encoding="utf-8"
)
def test_stuck_when_tail_repeats_same_step(tmp_path):
pdir = tmp_path / "meituan"
for i in range(20):
_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)]
def test_not_stuck_when_progressing(tmp_path):
pdir = tmp_path / "eleme"
_frame(pdir, 0, "set_address", "home")
for i in range(1, 8):
_frame(pdir, i, "enter_store", "store")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.readable is True
assert res.points == []
def test_adding_many_dishes_not_stuck_when_page_changes(tmp_path):
# add_one_dish 重复但 detected_page 在跳(换菜)=推进,不判卡死
pdir = tmp_path / "meituan"
for i in range(20):
_frame(pdir, i, "add_one_dish", "menu" if i % 2 == 0 else "dish_popup")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.points == []
def test_below_threshold_not_stuck(tmp_path):
pdir = tmp_path / "meituan"
for i in range(10): # < 15
_frame(pdir, i, "add_one_dish", "meal_detail_popup")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.points == []
def test_missing_dir_not_readable(tmp_path):
res = read_stuck_points(tmp_path / "nope", threshold=15, max_tail=40)
assert res.readable is False
assert res.points == []
def test_empty_dir_no_platform_frames_not_readable(tmp_path):
(tmp_path / "emptysub").mkdir()
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert res.readable is False
def test_per_platform_one_stuck_one_normal(tmp_path):
m = tmp_path / "meituan"
for i in range(18):
_frame(m, i, "add_one_dish", "meal_detail_popup")
e = tmp_path / "eleme"
_frame(e, 0, "set_address", "home")
for i in range(1, 6):
_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)]
def test_last_step_returns_busiest_platform_last_env(tmp_path):
m = tmp_path / "meituan"
for i in range(20):
_frame(m, i, "add_one_dish", "meal_detail_popup")
e = tmp_path / "eleme"
for i in range(3):
_frame(e, i, "enter_store", "store")
sp = last_step(tmp_path)
assert sp == StuckPoint("meituan", "add_one_dish", 20)
def test_dir_name_from_trace_url():
assert dir_name_from_trace_url("https://x/traces/20260804_1_abc/") == "20260804_1_abc"
assert dir_name_from_trace_url("https://x/traces/20260804_1_abc") == "20260804_1_abc"
assert dir_name_from_trace_url("") is None
assert dir_name_from_trace_url(None) is None
def test_corrupt_frame_does_not_crash(tmp_path):
# 截断的 UTF-8(pricebot 被 SIGTERM 打断的末帧)不应抛 UnicodeDecodeError
pdir = tmp_path / "meituan"
pdir.mkdir()
(pdir / "step_000.json").write_bytes(b'{"pipeline_step": "add\xff')
res = read_stuck_points(tmp_path, threshold=1, max_tail=40)
assert res.points == [] # 抠不出字段 → 不判卡死;关键是没崩溃
def _frame_ts(pdir: Path, idx: int, step: str, page: str, ts: str) -> None:
pdir.mkdir(parents=True, exist_ok=True)
body = {"pipeline_step": step, "detected_page": page, "timestamp": ts,
"windows": [{"n": ["x" * 200]}]}
(pdir / f"step_{idx:03d}.json").write_text(
json.dumps(body, ensure_ascii=False), encoding="utf-8"
)
def test_stuck_ms_computed_from_timestamps(tmp_path):
# 末段 16 帧都卡在 add_one_dish,timestamp 从 :00 到 :30(每帧+2s) → 卡住 30s
pdir = tmp_path / "meituan"
for i in range(16):
_frame_ts(pdir, i, "add_one_dish", "meal_detail_popup",
f"2026-08-04T12:00:{i * 2:02d}.000000")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert len(res.points) == 1
sp = res.points[0]
assert sp.frames == 16
assert sp.stuck_ms == 30000 # (第15帧:30 - 第0帧:00) = 30s
def test_stuck_ms_none_when_timestamp_missing(tmp_path):
# 帧无 timestamp 字段 → stuck_ms 为 None(不报错)
pdir = tmp_path / "meituan"
for i in range(16):
_frame(pdir, i, "add_one_dish", "meal_detail_popup") # 现有 _frame,无 timestamp
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert len(res.points) == 1
assert res.points[0].stuck_ms is None
def test_stuck_ms_none_when_clock_goes_backwards(tmp_path):
# 帧 timestamp 非单调(时钟回退):step_000=:30 ... step_015=:00 → 末帧早于段首 → 负时长 → 降级 None
pdir = tmp_path / "meituan"
for i in range(16):
_frame_ts(pdir, i, "add_one_dish", "meal_detail_popup",
f"2026-08-04T12:00:{(30 - i * 2):02d}.000000")
res = read_stuck_points(tmp_path, threshold=15, max_tail=40)
assert len(res.points) == 1
assert res.points[0].stuck_ms is None # 负时长降级为 None