#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 报告生成模块 - 生成Word和PDF格式的运维报告 """ import os from datetime import datetime from docx import Document from docx.shared import Inches, Pt, Cm, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.enum.table import WD_TABLE_ALIGNMENT import markdown from tag_names import get_tag_description def create_word_report(title, content, output_path=None): """ 创建Word格式报告 :param title: 报告标题 :param content: Markdown格式的报告内容 :param output_path: 输出文件路径 :return: 文件路径 """ doc = Document() # 设置中文字体 style = doc.styles['Normal'] font = style.font font.name = '微软雅黑' font.size = Pt(11) # 添加标题 heading = doc.add_heading(title, level=1) heading.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(20) # 添加报告元信息 doc.add_paragraph(f'生成时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}') doc.add_paragraph('生成系统:神东煤炭综采工作面数据管理与AI分析系统') doc.add_paragraph('') # 空行 # 解析Markdown内容 parse_markdown_to_docx(doc, content) # 保存 if output_path is None: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'report_{timestamp}.docx' output_path = os.path.join(os.path.dirname(__file__), 'outputs', filename) # 确保目录存在 os.makedirs(os.path.dirname(output_path), exist_ok=True) doc.save(output_path) return output_path def parse_markdown_to_docx(doc, markdown_text): """将Markdown文本解析并添加到Word文档""" lines = markdown_text.split('\n') current_list = None for line in lines: # 标题 if line.startswith('# '): if current_list: current_list = None heading = doc.add_heading(line[2:], level=1) for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(16) elif line.startswith('## '): if current_list: current_list = None heading = doc.add_heading(line[3:], level=2) for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(14) elif line.startswith('### '): if current_list: current_list = None heading = doc.add_heading(line[4:], level=3) for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(12) elif line.startswith('#### '): if current_list: current_list = None heading = doc.add_heading(line[5:], level=4) for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(11) elif line.startswith('- ') or line.startswith('* '): # 列表项 text = line[2:].strip() if current_list is None: current_list = True p = doc.add_paragraph(text, style='List Bullet') elif line.startswith('1. ') or line.startswith('1)'): # 有序列表 text = line[3:].strip() if line.startswith('1.') else line[3:].strip() p = doc.add_paragraph(text, style='List Number') elif line.startswith('**') and line.endswith('**'): # 加粗文本 text = line.strip('*') p = doc.add_paragraph() run = p.add_run(text) run.bold = True run.font.name = '微软雅黑' run.font.size = Pt(11) elif line.startswith('---') or line.startswith('==='): # 分隔线 doc.add_paragraph('') elif line.strip() == '': # 空行 pass else: # 普通文本 if current_list and line.strip(): # 如果之前在列表中,继续添加 p = doc.add_paragraph(line.strip(), style='List Bullet') else: current_list = None if line.strip(): p = doc.add_paragraph(line.strip()) return doc def create_data_table_report(unit_name, data, output_path=None): """ 创建包含数据表格的报告 :param unit_name: 单位名称 :param data: 数据列表 :param output_path: 输出路径 :return: 文件路径 """ doc = Document() # 设置样式 style = doc.styles['Normal'] font = style.font font.name = '微软雅黑' font.size = Pt(10) # 标题 title = f'{unit_name} 设备监测数据报告' heading = doc.add_heading(title, level=1) heading.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(18) doc.add_paragraph(f'报告时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}', style='Normal') doc.add_paragraph('') # 按设备分类 categories = {} for item in data: tag = item.get('tag_short_name', item.get('tag_name', '')) # 完整标签名(用于CSV名称映射查找) tag_full = item.get('tag_full_name', '') value = item.get('tag_value', item.get('value')) value_bool = item.get('tag_value_bool', item.get('value_bool')) unit = item.get('tag_unit', '') # 优先使用已有的中文名称,否则从映射表查找(优先用完整标签名) chinese_name = item.get('chinese_name') if not chinese_name or chinese_name == tag: # 先用完整标签名查找,再fallback到短名称 chinese_name = get_tag_description(tag_full if tag_full else tag) if 'Shearer' in tag or 'shearer' in tag: cat = '采煤机' elif any(x in tag for x in ['AFC', 'LOA', 'CRU']): cat = '三机' elif 'Emulsion' in tag or 'Spray' in tag: cat = '泵站' elif 'Loop' in tag: cat = '开关' else: cat = '其他' if cat not in categories: categories[cat] = [] display_value = value if value is not None else (value_bool if value_bool is not None else 'N/A') categories[cat].append({ 'tag': tag, 'chinese_name': chinese_name, 'value': display_value, 'unit': unit }) # 创建表格 for cat, items in categories.items(): doc.add_heading(cat, level=2) table = doc.add_table(rows=1, cols=3, style='Table Grid') table.alignment = WD_TABLE_ALIGNMENT.CENTER # 表头 hdr_cells = table.rows[0].cells hdr_cells[0].text = '参数名称' hdr_cells[1].text = '数值' hdr_cells[2].text = '单位' for cell in hdr_cells: for paragraph in cell.paragraphs: for run in paragraph.runs: run.font.name = '微软雅黑' run.font.size = Pt(10) run.bold = True # 数据行 for item in items: row_cells = table.add_row().cells # 使用中文名称显示,英文标签作为补充 row_cells[0].text = f"{item['chinese_name']}({item['tag']})" row_cells[1].text = str(item['value']) row_cells[2].text = item['unit'] for cell in row_cells: for paragraph in cell.paragraphs: for run in paragraph.runs: run.font.name = '微软雅黑' run.font.size = Pt(9) doc.add_paragraph('') # 保存 if output_path is None: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'{unit_name}_data_{timestamp}.docx' output_path = os.path.join(os.path.dirname(__file__), 'outputs', filename) os.makedirs(os.path.dirname(output_path), exist_ok=True) doc.save(output_path) return output_path def generate_full_report(unit_code, unit_name, data, ai_report, model_name, output_format='docx', output_path=None): """ 生成完整报告(数据+AI分析) :param unit_code: 单位代码 :param unit_name: 单位名称 :param data: 监测数据 :param ai_report: AI分析内容 :param model_name: AI模型名称 :param output_format: 输出格式 (docx/pdf) :param output_path: 输出路径 :return: 文件路径 """ doc = Document() # 设置样式 style = doc.styles['Normal'] font = style.font font.name = '微软雅黑' font.size = Pt(10) # 封面标题 title = f'{unit_name}\n综采工作面设备运维分析报告' heading = doc.add_heading(title, level=0) heading.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in heading.runs: run.font.name = '微软雅黑' run.font.size = Pt(22) doc.add_paragraph('') # 报告元信息表 meta_table = doc.add_table(rows=4, cols=2, style='Table Grid') meta_table.alignment = WD_TABLE_ALIGNMENT.CENTER meta_data = [ ('单位名称', unit_name), ('单位代码', unit_code), ('分析时间', datetime.now().strftime('%Y-%m-%d %H:%M:%S')), ('AI模型', model_name) ] for i, (key, value) in enumerate(meta_data): meta_table.rows[i].cells[0].text = key meta_table.rows[i].cells[1].text = str(value) for cell in meta_table.rows[i].cells: for paragraph in cell.paragraphs: paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in paragraph.runs: run.font.name = '微软雅黑' run.font.size = Pt(10) doc.add_paragraph('') # 数据概览表格 if data: doc.add_heading('一、监测数据概览', level=1) # 统计信息 total_points = len(data) normal_count = sum(1 for d in data if d.get('quality', 0) == 0) abnormal_count = total_points - normal_count stats_table = doc.add_table(rows=3, cols=2, style='Table Grid') stats_table.alignment = WD_TABLE_ALIGNMENT.CENTER stats_data = [ ('监测点总数', str(total_points)), ('正常数据', str(normal_count)), ('异常数据', str(abnormal_count)) ] for i, (key, value) in enumerate(stats_data): stats_table.rows[i].cells[0].text = key stats_table.rows[i].cells[1].text = value for cell in stats_table.rows[i].cells: for paragraph in cell.paragraphs: for run in paragraph.runs: run.font.name = '微软雅黑' run.font.size = Pt(10) doc.add_paragraph('') # AI分析报告 doc.add_heading('二、AI智能分析', level=1) parse_markdown_to_docx(doc, ai_report) # 保存 if output_path is None: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'{unit_name}_full_report_{timestamp}.{output_format}' output_path = os.path.join(os.path.dirname(__file__), 'outputs', filename) os.makedirs(os.path.dirname(output_path), exist_ok=True) if output_format == 'docx': doc.save(output_path) else: # PDF需要额外处理,先保存为docx再转换 temp_docx = output_path.replace(f'.{output_format}', '.docx') doc.save(temp_docx) # PDF转换需要额外依赖,这里先返回docx output_path = temp_docx return output_path