167 lines
6.5 KiB
Python
167 lines
6.5 KiB
Python
# -*- 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]}")
|