# -*- 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}, )