vent_deep_agent/tools/vent_plan_tools.py

# -*- coding: utf-8 -*-
"""
配风计划审查 - MCP 工具模块

通过 MCP 客户端查询外部数据源，供配风计划审查智能体（数据一致性审查）调用。

MCP 服务端提供的工具：
- query_monthly_mining_plan:      月度采掘计划查询
- query_working_face_procedure:    工作面作业规程查询
- query_gas_identification_report: 瓦斯等级鉴定报告查询
- query_ventilation_report:        测风报表查询

工具列表：
- query_mining_plan:        查询煤矿月度采掘计划
- query_face_procedure:     查询工作面作业规程
- query_gas_report:         查询瓦斯等级鉴定报告
- query_wind_report:        查询测风报表
"""

import json
from fastmcp import Client
from tools.mcp_logger import log_mcp_call


# ============================================================
# MCP 客户端辅助函数
# ============================================================

def _get_mcp_url() -> str:
    """获取 MCP 服务 URL。

    从项目根目录 .env 读取 MCP_BASE_URL 配置，拼接 /mcp 路径。
    """
    from dotenv import dotenv_values
    from pathlib import Path as _Path

    _env_file = _Path(__file__).parent.parent / ".env"
    _cfg = dotenv_values(str(_env_file))
    mcp_url = _cfg.get("MCP_BASE_URL", "http://localhost:8100").rstrip("/")
    return f"{mcp_url}/mcp"


async def _call_mcp_tool(tool_name: str, arguments: dict) -> str:
    """通过 FastMCP 异步客户端调用远程 MCP 工具。

    Args:
        tool_name: MCP 工具名称
        arguments: 工具参数字典

    Returns:
        JSON 字符串格式的查询结果
    """
    mcp_url = _get_mcp_url()
    try:
        client = Client(mcp_url)
        async with client:
            result = await client.call_tool(tool_name, arguments)
            raw = result.content[0].text
            log_mcp_call(tool_name, arguments, raw)
            return raw
    except Exception as e:
        log_mcp_call(tool_name, arguments, "", error=str(e))
        import traceback
        traceback.print_exc()
        return json.dumps(
            {"error": f"MCP调用失败: {e}", "tool": tool_name, "args": arguments},
            ensure_ascii=False,
        )


# ============================================================
# 配风计划审查 MCP 工具
# ============================================================

def _normalize_plan_month(raw: str) -> str:
    """规范化计划月份为 yyyy-MM 格式（匹配 MCP 服务端要求）。

    支持的输入格式：
    - "2026-6" / "2026-06" → "2026-06"
    - "2026年6月" → "2026-06"
    - "2026/6" → "2026-06"
    - "2026.6" → "2026-06"

    Args:
        raw: LLM 从 PDF 中提取的原始月份字符串

    Returns:
        str: yyyy-MM 格式的计划月份
    """
    import re
    raw = raw.strip().replace(" ", "")
    # 匹配 "2026年6月" 格式
    m = re.match(r"(\d{4})\s*年\s*(\d{1,2})\s*月?", raw)
    if m:
        return f"{m.group(1)}-{int(m.group(2)):02d}"
    # 匹配 "2026-6" / "2026-06" / "2026/6" / "2026.6" 格式
    m = re.match(r"(\d{4})\s*[-/\.]\s*(\d{1,2})", raw)
    if m:
        return f"{m.group(1)}-{int(m.group(2)):02d}"
    # 无法识别则原样返回（MCP 服务端做模糊匹配）
    return raw


# ============================================================
# 煤矿名称模糊重试机制
# ============================================================

def _generate_name_variations(mine_name: str, max_variations: int = 3) -> list:
    """生成煤矿名称的渐进放宽变体列表，用于 MCP 查询重试。

    典型场景：智能体从 PDF 中提取的矿名与数据库中的矿名存在细微差异，
    如 "合阳县金桥煤炭有限责任公司"（PDF 提取） vs "合阳县金桥煤炭有限公司"（DB 记录）。

    变体优先级（从最精确到最宽松）：
    Level 0: 原始名称
    Level 1: 替换「有限责任公司」→「有限公司」（最常见的不匹配模式）
    Level 2: 去除所有企业后缀（有限责任/股份/有限/公司）
    Level 3: 进一步去除行业词汇（煤炭/煤业/矿业/煤矿），仅保留地名+核心名

    Args:
        mine_name: 原始煤矿名称
        max_variations: 最大变体数（不含原始名称），默认 3

    Returns:
        list[str]: 名称变体列表，第一个元素始终是原始名称
    """
    import re

    variations = [mine_name.strip()]
    if not mine_name or len(mine_name.strip()) < 3:
        return variations

    name = mine_name.strip()

    # ── Level 1: 替换「有限责任公司」为「有限公司」──
    if "有限责任公司" in name:
        v = name.replace("有限责任公司", "有限公司")
        if v != name and v not in variations:
            variations.append(v)

    if len(variations) > max_variations:
        return variations[:max_variations + 1]

    # ── Level 2: 去除常见企业后缀 ──
    # 按长度降序匹配，避免短后缀先匹配导致长后缀残留
    entity_suffixes = [
        "股份有限公司", "有限责任公司",
        "股份公司", "有限公司",
        "集团公司", "总公司",
        "公司",
    ]
    v2 = name
    for suffix in entity_suffixes:
        if v2.endswith(suffix):
            v2 = v2[:-len(suffix)]
            break
    if v2 != name and v2 not in variations:
        variations.append(v2)

    if len(variations) > max_variations:
        return variations[:max_variations + 1]

    # ── Level 3: 进一步去除行业/类型词汇 ──
    industry_words = ["煤炭", "煤业", "矿业", "煤矿", "煤电", "煤化"]
    v3 = v2
    for w in industry_words:
        v3 = v3.replace(w, "")
    v3 = v3.strip()
    if v3 and v3 != name and v3 not in variations:
        variations.append(v3)

    return variations[:max_variations + 1]


def _has_data(result: str) -> bool:
    """检测 MCP 返回结果是否包含有效数据（非错误、非空）。

    判断依据：
    - 空字符串/纯空白 → 无数据
    - 包含 {"error": ...} → 无数据（MCP 调用异常）
    - JSON 对象所有值均为空/None/[] → 无数据
    - JSON 数组为空 → 无数据
    - 有效 JSON 且有实质内容 → 有数据
    - 非 JSON 长文本 → 可能有数据（保守判定为有数据）

    Args:
        result: MCP 工具返回的原始字符串

    Returns:
        bool: True 表示结果中包含有效数据
    """
    if not result or not result.strip():
        return False

    stripped = result.strip()

    try:
        data = json.loads(stripped)
    except (json.JSONDecodeError, TypeError):
        # 非 JSON 响应：若长度 > 50 字符则保守判定为有数据
        return len(stripped) > 50

    # 显式错误响应
    if isinstance(data, dict):
        if data.get("error"):
            return False
        # 检查是否有任何实质内容
        has_content = any(
            v is not None and v != [] and v != {} and v != ""
            for v in data.values()
        )
        return has_content

    if isinstance(data, list):
        return len(data) > 0

    # 其他类型（字符串、数字等）视为有数据
    return True


async def _query_mine_with_retry(
    tool_name: str,
    mine_name: str,
    extra_args: dict = None,
    max_retries: int = 3,
) -> str:
    """带名称放宽重试的 MCP 查询包装器。

    首次使用原始名称查询；若返回空结果，则逐步放宽煤矿名称并重试，
    最多重试 max_retries 次（即总共最多查询 1 + max_retries 次）。

    Args:
        tool_name: MCP 工具名称
        mine_name: 煤矿名称
        extra_args: 除 mine_name 外的额外参数（如 plan_month, year 等）
        max_retries: 最大重试次数，默认 3

    Returns:
        str: MCP 工具返回的 JSON 字符串（首次查到数据即返回）
    """
    names_to_try = _generate_name_variations(mine_name, max_retries)
    last_result = None

    for i, name in enumerate(names_to_try):
        args = {"mine_name": name}
        if extra_args:
            args.update(extra_args)

        if i == 0:
            print(f"[MCP Retry] 首次查询，mine_name='{name}'")
        else:
            print(f"[MCP Retry] 第 {i} 次重试，放宽名称: '{name}'")

        result = await _call_mcp_tool(tool_name, args)
        last_result = result

        if _has_data(result):
            if i > 0:
                print(f"[MCP Retry] ✓ 重试成功！使用放宽名称 '{name}' 查到数据")
            return result
        else:
            print(f"[MCP Retry] ✗ 名称 '{name}' 未查到数据，"
                  f"结果预览: {str(result)[:150]}")

    # 所有重试均未查到数据
    print(f"[MCP Retry] {len(names_to_try)} 次尝试均未查到有效数据，返回最后一次结果")
    return last_result if last_result else json.dumps(
        {"error": f"查询无数据: 已尝试 {len(names_to_try)} 个名称变体，均未匹配到数据",
         "names_tried": names_to_try},
        ensure_ascii=False,
    )


async def query_mining_plan(mine_name: str, plan_month: str) -> str:
    """查询煤矿月度采掘计划。

    调用远程 MCP 工具 query_monthly_mining_plan，按照煤矿名称模糊匹配 +
    计划月份精确查询，返回：
    - 主表信息（矿井基本信息、计划编制日期等）
    - 生产准备任务列表
    - 巷道掘进明细列表（含巷道名称、设计长度、支护方式等）

    用于配风计划审查「用风地点完整性」检查：比对采掘计划中的工作面清单
    与配风计划中列出的用风地点是否一致。

    内置名称放宽重试：若首次查询无数据，依次尝试去除企业后缀、行业词汇后重试（最多 3 次）。

    Args:
        mine_name: 煤矿企业名称（模糊匹配），如 "合阳县金桥煤炭有限公司"
        plan_month: 计划月份，格式 yyyy-M 或 yyyy-MM，如 "2026-6" 或 "2026-06"（自动规范化为 yyyy-MM）

    Returns:
        JSON格式的采掘计划数据
    """
    plan_month = _normalize_plan_month(plan_month)
    return await _query_mine_with_retry(
        "query_monthly_mining_plan",
        mine_name,
        extra_args={"plan_month": plan_month},
    )


async def query_face_procedure(mine_name: str) -> str:
    """查询工作面作业规程。

    调用远程 MCP 工具 query_working_face_procedure，按照矿井名称模糊匹配，
    返回最近创建的一条作业规程记录（按 create_time 降序取第一条）。
    包含工作面参数：最大控顶距、最小控顶距、平均采高、工作面长度等。

    用于配风计划审查「工作面参数一致性」检查：比对作业规程中的参数
    与配风计划中该地点列出的参数是否一致。

    内置名称放宽重试：若首次查询无数据，依次尝试去除企业后缀、行业词汇后重试（最多 3 次）。

    Args:
        mine_name: 矿井名称（模糊匹配），如 "合阳县金桥煤炭有限公司"

    Returns:
        JSON格式的作业规程数据
    """
    return await _query_mine_with_retry(
        "query_working_face_procedure",
        mine_name,
    )


async def query_gas_report(mine_name: str, year: int) -> str:
    """查询瓦斯等级鉴定报告。

    调用远程 MCP 工具 query_gas_identification_report，按照矿井名称模糊匹配 +
    鉴定年度精确查询，关联查询：
    - 主表：矿井瓦斯等级、鉴定日期、审批意见等
    - 明细表：各工作面瓦斯参数（绝对瓦斯涌出量、相对瓦斯涌出量、
      绝对二氧化碳涌出量、相对二氧化碳涌出量等）

    用于配风计划审查「瓦斯与二氧化碳数据一致性」检查：
    校验配风计划中各用风地点的瓦斯/CO2 涌出量是否与鉴定报告一致。

    内置名称放宽重试：若首次查询无数据，依次尝试去除企业后缀、行业词汇后重试（最多 3 次）。

    Args:
        mine_name: 矿井名称（模糊匹配），如 "合阳县金桥煤炭有限公司"
        year: 鉴定年度（精确匹配），如 2026

    Returns:
        JSON格式的瓦斯等级鉴定数据
    """
    # 确保 year 是 int 类型
    year_int = int(year) if not isinstance(year, int) else year
    return await _query_mine_with_retry(
        "query_gas_identification_report",
        mine_name,
        extra_args={"year": year_int},
    )


async def query_wind_report(mine_name: str, report_month: str) -> str:
    """查询测风报表。

    调用远程 MCP 工具 query_ventilation_report，按照煤矿名称模糊匹配 +
    报表月份精确查询，关联查询：
    - 主表：报表基本信息（编制日期、审核人等）
    - 风井风机表：各风井风机运行参数
    - 测点表：各测风点数据（风速、风量、温度、瓦斯浓度等）

    用于配风计划审查「风速与温度数据匹配」检查：
    根据当月温度监测数据，核验配风计划中温度系数选择是否合理。

    内置名称放宽重试：若首次查询无数据，依次尝试去除企业后缀、行业词汇后重试（最多 3 次）。

    Args:
        mine_name: 煤矿名称（模糊匹配），如 "合阳县金桥煤炭有限公司"
        report_month: 报表月份，格式 yyyy-MM，如 "2026-06"（自动规范化为 yyyy-MM）

    Returns:
        JSON格式的测风报表数据
    """
    report_month = _normalize_plan_month(report_month)
    return await _query_mine_with_retry(
        "query_ventilation_report",
        mine_name,
        extra_args={"report_month": report_month},
    )