cma-management/backend/app/utils/threshold_parser.py

# -*- coding: utf-8 -*-
"""阈值表达式统一解析与红黄绿判定 —— 单一事实源（Single Source of Truth）

背景（project-20260911-threshold-fix）：
    原先 kpis.py 与 alert_generator.py 各自实现一份阈值解析（正则相同、独立维护），
    且解析器不支持负数，导致 `>=-100000` 这类表达式静默失效。

表达式语法（label 的语义 = “该等级的触发条件”，不是区间标签）：
    >=N  <=N  >N  <N  =N  !=N    标量比较，N 支持负数与小数（如 `<-100000`、`>=52.33`）
    A~B                          闭区间包含，A <= 实际值 <= B（如 `52.33~107.67`）
    C1|C2                        多条件或，任一命中即为命中（如 `<=36|>=124`）
    允许空白：`>= 800`、`52.33 ~ 107.67`；空串/None 视为“无阈值”。

红黄绿判定优先级：red → green → yellow（理由见下方 calc_level_from_exprs）
"""

from __future__ import annotations

import re
from typing import List, Optional, Tuple, Union

# 标量比较：操作符 + 数值（负号/小数）
_SCALAR_RE = re.compile(r"(>=|<=|!=|>|<|=)\s*(-?\d+(?:\.\d+)?)")
# 区间：A~B（闭区间）
_RANGE_RE = re.compile(r"(-?\d+(?:\.\d+)?)\s*~\s*(-?\d+(?:\.\d+)?)")
# 多条件分隔（半角竖线 / 全角竖线）
_CLAUSE_SPLIT_RE = re.compile(r"\s*[|｜]\s*")

#: 单个子句取值：标量 -> float；区间 -> (low, high)
ThresholdValue = Union[float, Tuple[float, float]]
Clause = Tuple[str, ThresholdValue]

#: 支持的格式说明（前端 tooltip / 接口文档共用文案）
THRESHOLD_HELP = (
    "支持格式：>=800 / <=25 / >60 / <20 / =100 / !=0（数值可为负数，如 <-100000）；"
    "区间 52.33~107.67（含端点）；多条件用 | 连接，如 <=36|>=124。"
)
THRESHOLD_PLACEHOLDER = "如 >=800 / 52.33~107.67"
#: 判定优先级（严重度最高者优先；green 优先于 yellow 见 docstring 说明）
LEVEL_PRIORITY: Tuple[str, str, str] = ("red", "green", "yellow")


def _is_blank(expr) -> bool:
    return expr is None or not str(expr).strip()


def parse_clause(clause) -> Optional[Clause]:
    """解析单个子句；非法返回 None。"""
    s = str(clause).strip()
    if not s:
        return None
    m = _RANGE_RE.fullmatch(s)
    if m:
        lo, hi = float(m.group(1)), float(m.group(2))
        return ("~", (lo, hi))
    m = _SCALAR_RE.fullmatch(s)
    if m:
        return (m.group(1), float(m.group(2)))
    return None


def parse_clauses(expr) -> List[Clause]:
    """解析表达式为子句列表；任一子句非法则整体视为非法（返回空列表）。"""
    if _is_blank(expr):
        return []
    parts = _CLAUSE_SPLIT_RE.split(str(expr).strip())
    clauses: List[Clause] = []
    for p in parts:
        c = parse_clause(p)
        if c is None:
            return []
        clauses.append(c)
    return clauses


def parse_threshold(expr) -> Tuple[Optional[str], Optional[ThresholdValue]]:
    """解析阈值表达式，返回 (operator, value)。

    保持向后兼容的二元返回：解析失败 -> (None, None)。
    区间操作符为 "~"，value 为 (low, high)；多条件表达式返回首个命中子句之外的首个子句。
    示例:
        '>=32000000'      -> ('>=', 32000000.0)
        '>=-100000'       -> ('>=', -100000.0)
        '52.33~107.67'    -> ('~', (52.33, 107.67))
    """
    clauses = parse_clauses(expr)
    if not clauses:
        return None, None
    return clauses[0]


def is_valid_threshold(expr) -> bool:
    """表达式是否合法（空串视为合法=不设置阈值）。"""
    if _is_blank(expr):
        return True
    return bool(parse_clauses(expr))


def eval_clause(actual, clause: Clause) -> bool:
    op, val = clause
    if actual is None:
        return False
    if op == ">=":
        return actual >= val  # type: ignore[operator]
    if op == "<=":
        return actual <= val  # type: ignore[operator]
    if op == ">":
        return actual > val  # type: ignore[operator]
    if op == "<":
        return actual < val  # type: ignore[operator]
    if op == "=":
        return actual == val
    if op == "!=":
        return actual != val
    if op == "~":
        lo, hi = val  # type: ignore[misc]
        return lo <= actual <= hi
    return False


def eval_threshold(actual, expr) -> bool:
    """实际值是否命中阈值表达式（多条件任一命中即 True；非法表达式 -> False）。"""
    if actual is None or _is_blank(expr):
        return False
    clauses = parse_clauses(expr)
    return any(eval_clause(actual, c) for c in clauses)


def calc_level_from_exprs(actual, green=None, yellow=None, red=None) -> str:
    """红黄绿判定（统一入口），返回 'red' / 'yellow' / 'green' / 'info'。

    判定顺序 red → green → yellow：
      1. red 最严重，必须最先判 —— 否则宽黄灯（如 Y='>76'）会遮蔽红灯（如 R='>91'），
         红灯永远不可达（缺陷5：id=6 实测值 95 被判黄）。
      2. green 先于 yellow —— green 表达式（如 G='>=800'）是“主条件”，
         yellow（如 Y='>=700'）是更宽的兜底带；若 yellow 先判，值 900 会被黄灯截胡，
         与用户“>=800 即绿”的填写意图相反（id=1 实测）。
         对 moderate 型（绿=内区间、黄=外侧环带）两者天然互斥，顺序无影响。
      3. 三者均不命中 -> 'info'（无判定，与旧实现一致）。
    """
    if actual is None:
        return "info"
    exprs = {"green": green, "yellow": yellow, "red": red}
    for level in LEVEL_PRIORITY:
        if eval_threshold(actual, exprs[level]):
            return level
    return "info"


def calc_level(actual, kpi) -> str:
    """按 KPI 定义对象（含 threshold_green/yellow/red 属性）判定级别。"""
    return calc_level_from_exprs(
        actual,
        green=getattr(kpi, "threshold_green", None),
        yellow=getattr(kpi, "threshold_yellow", None),
        red=getattr(kpi, "threshold_red", None),
    )


def describe_level(actual, kpi) -> Tuple[str, Optional[str]]:
    """返回 (级别, 命中的表达式)。供预警消息构造使用；无命中 -> (level, None)。"""
    level = calc_level(actual, kpi)
    if level == "info":
        return level, None
    field = {"green": "threshold_green", "yellow": "threshold_yellow", "red": "threshold_red"}[level]
    return level, getattr(kpi, field, None)