#!/usr/bin/env python3 # -*- coding: utf-8 -*- """厂家振动报告 (docx) → 结构化提取 + 消费端格式的评估报告 (2026-09-12 用户令). ## 范围与诚实边界 (先读这段, 再看代码) 数据层「CMS 振动评估报告」的源件有两类, 处理方式**不同**: · CMS 原始测量导出 (Brande TCM `*_decode.json`) → 可重算, 走 scripts/vib_raw_build.py (索引/谱/六层链), 这一路是本目录下 `report_CMS振动状态评估报告_*.md` 的**正路**。 · 厂家月度评估报告 (上海电气 12 份用印版 PDF + 2026年07月 docx; 大生科技传动链 docx) → **PDF 全部是扫描件** (2026-09-12 用 pypdf 抽检: 12 页 12 图, extract_text 长度 0), 没有 OCR 就取不出数值 ⇒ 只登记归档, 不进判级; **docx 有文本层**, 本脚本把里面的逐台判级**逐字转录**成结构化件与一份 report md。 **转录纪律** (照抄本包既有做法, 防"看起来像算出来的"): 每个产物的 meta 里写清源文件+sha256+表头+行数; 取值一律来自报告原文, 不补不猜; 报告没给的列写 '—' 并在说明里点明"厂家未给"; 聚合量(如综合级) 必须标注它是**取严规则**得出的, 不是厂家判的。 ## 输出的两份东西 (都在 outputs/<场>/ 下, 与既有消费者兼容) windcms/厂家报告提取_<报告期>.json 逐台原文 + provenance (给机器读/复核) windcms/报告_CMS振动状态评估报告_<报告期>.md 与 windcms 自产报告**同构**: `## 附录 A` 下 `| WTGxx | 主轴承前 | 主轴承后 | 齿轮箱 | 发电机 | 综合 | …` (消费者: src/windscada/subsys/fusion.py::windcms_grades 取第 6 列=综合, src/windscada/taxonomy.py 取第 2..5 列)。日期用**报告期**, 于是它不会顶掉更新的自产报告, 但随包自产报告缺失时它就是最新版 (从零重算的机器上正好用得上)。 m5_cms_tcm/厂家报告提取_<报告期>.json + 报告_TCM传动链振动分析_<报告期>.md 大生科技 (TCM M-system 数据) 逐台状态等级/分析/建议, 原文透传; **不**冒充 handoff 判级。 用法: python scripts/vib_reports_build.py --dump # 只打印解析结果 (人工核对用, 不写盘) python scripts/vib_reports_build.py # 摄入并写产物 """ from __future__ import annotations import argparse import hashlib import json import pathlib import re import sys import time ROOT = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.console import soft # noqa: E402 soft() STATE_ORDER = ['危险', '报警', '良好', '优秀', '不可判'] # 取严: 越靠前越严重 (报告用四级 + 不可判) NO_DATA_MARKS = ('--', '—', '无数据', '通讯中断', '无通讯') def sha256(p: pathlib.Path) -> str: h = hashlib.sha256() with open(p, 'rb') as f: for chunk in iter(lambda: f.read(1 << 20), b''): h.update(chunk) return h.hexdigest()[:16] def grid_of(table): """docx 表 → 二维网格, **正确还原合并单元格**。 为什么不能直接用 `row.cells`: python-docx 对合并区会把同一个 `tc` 在相邻列/相邻行重复给出 —— 照抄等于把 "优秀 | 优秀 | 优秀" 压成 "优秀", 列就对不齐了 (第一版手工 dump 就吃了这个亏: WTG-01 行显示成 `WTG-01 | 优秀 | 无`, 而实际是 主轴承=优秀 齿轮箱=优秀 发电机=优秀 结论=无)。 规则: 同一行里 tc 与左邻相同 = 横向合并 (沿用左值); tc 与上方同行同列相同 = 纵向合并 (沿用上值)。 """ ncol = len(table.columns) out, tc_above = [], [None] * ncol for row in table.rows: vals, tcs, prev = [], [], None for ci, cell in enumerate(row.cells): tc = cell._tc if tc is prev: # 横向合并的续格 vals.append(vals[-1] if vals else '') tcs.append(tc) continue if tc is tc_above[ci] and ci < len(tc_above): # 纵向合并的续格 vals.append(out[-1][ci] if out else '') else: vals.append(cell.text.strip().replace('\n', ' ')) tcs.append(tc) prev = tc tc_above = tcs out.append(vals) return out def find_table(doc, must_have, header_hint=None): """按表头关键字找表 → (index, grid)。must_have: 表头行必须都含这些词。""" import docx for i, tb in enumerate(doc.tables): g = grid_of(tb) if g and all(any(k in c for c in g[0]) for k in must_have): return i, g return None, None def norm_turbine(s): m = re.search(r'WTG[\s_-]*0*(\d{1,2})', str(s).upper()) return f'WTG{int(m.group(1)):02d}' if m else None def state_of(s): """文本 → 五级之一; 认不出给 '不可判' (宁可说不可判, 不冒充优秀)。""" t = str(s or '').strip() if any(k in t for k in NO_DATA_MARKS) or t == '': return '不可判' for v in ('危险', '报警', '预警', '良好', '优秀'): if v in t: return {'预警': '报警'}.get(v, v) # 上海电气四级里没有"预警", 大生科技有; 此处按四级归一 if '异常' in t: return '不可判' return '不可判' def worst(*states): """取严 (综合列的口径; 厂家未给综合列时必须写明这是我们的聚合规则)。""" live = [s for s in states if s] return min(live, key=lambda s: STATE_ORDER.index(s) if s in STATE_ORDER else 99) if live else '不可判' def parse_sa(path: pathlib.Path): """上海电气月度振动分析报告: 表「机组号|主轴承|齿轮箱|发电机|诊断结论和维护建议」+ 总览句校验和。""" import docx d = docx.Document(str(path)) ti, g = find_table(d, ['机组号', '主轴承', '齿轮箱']) if g is None: return None head = g[0] idx = {name: next((i for i, c in enumerate(head) if name in c), None) for name in ('机组号', '主轴承', '齿轮箱', '发电机', '诊断')} rows = [] for r in g[1:]: t = norm_turbine(r[idx['机组号']] if idx['机组号'] is not None else '') if not t: continue rows.append(dict(turbine=t, 主轴承=state_of(r[idx['主轴承']]) if idx['主轴承'] is not None else '不可判', 齿轮箱=state_of(r[idx['齿轮箱']]) if idx['齿轮箱'] is not None else '不可判', 发电机=state_of(r[idx['发电机']]) if idx['发电机'] is not None else '不可判', 厂家结论=(r[idx['诊断']] if idx['诊断'] is not None and idx['诊断'] < len(r) else ''))) # 总览句 = 报告的**自带校验和** (2026-07 报告原文: 优秀34 良好3 预警0 无数据1 测点异常1)。 # ★它在**表格单元格**里, 不是段落 (第一版只扫 paragraphs → 拿到空串, 于是"校验和"形同虚设)。 overview, sums = '', {} texts = [p.text for p in d.paragraphs] for tb in d.tables: for row in tb.rows: for c in row.cells: texts.append(c.text) for t in texts: if '检测结果' in t and '机组' in t and '台' in t: overview = t.strip() break for k in ('优秀', '良好', '预警', '无数据', '测点异常'): # 原话是"运行状态优秀机组34台" —— 等级词与数字之间夹着"机组"二字, 只写 `{k}\s*(\d+)` 会漏掉优秀 m = re.search(rf'{k}[^0-9]{{0,6}}(\d+)\s*台', overview) if m: sums[k] = int(m.group(1)) period = re.search(r'(20\d\d)\s*年\s*(\d{1,2})\s*月', path.name) # 报告期: 文件名 2026年07月 → 用**月末** (报告覆盖整个月; 用月初会让它比同类报告显得更新) import calendar if period: y, mo = int(period.group(1)), int(period.group(2)) period = f'{y:04d}-{mo:02d}-{calendar.monthrange(y, mo)[1]:02d}' else: period = time.strftime('%Y-%m-%d') return dict(kind='上海电气月度', file=path.name, sha256=sha256(path), table_index=ti, header=[c for c in head], period=period, overview=overview, overview_counts=sums, rows=rows) def parse_ds(path: pathlib.Path): """大生科技传动链振动分析报告: 表「机组号|状态等级|分析与结论|建议」+ 数据窗 (报告正文)。""" import docx d = docx.Document(str(path)) ti, g = find_table(d, ['机组号', '状态等级', '分析']) if g is None: return None head = g[0] idx = {name: next((i for i, c in enumerate(head) if name in c), None) for name in ('机组号', '状态等级', '分析', '建议')} rows = [] for r in g[1:]: t = norm_turbine(r[idx['机组号']] if idx['机组号'] is not None else '') if not t: continue rows.append(dict(turbine=t, 状态等级=str(r[idx['状态等级']]).strip() if idx['状态等级'] is not None else '', 分析与结论=r[idx['分析']] if idx['分析'] is not None else '', 建议=r[idx['建议']] if idx['建议'] is not None else '')) txt = '\n'.join(p.text for p in d.paragraphs) # 数据窗: 正文原话 "本次分析采集了 A 至 B 期间的振动监测数据" (两个时间戳都要带时刻) win = re.search(r'(\d{4}年\d{1,2}月\d{1,2}日\s*\d{1,2}:\d{2}:\d{2})\s*至\s*' r'(\d{4}年\d{1,2}月\d{1,2}日\s*\d{1,2}:\d{2}:\d{2})', txt) rep = re.search(r'报告日期[:：]\s*([\d.]{8,10})', txt) period = (rep.group(1).replace('.', '-') if rep else time.strftime('%Y-%m-%d')) if re.fullmatch(r'\d{4}-\d{1,2}-\d{1,2}', period): y, m, dd = period.split('-') period = f'{y}-{int(m):02d}-{int(dd):02d}' return dict(kind='大生科技传动链', file=path.name, sha256=sha256(path), table_index=ti, header=[c for c in head], period=period, data_window=(f'{win.group(1)} ~ {win.group(2)}' if win else '未标'), rows=rows) def md_sa(ext, src_note): """上海电气提取 → 与 windcms 自产报告同构的评估报告 (附录 A 表)。""" L = [f'# CMS 振动状态评估报告 (厂家件转录) {ext["period"]}', '', f'> 来源: `{ext["file"]}` (sha256 {ext["sha256"]}, 表 {ext["table_index"]}) —— {src_note}', '', f'> 报告原文总览: {ext["overview"] or "(未找到总览句)"}', '', '> ★ 本表是从**厂家月度振动分析报告**逐字转录的（上海电气, 依据 VDI3834 与 NB/T 31129-2018）:', '> ① 厂家按「主轴承 / 齿轮箱 / 发电机」给级, **主轴承不分前后** —— 本表两列填同一值, 不是我们分的;', '> ② 厂家**未给综合列**, 本表「综合」= 三项**取严**(危险>报警>良好>优秀>不可判) 的聚合规则结果;', '> ③ 厂家未给「融合级/CMS 红黄/行动等级建议」, 一律写 `—` (不猜);', '> ④ 判级细节(谱/门槛)以厂家报告与 CMS 原始导出分析为准, 本表只承接"哪台什么级"。', '', '## 附录 A　38 台状态等级表（厂家报告转录，不代表确诊数量）', '', '| 机组号 | 主轴承前 | 主轴承后 | 齿轮箱 | 发电机 | 综合 | 融合级(模型) | 厂家结论与建议 |', '|---|---|---|---|---|---|---|---|'] for r in sorted(ext['rows'], key=lambda x: x['turbine']): zh = worst(r['主轴承'], r['齿轮箱'], r['发电机']) note = (r['厂家结论'] or '—').replace('|', '/') L.append(f"| {r['turbine']} | {r['主轴承']} | {r['主轴承']} | {r['齿轮箱']} | {r['发电机']} | " f"{zh} | — | {note} |") L += ['', '## 说明事项', '', '1. 本件是**厂家报告的转录**, 不是观澜自算结果; 观澜自算报告 (CMS 原始导出 → 六层链) 见同目录 `报告_CMS振动状态评估报告_*.md`。', '2. 厂家报告里的「无数据/通讯中断/测点异常」在本表记为 `不可判` —— 没测到 ≠ 正常。', f'3. 报告期为 {ext["period"]} (文件名月份, 取月末)。', ''] return '\n'.join(L) def md_ds(ext, src_note): L = [f'# TCM 传动链振动分析报告 (厂家件转录) {ext["period"]}', '', f'> 来源: `{ext["file"]}` (sha256 {ext["sha256"]}, 表 {ext["table_index"]}) —— {src_note}', '', f'> 数据窗: {ext["data_window"]} 报告日期: {ext["period"]}', '', '> ★ 原文透传: 状态等级/分析与结论/建议按厂家报告逐字记录。**本件不构成 handoff 判级** ——', '> handoff 的定谳链判级 (候选以上/证据窗) 是振动线接口件, 本报告只作 TCM 侧证据与交叉参考。', '', '| 机组 | 状态等级 | 分析与结论 | 建议 |', '|---|---|---|---|'] for r in sorted(ext['rows'], key=lambda x: x['turbine']): cells = [r['turbine'], r['状态等级'], r['分析与结论'] or '—', r['建议'] or '—'] L.append('| ' + ' | '.join(str(c).replace('|', '/') for c in cells) + ' |') L += ['', '## 说明事项', '', '1. 判据/图谱在厂家报告原文 (本包只落可解析的 docx; 若同批有 PDF 扫描件, 扫描件只归档不解析)。', '2. 状态等级为厂家四级口径 (优秀/良好/预警/报警) + "数据异常"(系统故障无通讯), 与观澜五级不是同一套, 消费时别直接比。', ''] return '\n'.join(L) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument('--farm', default='rudong') ap.add_argument('--dump', action='store_true', help='只打印, 不写盘') a = ap.parse_args() from src.windscada.config import raw_station_dir from src import paths as P station = pathlib.Path(raw_station_dir(a.farm)) sa_dir = station / 'windcms' / '厂家报告' / '上海电气_月度' ds_dir = station / 'm5_cms_tcm' / '厂家报告' out_cms = P.cms(a.farm) out_m5 = P.m5(a.farm) sa_files = sorted(sa_dir.glob('*.docx')) if sa_dir.is_dir() else [] ds_files = sorted(ds_dir.glob('*.docx')) if ds_dir.is_dir() else [] print(f'上海电气 docx: {len(sa_files)} 件 {sa_dir}') for p in sa_files: print(f' {p.name} {p.stat().st_size / 1048576:.2f} MB') print(f'大生科技 docx: {len(ds_files)} 件 {ds_dir}') for p in ds_files: print(f' {p.name} {p.stat().st_size / 1048576:.2f} MB') pdfs = sorted(sa_dir.glob('*.pdf')) if sa_dir.is_dir() else [] if pdfs: print(f'扫描件 PDF: {len(pdfs)} 件 (登记归档, 不解析)') if not sa_files and not ds_files: print('没有可解析的 docx 厂家报告 (只有扫描件 PDF 属正常) → 空跑退出') return 0 written = [] for p in sa_files: ext = parse_sa(p) if not ext: print(f'⚠ {p.name}: 没找到"机组检测结果列表"表 (跳过, 不猜)') continue zh = [worst(r['主轴承'], r['齿轮箱'], r['发电机']) for r in ext['rows']] print(f'\n== {p.name} ({ext["kind"]}, 表{ext["table_index"]}) ==') print(f' 报告期 {ext["period"]} 台数 {len(ext["rows"])} 原文总览: {ext["overview"]}') from collections import Counter got = Counter(zh) print(f' 转录结果取严分布: {dict(got)}') print(f' 厂家总览句: {ext["overview_counts"]}') if a.dump: for r in ext['rows'][:6]: print(' ', r) continue out_cms.mkdir(parents=True, exist_ok=True) jp = out_cms / f'厂家报告提取_{ext["period"]}.json' jp.write_text(json.dumps(ext, ensure_ascii=False, indent=1), encoding='utf-8') mp = out_cms / f'报告_CMS振动状态评估报告_{ext["period"]}.md' mp.write_text(md_sa(ext, '现场包 厂家月度评估报告'), encoding='utf-8') written += [jp, mp] print(f' → {jp.name}\n → {mp.name}') for p in ds_files: ext = parse_ds(p) if not ext: print(f'⚠ {p.name}: 没找到"信号分析与结论描述"表 (跳过, 不猜)') continue print(f'\n== {p.name} ({ext["kind"]}, 表{ext["table_index"]}) ==') print(f' 报告期 {ext["period"]} 数据窗 {ext["data_window"]} 台数 {len(ext["rows"])}') from collections import Counter print(f' 状态等级分布: {dict(Counter(r["状态等级"] for r in ext["rows"]))}') if a.dump: for r in ext['rows'][:4]: print(' ', {k: (v[:60] + '…' if isinstance(v, str) and len(v) > 60 else v) for k, v in r.items()}) continue out_m5.mkdir(parents=True, exist_ok=True) jp = out_m5 / f'厂家报告提取_{ext["period"]}.json' jp.write_text(json.dumps(ext, ensure_ascii=False, indent=1), encoding='utf-8') mp = out_m5 / f'报告_TCM传动链振动分析_{ext["period"]}.md' mp.write_text(md_ds(ext, '现场包 大生科技 TCM 传动链分析报告 (docx)'), encoding='utf-8') written += [jp, mp] print(f' → {jp.name}\n → {mp.name}') if not a.dump and pdfs: reg = dict(note='扫描件登记: 无文本层 → 只归档, 数值不进系统; 需数值须 OCR 或取厂家电子件', verified='2026-09-12 用 pypdf 抽检一件: 12 页 / 每页 1 图 / extract_text() 长度 0', files=[dict(name=q.name, bytes=q.stat().st_size, sha256=sha256(q)) for q in pdfs]) rp = sa_dir / '_扫描件清单.json' rp.write_text(json.dumps(reg, ensure_ascii=False, indent=1), encoding='utf-8') written.append(rp) print(f'\n扫描件登记 → {rp}') # 来源自登记 (见 src/derived_manifest.py): 转录件也是"从 data/raw 来的", 台账要记 raw-derived if not a.dump and written: try: from src.derived_manifest import record as _rec from src import paths as _P rels = {} for p in written: if p.suffix.lower() not in ('.json', '.md') or '_扫描件清单' in p.name: continue if p.parent == out_cms: rels[f'windcms/{p.name}'] = 'scripts/vib_reports_build.py (厂家 docx 报告逐台转录)' elif p.parent == out_m5: rels[f'm5_cms_tcm/{p.name}'] = 'scripts/vib_reports_build.py (厂家 docx 报告逐台转录)' if rels: _rec(_P.out_root(a.farm), rels, by='scripts/vib_reports_build.py') print(f'自登记 {len(rels)} 件 → {_P.out_root(a.farm) / "_derived_manifest.json"}') except Exception as _e: print(f'⚠ 来源自登记失败 ({type(_e).__name__}: {_e})') print(f'\n完成: 写出 {len(written)} 件') return 0 if __name__ == '__main__': sys.exit(main())