vent_agent_2.0/extensions/main_fan/result_output.py

"""主通风机结果平台映射：结构化事实决定编码，报告正文保留整体结论。"""
from datetime import datetime
from html import unescape
import re

from algorithms.main_fan.analysis import LOCAL_ZONE, local_time, same_mine
from algorithms.main_fan.equivalence import three_check_statuses
from runtime.result_models import PlatformJudgment, PlatformResultPayload


def judgment_result(assessment, analysis):
    """编码只采用三项业务规则；OCR说明、表图差异和补充指标单独呈现。"""
    assessment = assessment or {}
    if assessment.get("mode") == "sandbox_exploration":
        # 自主模式采用脚本交付的结论；此处只映射平台格式，不重跑业务计算。
        from extensions.main_fan.schemas import AgentAssessment
        submitted = AgentAssessment.model_validate(assessment["agent_result"])
        return submitted.result_code, submitted.summary
    decision = assessment.get("decision", {})
    basic = assessment.get("basic_comparison", {})
    switches = assessment.get("switch_comparison", {})
    items = three_check_statuses(basic, switches)
    code = ("NORMAL" if all(status == "pass" for status in items.values()) else
            "ABNORMAL" if decision.get("severe_findings") else
            "WARNING" if "fail" in items.values() else "UNDETERMINED")
    if code in {"WARNING", "ABNORMAL"}:
        reasons = []
        mismatches = [r["label"] for r in basic.get("rows", []) if r["status"] == "mismatch"]
        if mismatches:
            reasons.append("两台风机的" + "、".join(mismatches) + "不一致")
        for event in switches.get("events", []):
            metrics, timing, checks = event["metrics"], event["timing"], event["checks"]
            if checks.get("switch_interval") == "fail":
                reasons.append(f"切换间隔{metrics['switch_interval_minutes']:g}分钟，超过{timing['limit_minutes']:g}分钟")
            if checks.get("flow_change") == "fail":
                reasons.append(f"稳定风量差异{metrics['flow_deviation_percent']:.2f}%，超过{event['limits']['flow_deviation_percent']:g}%")
        prefix = "研判预警：" if code == "WARNING" else "研判异常："
        suffix = "。" if code == "WARNING" else "，其中已核实问题达到配置的严重异常条件。"
        return code, prefix + "；".join(dict.fromkeys(reasons)) + suffix
    if code == "NORMAL":
        events = switches.get("events", [])
        details = []
        if len(events) == 1:
            metrics = events[0]["metrics"]
            if metrics.get("switch_interval_minutes") is not None:
                details.append(f"切换间隔{metrics['switch_interval_minutes']:g}分钟")
            if metrics.get("flow_deviation_percent") is not None:
                details.append(f"稳定风量差异{metrics['flow_deviation_percent']:.2f}%")
        sentence = "研判正常：两台风机报告基础参数一致"
        return code, sentence + ("，" + "、".join(details) if details else "") + "，三项均通过，判定主备主要通风机在本次核验范围内具备同等能力。"
    scopes = {"report_parameters": "报告参数", "report_performance": "性能测定数据", "switch_performance": "切换工况"}
    compared = [scopes[key] for key in decision.get("reported_scope", []) if key in scopes]
    prefix = "已完成" + "、".join(compared) + "分析，" if compared else ""
    return code, prefix + "关键比较数据未能形成有效对照，本次不作主备主要通风机同等能力的确定结论。"


def report_text(markdown):
    """移除图像编码与 Markdown 标记，将每张表独立转换为可阅读的键值行。"""
    lines, headers = [], None
    for raw in markdown.splitlines():
        line = raw.strip()
        if re.match(r"^!\[[^]]*\]\(", line):
            headers = None
            continue
        if line.startswith("|") and line.endswith("|"):
            values = [value.strip() for value in line.strip("|").split("|")]
            if all(re.fullmatch(r":?-{3,}:?", value) for value in values):
                continue
            if headers is None:
                headers = values
                continue
            line = "；".join(f"{key}：{value}" for key, value in zip(headers, values))
        else:
            headers = None
            line = re.sub(r"^#{1,6}\s+", "", line)
        line = re.sub(r"\[([^]]+)\]\([^)]*\)", r"\1", line)
        line = line.replace("**", "").replace("`", "")
        line = unescape(line)
        if line or (lines and lines[-1]):
            lines.append(line)
    return "\n".join(lines).strip()


def _data_detail(data, assessment):
    if (assessment or {}).get("mode") == "sandbox_exploration":
        return assessment["agent_result"]["data_detail"]
    parts = []
    basic = (assessment or {}).get("basic_comparison", {})
    curves = (assessment or {}).get("performance_comparison", {}).get("results", [])
    events = (assessment or {}).get("switch_comparison", {}).get("events", [])
    pair = (assessment or {}).get("fan_pair", [])
    ids = {fan["fan_id"] for fan in pair}
    report_ids = {fan["report_id"] for fan in pair if fan.get("report_id")}
    reports = [r for r in data.reports if same_mine(r, data) and r.report_id in report_ids]
    sources = []
    if reports:
        sources.append(f"双机性能测定资料（{len(reports)}份）" if len(reports) == 2 else f"性能测定资料（{len(reports)}份）")
    if basic.get("backup_availability"):
        sources.append("设备台账")
    if events:
        if any(event.get("timing", {}).get("interval_minutes") is not None for event in events):
            sources.append("整机开停记录")
        if any(event.get("changes") for event in events):
            sources.append("运行监测序列")
    if sources:
        parts.append("资料来源：" + "、".join(sources))
    if events:
        used_series = [s for s in data.operating_series if same_mine(s, data) and s.fan_id in ids]
        times = [local_time(p.observed_at) for series in used_series for p in series.samples]
        if times:
            parts.append(f"监测时段：{min(times):%Y-%m-%d %H:%M:%S}至{max(times):%Y-%m-%d %H:%M:%S}")
    backup = basic.get("backup_availability")
    if backup:
        parts.append("备用状态：" + {"available": "可用", "unavailable": "不可用"}.get(backup["status"], backup["status"]))
    angles = sorted({c["blade_angle_deg"] for c in curves if c.get("blade_angle_deg") is not None})
    if angles:
        parts.append("测试叶片角度：" + "、".join(f"{value:g}°" for value in angles))
    if curves:
        parts.append("共同频率：" + "、".join(f"{value:g}Hz" for value in sorted({c["frequency_hz"] for c in curves})))
    for index, event in enumerate(events, 1):
        prefix = f"第{index}次切换：" if len(events) > 1 else ""
        metrics = event["metrics"]
        if metrics.get("switch_interval_minutes") is not None:
            parts.append(f"{prefix}切换间隔：{metrics['switch_interval_minutes']:g}分钟（阈值≤{event['timing']['limit_minutes']:g}分钟）")
        if "flow" in event.get("changes", {}):
            change = event["changes"]["flow"]
            parts.extend([f"{prefix}切换前稳定风量：{change['before']*60:g}m³/min",
                          f"{prefix}切换后稳定风量：{change['after']*60:g}m³/min",
                          f"{prefix}风量差异：{metrics['flow_deviation_percent']:.2f}%（阈值≤{event['limits']['flow_deviation_percent']:g}%）"])
    return "；".join(parts)


def build_payload(value, data, analysis, binding, *, task_id, checked_at=None):
    assessment = value.get("assessment") or {}
    pair = assessment.get("fan_pair", [])
    names = "、".join(fan["fan_name"] for fan in pair) or value["check_object"]
    ids = "、".join(fan["fan_id"] for fan in pair) or data.fan_id
    object_detail = "研判对象：" + names
    if ids:
        object_detail += "；风机标识：" + ids
    object_detail += "；研判内容：主备风机同等能力"
    code, short_result = judgment_result(assessment, analysis)
    return PlatformResultPayload(
        agentCode=binding.agent_code, agentName=binding.agent_name, taskId=task_id,
        checkedAt=checked_at or datetime.now(LOCAL_ZONE).strftime("%Y-%m-%d %H:%M:%S"),
        mineCode=value["mine_code"], mineName=value["mine_name"], objectDetail=object_detail,
        dataDetail=_data_detail(data, assessment),
        judgment=PlatformJudgment(resultCode=code, result=short_result,
            reason=report_text(value["result"]), evidenceLink=""),
    ).model_dump(mode="json")