# -*- coding: utf-8 -*- """报告导出 Word / Excel (2026-08-28 用户令: 导出 demo)。 汇报画布选中的块 → .docx / .xlsx。图表块不丢信息: 转成对应的数据表 — Word 里放位图既不可编辑也不能核对, 上报集团要的是能核的数字。 脱敏跟随受众设置 (上报版沿用服务端 _redact 的规则)。 """ import io, datetime as dt TITLE = '如东海上风电场 · 运行分析报告' def _fmt(v, n=0): try: return f'{float(v):,.{n}f}' except Exception: return str(v) def blocks_data(fl, blocks, narr=None): """把画布块转成 [(标题, 段落[], 表格{head, rows})] — Word/Excel 共用同一份中间表示。""" E = (fl.get('control') or {}).get('energy') or {} R = fl.get('rel') or {} M = ((fl.get('m8') or {}).get('mtbf')) or {} F = fl.get('faults') or {} FU = fl.get('fus') or {} BD = FU.get('链盘') or {} C = fl.get('control') or {} out = [] def add(t, paras=None, head=None, rows=None): out.append(dict(title=t, paras=paras or [], head=head, rows=rows or [])) for k in blocks: if k == 'kpi': add('一、核心指标', head=['指标', '数值', '口径'], rows=[ ['上网电量', f"{_fmt(E.get('act'))} MWh", f"理论可发 {_fmt(E.get('theo'))} − 损失 {_fmt(E.get('loss'))}"], ['等效满发小时', f"{_fmt(E.get('eflh'))} h", '上网电量 ÷ (台数×额定容量)'], ['时间可用率', f"{C.get('avail')}%", '分母已剔调度令与低风待机'], ['损失率', f"{E.get('loss_pct')}%", '损失 ÷ 理论可发'], ['平均停机间隔', f"{M.get('MTBF_h')} h", f"停机 {M.get('停机事件数')} 起, 单次均 {M.get('MDT_h')} h; 分母=窗内实测台时"], ['需行动机组', f"{len(BD.get('rows') or [])} 台", f"四源融合判级; 定论 {(FU.get('kpi') or {}).get('定论')} · 预警 {(FU.get('kpi') or {}).get('预警')} · 候选 {(FU.get('kpi') or {}).get('候选观察')}"], ]) elif k == 'water': rows = [['理论可发', _fmt(E.get('theo')), '100%']] for x in (E.get('items') or []): rows.append([x['k'], '−' + _fmt(x['loss']), f"{x['loss']/max(E.get('theo') or 1,1)*100:.1f}%"]) rows.append(['实际上网', _fmt(E.get('act')), f"{(E.get('act') or 0)/max(E.get('theo') or 1,1)*100:.1f}%"]) add('二、发电完成情况', [f"理论可发 {_fmt(E.get('theo'))} MWh = 上网 {_fmt(E.get('act'))} + 损失 {_fmt(E.get('loss'))} MWh, 损失率 {E.get('loss_pct')}%。"], ['项目', 'MWh', '占理论可发'], rows) elif k == 'emon': add('逐月发电', head=['月份', '上网 MWh', '损失 MWh', '损失率', '等效满发 h', '数据覆盖'], rows=[[x['m'], _fmt(x['act']), _fmt(x['loss']), f"{x['loss_pct']}%", _fmt(x['eflh']), f"{x['cov']*100:.0f}%" + ('(不满月)' if x['cov'] < 0.5 else '')] for x in (E.get('monthly') or [])]) elif k == 'blame': cat = lambda kk: '交易与调度侧' if ('限电' in kk or '调度' in kk) else ('设备侧' if kk == '停机' else '运行侧') g = {} for x in (E.get('items') or []): g[cat(x['k'])] = g.get(cat(x['k']), 0) + x['loss'] tot = sum(g.values()) or 1 add('损失责任分账', ['进设备考核的仅"故障与检修停机"; 限电属交易与调度侧。'], ['责任方', '损失 MWh', '占比'], [[a, _fmt(b), f'{b/tot*100:.0f}%'] for a, b in sorted(g.items(), key=lambda x: -x[1])]) elif k in ('ram', 'parts') and (R.get('表')): add('三、设备可靠性' if k == 'ram' else '部件停机明细', [f"总停机 {_fmt(R.get('总停机h'))} h = 设备类 {_fmt(R.get('设备类停机h'))} + 外部类 {_fmt(R.get('外部类停机h'))} (电网/风况/运维操作不计设备可靠性)。"], ['部件', '停机 h', '事件数', 'MTBF h', 'MDT h', '影响台数'], [[x['部件'] + ('(外部)' if x['外部'] else ''), _fmt(x['停机时长h']), x['事件数'], x['MTBF_h'] or '—', x['MDT_h'], x['影响台数']] for x in (R.get('表') or [])]) elif k == 'attrib' and R.get('表'): nr = next((x['停机时长h'] for x in R['表'] if x['部件'] == '无记录停机'), 0) add('归因覆盖', [f"无记录停机 {_fmt(nr)} h, 占总停机 " f"{nr/max(R.get('总停机h') or 1,1)*100:.1f}% — 属归因缺口, 非无故障; 成因待定。"]) elif k == 'pareto': add('四、故障统计', [f"窗内 {_fmt(F.get('total'))} 条报警 / {F.get('n_codes')} 个码。"], ['报警码与名称', '条数'], [[x['k'], _fmt(x['v'])] for x in (F.get('pareto_n') or [])]) elif k == 'quad': add('故障模式分型', ['条数高不等于问题严重: 单次时长在分钟级以下的属重复触发, 不应按条数排定优先级。'], ['码', '名称', '条数', '单次中位 s', '影响台数'], [[x['code'], x['k'], _fmt(x['n']), x['med'], x['nt']] for x in (F.get('quad') or [])[:12]]) elif k == 'frate': mo = F.get('monthly') or {} add('月度故障率', ['按天归一, 数据不满月单独标注。'], ['月份', '条数', '条/日', '数据覆盖'], [[m, _fmt(v), (mo.get('per_day') or [None])[i] if i < len(mo.get('per_day') or []) else '—', f"{(mo.get('cov') or [1])[i]*100:.0f}%" if i < len(mo.get('cov') or []) else '—'] for i, (m, v) in enumerate(zip(mo.get('months') or [], mo.get('vals') or []))]) elif k == 'units': add('五、重点机组与处置', [f"需跟踪 {len(BD.get('rows') or [])} 台 / 38。瓶颈: " + ', '.join(f"{v} 台处于「{kk}」环节" for kk, v in (BD.get('stuck') or {}).items()) + '。'], ['机组', '设备状态', '部件', '证据源', '六步进度', '当前环节'], [[r['t'].replace('WTG', '') + '#', r.get('ostate') or '—', (r.get('part') or '')[:22], '/'.join(r.get('srcs') or []) or '—', f"{r['done']}/6", r.get('stuck') or '全通'] for r in (BD.get('rows') or [])]) elif k == 'chain': add('决策链进度', ['六步: 证据 → 机制 → 判级 → 排期 → 动作 → 验收。'], ['环节', '停滞机组数'], [[kk, v] for kk, v in (BD.get('stuck') or {}).items()]) elif k == 'loop': add('六、检修执行与闭环', ['检出 → 换件 → 复测回落的历史例证。'], ['机组', '部件', '换件日期', '换前', '换后', '比值'], [[x['turbine'].replace('WTG', '') + '#', x['component'], x['replace_date'], x['pre'], x['post'], f"{x['ratio']}×"] for x in (BD.get('closed') or [])]) elif k == 'plan': add('七、下期计划与建议', [ f"1. 补机制定性 — {(BD.get('stuck') or {}).get('机制', 0)} 台已判级但根因未定, 排期与动作缺乏依据, 建议按预警级优先安排现场检查与取样。", '2. 回收 2025~2026 工单 — 台账止于 2024-11, 本期检修执行与闭环无法自动判定。', '3. 油液采样 — 目前多数机组末样时效偏旧, 仅可作回测锚, 建议纳入定期采样。', '4. 限电侧 — 损失主要发生在交易与调度侧, 设备侧改善空间有限。']) elif k == 'caliber': part = F.get('partial') or [] add('八、口径与数据边界', [ f"统计期 {fl.get('win')}" + (f"; 其中 {'/'.join(part)} 数据不满月, 月度对比已按天归一, 环比不纳入残月。" if part else '。'), '时间可用率分母已剔除调度令与低风待机; 损失率分母为理论可发, 两者分母不同, 不可直接互推。', f"停机口径: 可靠性按 ≥1h 段计 {_fmt(R.get('总停机h'))} h, 电量账按逐条工况态计; 两者差额为短于 1h 的停机段, 不可相减。", '工单台账 2020-01 ~ 2024-11, 现场未提供 2025~2026, 本报告不对本期检修执行下结论。', '判级由振动/温度/润滑/油液四源融合得出; 行动等级由综合报告出具。']) elif k == 'narr' and narr: add('运行总结', [narr]) return out def to_docx(fl, blocks, win, aud='in', narr=None): from docx import Document from docx.shared import Pt from docx.enum.text import WD_ALIGN_PARAGRAPH doc = Document() st = doc.styles['Normal'] st.font.name = 'Microsoft YaHei' st.font.size = Pt(10.5) h = doc.add_heading(TITLE, level=0) h.alignment = WD_ALIGN_PARAGRAPH.CENTER sub = doc.add_paragraph(f"统计期 {win} ｜ 38 × SWT-4.0-130" + (' ｜ 上报版(判据门槛与特征频率已脱敏)' if aud == 'up' else '')) sub.alignment = WD_ALIGN_PARAGRAPH.CENTER for b in blocks_data(fl, blocks, narr): doc.add_heading(b['title'], level=1) for p in b['paras']: doc.add_paragraph(p) if b['head'] and b['rows']: t = doc.add_table(rows=1, cols=len(b['head'])) t.style = 'Light Grid Accent 1' for i, c in enumerate(b['head']): t.rows[0].cells[i].text = str(c) for r in b['rows']: cells = t.add_row().cells for i, c in enumerate(r[:len(b['head'])]): cells[i].text = str(c) doc.add_paragraph() f = doc.add_paragraph(f"数据源 SCADA 10min · 报警事件 · 工单台账 · CMS 振动 · 油液化验　｜　出件 {dt.datetime.now():%Y-%m-%d}") f.runs[0].font.size = Pt(8.5) buf = io.BytesIO(); doc.save(buf); return buf.getvalue() def to_xlsx(fl, blocks, win, aud='in', narr=None): from openpyxl import Workbook from openpyxl.styles import Font, PatternFill wb = Workbook(); wb.remove(wb.active) hdr_f = Font(bold=True, color='FFFFFF') hdr_b = PatternFill('solid', fgColor='1E4D6B') for b in blocks_data(fl, blocks, narr): name = b['title'].split('、')[-1][:28] or 'sheet' ws = wb.create_sheet(name[:31]) r = 1 for p in b['paras']: ws.cell(r, 1, p); r += 1 if b['paras']: r += 1 if b['head']: for i, c in enumerate(b['head'], 1): cell = ws.cell(r, i, c); cell.font = hdr_f; cell.fill = hdr_b r += 1 for row in b['rows']: for i, c in enumerate(row[:len(b['head'])], 1): ws.cell(r, i, c) r += 1 for i, c in enumerate(b['head'], 1): w = max([len(str(c))] + [len(str(x[i-1])) for x in b['rows'] if len(x) >= i]) + 2 ws.column_dimensions[ws.cell(1, i).column_letter].width = min(w * 1.6, 52) if not wb.sheetnames: wb.create_sheet('空') buf = io.BytesIO(); wb.save(buf); return buf.getvalue()