easy-ml/monitor/report_generate.py

# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2024/11/8
@desc: 
"""
import os
from typing import Dict

import openpyxl
import pandas as pd
from docx import Document
from docx.enum.table import WD_TABLE_ALIGNMENT, WD_CELL_VERTICAL_ALIGNMENT
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from docx.shared import Inches, Cm, Pt
from openpyxl.worksheet.worksheet import Worksheet

from commom import GeneralException, f_get_datetime
from config import BaseConfig
from entitys import MetricFucResultEntity
from enums import ResultCodesEnum, PlaceholderPrefixEnum


class ReportWord():

    @staticmethod
    def _set_cell_width(table, table_cell_width):
        # 固定宽度
        for column in table.columns:
            if table_cell_width is not None:
                column.width = Cm(table_cell_width)
                continue
        # 自动调整宽度
        max_text_len_list = []
        a4_width = 21 - 2  # * 3.18
        for column in table.columns:
            max_text_len = 0
            for cell in column.cells:
                cell_text_len = ReportWord._get_text_length(cell.text)
                max_text_len = cell_text_len if cell_text_len > max_text_len else max_text_len
            max_text_len_list.append(max_text_len)

        # 按比例分配宽度
        cell_width_unit = a4_width / sum(max_text_len_list)
        cell_widths = [c * cell_width_unit for c in max_text_len_list]
        min_cell_width = 1
        # 限制最小宽度
        adjusted_cell_widths = [max(c, min_cell_width) for c in cell_widths]
        adjusted_width = sum(adjusted_cell_widths)
        if adjusted_width > a4_width:
            excess_width = adjusted_width - a4_width
            excess_width_per_column = excess_width / len(table.columns)
            # 减去多的宽度
            adjusted_cell_widths = [max(min_cell_width, c - excess_width_per_column) for c in
                                    adjusted_cell_widths]

        for idx, column in enumerate(table.columns):
            column.width = Cm(adjusted_cell_widths[idx])

    @staticmethod
    def _set_cell_format(cell, font_size=None):
        for paragraph in cell.paragraphs:
            paragraph.alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
            # 首行缩进改0
            paragraph_format = paragraph.paragraph_format
            paragraph_format.first_line_indent = 0
            paragraph_format.element.pPr.ind.set(qn("w:firstLineChars"), '0')
            for run in paragraph.runs:
                # 判断文本是否包含中文
                if any('\u4e00' <= char <= '\u9fff' for char in run.text):
                    run.font.name = '宋体'  # 设置中文字体为宋体
                else:
                    run.font.name = 'Times New Roman'  # 设置英文字体为Times New Roman
                if font_size is not None:
                    run.font.size = Pt(font_size)
        cell.vertical_alignment = WD_CELL_VERTICAL_ALIGNMENT.CENTER

    @staticmethod
    def _merge_cell_column(pre_cell, curr_cell, table_font_size, table_cell_width):
        if curr_cell.text == pre_cell.text:
            column_name = curr_cell.text
            pre_cell.merge(curr_cell)
            pre_cell.text = column_name
            for run in pre_cell.paragraphs[0].runs:
                run.bold = True
            ReportWord._set_cell_format(pre_cell, table_font_size)

    @staticmethod
    def _set_table_singleBoard(table):
        # 将table 的所有单元格四个边设置为 0.5 镑, 黑色, 实线

        def _set_table_boarder(table, **kwargs):
            """
            Set table`s border
            Usage:
            set_table_border(
                cell,
                top={"sz": 12, "val": "single", "color": "#FF0000"},
                bottom={"sz": 12, "color": "#00FF00", "val": "single"},
                left={"sz": 24, "val": "dashed"},
                right={"sz": 12, "val": "dashed"},
            )
            """
            borders = OxmlElement('w:tblBorders')
            for tag in ('bottom', 'top', 'left', 'right', 'insideV', 'insideH'):
                edge_data = kwargs.get(tag)
                if edge_data:
                    any_border = OxmlElement(f'w:{tag}')
                    for key in ["sz", "val", "color", "space", "shadow"]:
                        if key in edge_data:
                            any_border.set(qn(f'w:{key}'), str(edge_data[key]))
                    borders.append(any_border)
                    table._tbl.tblPr.append(borders)

        return _set_table_boarder(
            table,
            top={"sz": 4, "val": "single", "color": "#000000"},
            bottom={"sz": 4, "val": "single", "color": "#000000"},
            left={"sz": 4, "val": "single", "color": "#000000"},
            right={"sz": 4, "val": "single", "color": "#000000"},
            insideV={"sz": 4, "val": "single", "color": "#000000"},
            insideH={"sz": 4, "val": "single", "color": "#000000"}
        )

    @staticmethod
    def _get_placeholder(placeholder_prefix_enum: PlaceholderPrefixEnum, metric_code: str):
        return "{{" + f"{placeholder_prefix_enum.value}{metric_code}" + "}}"

    @staticmethod
    def _fill_value_placeholder(doc: Document, metric_value_dict: Dict[str, MetricFucResultEntity]):
        # 替换指标
        for paragraph in doc.paragraphs:
            text = paragraph.text
            for metric_code, metric_fuc_entity in metric_value_dict.items():
                placeholder = ReportWord._get_placeholder(PlaceholderPrefixEnum.VALUE, metric_code)
                metric_value = metric_fuc_entity.value
                if metric_value is None:
                    continue
                text = text.replace(placeholder, str(metric_value))
            # 段落中多个runs时执行，最后一个run改成替换好的文本，其他run置空
            if len(paragraph.runs[:-1]) > 0:
                for run in paragraph.runs[:-1]:
                    run.text = ''
                paragraph.runs[-1].text = text

    @staticmethod
    def _get_text_length(text):
        return sum(2 if '\u4e00' <= char <= '\u9fff' else 1 for char in text)

    @staticmethod
    def _fill_table_placeholder(doc: Document, metric_value_dict: Dict[str, MetricFucResultEntity]):
        # 替换表格
        for paragraph in doc.paragraphs:
            for metric_code, metric_fuc_entity in metric_value_dict.items():
                placeholder = ReportWord._get_placeholder(PlaceholderPrefixEnum.TABLE, metric_code)
                metric_table = metric_fuc_entity.table
                table_font_size = metric_fuc_entity.table_font_size
                table_autofit = metric_fuc_entity.table_autofit
                table_cell_width = metric_fuc_entity.table_cell_width
                if metric_table is None:
                    continue
                if not placeholder in paragraph.text:
                    continue
                # 清除占位符
                for run in paragraph.runs:
                    run.text = run.text.replace(placeholder, "")
                table = doc.add_table(rows=metric_table.shape[0] + 1, cols=metric_table.shape[1])
                table.alignment = WD_TABLE_ALIGNMENT.CENTER
                paragraph._element.addnext(table._element)

                # 列名
                for column_idx, column_name in enumerate(metric_table.columns):
                    cell = table.cell(0, column_idx)
                    cell.text = str(column_name)
                    for run in cell.paragraphs[0].runs:
                        run.bold = True
                    ReportWord._set_cell_format(cell, table_font_size)
                    # 合并相同的列名
                    if column_idx != 0 and BaseConfig.merge_table_column:
                        pre_cell = table.cell(0, column_idx - 1)
                        ReportWord._merge_cell_column(pre_cell, cell, table_font_size, table_cell_width)
                # 值
                for row_idx, row in metric_table.iterrows():
                    for column_idx, value in enumerate(row):
                        cell = table.cell(row_idx + 1, column_idx)
                        value = str(value) if pd.notna(value) else '/'
                        cell.text = str(value)
                        ReportWord._set_cell_format(cell, table_font_size)
                        # 合并第一行数据也为列的情况
                        if row_idx == 0:
                            ReportWord._merge_cell_column(table.cell(0, column_idx), cell, table_font_size,
                                                          table_cell_width)

                ReportWord._set_table_singleBoard(table)
                ReportWord._set_cell_width(table, table_cell_width)
                # 禁止自动调整表格
                if len(metric_table.columns) <= 20 and not table_autofit:
                    table.autofit = False

    @staticmethod
    def _fill_image_placeholder(doc: Document, metric_value_dict: Dict[str, MetricFucResultEntity]):
        # 替换图片
        for paragraph in doc.paragraphs:
            for metric_code, metric_fuc_entity in metric_value_dict.items():
                placeholder = ReportWord._get_placeholder(PlaceholderPrefixEnum.IMAGE, metric_code)
                image_path = metric_fuc_entity.image_path
                image_size = metric_fuc_entity.image_size
                if image_path is None:
                    continue
                if not placeholder in paragraph.text:
                    continue
                if isinstance(image_path, str):
                    image_path = [image_path]
                for path in image_path:
                    if not os.path.exists(path):
                        raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"文件【{image_path}】不存在")
                # 清除占位符
                for run in paragraph.runs:
                    if placeholder not in run.text:
                        continue
                    run.text = run.text.replace(placeholder, "")
                    for path in image_path:
                        run.add_picture(path, width=Inches(image_size))

    @staticmethod
    def generate_report(metric_value_dict: Dict[str, MetricFucResultEntity], template_path: str, save_path=None):
        if os.path.exists(template_path):
            doc = Document(template_path)
        else:
            raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"监控模板文件【{template_path}】不存在")

        ReportWord._fill_value_placeholder(doc, metric_value_dict)
        ReportWord._fill_table_placeholder(doc, metric_value_dict)
        ReportWord._fill_image_placeholder(doc, metric_value_dict)
        new_path = template_path.replace(".docx", f"{f_get_datetime()}.docx")
        if save_path is not None:
            new_path = save_path
        doc.save(f"./{new_path}")


class ReportExcel():

    @staticmethod
    def _fill_value_placeholder(worksheet: Worksheet, metric_value_dict: Dict[str, MetricFucResultEntity]):
        # 替换指标,检查每个单元格并替换
        for metric_code, metric_fuc_entity in metric_value_dict.items():
            metric_value = metric_fuc_entity.value
            if metric_value is None:
                continue
            placeholder = ReportWord._get_placeholder(PlaceholderPrefixEnum.VALUE, metric_code)
            for row in worksheet.rows:
                for cell in row:
                    if placeholder in str(cell.value):
                        cell.value = str(cell.value).replace(placeholder, str(metric_value))

    @staticmethod
    def _fill_table_placeholder(worksheet: Worksheet, metric_value_dict: Dict[str, MetricFucResultEntity]):
        # 替换表格
        for metric_code, metric_fuc_entity in metric_value_dict.items():
            metric_table = metric_fuc_entity.table
            if metric_table is None:
                continue
            placeholder = ReportWord._get_placeholder(PlaceholderPrefixEnum.TABLE, metric_code)
            # 定位占位符位置
            start_row = 1
            start_col = 1
            end_flag = False
            for row in worksheet.rows:
                start_col = 1
                for cell in row:
                    if placeholder in str(cell.value):
                        end_flag = True
                        break
                    start_col += 1
                if end_flag:
                    break
                start_row += 1
            # 无占位符则跳过
            if not end_flag:
                continue

            for row_idx, row in metric_table.iterrows():
                for column_idx, value in enumerate(row):
                    worksheet.cell(row=start_row + row_idx, column=start_col + column_idx, value=str(value))

    @staticmethod
    def generate_report(metric_value_dict: Dict[str, MetricFucResultEntity], template_path: str, save_path=None):
        if os.path.exists(template_path):
            workbook = openpyxl.load_workbook(template_path)
            sheet_names = workbook.sheetnames
            worksheet = workbook[sheet_names[0]]

        else:
            raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"监控模板文件【{template_path}】不存在")

        ReportExcel._fill_value_placeholder(worksheet, metric_value_dict)
        ReportExcel._fill_table_placeholder(worksheet, metric_value_dict)
        new_path = template_path.replace(".xlsx", f"{f_get_datetime()}.xlsx")
        if save_path is not None:
            new_path = save_path
        workbook.save(f"./{new_path}")


if __name__ == "__main__":
    pass