#!/usr/bin/env python3 from __future__ import annotations import argparse import json import shutil import subprocess import sys import tempfile import time from collections import Counter, defaultdict from datetime import date from pathlib import Path from urllib.parse import quote from urllib.request import urlopen from PIL import Image from docx import Document from docx.enum.section import WD_ORIENT from docx.enum.table import WD_ALIGN_VERTICAL, WD_TABLE_ALIGNMENT from docx.enum.text import WD_ALIGN_PARAGRAPH, WD_LINE_SPACING from docx.oxml import OxmlElement from docx.oxml.ns import qn from docx.shared import Cm, Pt, RGBColor ROOT = Path(__file__).resolve().parent OUT_DIR = ROOT / "总结文档" FIG_DIR = OUT_DIR / "figures" DEFAULT_DOCX = OUT_DIR / "组态数据高质量演化过程故障标签手册.docx" BASE_URL = "http://127.0.0.1:5173/preview.html" API_URL = "http://127.0.0.1:8000/api/tag-manual/labels" CHROME = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" BODY_FONT = "宋体" HEAD_FONT = "黑体" BODY_SIZE = 12.0 TABLE_SIZE = 9.0 DEVICE_ORDER = {"7号机": 7, "8号机": 8, "9号机": 9} STAGE_ORDER = {"轻微": 0, "异常": 1, "严重异常": 2} KIND_ORDER = {"pressure": 0, "vibration": 1} KIND_LABEL = {"pressure": "润滑油压力", "vibration": "振动"} def set_run_font(run, name: str = BODY_FONT, size: float = BODY_SIZE, bold: bool = False) -> None: run.font.name = name run.font.size = Pt(size) run.font.bold = bold run.font.color.rgb = RGBColor(0, 0, 0) run._element.get_or_add_rPr().rFonts.set(qn("w:eastAsia"), name) def set_style_font(style, name: str, size: float, bold: bool) -> None: style.font.name = name style.font.size = Pt(size) style.font.bold = bold style.font.color.rgb = RGBColor(0, 0, 0) style.element.get_or_add_rPr().rFonts.set(qn("w:eastAsia"), name) def configure_document(doc: Document) -> None: section = doc.sections[0] section.orientation = WD_ORIENT.LANDSCAPE section.page_width = Cm(29.7) section.page_height = Cm(21.0) section.top_margin = Cm(2.0) section.bottom_margin = Cm(2.0) section.left_margin = Cm(2.0) section.right_margin = Cm(2.0) normal = doc.styles["Normal"] set_style_font(normal, BODY_FONT, BODY_SIZE, False) normal.paragraph_format.line_spacing_rule = WD_LINE_SPACING.ONE_POINT_FIVE normal.paragraph_format.space_before = Pt(0) normal.paragraph_format.space_after = Pt(0) heading1 = doc.styles["Heading 1"] set_style_font(heading1, HEAD_FONT, 16, True) heading1.paragraph_format.space_before = Pt(18) heading1.paragraph_format.space_after = Pt(10) heading1.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE heading1.paragraph_format.keep_with_next = True heading2 = doc.styles["Heading 2"] set_style_font(heading2, HEAD_FONT, 14, True) heading2.paragraph_format.space_before = Pt(12) heading2.paragraph_format.space_after = Pt(6) heading2.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE heading2.paragraph_format.keep_with_next = True add_footer(doc) def add_footer(doc: Document) -> None: section = doc.sections[0] footer = section.footer paragraph = footer.paragraphs[0] paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run("组态数据高质量演化过程故障标签手册 第 ") set_run_font(run, BODY_FONT, 9) field = OxmlElement("w:fldSimple") field.set(qn("w:instr"), "PAGE") paragraph._p.append(field) tail = paragraph.add_run(" 页") set_run_font(tail, BODY_FONT, 9) def body_paragraph(doc: Document, text: str, bold_prefix: str = "") -> None: paragraph = doc.add_paragraph() paragraph.paragraph_format.line_spacing_rule = WD_LINE_SPACING.ONE_POINT_FIVE paragraph.paragraph_format.space_before = Pt(0) paragraph.paragraph_format.space_after = Pt(6) if bold_prefix and text.startswith(bold_prefix): first = paragraph.add_run(bold_prefix) set_run_font(first, BODY_FONT, BODY_SIZE, True) rest = paragraph.add_run(text[len(bold_prefix):]) set_run_font(rest, BODY_FONT, BODY_SIZE) else: run = paragraph.add_run(text) set_run_font(run, BODY_FONT, BODY_SIZE) def set_cell_margins(cell, left: int = 60, right: int = 60, top: int = 30, bottom: int = 30) -> None: tc_pr = cell._tc.get_or_add_tcPr() mar = tc_pr.first_child_found_in("w:tcMar") if mar is None: mar = OxmlElement("w:tcMar") tc_pr.append(mar) for key, value in (("w:left", left), ("w:right", right), ("w:top", top), ("w:bottom", bottom)): node = mar.find(qn(key)) if node is None: node = OxmlElement(key) mar.append(node) node.set(qn("w:w"), str(value)) node.set(qn("w:type"), "dxa") def fix_table_layout(table, widths: list[float]) -> None: tbl_pr = table._tbl.tblPr layout = OxmlElement("w:tblLayout") layout.set(qn("w:type"), "fixed") tbl_pr.append(layout) tbl_w = tbl_pr.find(qn("w:tblW")) if tbl_w is None: tbl_w = OxmlElement("w:tblW") tbl_pr.append(tbl_w) tbl_w.set(qn("w:w"), str(int(sum(widths) * 567))) tbl_w.set(qn("w:type"), "dxa") for index, column in enumerate(table._tbl.tblGrid.gridCol_lst): column.set(qn("w:w"), str(int(widths[index] * 567))) def no_split(row) -> None: tr_pr = row._tr.get_or_add_trPr() cant = OxmlElement("w:cantSplit") tr_pr.append(cant) def repeat_header(row) -> None: tr_pr = row._tr.get_or_add_trPr() header = OxmlElement("w:tblHeader") tr_pr.append(header) def add_table( doc: Document, headers: list[str], rows: list[list[str]], widths: list[float], font_size: float = TABLE_SIZE, align_center: bool = False, ) -> None: table = doc.add_table(rows=1, cols=len(headers)) table.style = "Table Grid" table.alignment = WD_TABLE_ALIGNMENT.CENTER table.autofit = False fix_table_layout(table, widths) header = table.rows[0] repeat_header(header) no_split(header) for index, text in enumerate(headers): cell = header.cells[index] cell.width = Cm(widths[index]) set_cell_margins(cell) cell.vertical_alignment = WD_ALIGN_VERTICAL.CENTER paragraph = cell.paragraphs[0] paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER paragraph.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE run = paragraph.add_run(text) set_run_font(run, BODY_FONT, font_size, True) for values in rows: row = table.add_row() no_split(row) for index, text in enumerate(values): cell = row.cells[index] cell.width = Cm(widths[index]) set_cell_margins(cell) cell.vertical_alignment = WD_ALIGN_VERTICAL.CENTER paragraph = cell.paragraphs[0] paragraph.alignment = ( WD_ALIGN_PARAGRAPH.CENTER if align_center or index > 1 else WD_ALIGN_PARAGRAPH.LEFT ) paragraph.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE run = paragraph.add_run(str(text)) set_run_font(run, BODY_FONT, font_size) doc.add_paragraph() def add_attr_table(doc: Document, fields: list[dict]) -> None: widths = [2.0, 6.5, 2.0, 6.5, 2.0, 6.5] table = doc.add_table(rows=0, cols=6) table.style = "Table Grid" table.alignment = WD_TABLE_ALIGNMENT.CENTER table.autofit = False fix_table_layout(table, widths) row = None pair = 0 def new_row(): nonlocal row, pair row = table.add_row() no_split(row) pair = 0 for index, width in enumerate(widths): cell = row.cells[index] cell.width = Cm(width) set_cell_margins(cell) cell.vertical_alignment = WD_ALIGN_VERTICAL.CENTER for field in fields: label = str(field.get("label") or field.get("key") or "") value = str(field.get("value") or "—") span = int(field.get("span") or 1) if span >= 2: if row is not None and pair > 0: row = None new_row() label_cell = row.cells[0] value_cell = row.cells[1].merge(row.cells[5]) for cell, text, bold in ((label_cell, label, True), (value_cell, value, False)): cell.vertical_alignment = WD_ALIGN_VERTICAL.CENTER paragraph = cell.paragraphs[0] paragraph.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE run = paragraph.add_run(text) set_run_font(run, BODY_FONT, TABLE_SIZE, bold) row = None pair = 0 continue if row is None or pair >= 3: new_row() label_cell = row.cells[pair * 2] value_cell = row.cells[pair * 2 + 1] for cell, text, bold in ((label_cell, label, True), (value_cell, value, False)): cell.vertical_alignment = WD_ALIGN_VERTICAL.CENTER paragraph = cell.paragraphs[0] paragraph.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE run = paragraph.add_run(text) set_run_font(run, BODY_FONT, TABLE_SIZE, bold) pair += 1 doc.add_paragraph() def load_labels() -> list[dict]: with urlopen(API_URL, timeout=20) as response: payload = json.load(response) return payload["items"] def fetch_ready() -> None: for name, url in ( ("后端服务", "http://127.0.0.1:8000/api/tag-manual/labels"), ("前端预览", "http://127.0.0.1:5173/preview.html"), ): try: with urlopen(url, timeout=5): pass except Exception as error: raise SystemExit( f"{name}未启动或不可访问:{error}\n" "请先启动:uvicorn app.main:app --app-dir backend --port 8000\n" "并启动:cd frontend && npm run dev" ) from error def screenshot(label: dict, profile_dir: Path, force: bool = False) -> Path: FIG_DIR.mkdir(parents=True, exist_ok=True) jpg_path = FIG_DIR / f"{label['key'].replace(':', '_')}.jpg" if jpg_path.exists() and not force: return jpg_path png_path = FIG_DIR / f"{label['key'].replace(':', '_')}.png" url = f"{BASE_URL}?key={quote(label['key'])}&chart=1" command = [ CHROME, "--headless=new", "--disable-gpu", "--hide-scrollbars", "--window-size=1800,900", "--force-device-scale-factor=2", "--virtual-time-budget=20000", "--run-all-compositor-stages-before-draw", f"--user-data-dir={profile_dir}", f"--screenshot={png_path}", url, ] for attempt in range(2): subprocess.run( command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120, ) if png_path.exists() and png_path.stat().st_size > 5000: break time.sleep(1) if not png_path.exists() or png_path.stat().st_size <= 5000: raise RuntimeError(f"截图失败:{label['key']}") image = Image.open(png_path).convert("RGB") image = image.resize((2200, 1100), Image.LANCZOS) image.save(jpg_path, quality=90, optimize=True) png_path.unlink(missing_ok=True) return jpg_path def cycle_name(label: dict) -> str: cycle = label.get("cycle") or "" unit = label.get("device") or "" prefix = f"{unit}-" return cycle[len(prefix):] if cycle.startswith(prefix) else cycle def sort_key(label: dict) -> tuple: unit = label.get("device") or "" return ( DEVICE_ORDER.get(unit, 99), KIND_ORDER.get(label.get("kind"), 9), label.get("start") or "", ) def add_cover(doc: Document, labels: list[dict]) -> None: for _ in range(4): doc.add_paragraph() paragraph = doc.add_paragraph() paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run("组态数据高质量演化过程") set_run_font(run, HEAD_FONT, 26, True) paragraph = doc.add_paragraph() paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run("故障标签手册") set_run_font(run, HEAD_FONT, 26, True) doc.add_paragraph() paragraph = doc.add_paragraph() paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run("润滑油压力故障 · 振动故障") set_run_font(run, BODY_FONT, 16, False) for _ in range(6): doc.add_paragraph() paragraph = doc.add_paragraph() paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run(f"共 {len(labels)} 条演化标签") set_run_font(run, BODY_FONT, 12, False) paragraph = doc.add_paragraph() paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER run = paragraph.add_run(f"{date.today().year} 年 {date.today().month} 月") set_run_font(run, BODY_FONT, 12, False) doc.add_page_break() def add_reading_guide(doc: Document) -> None: doc.add_heading("文档阅读说明", level=1) body_paragraph( doc, "用途:本手册汇总设备运行过程中由规则扫描形成的润滑油压力、振动两类演化标签," "逐条给出标签属性、演化阶段和标签前后运行曲线,用于复核、诊断和维修决策辅助。", "用途:", ) body_paragraph( doc, "阅读方法:先阅读标签数量总结,了解总体分布和演化过程;再按章节、机组或阶段查阅详情。" "在 Word 中打开导航窗格,可通过一级、二级标题直接跳转到对应章节或标签。", "阅读方法:", ) body_paragraph( doc, "排序方式:详情部分分为润滑油压力、振动两章,章内按机组和开始时间排序;" "同一标签的所有属性集中在一张表中,图表位于属性表下方。", "排序方式:", ) body_paragraph( doc, "数据口径:标签来自演化阶段统计记录;曲线来自设备运行监测数据,只使用运行样本," "停机、坏点和无效值不参与曲线绘制。标签是规则扫描结果,不等同于物理故障结论。", "数据口径:", ) doc.add_heading("图形解释", level=2) rows = [ ["红色背景", "当前标签对应的故障阶段,画在当前类型子图上。"], ["淡黄色背景", "总表中其他故障标签与当前展示窗口重合的时间段;与红色故障段重叠的部分不重复标注。"], ["竖向虚线", "停机边界。两条虚线为一组,左侧为关机,右侧为开机。"], ["标题下备注", "故障段时间、图表展示时间,以及各停机段的关机时间和开机时间。"], ["相对基线子图", "润滑油压力按周期基线计算百分比;振动按近期基线计算倍数;虚线为基线。"], ["原值子图", "测点实际值,用于核对曲线形态和数值范围。"], ["右上角角标", "标签阶段:轻微、异常或严重异常。"], ["底部滑块", "时间轴缩放控件,默认显示全部时间范围。"], ["子图", "每个子图只画一条曲线,所有子图共用同一时间轴。"], ] add_table(doc, ["图形元素", "含义"], rows, [5.0, 20.5], font_size=9.5) body_paragraph( doc, "展示范围:故障段前后各补充 6 小时有效运行数据。若期间发生停机,会继续向更早或更晚的" "运行数据取数,停机段压缩为空档并用竖向虚线标出。停机后 20 分钟的启动瞬态在故障段外" "不展示,故障段内的数据不剔除。", "展示范围:", ) body_paragraph( doc, "本版为正文定稿版,仅保留一条示例曲线。其余标签的曲线图待图表调整完成后统一补充。", "图表说明:", ) def add_summary(doc: Document, labels: list[dict]) -> None: doc.add_heading("一、标签数量总结", level=1) body_paragraph( doc, f"本手册共收录 {len(labels)} 条演化标签,覆盖润滑油压力故障和振动故障两类," "按机组、阶段和周期编号统计如下。", ) kind_counts = Counter(label["kindLabel"] for label in labels) device_kind = Counter((label["device"], label["kindLabel"]) for label in labels) stage_kind = Counter((label["stage"], label["kindLabel"]) for label in labels) device_stage = Counter((label["device"], label["stage"]) for label in labels) doc.add_heading("1.1 按类型统计", level=2) rows = [ [name, str(kind_counts.get(name, 0)), f"{kind_counts.get(name, 0) / len(labels) * 100:.1f}%"] for name in ("润滑油压力", "振动") ] add_table(doc, ["类型", "数量", "占比"], rows, [7.0, 5.0, 5.0], align_center=True) doc.add_heading("1.2 按机组统计", level=2) rows = [] for device in ("7号机", "8号机", "9号机"): pressure = device_kind.get((device, "润滑油压力"), 0) vibration = device_kind.get((device, "振动"), 0) rows.append([device, str(pressure), str(vibration), str(pressure + vibration)]) rows.append([ "合计", str(kind_counts.get("润滑油压力", 0)), str(kind_counts.get("振动", 0)), str(len(labels)), ]) add_table(doc, ["机组", "润滑油压力", "振动", "合计"], rows, [6.0, 6.0, 6.0, 6.0], align_center=True) doc.add_heading("1.3 按阶段统计", level=2) rows = [] for stage in ("轻微", "异常", "严重异常"): pressure = stage_kind.get((stage, "润滑油压力"), 0) vibration = stage_kind.get((stage, "振动"), 0) rows.append([stage, str(pressure), str(vibration), str(pressure + vibration)]) rows.append([ "合计", str(kind_counts.get("润滑油压力", 0)), str(kind_counts.get("振动", 0)), str(len(labels)), ]) add_table(doc, ["阶段", "润滑油压力", "振动", "合计"], rows, [6.0, 6.0, 6.0, 6.0], align_center=True) doc.add_heading("1.4 按机组和阶段统计", level=2) rows = [] for device in ("7号机", "8号机", "9号机"): values = [device_stage.get((device, stage), 0) for stage in ("轻微", "异常", "严重异常")] rows.append([device, *[str(value) for value in values], str(sum(values))]) total_stage = Counter(label["stage"] for label in labels) rows.append([ "合计", *[str(total_stage.get(stage, 0)) for stage in ("轻微", "异常", "严重异常")], str(len(labels)), ]) add_table( doc, ["机组", "轻微", "异常", "严重异常", "合计"], rows, [6.0, 4.5, 4.5, 5.0, 5.0], align_center=True, ) doc.add_heading("1.5 按周期编号统计", level=2) grouped: dict[tuple, list[dict]] = defaultdict(list) for label in labels: grouped[(label["device"], label["kind"], cycle_name(label))].append(label) rows = [] for (device, kind, cycle), items in sorted( grouped.items(), key=lambda item: ( DEVICE_ORDER.get(item[0][0], 99), KIND_ORDER.get(item[0][1], 9), item[0][2], ), ): stages = Counter(item["stage"] for item in items) starts = min(item["start"] for item in items) ends = max(item["end"] for item in items) rows.append([ device, KIND_LABEL.get(kind, kind), cycle, str(stages.get("轻微", 0)), str(stages.get("异常", 0)), str(stages.get("严重异常", 0)), str(len(items)), starts[:16], ends[:16], ]) add_table( doc, ["机组", "类型", "周期编号", "轻微", "异常", "严重异常", "合计", "起始", "结束"], rows, [2.2, 2.4, 3.0, 1.4, 1.4, 1.8, 1.4, 4.8, 4.8], font_size=8.0, align_center=True, ) doc.add_heading("1.6 总体演化过程", level=2) pressure = kind_counts.get("润滑油压力", 0) vibration = kind_counts.get("振动", 0) slight = total_stage.get("轻微", 0) abnormal = total_stage.get("异常", 0) severe = total_stage.get("严重异常", 0) min_start = min(label["start"] for label in labels) max_end = max(label["end"] for label in labels) body_paragraph( doc, f"标签时间覆盖 {min_start[:10]} 至 {max_end[:10]}。润滑油压力 {pressure} 条," f"占 {pressure / len(labels) * 100:.1f}%;振动 {vibration} 条,占 " f"{vibration / len(labels) * 100:.1f}%。", ) body_paragraph( doc, f"按阶段看,轻微 {slight} 条、异常 {abnormal} 条、严重异常 {severe} 条。" "多数标签处于轻微阶段,异常和严重异常数量较少,符合演化过程由轻到重逐步发展的特点。", ) body_paragraph( doc, "润滑油压力标签以轻微阶段为主,异常和严重异常主要出现在持续下降过程中;" "振动标签在轻微、异常和严重异常三个阶段均有分布,说明振动异常既可能表现为短时冲击," "也可能持续发展。", ) body_paragraph( doc, "阅读详情时,应结合周期基线、相对变化、连续有效小时数、报警小时数和阶段触发依据," "判断标签是单次波动还是持续演化,并结合现场维修记录进行确认。", ) rows = [ ["轻微", "相对基线偏离达到轻微门槛,持续时间较短或尚未形成连续确认。"], ["异常", "相对基线偏离扩大,连续有效小时数增加,形成异常阶段。"], ["严重异常", "达到严重门槛或触发低低报、高高报,应优先复核。"], ] add_table(doc, ["阶段", "含义"], rows, [4.0, 21.5], font_size=9.5) def add_detail_chapters( doc: Document, labels: list[dict], profile_dir: Path, force: bool, limit: int, single_image: str, ) -> None: chapters = ( ("pressure", "二、润滑油压力故障标签详情", "润滑油压力"), ("vibration", "三、振动故障标签详情", "振动"), ) for chapter_index, (kind, title, kind_label) in enumerate(chapters): items = sorted( [label for label in labels if label.get("kind") == kind], key=sort_key, ) if limit: items = items[:limit] doc.add_page_break() doc.add_heading(title, level=1) stage_counts = Counter(item.get("stage") for item in items) body_paragraph( doc, f"本章共 {len(items)} 条{kind_label}故障标签,其中轻微 {stage_counts.get('轻微', 0)} 条、" f"异常 {stage_counts.get('异常', 0)} 条、严重异常 {stage_counts.get('严重异常', 0)} 条。" "章内按机组和开始时间排列;每个二级标题对应一条标签,标题下方为属性表。" "本版仅在一条标签下放置示例曲线,其余曲线待统一补充。", ) for index, label in enumerate(items, start=1): print(f"[{kind} {index}/{len(items)}] {label['title']}") if index > 1: doc.add_page_break() heading = ( f"{label['device']}-{label['kindLabel']}-{cycle_name(label)}-{label['start'][:16]}" ) doc.add_heading(heading, level=2) add_attr_table(doc, label.get("fields") or []) if single_image and label["key"] == single_image: image_path = screenshot(label, profile_dir, force=force) picture_paragraph = doc.add_paragraph() picture_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER picture_paragraph.paragraph_format.space_before = Pt(4) picture_paragraph.paragraph_format.space_after = Pt(2) picture_paragraph.paragraph_format.keep_with_next = True run = picture_paragraph.add_run() run.add_picture(str(image_path), width=Cm(24.0)) caption = doc.add_paragraph() caption.alignment = WD_ALIGN_PARAGRAPH.CENTER caption.paragraph_format.line_spacing_rule = WD_LINE_SPACING.SINGLE caption.paragraph_format.space_after = Pt(6) run = caption.add_run("图:标签前后运行曲线(红色为故障段,竖向虚线为关机/开机边界)") set_run_font(run, BODY_FONT, 9.0) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--limit", type=int, default=0, help="只生成前 N 条,用于冒烟测试") parser.add_argument("--output", type=Path, default=DEFAULT_DOCX) parser.add_argument("--force", action="store_true", help="强制重新截图") parser.add_argument( "--single-image", default="pressure:44", help="只插入该标签的曲线图;传空字符串表示全部插入", ) args = parser.parse_args() fetch_ready() labels = load_labels() print(f"标签总数:{len(labels)}") args.output.parent.mkdir(parents=True, exist_ok=True) profile_dir = Path(tempfile.mkdtemp(prefix="tag_manual_chrome_")) try: doc = Document() configure_document(doc) add_cover(doc, labels) add_reading_guide(doc) add_summary(doc, labels) add_detail_chapters( doc, labels, profile_dir, args.force, args.limit, args.single_image, ) doc.save(args.output) print(f"已生成:{args.output}") print(f"截图目录:{FIG_DIR}") finally: shutil.rmtree(profile_dir, ignore_errors=True) if __name__ == "__main__": main()