"""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