347 lines
12 KiB
Python
347 lines
12 KiB
Python
#!/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 |