vent_agent_2.0/extensions/main_fan/exploration.py

"""Skill 与沙箱工作台：宿主取数、保护输入、核验交付并归档，不执行业务分析。"""
import hashlib
import json
import math
from pathlib import Path
import re

from agents.workspace import session_root
from algorithms.main_fan.analysis import same_mine
from algorithms.main_fan.models import ScreeningInput, Device, MonitorPoint, OperatingSeries, SwitchRecord, FanStateRecord
from db import main_fan_store, business_store, db_conn
from extensions import get_service
from extensions.registry import load_extensions
from extensions.main_fan import data_source
from extensions.main_fan.pipeline import load_rules
from extensions.main_fan.schemas import AgentAssessment, MainFanInput
from extensions.main_fan.targets import load_catalog
from extensions.main_fan.service import _save_result
from runtime.business_models import RemoteFile
from runtime.python_sandbox import workspace_file, digest, capability, execution_receipt

MODE = 'sandbox_exploration'
QUERY_COLLECTIONS = ('timeseries_queries', 'sensor_queries', 'equipment_queries', 'history_queries', 'report_queries')


def _write(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + '.writing')
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False), encoding='utf-8')
    temporary.replace(path)


def _load(identifier, user, session, allow_completed=False):
    from extensions.main_fan.batch_context import current_scope
    scope = current_scope.get()
    if scope is not None and scope.active_workflow_id and identifier != scope.active_workflow_id:
        raise ValueError('当前矿级会话已进入新的阶段或通风井，请使用本轮workflow_id：' + scope.active_workflow_id)
    workflow = main_fan_store.workflow(identifier, user, session)
    if workflow['context'].get('mode') != MODE:
        raise ValueError('该任务不是当前沙箱探索模式，请重新 prepare_main_fan_task')
    if workflow['status'] == 'completed' and not allow_completed:
        raise ValueError('该探索任务已归档；新的研判请新建任务')
    return workflow


def _data(context):
    # 始终导出可信原始输入；Agent 提取的事实和中间数据保存在自己的派生文件中。
    return ScreeningInput.model_validate({key: value for key, value in context['task_input'].items()
                                         if key in ScreeningInput.model_fields})


def _fan_names(context, data):
    names = dict(context.get('fan_names', {}))
    names.update({row.device_code: row.name for row in data.devices if same_mine(row, data)})
    if data.fan_id and data.fan_name:
        names[data.fan_id] = data.fan_name
    return names


def _persist(workflow, user, session, operation, result, *, status='completed', export=False):
    context = workflow['context']
    stages = [*workflow['stages'], {'name': operation, 'status': status, 'result': result}]
    if len(stages) > 128:
        raise ValueError('探索操作已达上限，请提交已有结果或如实说明未完成原因')
    base = session_root(user, session) / 'workspace/exploration' / workflow['id']
    if export:
        data = _data(context)
        files = {'input.json': context['task_input'], 'data.json': data.model_dump(mode='json'),
                 'rules.json': context['policy'], 'submission.schema.json': AgentAssessment.model_json_schema(),
                 'identities.json': {'mine_code': data.mine_code, 'mine_name': data.mine_name,
                                    'fan_pair': data.fan_pair, 'fan_names': _fan_names(context, data)}}
        context['exports'] = {}
        for name, value in files.items():
            _write(base / name, value)
            context['exports']['/workspace/exploration/' + workflow['id'] + '/' + name] = digest(base / name)
    _write(base / 'operations.json', stages)
    main_fan_store.update_workflow(workflow['id'], context=context, stages=stages)
    workflow['stages'] = stages
    return base


def prepare(target, user, session, trusted=None, job_id=None):
    request = MainFanInput.model_validate(trusted if trusted is not None else target)
    inputs = request.model_dump(mode='json')
    for material in inputs['materials']:
        path = workspace_file(user, session, material.get('path', ''))
        if not path.is_file():
            raise ValueError('本次附件不存在：' + material.get('path', ''))
        material['sha256'] = digest(path)
    fingerprint = hashlib.sha256(json.dumps(inputs, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
    existing = db_conn.query(
        "SELECT id FROM main_fan_workflows WHERE user_name=? AND session_id=? AND status='running' ORDER BY created_at DESC LIMIT 10",
        (user, session))
    for row in existing:
        workflow = main_fan_store.workflow(row['id'], user, session)
        if (workflow['context'].get('mode') == MODE
                and workflow['context'].get('input_fingerprint') == fingerprint
                and workflow['context'].get('job_id') == job_id):
            validate_inputs(workflow, user, session)
            return manifest(workflow, user, session)
    context = {'mode': MODE, 'task_input': inputs, 'job_id': job_id,
               'input_fingerprint': fingerprint, 'policy': load_rules().model_dump(mode='json'),
               'fan_names': {item.fan_id: item.fan_name for item in load_catalog().targets
                             if item.mine_code == request.mine_code and item.mine_name == request.mine_name},
               'gateway_loaded': False, 'python_runs': []}
    identifier = main_fan_store.create_workflow(user, session, context, [])
    workflow = main_fan_store.workflow(identifier, user, session)
    _persist(workflow, user, session, '准备本轮资料', {'material_count': len(inputs['materials'])}, export=True)
    return manifest(workflow, user, session)


def manifest(workflow, user, session):
    context = workflow['context']
    prefix = '/workspace/exploration/' + workflow['id']
    return {'ok': True, 'mode': MODE, 'workflow_id': workflow['id'],
            'mine_code': context['task_input']['mine_code'], 'mine_name': context['task_input']['mine_name'],
            'input_path': prefix + '/input.json', 'data_path': prefix + '/data.json',
            'identities_path': prefix + '/identities.json', 'rules_path': prefix + '/rules.json',
            'submission_schema_path': prefix + '/submission.schema.json',
            'materials': context['task_input']['materials'], 'use_gateway': context['task_input']['use_gateway'],
            'python': capability(), 'python_paths': python_paths(workflow),
            **{key: context.get(key, []) for key in QUERY_COLLECTIONS},
            'prepared_data': context.get('prepared_data'),
            'operations': [step['name'] for step in workflow['stages']]}


def python_paths(workflow):
    """直接给模型代码中可使用的路径，避免误用工具虚拟路径。"""
    prefix = 'exploration/' + workflow['id']
    return {'cwd': '本会话workspace根目录', 'input': prefix + '/input.json', 'data': prefix + '/data.json',
            'rules': prefix + '/rules.json', 'identities': prefix + '/identities.json',
            'materials': [item['path'].removeprefix('/workspace/') for item in workflow['context']['task_input']['materials']],
            'report_example': 'report.md', 'assessment_example': 'assessment.json',
            'note': '以上是Python代码实际使用的相对路径。工具参数仍用/workspace/虚拟路径；不要缩成单独input.json。'}


def validate_inputs(workflow, user, session):
    context = workflow['context']
    fingerprints = {item['path']: item['sha256'] for item in context['task_input']['materials']}
    fingerprints.update(context['exports'])
    fingerprints.update(context.get('query_exports', {}))
    for virtual, expected in fingerprints.items():
        path = workspace_file(user, session, virtual)
        if not path.is_file() or digest(path) != expected:
            raise ValueError('原件或可信任务文件已被修改：' + virtual + '；请保留输入，将派生结果另存')


async def fetch(identifier, user, session):
    workflow = _load(identifier, user, session)
    validate_inputs(workflow, user, session)
    context = workflow['context']
    if context['gateway_loaded']:
        return manifest(workflow, user, session)
    request = context['task_input']
    if not request['use_gateway']:
        raise ValueError('本次输入未启用网关，请使用已提供的原件与时序文件')
    gateway = get_service('shaanxi_gateway')
    spec = load_extensions().business('main_fan')
    incoming = await data_source.fetch(gateway, spec.gateway.agent_code, request['mine_code'])
    for key, schema in (('devices', Device), ('points', MonitorPoint), ('operating_series', OperatingSeries),
                        ('switch_records', SwitchRecord), ('fan_state_records', FanStateRecord)):
        request[key].extend(schema.model_validate(row).model_dump(mode='json') for row in incoming.get(key, []))
    known = {item.get('file_id') for item in request['materials']}
    for raw in incoming['files']:
        source = RemoteFile.model_validate(raw)
        if source.file_id in known:
            continue
        filename = re.sub(r'[^\w.\-\u4e00-\u9fff]', '_', Path(source.filename).name)
        key = hashlib.sha256((source.file_id + '\0' + source.version).encode()).hexdigest()[:12]
        virtual = '/workspace/uploads/' + key + '_' + filename
        local = workspace_file(user, session, virtual)
        local.parent.mkdir(parents=True, exist_ok=True)
        metadata = await gateway.download(spec.gateway.agent_code, source, local)
        if context['job_id']:
            business_store.record_file(context['job_id'], metadata)
        request['materials'].append({'file_id': source.file_id, 'version': source.version,
            'filename': source.filename, 'path': virtual, 'sha256': digest(local)})
        known.add(source.file_id)
    context['task_input'] = MainFanInput.model_validate(request).model_dump(mode='json')
    context['gateway_loaded'] = True
    _persist(workflow, user, session, '读取网关资料', {'material_count': len(request['materials'])}, export=True)
    return manifest(workflow, user, session)


def record_python(identifier, user, session, receipt):
    workflow = _load(identifier, user, session)
    summary = {key: receipt[key] for key in ('run_id', 'script_path', 'script_sha256', 'exit_code', 'timed_out', 'outputs')}
    summary['receipt_path'] = receipt['receipt_path']
    workflow['context']['python_runs'].append(summary)
    _persist(workflow, user, session, '执行自编 Python', summary,
             status='completed' if receipt['exit_code'] == 0 and not receipt['timed_out'] else 'failed')


def _read_json(path):
    def invalid_constant(value):
        raise ValueError('结果 JSON 不接受非有限数值：' + value)
    def finite_float(value):
        number = float(value)
        if not math.isfinite(number):
            invalid_constant(value)
        return number
    return json.loads(path.read_text(encoding='utf-8-sig'), parse_constant=invalid_constant, parse_float=finite_float)


def submit(identifier, user, session, report_path, calculations_path=None, python_run_ids=(), artifacts=()):
    workflow = _load(identifier, user, session, allow_completed=True)
    if workflow['status'] == 'completed':
        return workflow['result']
    validate_inputs(workflow, user, session)
    context = workflow['context']
    if (context['task_input']['use_gateway'] and not context['gateway_loaded']
            and not any(context.get(key) for key in QUERY_COLLECTIONS)):
        raise ValueError('本轮要求网关资料，请先调用 fetch_main_fan_data 或实时/时序查询工具取得实际输入')
    report = workspace_file(user, session, report_path)
    if not report.is_file() or report.suffix.lower() != '.md' or report.stat().st_size > 1024 * 1024:
        raise ValueError('报告须为工作区内不超过 1MB 的 Markdown 文件')
    body = report.read_text(encoding='utf-8-sig').strip()
    if not body or 'data:image/' in body or ';base64,' in body:
        raise ValueError('报告正文为空或包含内嵌图像编码')
    if not calculations_path:
        raise ValueError('请提供沙箱脚本生成的研判 JSON calculations_path，格式见 submission.schema.json')
    calculations = workspace_file(user, session, calculations_path)
    if not calculations.is_file() or calculations.stat().st_size > 4 * 1024 * 1024:
        raise ValueError('研判 JSON 不存在或超过 4MB')
    submitted = AgentAssessment.model_validate(_read_json(calculations))
    if not python_run_ids or len(python_run_ids) > 40 or len(artifacts) > 20:
        raise ValueError('提交须关联本轮成功执行记录（1～40次），附件最多20个')
    known_runs = {item['run_id'] for item in context['python_runs']}
    if not set(python_run_ids).issubset(known_runs):
        raise ValueError('执行记录不属于本轮任务，请用 workflow_id 执行脚本')
    receipts = [execution_receipt(user, session, value) for value in dict.fromkeys(python_run_ids)]
    outputs = {item['path'] for receipt in receipts for item in receipt['outputs']}
    delivery = list(dict.fromkeys([report_path, calculations_path, *artifacts]))
    if not set(delivery).issubset(outputs):
        raise ValueError('报告、研判 JSON 和附件必须是成功沙箱执行的声明输出，请重新执行生成脚本')
    sources = {item['path']: item['sha256'] for item in context['task_input']['materials']}
    prefix = '/workspace/exploration/' + identifier
    sources.update({path: sha for path, sha in context['exports'].items()
                    if path in {prefix + '/input.json', prefix + '/data.json'}})
    sources.update(context.get('query_exports', {}))
    bound_evidence = []
    for evidence in submitted.evidence:
        if evidence.source not in sources:
            raise ValueError('证据来源不属于本轮原件或原始输入：' + evidence.source)
        bound_evidence.append({**evidence.model_dump(), 'sha256': sources[evidence.source]})
    data = _data(context)
    names = _fan_names(context, data)
    pair = [{'fan_id': fan_id, 'fan_name': names.get(fan_id, fan_id)} for fan_id in data.fan_pair]
    assessment = {'mode': MODE, 'agent_result': submitted.model_dump(mode='json'),
        'decision': {'result_code': submitted.result_code, 'summary': submitted.summary},
        'fan_pair': pair, 'evidence': bound_evidence,
        'exploration': {'mode': MODE, 'operations': [step['name'] for step in workflow['stages']],
            'calculations_path': calculations_path, 'calculations': submitted.model_dump(mode='json'),
            'execution_history': context['python_runs'],
            **{key: context.get(key, []) for key in QUERY_COLLECTIONS},
            'python_runs': [{'run_id': receipt['run_id'], 'script_path': receipt['script_path'],
                            'script_sha256': receipt['script_sha256'], 'outputs': receipt['outputs']} for receipt in receipts]}}
    attachments = []
    for virtual in delivery:
        path = workspace_file(user, session, virtual)
        if not path.is_file():
            raise ValueError('提交的附件不存在：' + virtual)
        attachments.append({'filename': path.name, 'path': virtual})
    source_list = [{'title': item.get('filename') or Path(item['path']).name, 'url': item['path'], 'type': 'file'}
                   for item in context['task_input']['materials']]
    if any(item['source'].endswith(('/input.json', '/data.json')) for item in bound_evidence):
        source_list.append({'title': '本轮原始输入与监测数据', 'url': prefix + '/input.json', 'type': 'file'})
    for source in dict.fromkeys(item['source'] for item in bound_evidence):
        if source in context.get('query_exports', {}):
            source_list.append({'title': '本轮网关取数 · ' + Path(source).name, 'url': source, 'type': 'file'})
    value = {'status': submitted.status, 'mine_code': data.mine_code, 'mine_name': data.mine_name,
        'check_object': '、'.join(fan['fan_name'] for fan in pair) or data.fan_name or '主备风机',
        'result_title': data.mine_name + ' 主备风机研判报告', 'result': body, 'result_format': 'report',
        'workflow_id': identifier, 'sources': source_list, 'artifacts': attachments,
        'check_process': ' → '.join(step['name'] for step in workflow['stages']),
        'missing_materials': submitted.missing_materials, 'clues': [], 'analysis': None,
        'assessment': assessment, 'judgment_scope': 'main_fan_consistency'}
    # 批次工具的原件和覆盖信息由宿主附加，避免报告输出遗漏查询失败或模拟来源。
    if context.get('history_manifest') is not None:
        value['assessment']['history_coverage'] = {key: context['history_manifest'][key]
            for key in ('cutoff', 'lookback_days', 'point_count', 'expected_segments',
                        'completed_segments', 'failed_segments', 'coverage_complete')}
        value['assessment']['report_archives'] = context.get('report_archives', [])
    with db_conn.transaction():
        saved = _save_result(value, data, None, user, session, context['job_id'])
        stages = [*workflow['stages'], {'name': '核验交付并归档', 'status': 'completed',
            'result': {'analysis_id': saved['analysis_id'], 'result_code': submitted.result_code}}]
        main_fan_store.update_workflow(identifier, context=context, stages=stages, result=saved, status='completed')
    return saved