vent_agent_2.0/algorithms/airflow_review/contracts.py

"""风量研判的公共契约：枚举、常量与规范化函数。

字段名与《矿井风量研判-研发需求》§2.7 一致；这里只做定义与规范化，不做
业务判断，避免规则分散到多个模块。
"""
from __future__ import annotations

import re
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable, Mapping

ENGINE_VERSION = "airflow-review-2.0.0"
RULE_VERSION_DEFAULT = "AIR-RULE-2026.1"

# 矿井所在地时区（布尔台矿位于内蒙古鄂尔多斯，东八区）
MINE_TZ = timezone(timedelta(hours=8))

# ---- 目标类型 -------------------------------------------------------------
TARGET_MINE_INTAKE = "mine_total_intake"
TARGET_MINE_RETURN = "mine_total_return"
TARGET_RETURN_SYSTEM = "return_air_system"
TARGET_WORKPLACE = "workplace"
TARGET_TYPES = (TARGET_MINE_INTAKE, TARGET_MINE_RETURN, TARGET_RETURN_SYSTEM, TARGET_WORKPLACE)

# 配风计划侧的层级（§2.7.2），比实测侧少了总进风/总回风两个口径
PLAN_TARGET_TYPES = (TARGET_WORKPLACE, TARGET_RETURN_SYSTEM, "mine_total")

# ---- 地点类型 -------------------------------------------------------------
LOC_WORKPLACE_COAL = "采煤工作面"
LOC_WORKPLACE_HEADING = "掘进工作面"
LOC_CHAMBER = "硐室"
LOC_OTHER = "其他巷道"
LOC_STANDBY = "备用工作面"
LOCATION_TYPES = (LOC_WORKPLACE_COAL, LOC_WORKPLACE_HEADING, LOC_CHAMBER, LOC_OTHER, LOC_STANDBY)

# 视为工作面的类型（适用 75% 红线）；硐室与其他巷道不在其中
WORKPLACE_LOCATION_TYPES = (LOC_WORKPLACE_COAL, LOC_WORKPLACE_HEADING, LOC_STANDBY)

# ---- 问题类型（§2.7.4，七种枚举）----------------------------------------
FINDING_PLAN_MISMATCH = "plan_calculation_mismatch"
FINDING_TOTAL_SHORTAGE = "total_airflow_shortage"
FINDING_RETURN_SHORTAGE = "return_system_airflow_shortage"
FINDING_WORKPLACE_BELOW_75 = "workplace_airflow_below_75_percent"
FINDING_TIME_ALIGNMENT = "time_alignment_failed"
FINDING_DATA_CONFLICT = "data_conflict"
FINDING_DATA_INSUFFICIENT = "data_insufficient"
FINDING_TYPES = (
    FINDING_PLAN_MISMATCH,
    FINDING_TOTAL_SHORTAGE,
    FINDING_RETURN_SHORTAGE,
    FINDING_WORKPLACE_BELOW_75,
    FINDING_TIME_ALIGNMENT,
    FINDING_DATA_CONFLICT,
    FINDING_DATA_INSUFFICIENT,
)

# ---- 状态枚举 -------------------------------------------------------------
VERIFY_PENDING = "pending"
VERIFY_VALID = "valid"
VERIFY_INVALID = "invalid"
VERIFY_MANUAL = "manual_review"

ALIGN_ALIGNED = "aligned"
ALIGN_PARTIAL = "partial"
ALIGN_FAILED = "failed"
ALIGN_MANUAL = "manual_review"

COMPARE_NORMAL = "normal"
COMPARE_INSUFFICIENT = "insufficient"
COMPARE_MAJOR = "major_risk"
COMPARE_NOT_COMPARABLE = "not_comparable"

SEV_MAJOR = "major"
SEV_GENERAL = "general"
SEV_ATTENTION = "attention"
SEV_MANUAL = "manual_review"
SEVERITIES = (SEV_MAJOR, SEV_GENERAL, SEV_ATTENTION, SEV_MANUAL)

# 来源优先级（§2.6 第 11 条）：正式测风报表为主，安全监控用于交叉校核
SOURCE_MEASUREMENT = "measurement_report"
SOURCE_MONITORING = "monitoring"
SOURCE_PLAN = "plan"

# 数值容差默认值（§2.3 工具 4 输入示例）
DEFAULT_TOLERANCE = {"absolute_m3_min": 1.0, "relative_percent": 0.5}


class ContractError(ValueError):
    """输入不满足契约时抛出；调用方应转为 manual_review 或可读错误返回。"""


def now_iso(tz: timezone | None = None) -> str:
    return datetime.now(tz or MINE_TZ).isoformat(timespec="seconds")


_ISO_PAD_DATE = re.compile(r"^(\d{4})-(\d{1,2})-(\d{1,2})")
_ISO_PAD_TIME = re.compile(r"[T ](\d{1,2}):(\d{2})")


def _normalize_iso(text: str) -> str:
    """把材料里常见的非 ISO 时间写法收成 ISO，好让 fromisoformat 认得出。

    ``2024年12月1日`` → ``2024-12-01``；``2024/12/01`` → ``2024-12-01``；
    ``2024-12-01 9:00`` → ``2024-12-01T09:00``。补零是必需的——fromisoformat 不接受
    ``2024-12-1`` 这种单位数月日。
    """
    collapsed = re.sub(r"\s*([年月日])\s*", r"\1", text)
    normalized = collapsed.replace("年", "-").replace("月", "-").replace("日", " ")
    if re.match(r"^\d{4}/", normalized):
        normalized = normalized.replace("/", "-")
    normalized = re.sub(r"\s+", " ", normalized).strip()
    normalized = _ISO_PAD_DATE.sub(
        lambda match: f"{match.group(1)}-{int(match.group(2)):02d}-{int(match.group(3)):02d}", normalized)
    normalized = _ISO_PAD_TIME.sub(
        lambda match: f"T{int(match.group(1)):02d}:{match.group(2)}", normalized)
    return normalized


def parse_time(value: Any) -> datetime | None:
    """解析时间；无时区信息时按矿井时区补全，无法解析返回 None。

    容忍国内材料里常见的非 ISO 写法（``2024年12月1日``、``2024/12/01``、``9:00``）——
    计划和测风报表里这么写很普遍，认不出来就会被误判成"时间无法对齐"。
    """
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, str) and value.strip():
        text = _normalize_iso(value.strip().replace("Z", "+00:00"))
        try:
            parsed = datetime.fromisoformat(text)
        except ValueError:
            return None
    else:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=MINE_TZ)
    return parsed


def to_m3_min(value: Any, unit: str | None = None) -> tuple[float | None, str | None]:
    """把风量统一换算为 m3/min。

    返回 (值, 告警)。单位缺失或无法识别时告警，不猜测——调用方据此转人工复核。
    """
    number = _as_float(value)
    if number is None:
        return None, "风量数值无法解析"
    suffix = (unit or "m3/min").strip().lower().replace("³", "3").replace(" ", "")
    factors = {
        "m3/min": 1.0, "m³/min": 1.0, "m3min": 1.0,
        "m3/s": 60.0, "m³/s": 60.0,
        "m3/h": 1.0 / 60.0, "m³/h": 1.0 / 60.0,
        "m3/min.1": 1.0,
    }
    factor = factors.get(suffix)
    if factor is None:
        return None, f"未识别的风量单位：{unit}"
    return round(number * factor, 4), None


def _as_float(value: Any) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str):
        text = value.strip().replace(",", "").replace("，", "")
        try:
            return float(text)
        except ValueError:
            return None
    return None


def window_contains(outer: dict[str, Any] | None, inner: dict[str, Any] | None) -> bool:
    """判断 inner 时间窗是否完全落在 outer 内；任一侧缺时间返回 False。"""
    if not outer or not inner:
        return False
    o_start, o_end = parse_time(outer.get("start")), parse_time(outer.get("end"))
    i_start, i_end = parse_time(inner.get("start")), parse_time(inner.get("end"))
    if None in (o_start, o_end, i_start, i_end):
        return False
    return o_start <= i_start and i_end <= o_end


def window_overlaps(left: dict[str, Any] | None, right: dict[str, Any] | None) -> bool:
    if not left or not right:
        return False
    l_start, l_end = parse_time(left.get("start")), parse_time(left.get("end"))
    r_start, r_end = parse_time(right.get("start")), parse_time(right.get("end"))
    if None in (l_start, l_end, r_start, r_end):
        return False
    return l_start <= r_end and r_start <= l_end


def ratio(actual: float | None, required: float | None) -> float | None:
    """实际/需风量比例；需风量为 0 或缺失时返回 None（不得据此判定正常）。"""
    if actual is None or required in (None, 0):
        return None
    return round(actual / required, 4)


def unique(items: Iterable[str]) -> list[str]:
    seen: dict[str, None] = {}
    for item in items:
        if item:
            seen.setdefault(item, None)
    return list(seen)


def as_list(value: Any) -> list[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def text(value: Any) -> str:
    return "" if value is None else str(value).strip()


# ---- 输入归一化 -----------------------------------------------------------
# 这些函数只为"让工具不因模型写法不同而崩"而存在。模型会把结构字段写成字符串、
# 数组或省略键（实测把 required_effective_window 传成 "2024-12-01 ~ 2024-12-31"），
# 直接 dict() 会抛 ValueError 并炸掉整轮对话。归一化之后要么得到可用值，要么得到
# 空值并由调用方按"依据不足/时间无法对齐"转人工复核——绝不抛异常。

_DATE_TOKEN = re.compile(
    r"\d{4}[-/年]\d{1,2}(?:[-/月]\d{1,2})?"
    r"(?:[T ]\d{1,2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?"
)


def as_mapping(value: Any) -> dict[str, Any]:
    """把"应该是对象"的输入收成 dict；收不成返回空字典。"""
    if isinstance(value, Mapping):
        return dict(value)
    if isinstance(value, (list, tuple)) and value:
        pairs = [item for item in value if isinstance(item, (list, tuple)) and len(item) == 2]
        if len(pairs) == len(value):
            return {str(key): item for key, item in pairs}
    return {}


def as_window(value: Any) -> dict[str, Any]:
    """把时间窗输入收成 ``{start, end}``。

    接受对象、``[[start, a], [end, b]]`` 形式，以及常见的字符串写法——按其中的日期
    记号抽取：抽到两个即起止，抽到一个按点窗口处理并留痕，抽不到返回空字典（调用方
    会按"时间无法对齐"转人工复核）。用抽记号而不是切分隔符，是为了不被日期里的
    ``-``、``/`` 误伤。
    """
    mapping = as_mapping(value)
    if mapping:
        return mapping
    tokens: list[str] = []
    if isinstance(value, str):
        tokens = _DATE_TOKEN.findall(value)
    elif isinstance(value, (list, tuple)):
        tokens = [str(item) for item in value if parse_time(item) is not None]
    if len(tokens) >= 2:
        start, end = tokens[0], tokens[-1]
        if parse_time(start) and parse_time(end):
            return {"start": start, "end": end, "normalized_from": "string" if isinstance(value, str) else "sequence"}
    elif len(tokens) == 1 and parse_time(tokens[0]):
        return {"start": tokens[0], "end": tokens[0], "normalized_from": "single_token"}
    return {}


def mappings(items: Any) -> list[dict[str, Any]]:
    """把"应该是对象数组"的输入收成对象数组，跳过不是对象的元素。"""
    return [dict(item) for item in as_list(items) if isinstance(item, Mapping)]