feat: 完成 issue #75 [Ti-1] 交接班摘要生成功能

2026-08-05 04:19:36 +08:00
parent 95dace5ef5
commit e2d24f1839
3 changed files with 296 additions and 0 deletions
@@ -0,0 +1,166 @@
+# -*- coding: utf-8 -*-
+"""Template-Ti 一期 · 交接班摘要生成功能 —— issue #75。
+
+父 Issue #11「④ LLM 报警解释 / 交接班 / NL 查询」子任务：
+把班次事件记录生成为**结构化交接班摘要**（本地规则，可离线测试）：
+
+- 事件归类：production（生产概况）/ equipment（设备运行）/ alarm（异常与处置）/
+  safety（安全注意事项）/ todo（待办事项）；
+- 异常处置校验：alarm 事件必须含处置记录，缺失 → `needs_followup=True`
+  （对齐交接班规范：异常事项须标注发生时间与处理人，PRD 5.4）；
+- 章节输出：固定五节 Markdown 摘要（to_text），供交接班对话/报表使用；
+- 与 `TiScenarioRunner.generate_handover`（LLM + RAG 润色）互补：
+  本模块产出结构化底稿，LLM 层可再润色/补充。
+
+纯本地规则实现（无 LLM 依赖）；事件来源：DCS 告警、人工录入、巡检记录。
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from datetime import date as _date
+from typing import Dict, List, Optional
+
+#: 默认配置资产路径（相对本模块）
+DEFAULT_CONFIG_PATH = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)), "config",
+    "handover.template.yaml")
+
+#: 默认章节定义（key → 标题）
+DEFAULT_SECTIONS: List[Dict[str, str]] = [
+    {"key": "production", "title": "生产概况"},
+    {"key": "equipment", "title": "设备运行状态"},
+    {"key": "alarm", "title": "异常与处置"},
+    {"key": "safety", "title": "安全注意事项"},
+    {"key": "todo", "title": "待办事项"},
+]
+
+
+@dataclass
+class HandoverReport:
+    """结构化交接班摘要。"""
+
+    shift: str
+    date: str
+    sections: Dict[str, List[str]] = field(default_factory=dict)
+    needs_followup: bool = False
+    followup_items: List[str] = field(default_factory=list)
+
+    def to_text(self, section_titles: Optional[Dict[str, str]] = None) -> str:
+        """输出 Markdown 交接班摘要文本。"""
+        titles = section_titles or {s["key"]: s["title"]
+                                    for s in DEFAULT_SECTIONS}
+        lines = [f"# 交接班摘要（{self.date} {self.shift}）"]
+        for key, title in titles.items():
+            items = self.sections.get(key, [])
+            lines.append(f"\n## {title}")
+            lines.append("\n".join(f"- {it}" for it in items)
+                         if items else "- （无记录）")
+        if self.needs_followup:
+            lines.append("\n> ⚠ 待跟进：异常事项缺少处置记录，请人工确认。")
+        return "\n".join(lines)
+
+    def to_dict(self) -> dict:
+        return {
+            "shift": self.shift, "date": self.date,
+            "sections": self.sections,
+            "needs_followup": self.needs_followup,
+            "followup_items": self.followup_items,
+        }
+
+
+class HandoverGenerator:
+    """交接班摘要生成器（事件 → 结构化报告，本地规则）。"""
+
+    #: 事件类型 → 章节 key（未知类型归入待办）
+    _TYPE_TO_SECTION = {
+        "production": "production", "设备": "production",
+        "equipment": "equipment", "设备运行": "equipment",
+        "alarm": "alarm", "异常": "alarm", "报警": "alarm",
+        "safety": "safety", "安全": "safety",
+        "todo": "todo", "待办": "todo",
+    }
+
+    def __init__(
+        self,
+        sections: Optional[List[Dict[str, str]]] = None,
+        alarm_requires_disposal: bool = True,
+        default_shift: str = "甲班",
+    ) -> None:
+        self.sections = sections or DEFAULT_SECTIONS
+        self.alarm_requires_disposal = alarm_requires_disposal
+        self.default_shift = default_shift
+
+    # ------------------------------------------------------------------
+    @classmethod
+    def from_template_config(cls, path: str = DEFAULT_CONFIG_PATH) -> "HandoverGenerator":
+        """从模板配置资产加载（config/handover.template.yaml）。"""
+        import yaml
+
+        with open(path, "r", encoding="utf-8") as fh:
+            raw = yaml.safe_load(fh) or {}
+        cfg = raw.get("handover", {}) or {}
+        return cls(
+            sections=cfg.get("sections") or DEFAULT_SECTIONS,
+            alarm_requires_disposal=bool(
+                cfg.get("alarm_requires_disposal", True)),
+            default_shift=cfg.get("default_shift", "甲班"),
+        )
+
+    # ------------------------------------------------------------------
+    def generate(
+        self,
+        events: List[dict],
+        shift: Optional[str] = None,
+        date: Optional[str] = None,
+    ) -> HandoverReport:
+        """按事件记录生成交接班摘要。
+
+        Args:
+            events: `[{type, text, time?, operator?}]`；type 见 `_TYPE_TO_SECTION`；
+            shift: 班次（缺省 = 配置默认）；
+            date: 日期（缺省 = 今天）。
+        """
+        sections: Dict[str, List[str]] = {s["key"]: []
+                                          for s in self.sections}
+        report = HandoverReport(
+            shift=shift or self.default_shift,
+            date=date or _date.today().isoformat(),
+            sections=sections,
+        )
+        for ev in events:
+            section = self._section_for(ev.get("type", ""))
+            text = self._format_event(ev)
+            sections.setdefault(section, []).append(text)
+            if section == "alarm":
+                self._check_disposal(ev, report)
+        if not sections.get("alarm"):
+            sections["alarm"] = ["本班无异常事项"]
+        return report
+
+    # ------------------------------------------------------------------
+    def _section_for(self, event_type: str) -> str:
+        return self._TYPE_TO_SECTION.get(event_type.strip(), "todo")
+
+    @staticmethod
+    def _format_event(ev: dict) -> str:
+        time_ = ev.get("time")
+        operator = ev.get("operator")
+        text = str(ev.get("text", "")).strip()
+        parts = []
+        if time_:
+            parts.append(f"[{time_}]")
+        parts.append(text or "（无描述）")
+        if operator:
+            parts.append(f"（处理人: {operator}）")
+        return " ".join(parts)
+
+    def _check_disposal(self, ev: dict, report: HandoverReport) -> None:
+        """异常事件须含处置记录（disposal 或 operator），否则标记待跟进。"""
+        if not self.alarm_requires_disposal:
+            return
+        disposal = ev.get("disposal") or ev.get("operator")
+        if not disposal:
+            report.needs_followup = True
+            report.followup_items.append(
+                f"异常事件缺处置记录: {ev.get('text', '')[:40]}")