shaguabijia-app-server/app/services/trace_stuck.py

"""比价卡死定位:读 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
    detected_page: str | None = None  # 末帧所在页面(pricebot detected_page 原值);无则 None
    dwell_ms: int | None = None  # 末帧所在屏停留时长(ms);末帧路径(failed/兜底)用,无 ts → None

    def label(self) -> str:
        p = PLATFORM_LABELS.get(self.platform, self.platform)
        s = PIPELINE_STEP_LABELS.get(self.pipeline_step, self.pipeline_step)
        base = f"{p}·{s}"
        if self.detected_page:
            base += f"·{self.detected_page}"  # 页面暂用 pricebot 原值(英文),无中文映射
        return base


@dataclass(frozen=True)
class StuckResult:
    readable: bool               # trace 是否读到(区分「读不到」与「读到但没卡」)
    points: list[StuckPoint]     # 卡死平台列表;readable=True 且空 = 确认没卡死
    last: StuckPoint | None = None  # 末帧(帧数最多平台的末帧,带 detected_page);读不到 → None


def dir_name_from_trace_url(trace_url: str | None) -> str | None:
    """.../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 _last_segment(
    step_files: list[Path], max_tail: int
) -> tuple[str, str | None, int, int | None] | None:
    """末帧往前数连续同 (pipeline_step, detected_page) 的一段。

    返回 (pipeline_step, detected_page, count, dwell_ms);末帧 pipeline_step 抠不出 → None。
    dwell_ms = 段末帧ts − 段首帧ts(ms);两端 ts 不全可解析、或负(帧钟回退) → None。
    这是「末段停留」的唯一算法,卡死判据(≥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
    t_last, t_first = _parse_ts(seg_ts[0]), _parse_ts(seg_ts[-1])
    dwell_ms = round((t_last - t_first).total_seconds() * 1000) if t_last and t_first else None
    if dwell_ms is not None and dwell_ms < 0:
        dwell_ms = None  # 时钟不单调(帧 timestamp 回退)→ 降级为不显示时长
    return last_ps, last_pg, len(seg_ts), dwell_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
        last: StuckPoint | None = None
        best_n = -1
        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
            seg = _last_segment(step_files, max_tail)  # 每平台只扫一次末段
            if seg is None:
                # 末帧抠不出:整段跳过(不判卡死、也不当末帧候选)——与旧版 _platform_stuck→None
                # + 独立 _read_head(末帧)→ps None 两处一并跳过等价(旧版两者都 key off 末帧)
                continue
            ps, pg, count, dwell_ms = seg
            if count >= threshold:
                points.append(StuckPoint(pdir.name, ps, count, dwell_ms, detected_page=pg))
            # 末帧:取帧数最多平台的末帧(与 last_step 同口径),带 detected_page + 末段停留
            if len(step_files) > best_n:
                best_n = len(step_files)
                last = StuckPoint(
                    pdir.name, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms
                )
        if not any_frames:
            return StuckResult(readable=False, points=[])
        return StuckResult(readable=True, points=points, last=last)
    except OSError:
        return StuckResult(readable=False, points=[])


def last_step(trace_dir: Path, *, max_tail: int) -> StuckPoint | None:
    """failed 用:取帧数最多平台的末帧环节(不要求原地打转)。读不到 → None。
    附末段停留 dwell_ms(末帧所在屏停留时长);无 ts/时钟回退 → 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
        seg = _last_segment(step_files, max_tail)
        if seg is None:
            return None
        ps, pg, _count, dwell_ms = seg
        # frames 仍=总帧数(选平台口径不变);dwell_ms=末帧所在屏停留
        return StuckPoint(platform, ps, len(step_files), detected_page=pg, dwell_ms=dwell_ms)
    except OSError:
        return None