vent_agent_2.0/extensions/main_fan/batch_tools.py

"""矿级 Agent 的发现提交、批量取数与受控报告复用。"""
import asyncio
import json

import httpx
from langchain.tools import ToolRuntime
from langchain_core.messages import ToolMessage
from langgraph.types import Command

from db import main_fan_store
from extensions import define_tool
from extensions.main_fan import exploration, history, report_archive
from extensions.main_fan.batch_context import scope_for
from extensions.main_fan.batch_models import MinePlan
from extensions.main_fan.exploration_tools import _action, _identity, StringList, WorkspacePath
from extensions.main_fan.schemas import MainFanResult
from extensions.main_fan.source_policy import validate_history_sources
from runtime.python_sandbox import workspace_file


def checked_plan(workflow, user, session, plan):
    known = {}
    context = workflow['context']
    for system, collection in (('safety', 'sensor_queries'), ('equipment', 'equipment_queries')):
        queries = context.get(collection, [])
        if not queries:
            raise ValueError('矿级发现须查询安全监控与重大设备两个实时列表，保留各系统可用情况')
        if any(row.get('next_page') and not any(other.get('page_num') == row['next_page'] for other in queries)
               for row in queries):
            raise ValueError('实时列表还有未获取分页，请取齐后再提交矿级计划')
        for query in queries:
            body = json.loads(workspace_file(user, session, query['data_path']).read_text())['payload']
            rows = body.get('rows') if isinstance(body, dict) else body
            if rows is None and isinstance(body, dict):
                data = body.get('data')
                rows = data.get('rows') if isinstance(data, dict) else data
            for row in rows or []:
                if isinstance(row, dict) and row.get('nodeId'):
                    known.setdefault((system, str(row['nodeId'])), []).append(row)
    for shaft in plan.shafts:
        if shaft.identity_status == 'confirmed' and not any(
                shaft.shaft_name in str(value)
                for rows in known.values() for row in rows for value in row.values() if isinstance(value, str)):
            raise ValueError('已确认通风井的名称须能在本矿实际资料中找到原文。仅有1#、2#主通风机不能确认井名；请保留一个unresolved-main主备系统（通风井归属待确认）。')
        if shaft.identity_status == 'candidate' and (not shaft.shaft_id.startswith('unresolved-')
                or not any(word in shaft.shaft_name for word in ('待确认', '候选'))):
            raise ValueError('未确认通风井时请使用unresolved-内部标识和带“待确认”的名称；风机号不能当作通风井号')
        for point in shaft.points:
            if (point.system, point.node_id) not in known:
                raise ValueError('计划包含未从本矿实际列表取得的测点：' + point.node_id)
            rows = known[(point.system, point.node_id)]
            if point.role == 'flow':
                valid = any('风量' in str(row.get('nodeTypeName', row.get('sensorName', ''))) for row in rows)
            elif point.system == 'safety':
                valid = any(str(row.get('sensorTypeId', row.get('SensorTypeID', ''))) == '1010' for row in rows)
            else:
                valid = any(any(word in str(row.get('nodeTypeName', '')) for word in ('运行状态', '开停', '启停')) for row in rows)
            if not valid:
                raise ValueError('测点实际类型与声明用途不符：' + point.node_id + '。只选主通风机开停/运行状态与风量，电流、振动、温度等不能声明为fan_state。')
        validate_history_sources(shaft.points)
    return plan


@define_tool(group='主通风机', label='提交矿级通风井与测点计划')
@_action
async def submit_main_fan_mine_plan(workflow_id: str, plan: MinePlan, runtime: ToolRuntime):
    """仅本地批次发现阶段：取得本矿两个实时列表及分页后，提交全部通风井和历史测点计划并结束本轮。

    plan含coverage(confirmed/partial)、shafts、limitations。每井含shaft_id(内部安全标识)、shaft_name、
    identity_status(confirmed/candidate)、identity_reason、fan_pair(最多两个设备标识)、points及limitations。
    每个point含system(safety/equipment)、node_id(实际列表编号)、role(fan_state/flow)、fan_id、label、
    unit、mapping_confirmed、mapping_reason。fan_state只用safety，flow只用equipment；重大设备运行状态不入计划。
    一级/二级须分别保留，不默认等于整机；不明通风井可用
    candidate及内部标识unresolved-main，不编造正式井号。所有测点须来自本轮本矿实际返回。
    """
    _, user, session = _identity(runtime)
    scope = scope_for(user, session)
    if scope.shaft is not None:
        raise ValueError('当前已是通风井分析阶段，请提交井级报告')
    workflow = exploration._load(workflow_id, user, session)
    exploration.validate_inputs(workflow, user, session)
    plan = checked_plan(workflow, user, session, plan)
    value = {'status': 'needs_materials', 'mine_code': scope.mine['mine_id'],
        'mine_name': scope.mine['mine_name'], 'result_title': '矿级通风井与测点发现', 'check_object': '主通风机测点目录',
        'result': '已保存本矿实际测点与通风井计划，等待逐井分析。',
        'workflow_id': workflow_id, 'assessment': {'mode': 'mine_discovery', 'plan': plan.model_dump(mode='json')}}
    value = main_fan_store.save_analysis(user, session, value)
    value = MainFanResult.model_validate(value).model_dump(mode='json')
    main_fan_store.update_workflow(workflow_id, result=value, status='completed')
    return Command(update={'structured_response': value, 'messages': [ToolMessage(
        content=json.dumps({'ok': True, 'shaft_count': len(plan.shafts), 'message': '计划已保存，宿主将继续派发本矿逐井分析。'}, ensure_ascii=False),
        tool_call_id=runtime.tool_call_id, name='submit_main_fan_mine_plan')]})


@define_tool(group='主通风机', label='按天回查全部开停和风量历史')
@_action
async def query_main_fan_history(workflow_id: str, runtime: ToolRuntime) -> dict:
    """对宿主确认的本井计划测点，从批次截止时刻向前按天回查40天。一次工具调用完成传输与归档。

    每个日窗口的请求、响应及完整数据压缩为一个ZIP；先读agent_data_path里的观测状态区间、
    变化边界、缺口和风量计数。不把一级/二级直接当整机。需要复核变化或计算风量窗口时，
    自编Python按manifest的archive.path及members读取ZIP中的对应段，避免打印全部采样。
    开停仅查安全监控，重大设备仅查风量；均查满，失败段再次调用可补取，成功段不重复请求。
    """
    _, user, session = _identity(runtime)
    return await history.query(workflow_id, user, session)


@define_tool(group='主通风机', label='获取测定PDF及可复用报告档案')
@_action
async def fetch_main_fan_reports(workflow_id: str, runtime: ToolRuntime) -> dict:
    """按本井通过报告接口查询并下载PDF，计算实际SHA256，取得同矿同井的成功历史提取档案。

    mode=analyze须按Skill完成全页提取；mode=reuse读取archive_python_path中的facts，不再渲染识别。
    is_mock=true表示既有测试原件，须读取原件真实矿名并核对目标归属，不得把测试报告改写为其他矿的正式报告。
    """
    _, user, session = _identity(runtime)
    try:
        return await report_archive.fetch(workflow_id, user, session)
    except httpx.HTTPError as exc:
        raise ValueError('报告接口查询或下载失败：' + type(exc).__name__) from None


@define_tool(group='主通风机', label='保存已完整提取的报告档案')
@_action
async def archive_main_fan_report(workflow_id: str, facts_path: WorkspacePath,
                                   python_run_ids: StringList, runtime: ToolRuntime) -> dict:
    """保存成功沙箱输出，供下次相同报告复用；调用后仍继续当期时序分析，不结束Agent。

    facts_path为JSON，字段严格为source_sha256、page_count、read_pages(完整1..N)、
    page_sources(物理页码到各页独立图片/文本路径，必须有实际完整read_file记录)、
    reports(每份报告的真实身份、原页、参数及性能事实，合订文件不得漏后面的报告)、analysis、limitations。
    必须真实读完全部页，不能用空对象冒充提取。首次解析完成后即保存；当期历史分析失败不影响已成功的报告档案。
    """
    _, user, session = _identity(runtime)
    return await asyncio.to_thread(report_archive.save, workflow_id, user, session, facts_path, python_run_ids)