guanlan-src/scripts/vib_reports_build.py

#!/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())