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

"""层级关系校验：用风地点 → 回风系统 → 矿井总需风量（§2.6 第 4、5 条）。

只做结构与归属判断，不算风量：遗漏归属、重复归属、循环引用、重复汇总、
汇总关系缺失，以及单一回风系统与矿井总需风量的等值关系识别。
"""
from __future__ import annotations

from typing import Any, Mapping

from algorithms.airflow_review.contracts import (
    TARGET_RETURN_SYSTEM,
    TARGET_WORKPLACE,
    as_list,
    text,
    unique,
)

LEVEL_WORKPLACE = "workplace"
LEVEL_RETURN = "return_air_system"
LEVEL_MINE = "mine_total"

_LEVEL_BY_TYPE = {
    TARGET_WORKPLACE: LEVEL_WORKPLACE,
    TARGET_RETURN_SYSTEM: LEVEL_RETURN,
    "mine_total": LEVEL_MINE,
}


def build_hierarchy(items: list[Mapping[str, Any]]) -> dict[str, Any]:
    """建立层级模型并返回结构问题清单。"""
    nodes: dict[str, dict[str, Any]] = {}
    duplicates: set[str] = set()
    for item in items:
        target_id = text(item.get("target_id"))
        if not target_id:
            continue
        if target_id in nodes:
            duplicates.add(target_id)
            continue
        target_type = text(item.get("target_type"))
        nodes[target_id] = {
            "target_id": target_id,
            "target_name": item.get("target_name") or target_id,
            "target_type": target_type,
            "level": _LEVEL_BY_TYPE.get(target_type, ""),
            "parent_id": text(item.get("parent_return_system_id") or item.get("parent_mine_total_id")
                              or item.get("parent_target_id")),
            "component_ids": [text(cid) for cid in _component_ids(item)],
            "planned_required_airflow_m3_min": item.get("planned_required_airflow_m3_min"),
        }

    issues: list[dict[str, Any]] = []
    for target_id in sorted(duplicates):
        issues.append(_issue("duplicate_target_id", "major", target_id,
                             f"同一对象在计划中出现多次：{target_id}，无法确定唯一需风量"))

    # 归属校验：每个子对象必须且只能归属一个父对象
    parent_claims: dict[str, set[str]] = {}
    for node in nodes.values():
        if not node["parent_id"]:
            continue
        parent_claims.setdefault(node["parent_id"], set()).add(node["target_id"])
    for node in nodes.values():
        if node["level"] == LEVEL_MINE:
            continue
        parent_id = node["parent_id"]
        if not parent_id:
            issues.append(_issue("orphan_target", "major", node["target_id"],
                                 f"{node['target_name']} 未标明所属"
                                 f"{'回风系统' if node['level'] == LEVEL_WORKPLACE else '矿井总需风量'}，无法归属"))
            continue
        parent = nodes.get(parent_id)
        if parent is None:
            issues.append(_issue("missing_parent", "major", node["target_id"],
                                 f"{node['target_name']} 的上级对象 {parent_id} 在计划中不存在"))
            continue
        if parent["level"] == node["level"]:
            issues.append(_issue("level_conflict", "major", node["target_id"],
                                 f"{node['target_name']} 与上级 {parent_id} 处于同一层级"))
        if node["target_id"] not in parent["component_ids"]:
            issues.append(_issue("missing_membership", "general", node["target_id"],
                                 f"{node['target_name']} 声明归属 {parent_id}，但该上级未将其列入组成对象"))

    for parent_id, children in parent_claims.items():
        owner = nodes.get(parent_id)
        if owner is None:
            continue
        for child_id in sorted(children):
            claims = [pid for pid, cids in parent_claims.items() if child_id in cids]
            if len(claims) > 1:
                issues.append(_issue("duplicate_membership", "major", child_id,
                                     f"对象 {child_id} 被 {('/'.join(sorted(claims)))} 重复归属"))

    # 汇总关系校验
    for node in nodes.values():
        if node["level"] == LEVEL_WORKPLACE:
            continue
        component_ids = node["component_ids"]
        if not component_ids:
            issues.append(_issue("missing_aggregation", "major", node["target_id"],
                                 f"{node['target_name']} 未列出组成对象，无法校验汇总关系"))
        seen: dict[str, int] = {}
        for child_id in component_ids:
            seen[child_id] = seen.get(child_id, 0) + 1
        for child_id, count in sorted(seen.items()):
            if count > 1:
                issues.append(_issue("duplicate_in_aggregation", "major", node["target_id"],
                                     f"{node['target_name']} 重复列入 {child_id} {count} 次，汇总会重复计算"))
            if child_id not in nodes:
                issues.append(_issue("unknown_component", "general", node["target_id"],
                                     f"{node['target_name']} 列出的组成对象 {child_id} 未在计划中定义"))

    cycles = _find_cycles(nodes)
    for cycle in cycles:
        issues.append(_issue("cyclic_reference", "major", cycle[0],
                             "层级存在循环引用：" + " → ".join(cycle)))

    return_systems = [node for node in nodes.values() if node["level"] == LEVEL_RETURN]
    mine_totals = [node for node in nodes.values() if node["level"] == LEVEL_MINE]
    if not return_systems:
        issues.append(_issue("no_return_system", "major", "",
                             "计划未列出任何回风系统，层级关系不成立"))
    if not mine_totals:
        issues.append(_issue("no_mine_total", "major", "", "计划未列出矿井总需风量"))

    equivalence = _single_return_equivalence(return_systems, mine_totals)
    return {
        "nodes": nodes,
        "roots": [node["target_id"] for node in mine_totals],
        "levels": {
            LEVEL_WORKPLACE: sorted(node["target_id"] for node in nodes.values() if node["level"] == LEVEL_WORKPLACE),
            LEVEL_RETURN: sorted(node["target_id"] for node in return_systems),
            LEVEL_MINE: sorted(node["target_id"] for node in mine_totals),
        },
        "issues": issues,
        "single_return_system": len(return_systems) == 1,
        "single_return_equivalence": equivalence,
        "blocking_targets": unique([issue["target_id"] for issue in issues
                                    if issue["severity"] == "major" and issue["target_id"]]),
    }


def ancestors_of(target_id: str, nodes: Mapping[str, Mapping[str, Any]]) -> list[str]:
    """向上追溯到矿井总需风量，用于"计算错误影响上层对象"的阻断传播。"""
    chain: list[str] = []
    seen: set[str] = set()
    current = target_id
    while current and current not in seen:
        seen.add(current)
        node = nodes.get(current)
        if node is None:
            break
        parent = text(node.get("parent_id"))
        if not parent or parent in seen:
            break
        chain.append(parent)
        current = parent
    return chain


def _single_return_equivalence(
    return_systems: list[dict[str, Any]],
    mine_totals: list[dict[str, Any]],
) -> dict[str, Any]:
    """单一回风系统时，回风系统需风量与矿井总需风量应等值并可复用同一基准。"""
    if len(return_systems) != 1 or len(mine_totals) != 1:
        return {"applicable": False, "equivalent": False, "reason": "非单一回风系统，不适用等值关系"}
    system_value = return_systems[0].get("planned_required_airflow_m3_min")
    mine_value = mine_totals[0].get("planned_required_airflow_m3_min")
    tail = return_systems[0].get("component_ids") or []
    if not tail:
        return {"applicable": True, "equivalent": False,
                "return_system_id": return_systems[0]["target_id"],
                "mine_total_id": mine_totals[0]["target_id"],
                "system_value": system_value, "mine_value": mine_value,
                "reason": "单一回风系统，但矿井总需风量未将回风系统列为组成对象"}
    if isinstance(system_value, (int, float)) and isinstance(mine_value, (int, float)):
        equivalent = abs(float(system_value) - float(mine_value)) <= max(1.0, abs(float(mine_value)) * 0.005)
    else:
        equivalent = False
    return {
        "applicable": True,
        "equivalent": equivalent,
        "return_system_id": return_systems[0]["target_id"],
        "mine_total_id": mine_totals[0]["target_id"],
        "system_value": system_value,
        "mine_value": mine_value,
        "reuse_key": f"reuse:{mine_totals[0]['target_id']}" if equivalent else "",
        "reason": "单一回风系统，需风量与矿井总需风量等值，可复用同一基准"
        if equivalent else "单一回风系统，但回风系统需风量与矿井总需风量的计划值不一致",
    }


def _component_ids(item: Mapping[str, Any]) -> list[str]:
    """原样保留组成对象列表：这里不能去重，否则"重复汇总"永远检测不出来。"""
    return [text(cid) for cid in as_list(
        item.get("component_workplace_ids") or item.get("component_return_system_ids")
        or item.get("component_target_ids"))]


def _find_cycles(nodes: Mapping[str, Mapping[str, Any]]) -> list[list[str]]:
    cycles: list[list[str]] = []
    reported: set[frozenset[str]] = set()
    for start in nodes:
        path: list[str] = []
        current = start
        while current and current in nodes:
            if current in path:
                cycle = path[path.index(current):] + [current]
                key = frozenset(cycle)
                if key not in reported:
                    reported.add(key)
                    cycles.append(cycle)
                break
            path.append(current)
            current = text(nodes[current].get("parent_id"))
    return cycles


def _issue(code: str, severity: str, target_id: str, message: str) -> dict[str, Any]:
    return {"code": code, "severity": severity, "target_id": target_id, "message": message}