vent_agent_2.0/extensions/main_fan/targets.py

"""服务端目标目录：将业务名称解析为真实身份，技术参数不交给模型猜测。"""
import json
from pathlib import Path
import re
import unicodedata

from pydantic import BaseModel, ConfigDict, Field

import config
from algorithms.main_fan.analysis import fan_numbers


class TargetResolutionError(ValueError):
    """需要用户明确业务对象，不能继续取数。"""


class FanTarget(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    mine_code: str = Field(min_length=1)
    mine_name: str = Field(min_length=1)
    fan_id: str = Field(min_length=1)
    fan_name: str = Field(min_length=1)
    mine_aliases: list[str] = Field(default_factory=list)
    aliases: list[str] = Field(default_factory=list)
    fan_pair: list[str] = Field(default_factory=list, max_length=2)


class TargetCatalog(BaseModel):
    model_config = ConfigDict(extra="forbid")
    gateway_url: str = ""
    targets: list[FanTarget] = Field(default_factory=list, max_length=500)


def load_catalog():
    filename = config.get_str("MAIN_FAN_TARGETS_FILE")
    if not filename:
        return TargetCatalog()
    path = Path(filename)
    if not path.is_absolute():
        path = config.ROOT / path
    catalog = TargetCatalog.model_validate(json.loads(path.read_text(encoding="utf-8")))
    if catalog.gateway_url and catalog.gateway_url.rstrip("/") != config.get_str("SHANXI_GATEWAY_URL").rstrip("/"):
        raise ValueError("主通风机目标目录绑定的网关与当前配置不一致，请配置对应环境的目标目录")
    identities = [(t.mine_code, t.fan_id) for t in catalog.targets]
    if len(set(identities)) != len(identities):
        raise ValueError("主通风机目标目录存在重复身份")
    return catalog


def _key(value):
    value = re.sub(r"\s+", "", unicodedata.normalize("NFKC", value)).casefold()
    value = re.sub(r"^(?:(?:请|帮我|麻烦|对|审核|审查|检查|研判|检测|评估|分析|一下))+", "", value)
    value = re.sub(r"(?:进行)?(?:审核|审查|检查|研判|检测|评估|分析)(?:一下)?[。.!！?？]*$", "", value)
    value = value.strip("。.!！?？:：，,").replace("的", "")
    for cn, number in {"一":"1", "二":"2", "三":"3", "四":"4", "五":"5",
                       "六":"6", "七":"7", "八":"8", "九":"9", "十":"10"}.items():
        value = value.replace(cn+"号", number+"号")
    value = re.sub(r"(\d+)#", r"\1号", value)
    return re.sub("主通风机|通风机|主扇|风机", "风机", value)


def _fan_aliases(target):
    names = [target.fan_name]
    numbers = fan_numbers(target.fan_name)
    if len(numbers) == 1:
        number = next(iter(numbers))
        names.extend([f"{number}号", f"{number}号通风机", f"主通{number}号", f"主通{number}号风机"])
    return names


def _matches(target, selector):
    mines = [target.mine_name, *target.mine_aliases]
    fans = _fan_aliases(target)
    names = [target.fan_id, *target.aliases, *fans, *mines, "主通风机",
             *(mine+"主通风机" for mine in mines),
             *(mine+fan for mine in mines for fan in fans)]
    return _key(selector) in {_key(name) for name in names if name}


def resolve_request(raw):
    if isinstance(raw, str) and raw.lstrip().startswith("{"):
        try:
            raw = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise TargetResolutionError("目标输入的JSON格式不完整，请传目标名称或target对象。") from exc
    if isinstance(raw, str):
        data = {"target": raw}
    elif isinstance(raw, dict):
        data = dict(raw)
    else:
        raise TargetResolutionError("请说明要审核的矿井和主通风机。")
    for canonical, alias in (("mine_code","mineCode"),("mine_name","mineName"),("fan_id","fanId")):
        if alias in data:
            if canonical in data and data[canonical] != data[alias]:
                raise TargetResolutionError("输入的矿井或风机身份存在冲突，请确认业务对象。")
            data[canonical] = data.pop(alias)
    identity_fields = ("mine_code", "mine_name", "fan_id", "fan_name")
    if any(key in data and not isinstance(data[key], str) for key in identity_fields):
        raise TargetResolutionError("矿井与风机身份必须为文字，请确认业务对象。")
    provided = {key: value.strip() for key in identity_fields if isinstance((value := data.get(key)), str) and value.strip()}
    selector = data.get("target", "")
    if not isinstance(selector, str):
        raise TargetResolutionError("目标名称必须为文字。")
    selector = selector.strip()
    compact = bool(selector) or not (provided.get("mine_code") and provided.get("mine_name"))
    if compact:
        catalog = load_catalog()
        candidates = [t for t in catalog.targets if not selector or _matches(t, selector)]
        if not selector and not provided:
            candidates = []
        for key, value in provided.items():
            if key in {"mine_code", "fan_id"}:
                candidates = [t for t in candidates if getattr(t, key) == value]
            elif key == "mine_name":
                candidates = [t for t in candidates if _key(value) in {_key(n) for n in [t.mine_name,*t.mine_aliases]}]
            else:
                candidates = [t for t in candidates if _key(value) in {_key(n) for n in _fan_aliases(t)}]
        if len(candidates) > 1:
            pairs = {(t.mine_code, tuple(sorted(t.fan_pair))) for t in candidates if len(t.fan_pair) == 2}
            if len(pairs) == 1 and all(len(t.fan_pair) == 2 for t in candidates):
                candidates = [sorted(candidates, key=lambda t: t.fan_id)[0]]
            else:
                names = "；".join(t.mine_name+" "+t.fan_name for t in candidates[:8])
                raise TargetResolutionError("请明确要审核哪组主通风机："+names)
        if not candidates:
            raise TargetResolutionError("未唯一匹配已配置的矿井主通风机，请确认矿井名称和风机编号；新增矿井需管理员补充身份映射。")
        target = candidates[0]
        data.update({key: getattr(target, key) for key in identity_fields})
        if target.fan_pair and not data.get("fan_pair"):
            data["fan_pair"] = target.fan_pair
        if "use_gateway" not in data:
            supplied = any(data.get(key) for key in ("materials", "reports", "devices", "points"))
            data["use_gateway"] = config.get_bool("SHANXI_GATEWAY_ENABLED", False) and not supplied
        if "document_cache_policy" not in data:
            data["document_cache_policy"] = config.get_str("MAIN_FAN_DEFAULT_DOCUMENT_CACHE_POLICY", "refresh")
    if not data.get("mine_code") or not data.get("mine_name"):
        raise TargetResolutionError("尚未确定矿井身份，请明确矿井名称和主通风机。")
    return data