vent_agent_2.0/extensions/main_fan/service.py

"""取数、文档解析、确定性核验和结果落库；模型只负责业务编排。"""
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")