"""主通风机结果平台映射：结构化事实决定编码，报告正文保留整体结论。""" 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")