"""取数、文档解析、确定性核验和结果落库；模型只负责业务编排。""" import asyncio import hashlib import json from pathlib import Path import re from subprocess import SubprocessError import config from agents.workspace import session_root from algorithms.main_fan.analysis import screen, same_mine from algorithms.main_fan.models import ScreeningInput, Screening, DocumentFacts, Finding, 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.documents import read_document, fingerprint from extensions.main_fan.reporting import render_result from extensions.main_fan.schemas import MainFanInput, MainFanResult from extensions.main_fan.pipeline import load_rules, new_stages, run_pipeline from algorithms.main_fan.equivalence import normalize_and_validate_fan_data from runtime.business_models import Clue, RemoteFile from extensions.main_fan.result_output import build_payload def _save_result(value, data, analysis, user_name, session_id, job_id=None): binding = load_extensions().business("main_fan").result_binding value["platform_payload"] = build_payload(value, data, analysis, binding, task_id=job_id or db_conn.new_id()) with db_conn.transaction(): saved = main_fan_store.save_analysis(user_name, session_id, value) result = MainFanResult.model_validate(saved).model_dump(mode="json") if (not job_id and result["status"] == "completed" and config.get_bool("RESULT_PLATFORM_ENABLED", False) and config.get_bool("MAIN_FAN_AUTO_PUBLISH", True)): business_store.record_completed_result("main_fan", user_name, session_id, data.model_dump(mode="json"), result, config.get_str("RESULT_PLATFORM_URL")) return result def workspace_file(root, virtual): if not isinstance(virtual, str) or not virtual.startswith("/workspace/"): raise ValueError("附件必须位于本会话 /workspace/ 中") local = (root / virtual.lstrip("/")).resolve() if not local.is_relative_to((root / "workspace").resolve()) or not local.is_file(): raise ValueError("附件不存在或越出当前会话") return local async def parse_fan_evidence_documents(request, user_name, session_id, *, gateway=None, document_reader=None, job_id=None, document_audit_dir=None, document_progress=None): root = session_root(user_name, session_id) warnings, file_paths, sources, fingerprints = [], [], [], [] if request.use_gateway: gateway = gateway or get_service("shaanxi_gateway") spec = load_extensions().business("main_fan") incoming = await data_source.fetch(gateway, spec.gateway.agent_code, request.mine_code) # 不静默覆盖显式输入，同编码冲突会在算法中被识别。 request.devices.extend(Device.model_validate(r) for r in incoming["devices"]) request.points.extend(MonitorPoint.model_validate(r) for r in incoming["points"]) for kind in ('operating_series','switch_records','fan_state_records'): schema={'operating_series':OperatingSeries,'switch_records':SwitchRecord,'fan_state_records':FanStateRecord}[kind] getattr(request,kind).extend(schema.model_validate(r) for r in incoming.get(kind,[])) existing_ids = {m.get("file_id") for m in request.materials} for raw in incoming["files"]: source = RemoteFile.model_validate(raw) if source.file_id in existing_ids: continue filename = re.sub(r"[^\w.\-\u4e00-\u9fff]", "_", Path(source.filename).name) key = hashlib.sha256(f"{source.file_id}\0{source.version}".encode()).hexdigest()[:12] target = root / "workspace" / "uploads" / f"{key}_{filename}" metadata = await gateway.download(spec.gateway.agent_code, source, target) if job_id: await asyncio.to_thread(business_store.record_file, job_id, metadata) request.materials.append({"file_id": source.file_id, "path": f"/workspace/uploads/{target.name}", "filename": source.filename, "sha256": metadata["sha256"]}) if request.measurements: warnings.append("旧 measurements 字段未明确设备、单位及来源，未猜测其含义；请使用 points。") # 先重验字典输入；上方扩展来自已验证网关 schema。 core = request.model_dump(mode="json") facts = DocumentFacts() document_audits = [] for material in request.materials: virtual = material.get("path", "") local = workspace_file(root, virtual) fingerprints.append({"path":virtual,"sha256":await asyncio.to_thread(fingerprint,local)}) file_paths.append(virtual) try: parsed = (await document_reader(local, virtual) if document_reader else await read_document(local, virtual, user_name=user_name, session_id=session_id, cache_policy=request.document_cache_policy, audit_dir=Path(document_audit_dir)/hashlib.sha256(virtual.encode()).hexdigest()[:12] if document_audit_dir else None, progress=document_progress)) facts.reports.extend(parsed.reports) facts.devices.extend(parsed.devices) facts.points.extend(parsed.points) warnings.extend(parsed.warnings) if parsed.extraction_audit: document_audits.append(parsed.extraction_audit) if any(same_mine(item,request) for item in [*parsed.reports,*parsed.devices,*parsed.points]): sources.append({"title":material.get("filename") or Path(virtual).name,"url":virtual,"type":"file","type_cn":"文件"}) except (ValueError, SubprocessError) as exc: detail = str(exc)[:160] if isinstance(exc, ValueError) else type(exc).__name__ warnings.append(f"附件 {Path(virtual).name} 未完成提取：{detail}") core["reports"].extend(r.model_dump(mode="json") for r in facts.reports) core["devices"].extend(r.model_dump(mode="json") for r in facts.devices) core["points"].extend(r.model_dump(mode="json") for r in facts.points) # 时序证据也进入来源栏；合成数据的标签由模拟网关显式提供。 for item in [*request.operating_series,*request.fan_state_records,*request.switch_records]: evidence = item.evidence if same_mine(item,request) and evidence.source.startswith(('http://','https://')) and not any(s['url'] == evidence.source for s in sources): sources.append({'title':evidence.label or ('运行时序模拟数据' if evidence.method == 'synthetic' else '运行监测数据'), 'url':evidence.source,'type':'api','type_cn':'模拟监测数据' if evidence.method == 'synthetic' else '监测数据'}) allowed = {key: value for key, value in core.items() if key in ScreeningInput.model_fields} screening_input = ScreeningInput.model_validate(allowed) return screening_input,warnings,file_paths,sources,fingerprints,document_audits def _screen(data): return screen(data, freshness_seconds=config.get_int("MAIN_FAN_FRESHNESS_SECONDS", 600), sync_seconds=config.get_int("MAIN_FAN_SYNC_SECONDS", 120)) async def analyze(data, user_name, session_id, *, gateway=None, document_reader=None, job_id=None, document_audit_dir=None, document_progress=None): from extensions.main_fan.targets import load_catalog request=MainFanInput.model_validate(data) stages=new_stages() workflow_id=await asyncio.to_thread(main_fan_store.create_workflow,user_name,session_id, {"request":request.model_dump(mode="json")},stages) if request.paired_assessment else "" try: if workflow_id: stages[0]['status']='running' await asyncio.to_thread(main_fan_store.update_workflow,workflow_id,stages=stages) core,warnings,file_paths,sources,fingerprints,document_audits=await parse_fan_evidence_documents(request,user_name,session_id, gateway=gateway,document_reader=document_reader,job_id=job_id,document_audit_dir=document_audit_dir,document_progress=document_progress) except Exception as exc: if workflow_id: stages[0].update(status='failed',error=str(exc)[:200]) await asyncio.to_thread(main_fan_store.update_workflow,workflow_id,stages=stages,status='failed') raise if request.paired_assessment: policy=load_rules() names={t.fan_id:t.fan_name for t in load_catalog().targets if t.mine_code==core.mine_code and t.mine_name==core.mine_name} dataset=normalize_and_validate_fan_data(core,policy,names) screens=[] for fan in dataset.fans: selected=core.model_copy(deep=True,update={"fan_id":fan['fan_id'],"fan_name":fan['fan_name'], "reports":[fan['report']] if fan['report'] else []}) screens.append(_screen(selected)) if not screens: screens=[_screen(core)] analysis=Screening(engineering_status='review_required' if any(s.has_in_scope_data for s in screens) else 'insufficient_data', has_in_scope_data=any(s.has_in_scope_data for s in screens),findings=[f for s in screens for f in s.findings], associations=[a for s in screens for a in s.associations],calculations=[c for s in screens for c in s.calculations], missing_materials=list(dict.fromkeys(m for s in screens for m in s.missing_materials))) else: analysis=_screen(core) analysis.document_audits = document_audits analysis.findings.extend(Finding(code="document_extraction", severity="review", message=value) for value in warnings) if warnings: analysis.missing_materials.append("核验文档识别结果及原件") assessment=None if request.paired_assessment: context={"data":core.model_dump(mode="json"),"analysis":analysis.model_dump(mode="json"),"fan_names":names, "policy":policy.model_dump(mode="json"),"sources":sources,"fingerprints":fingerprints,"document_audits":document_audits} await asyncio.to_thread(main_fan_store.update_workflow,workflow_id,context=context) def save_progress(state): main_fan_store.update_workflow(workflow_id,stages=state,status='failed' if any(s['status']=='failed' for s in state) else 'running') assessment,body,stages=await asyncio.to_thread(run_pipeline,context,stages=stages,save=save_progress) else: body=render_result(core,analysis) if not analysis.has_in_scope_data and any('未完成提取' in warning for warning in warnings): body='原件识别未完成，请恢复文档识别服务后重新执行本次研判。' # 先持久化原件再用结果 schema 校验；模型无法通过自报 completed 绕过这一步。 value = dict(status="completed" if analysis.has_in_scope_data else "needs_materials", mine_code=request.mine_code, mine_name=request.mine_name, check_object="、".join(f['fan_name'] for f in assessment['fan_pair']) if assessment and assessment['fan_pair'] else f"矿井主通风机 {request.fan_name or request.fan_id}".strip(), result_title=f"{request.mine_name} 主通风机与备用风机一致性研判报告" if request.paired_assessment else f"{request.mine_name} {request.fan_name or request.fan_id}研判报告", result=body, result_format="report", sources=sources,assessment=assessment,workflow_id=workflow_id, check_process="文档与运行数据接入 → 双机身份及单位核验 → 报告参数比较 → 性能数据比较 → 切换稳定工况分析 → 规则汇总 → 报告归档" if request.paired_assessment else "单机资料核验与公式计算", alarm_level=None, missing_materials=[] if request.paired_assessment else analysis.missing_materials, clues=[Clue(clueType="text", clueName=f.code, clueContent=f.message).model_dump(by_alias=True) for f in analysis.findings] + [Clue(clueType="table", clueName="测点关联与公式复核", clueContent=json.dumps({"associations": [a.model_dump() for a in analysis.associations], "calculations": [c.model_dump() for c in analysis.calculations]}, ensure_ascii=False)).model_dump(by_alias=True)] + [Clue(clueType="file", clueName=Path(p).name, clueContent=p).model_dump(by_alias=True) for p in dict.fromkeys([s['url'] for s in sources] if request.paired_assessment else file_paths)], judgment_scope="main_fan_consistency" if request.paired_assessment else "preliminary_screening", analysis=analysis.model_dump(mode="json")) result = await asyncio.to_thread(_save_result, value, core, analysis, user_name, session_id, job_id) if workflow_id: await asyncio.to_thread(main_fan_store.update_workflow,workflow_id,stages=stages,result=result,status='completed') return result async def resume_assessment(workflow_id, start_stage, user_name, session_id, *, job_id=None): workflow=await asyncio.to_thread(main_fan_store.workflow,workflow_id,user_name,session_id) context=workflow['context'] if 'data' not in context: raise ValueError("资料准备未完成，请重新执行原始研判请求") root=session_root(user_name,session_id) for record in context.get('fingerprints',[]): if await asyncio.to_thread(fingerprint,workspace_file(root,record['path'])) != record['sha256']: raise ValueError("原文件已变化，请重新获取和识别资料，不能沿用旧阶段") def save_progress(stages): main_fan_store.update_workflow(workflow_id,stages=stages,status='failed' if any(s['status']=='failed' for s in stages) else 'running') assessment,body,stages=await asyncio.to_thread(run_pipeline,context,stages=workflow['stages'],start_stage=start_stage,save=save_progress) previous=workflow['result'] or {} value={**previous,'status':'completed' if context['analysis']['has_in_scope_data'] else 'needs_materials', 'mine_code':context['data']['mine_code'],'mine_name':context['data']['mine_name'], 'check_object':'、'.join(f['fan_name'] for f in assessment['fan_pair']) or '矿井主通风机', 'result_title':context['data']['mine_name']+' 主通风机与备用风机一致性研判报告', 'result_format':'report','result':body,'assessment':assessment,'workflow_id':workflow_id, 'analysis':context['analysis'],'sources':context['sources'],'missing_materials':[]} value.pop('analysis_id',None) data=ScreeningInput.model_validate(context['data']) analysis=Screening.model_validate(context['analysis']) result=await asyncio.to_thread(_save_result,value,data,analysis,user_name,session_id,job_id) await asyncio.to_thread(main_fan_store.update_workflow,workflow_id,stages=stages,result=result,status='completed') return result def validate_result(value, user_name, session_id): canonical = main_fan_store.canonical_result(value, user_name, session_id) return MainFanResult.model_validate(canonical).model_dump(mode="json")