320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""比价失败报警后台任务:周期扫 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,判出原地卡点就带卡点报 T5,否则(可读没卡点/读不到)一律回退
|
|
耗时/步数兜底——超长放弃照报(卡点列留空);failed 命中后附末段卡点。
|
|
|
|
trace 读取限量 max_trace_reads 次/轮;任何 trace 异常都在 trace_stuck 内部降级为「读不到」,
|
|
cancelled 因而走耗时兜底、failed 不附卡点,绝不影响报警发送。trace 只做「锦上添花」标注卡点,
|
|
绝不因「可读但没判出卡点」把超长放弃吞掉(线上 trace 几乎总可读,否则 total_ms 阈值形同虚设)。
|
|
"""
|
|
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 and res.points:
|
|
stuck = "、".join(_fmt_stuck(sp) for sp in res.points)
|
|
hit = _dc_replace(make_hit(rec, "T5", "深度放弃"), stuck_point=stuck)
|
|
else:
|
|
# trace 判出卡点 → 上面带卡点报。其余一律回退耗时/步数兜底:可读但没判出卡点、
|
|
# 读不到、无 trace,都过 total_ms/step 阈值——超长放弃照报,不再因「trace 可读但不
|
|
# 原地卡」把超长放弃整条吞掉(线上 trace 几乎总可读,否则耗时阈值形同虚设)。
|
|
hit = classify_cancelled_fallback(
|
|
rec,
|
|
cancelled_ms_threshold=cancelled_ms_threshold,
|
|
cancelled_step_threshold=cancelled_step_threshold,
|
|
)
|
|
# 没卡点但命中(超阈值放弃)→ 用末帧标「退出前在哪屏」(平台·环节·页面)
|
|
if hit is not None and res is not None and res.last is not None:
|
|
hit = _dc_replace(hit, stuck_point=res.last.label())
|
|
if hit is not None:
|
|
hits.append(hit)
|
|
else:
|
|
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,
|
|
cancelled_ms_threshold=settings.COMPARE_ALERT_CANCELLED_MS_THRESHOLD,
|
|
cancelled_step_threshold=settings.COMPARE_ALERT_CANCELLED_STEP_THRESHOLD,
|
|
)
|
|
|
|
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
|