vent_agent_2.0/algorithms/airflow_review/rules/loader.py

"""规则库加载与查询：公式、参数派生、验算项、阈值、表中取值。

规则以 JSON 文件承载并版本化；加载器只做「取规则」和「按事实选变体」，
不做业务判断，保证同一 (规则版本, 参数事实) 必然得到同一公式与系数。
"""
from __future__ import annotations

import json
from functools import lru_cache
from pathlib import Path
from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    RULE_VERSION_DEFAULT,
    parse_time,
)
from algorithms.airflow_review.engine.expression import try_evaluate

RULES_DIR = Path(__file__).resolve().parent


class RuleError(ValueError):
    """规则库缺失、结构不合法或规则版本不可用。"""


class RuleSet:
    def __init__(self, data: dict[str, Any], source: str = "") -> None:
        if not data.get("rule_version"):
            raise RuleError("规则库缺少 rule_version")
        self._data = data
        self._source = source

    # ---- 基本信息 ---------------------------------------------------------
    @property
    def rule_version(self) -> str:
        return self._data["rule_version"]

    @property
    def source(self) -> str:
        return self._source

    @property
    def title(self) -> str:
        return self._data.get("title", "")

    def effective_window(self) -> dict[str, Any]:
        return {"start": self._data.get("effective_from"), "end": self._data.get("effective_to")}

    def is_effective_at(self, moment: Any) -> bool:
        parsed = parse_time(moment)
        if parsed is None:
            return False
        start = parse_time(self._data.get("effective_from"))
        end = parse_time(self._data.get("effective_to"))
        if start and parsed < start:
            return False
        if end and parsed > end:
            return False
        return True

    # ---- 常数与表 ---------------------------------------------------------
    def constant(self, name: str) -> Any:
        entry = (self._data.get("constants") or {}).get(name)
        if entry is None:
            raise RuleError(f"规则库缺少常数：{name}")
        return entry.get("value") if isinstance(entry, dict) else entry

    def constants(self) -> dict[str, Any]:
        return {name: (entry.get("value") if isinstance(entry, dict) else entry)
                for name, entry in (self._data.get("constants") or {}).items()}

    def table(self, name: str) -> list[dict[str, Any]]:
        table = (self._data.get("tables") or {}).get(name)
        if not table:
            raise RuleError(f"规则库缺少参数表：{name}")
        return list(table.get("rows") or [])

    def table_bounds(self, name: str, by_field: str) -> tuple[str, str]:
        """取区间字段名。表中字段名不必与 by 字段同名（如 min_temp vs max_temperature），
        因此允许在表定义里显式声明 bounds，未声明时回落 min_/max_ + by 字段名。"""
        table = (self._data.get("tables") or {}).get(name) or {}
        bounds = table.get("bounds") or {}
        return (str(bounds.get("low") or f"min_{by_field}"),
                str(bounds.get("high") or f"max_{by_field}"))

    def table_title(self, name: str) -> str:
        return str(((self._data.get("tables") or {}).get(name) or {}).get("title") or name)

    def lookup_table(
        self,
        name: str,
        by_field: str,
        value: Any,
        pick: Any,
    ) -> tuple[float | None, dict[str, Any] | None]:
        """按 by_field 命中行并取系数；pick 为字段名或 [低值字段, 高值字段]（区间内线性内插）。

        value 为字符串时按等值匹配（如硐室类型 matching room_type），为数值时按区间匹配。
        """
        low_field, high_field = self.table_bounds(name, by_field)
        row = self._match_row(self.table(name), by_field, low_field, high_field, value)
        if row is None:
            return None, None
        if isinstance(pick, (list, tuple)) and len(pick) == 2:
            low, high = row.get(pick[0]), row.get(pick[1])
            if low is None or high is None:
                return None, row
            if not isinstance(value, (int, float)):
                return float(low), row
            span_low, span_high = row.get(low_field), row.get(high_field)
            if None in (span_low, span_high) or span_high == span_low:
                return float(low), row
            weight = min(max((float(value) - float(span_low)) / (float(span_high) - float(span_low)), 0.0), 1.0)
            return round(float(low) + (float(high) - float(low)) * weight, 4), row
        picked = row.get(str(pick))
        if picked is None:
            return None, row
        return float(picked), row

    @staticmethod
    def _match_row(
        rows: list[dict[str, Any]],
        by_field: str,
        low_field: str,
        high_field: str,
        value: Any,
    ) -> dict[str, Any] | None:
        if not isinstance(value, (int, float)):
            wanted = str(value)
            for row in rows:
                if str(row.get(by_field)) == wanted:
                    return row
            for row in rows:
                if wanted and wanted in str(row.get(by_field) or ""):
                    return row
            return None
        numeric = float(value)
        for row in rows:
            low, high = row.get(low_field), row.get(high_field)
            if low is not None and numeric < float(low):
                continue
            if high is not None and numeric >= float(high):
                continue
            return row
        # 落在最后一档之外（例如超过最高温度），按边界档处理并留痕
        for row in reversed(rows):
            low = row.get(low_field)
            if low is None or numeric >= float(low):
                return row
        return rows[0] if rows else None

    # ---- 公式 -------------------------------------------------------------
    def formulas(self, location_type: str) -> list[dict[str, Any]]:
        return list((self._data.get("formulas") or {}).get(location_type) or [])

    def formulas_for_target(self, target_type: str) -> list[dict[str, Any]]:
        return list((self._data.get("formulas") or {}).get(target_type) or [])

    @staticmethod
    def select_formula(formula: dict[str, Any], facts: Mapping[str, Any]) -> dict[str, Any]:
        """按事实选择公式变体；返回 {expression, variant, note}。"""
        if formula.get("applies_to") and not _when_matches(formula["applies_to"], facts):
            return {"expression": None, "variant": None, "reason": "公式适用范围不匹配"}
        selected = {"expression": formula.get("expression"), "variant": None, "note": formula.get("note", "")}
        for variant in formula.get("variants") or []:
            if _when_matches(variant.get("when") or {}, facts):
                selected = {"expression": variant.get("expression"), "variant": variant.get("when"),
                            "note": variant.get("note", formula.get("note", ""))}
                break
        return selected

    def verification(self, location_type: str) -> list[dict[str, Any]]:
        return list((self._data.get("verification") or {}).get(location_type) or [])

    # ---- 最低风量与最大风速 ----------------------------------------------
    def min_airflow(self, location_type: str, facts: Mapping[str, Any]) -> dict[str, Any] | None:
        for entry in (self._data.get("min_airflow") or {}).get(location_type) or []:
            if _when_matches(entry.get("when") or {}, facts):
                return entry
        return None

    def max_air_velocity(self, location_type: str, tag: str | None = None) -> dict[str, Any] | None:
        table = self._data.get("max_air_velocity") or {}
        if tag and f"{location_type}:{tag}" in table:
            return table[f"{location_type}:{tag}"]
        return table.get(location_type)

    # ---- 阈值 -------------------------------------------------------------
    def thresholds(self) -> dict[str, Any]:
        return dict(self._data.get("thresholds") or {})

    def parameter_markers(self) -> dict[str, dict[str, Any]]:
        """列出会影响公式选择的"口径标记"及其允许取值。

        这些标记（如 section_area_basis、roadway_type、explosive_grade）决定用哪个公式
        变体，取错值会静默算出错误结果，因此必须把允许取值明确告诉提取方——否则模型会
        写一个同义但不同的说法（实测遇到过把"有效断面积"写成"平均有效净断面积"），
        导致变体不命中、复算出与计划不符的值，把正确计划误判成计算错误。
        """
        markers: dict[str, dict[str, Any]] = {}
        for bucket in (self._data.get("formulas") or {}):
            for formula in self.formulas(bucket):
                for key, values in (formula.get("marker_values") or {}).items():
                    entry = markers.setdefault(key, {"allowed": [], "affects": []})
                    for value in values:
                        if value not in entry["allowed"]:
                            entry["allowed"].append(value)
                    if formula["id"] not in entry["affects"]:
                        entry["affects"].append(formula["id"])
                for variant in formula.get("variants") or []:
                    for key, expected in (variant.get("when") or {}).items():
                        entry = markers.setdefault(key, {"allowed": [], "affects": []})
                        for value in (expected if isinstance(expected, (list, tuple)) else [expected]):
                            if value not in entry["allowed"]:
                                entry["allowed"].append(value)
                        if formula["id"] not in entry["affects"]:
                            entry["affects"].append(formula["id"])
                # 表驱动的标记：允许取值就是该表那一列的取值集合
                for spec in (formula.get("params") or {}).values():
                    derive = (spec or {}).get("derive") or {}
                    if "table" not in derive or not derive.get("by"):
                        continue
                    try:
                        rows = self.table(derive["table"])
                    except RuleError:
                        continue
                    values = [row.get(derive["by"]) for row in rows]
                    values = [value for value in values if value is not None]
                    if not values:
                        continue
                    entry = markers.setdefault(derive["by"], {"allowed": [], "affects": []})
                    for value in values:
                        if value not in entry["allowed"]:
                            entry["allowed"].append(value)
                    entry["table"] = derive["table"]
        for entry in markers.values():
            entry["note"] = ("取值必须严格来自 allowed；确实无法判断时留 null——"
                             "工具会用计划自己写出的结果反推它采用的是哪个变体")
        return markers

    def tolerance(self, override: Mapping[str, Any] | None = None) -> dict[str, float]:
        base = dict((self.thresholds().get("tolerance") or {}))
        base.pop("note", None)
        for key, value in (override or {}).items():
            if isinstance(value, (int, float)):
                base[key] = float(value)
        base.setdefault("absolute_m3_min", 1.0)
        base.setdefault("relative_percent", 0.5)
        return base

    def compare_policy(self, target_type: str) -> dict[str, Any]:
        policies = self.thresholds().get("compare") or {}
        if target_type not in policies:
            raise RuleError(f"规则库缺少比对阈值策略：{target_type}")
        return dict(policies[target_type])

    def reconcile_policy(self) -> dict[str, Any]:
        return dict(self.thresholds().get("reconcile") or {})


def _when_matches(when: Mapping[str, Any], facts: Mapping[str, Any]) -> bool:
    for key, expected in when.items():
        actual = facts.get(key)
        if isinstance(expected, (list, tuple)):
            if actual not in expected:
                return False
        elif actual != expected:
            return False
    return True


def _rules_path(rule_version: str) -> Path:
    return RULES_DIR / f"airflow-rules-{rule_version.replace('AIR-RULE-', '')}.json"


MASTER_DATA_FILE = RULES_DIR / "master_data.json"


@lru_cache(maxsize=1)
def load_master_data() -> dict[str, Any]:
    """加载主数据（地点分类、别名、回风系统标识）；缺失时返回空壳而非报错。"""
    if not MASTER_DATA_FILE.is_file():
        return {"version": "MASTER-未配置", "location_classification": [], "aliases": {}}
    return json.loads(MASTER_DATA_FILE.read_text(encoding="utf-8"))


def classify_location(name: str, master: Mapping[str, Any] | None = None) -> tuple[str, str]:
    """按主数据把地点名判成地点类型；命中不到返回 ("", 原因)，由调用方转待复核。"""
    data = master if master is not None else load_master_data()
    text = (name or "").strip()
    if not text:
        return "", "地点名为空"
    rules = sorted(data.get("location_classification") or [], key=lambda item: item.get("priority", 99))
    for rule in rules:
        for keyword in rule.get("match") or []:
            if keyword and keyword in text:
                return str(rule.get("location_type") or ""), f"命中关键词「{keyword}」({rule.get('id')})"
    return "", "未命中任何地点分类规则，需人工确认"


def normalize_location_name(name: str, master: Mapping[str, Any] | None = None) -> str:
    """别名归一：去掉空白与全角空格，保留原文以便回溯。"""
    text = (name or "").replace("\u3000", " ").strip()
    for _ in range(4):
        text = text.replace("  ", " ")
    return text


def available_versions() -> list[str]:
    return sorted(path.stem.replace("airflow-rules-", "AIR-RULE-") for path in RULES_DIR.glob("airflow-rules-*.json"))


@lru_cache(maxsize=8)
def load_rules(rule_version: str | None = None, effective_date: Any = None) -> RuleSet:
    """加载规则库；effective_date 用于校验规则在目标时间窗内有效。"""
    version = rule_version if rule_version and rule_version != "auto" else RULE_VERSION_DEFAULT
    path = _rules_path(version)
    if not path.is_file():
        raise RuleError(f"规则版本不可用：{version}；可选 {available_versions()}")
    data = json.loads(path.read_text(encoding="utf-8"))
    rules = RuleSet(data, source=path.name)
    if effective_date and not rules.is_effective_at(effective_date):
        raise RuleError(f"规则版本 {version} 在 {effective_date} 不生效，需人工确认适用版本")
    return rules


def derive_params(
    formula: dict[str, Any],
    facts: Mapping[str, Any],
    rules: RuleSet,
) -> tuple[dict[str, Any], list[str], list[dict[str, Any]]]:
    """按公式的参数声明补齐派生参数。

    返回 (可用参数, 缺失参数名, 派生过程)。派生过程要写进计算链，让人能看出
    每个系数是从哪张表、按哪个事实取的——这是"结论可追溯"的一半。
    """
    params = dict(facts)
    missing: list[str] = []
    trace: list[dict[str, Any]] = []
    for name, spec in (formula.get("params") or {}).items():
        if params.get(name) is not None:
            continue
        spec = spec or {}
        default_from = spec.get("default_from")
        if default_from and default_from.startswith("constants."):
            params[name] = rules.constant(default_from.split(".", 1)[1])
            trace.append({"param": name, "from": default_from, "value": params[name]})
            continue
        derive = spec.get("derive") or {}
        if "table" in derive:
            by_field = derive.get("by")
            driver = params.get(by_field)
            if driver is None:
                missing.append(name)
                continue
            value, row = rules.lookup_table(derive["table"], by_field, driver, derive.get("pick"))
            if value is None:
                missing.append(name)
                continue
            params[name] = value
            trace.append({"param": name, "table": derive["table"],
                          "table_title": rules.table_title(derive["table"]),
                          "by": by_field, "driver": driver, "value": value, "row": row})
            continue
        if "expression" in derive:
            value, reason = try_evaluate(derive["expression"], params)
            if value is None:
                missing.append(name)
                continue
            params[name] = round(value, 4)
            trace.append({"param": name, "expression": derive["expression"], "value": params[name]})
            continue
        if name in (formula.get("requires") or []):
            missing.append(name)
    for name in formula.get("requires") or []:
        if params.get(name) is None and name not in missing:
            missing.append(name)
    return params, missing, trace