vent_agent_2.0/algorithms/airflow_review/engine/matching.py

"""地点名称匹配：把测风报表的实测地点与配风计划的用风地点对应起来。

这是"参数一致性规则"（§2.6 第 10 条）的落地点：计划值、实测值、复算值必须指向
同一对象。逐字符相等几乎不可能（计划写"42 上208 主运顺槽工作面局扇安设点"，
报表写"42上208主运顺槽风桥段回风"），因此按三级策略逐步放宽：

1. 归一化后完全相等；
2. 去掉测点后缀（回风/入风/上风侧/下风侧/测点/站）后的核心名相等；
3. 一方包含另一方核心名，且核心名长度达到门槛。

匹配不上的**不许猜**：进 unmatched 并转人工复核（§2.8 第 1 条）。
"""
from __future__ import annotations

import re
from typing import Any, Mapping

from algorithms.airflow_review.rules.loader import load_master_data

# 测点位置与口径后缀：报表常在名称后加这些词，计划里换一种写法
# （如报表"明安木独回风立井总回风" vs 计划"明安木独回风立井回风系统"）。
_MEASUREMENT_SUFFIX = ("回风上风侧", "回风下风侧", "风桥段回风上风侧", "风桥段回风", "回风系统",
                       "总回风", "回风", "主要通风机", "主扇",
                       "入风", "进风", "上风侧", "下风侧", "测点", "测风站", "站")
_MIN_CONTAINMENT_LEN = 6
_NOISE = re.compile(r"[\s\u3000（）()、,，.。\-—_/]")


def normalize_name(name: Any) -> str:
    """归一化：去空白与标点、统一全角数字、压缩"综采（放）"之类写法差异。"""
    text = _NOISE.sub("", str(name or "").lower())
    text = text.replace("（", "").replace("）", "")
    return text


def core_name(name: Any) -> str:
    """核心名：去掉测点位置后缀，用于"同一地点的不同测点"归并。"""
    text = normalize_name(name)
    changed = True
    while changed:
        changed = False
        for suffix in _MEASUREMENT_SUFFIX:
            if text.endswith(suffix) and len(text) > len(suffix) + 2:
                text = text[: -len(suffix)]
                changed = True
    return text


def match_locations(
    dataset_locations: list[Mapping[str, Any]],
    verified_items: list[Mapping[str, Any]],
    *,
    aliases: Mapping[str, list[str]] | None = None,
) -> dict[str, Any]:
    """返回匹配对与未匹配清单；每个实测地点最多匹配一个计划对象。"""
    alias_map = _alias_index(aliases or (load_master_data().get("aliases") or {}))
    plan_index: dict[str, list[dict[str, Any]]] = {}
    core_index: dict[str, list[dict[str, Any]]] = {}
    for item in verified_items:
        target_id = str(item.get("target_id") or "")
        if not target_id:
            continue
        entry = dict(item)
        for key in _plan_keys(entry, alias_map):
            plan_index.setdefault(key, []).append(entry)
        core = core_name(entry.get("target_name") or target_id)
        if core:
            core_index.setdefault(core, []).append(entry)

    pairs: list[dict[str, Any]] = []
    unmatched_dataset: list[dict[str, Any]] = []
    used: set[str] = set()
    for location in dataset_locations:
        name = location.get("location_name")
        found, strategy = _lookup(name, plan_index, core_index, alias_map, used)
        if found is None:
            unmatched_dataset.append({
                "location_id": location.get("location_id"),
                "location_name": name,
                "airflow_m3_min": location.get("airflow_m3_min"),
                "reason": "在核验通过的配风计划对象中找不到对应地点",
                "action": "请人工确认该测点对应计划中的哪个用风地点",
            })
            continue
        used.add(str(found.get("target_id")))
        pairs.append({
            "location_id": location.get("location_id"),
            "location_name": name,
            "location_type": location.get("location_type"),
            "measurement_point": location.get("measurement_point"),
            "actual_airflow_m3_min": location.get("airflow_m3_min"),
            "target_id": found.get("target_id"),
            "target_name": found.get("target_name"),
            "target_type": found.get("target_type"),
            "matched_by": strategy,
            "match_confidence": 1.0 if strategy == "exact" else (0.9 if strategy == "core" else 0.75),
        })
    unmatched_plan = [dict(item) for item in verified_items
                      if str(item.get("target_id")) not in used
                      and str(item.get("target_type")) == "workplace"]
    return {
        "pairs": pairs,
        "unmatched_dataset": unmatched_dataset,
        "unmatched_plan_workplaces": unmatched_plan,
        "matched_count": len(pairs),
        "match_rate": round(len(pairs) / len(dataset_locations), 4) if dataset_locations else 0.0,
    }


def match_return_systems(
    dataset_systems: list[Mapping[str, Any]],
    verified_items: list[Mapping[str, Any]],
) -> dict[str, Any]:
    """回风井 ↔ 回风系统匹配：回风系统以回风井为单位，按核心名包含关系对应。"""
    plan_systems = [dict(item) for item in verified_items
                    if str(item.get("target_type")) == "return_air_system"]
    pairs: list[dict[str, Any]] = []
    unmatched: list[dict[str, Any]] = []
    used: set[str] = set()
    for system in dataset_systems:
        name = core_name(system.get("location_name"))
        found = None
        for item in plan_systems:
            candidate = core_name(item.get("target_name"))
            if not candidate or not name:
                continue
            if candidate == name or candidate in name or name in candidate:
                found = item
                break
        if found is None:
            unmatched.append({"location_name": system.get("location_name"),
                              "reason": "该回风井在核验通过的回风系统中找不到对应对象",
                              "action": "请人工确认回风井与回风系统的对应关系"})
            continue
        used.add(str(found.get("target_id")))
        pairs.append({
            "location_name": system.get("location_name"),
            "actual_airflow_m3_min": system.get("airflow_m3_min"),
            "target_id": found.get("target_id"),
            "target_name": found.get("target_name"),
            "target_type": "return_air_system",
        })
    return {
        "pairs": pairs,
        "unmatched_dataset": unmatched,
        "unmatched_plan_systems": [item for item in plan_systems
                                   if str(item.get("target_id")) not in used],
    }


def _lookup(
    name: Any,
    plan_index: Mapping[str, list[dict[str, Any]]],
    core_index: Mapping[str, list[dict[str, Any]]],
    alias_map: Mapping[str, list[str]],
    used: set[str],
) -> tuple[dict[str, Any] | None, str]:
    keys = _plan_keys({"target_name": name}, alias_map)
    for key in keys:
        for candidate in plan_index.get(key, []):
            if str(candidate.get("target_id")) not in used:
                return candidate, "exact"
    core = core_name(name)
    if not core:
        return None, ""
    for candidate in core_index.get(core, []):
        if str(candidate.get("target_id")) not in used:
            return candidate, "core"
    if len(core) >= _MIN_CONTAINMENT_LEN:
        for candidate_core, candidates in core_index.items():
            if not candidate_core or len(candidate_core) < _MIN_CONTAINMENT_LEN:
                continue
            if core in candidate_core or candidate_core in core:
                for candidate in candidates:
                    if str(candidate.get("target_id")) not in used:
                        return candidate, "containment"
    return None, ""


def _plan_keys(item: Mapping[str, Any], alias_map: Mapping[str, list[str]]) -> list[str]:
    names = [item.get("target_name") or ""]
    for name in list(names):
        names.extend(alias_map.get(normalize_name(name), []))
    return [normalize_name(name) for name in names if name]


def _alias_index(aliases: Mapping[str, Any]) -> dict[str, list[str]]:
    """把别名表展开成 归一化名 → 等价写法 的索引，两个方向都建。"""
    index: dict[str, list[str]] = {}
    for canonical, variants in aliases.items():
        if canonical in ("note", "示例") or not isinstance(variants, list):
            continue
        group = [canonical, *[str(item) for item in variants]]
        for name in group:
            key = normalize_name(name)
            bucket = index.setdefault(key, [])
            for other in group:
                if other != name and other not in bucket:
                    bucket.append(other)
    return index