RailWise KB已发布

监测报告自动生成模板

RailWise监测报告自动生成标准化代码模板，支持日报、周报、月报、专项报告等多种报告类型，实现从数据到Word/PDF报告的自动化生成流程。

复核 2026-07-09入门公开可引用RailWise 技术团队

template-doc

监测报告自动生成模板

1. 模板概述

1.1 用途

本模板提供监测报告从数据查询到文档生成的完整自动化流程，支持：

监测日报：每日监测数据汇总、变化趋势分析、预警状态汇总
监测周报：周累计变化量统计、速率分析、本周/上周对比
监测月报：月度变形趋势、控制值对比、阶段评估
专项报告：特定事件分析（如盾构穿越、基坑开挖关键期）
预警快报：超限预警即时报告，4小时内输出

1.2 适用场景

报告类型	频率	触发条件	输出格式
监测日报	每日	定时（每日8:00）	Word/PDF
监测周报	每周	定时（每周一8:00）	Word/PDF
监测月报	每月	定时（每月1日）	Word/PDF
专项报告	按需	关键施工节点	Word/PDF
预警快报	即时	监测值超预警值	Word/PDF

1.3 输入输出

输入：监测数据库、项目配置、阈值参数、模板文件
输出：结构化Word文档（.docx）、PDF文档、HTML预览
中间产物：数据汇总表、趋势图、统计指标

2. 代码示例

2.1 核心报告生成引擎

"""
RailWise 监测报告自动生成引擎
支持日报、周报、月报、专项报告、预警快报

作者: RailWise技术部
版本: 1.0.0
"""

import json
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union

import matplotlib.pyplot as plt
import pandas as pd
from docx import Document
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.shared import Inches, Pt, RGBColor
from docx.oxml.ns import qn
from jinja2 import Environment, FileSystemLoader

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("RailWise.ReportGenerator")


class ReportType(Enum):
    """报告类型枚举"""
    DAILY = "daily"           # 监测日报
    WEEKLY = "weekly"         # 监测周报
    MONTHLY = "monthly"       # 监测月报
    SPECIAL = "special"       # 专项报告
    ALERT = "alert"           # 预警快报


class AlertLevel(Enum):
    """预警等级枚举"""
    NORMAL = "正常"           # 正常
    YELLOW = "黄色预警"       # 黄色预警
    ORANGE = "橙色预警"       # 橙色预警
    RED = "红色预警"          # 红色预警


@dataclass
class ReportConfig:
    """报告配置参数"""
    project_id: str
    project_name: str
    report_type: ReportType
    report_date: datetime
    author: str = "RailWise监测系统"
    reviewer: str = ""
    approver: str = ""
    template_dir: Path = field(default_factory=lambda: Path("./templates"))
    output_dir: Path = field(default_factory=lambda: Path("./reports"))
    logo_path: Optional[Path] = None
    company_name: str = "宁波睿威工程技术有限公司"


@dataclass
class MonitoringSummary:
    """监测数据汇总统计"""
    total_points: int = 0
    active_points: int = 0
    alert_points: int = 0
    normal_points: int = 0
    max_displacement: float = 0.0
    max_rate: float = 0.0
    avg_displacement: float = 0.0
    data_completeness: float = 0.0
    alert_summary: Dict[str, int] = field(default_factory=dict)


@dataclass
class PointStatus:
    """单个测点状态"""
    point_id: str
    point_name: str
    point_type: str
    current_value: float
    cumulative_value: float
    change_rate: float
    alert_level: AlertLevel
    threshold: float
    trend: str = "稳定"  # 稳定/上升/下降/波动


class ReportGenerator(ABC):
    """报告生成器抽象基类"""

    def __init__(self, config: ReportConfig):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)
        self.template_env = Environment(
            loader=FileSystemLoader(str(config.template_dir))
        )

    @abstractmethod
    def gather_data(self) -> Dict[str, Any]:
        """收集报告所需数据"""
        pass

    @abstractmethod
    def analyze_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """分析数据，生成统计指标"""
        pass

    @abstractmethod
    def generate_content(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
        """生成报告内容结构"""
        pass

    def create_word_report(self, content: Dict[str, Any]) -> Path:
        """生成Word格式报告"""
        doc = Document()

        # 设置默认字体
        self._set_default_font(doc)

        # 1. 封面
        self._add_cover(doc, content)

        # 2. 目录
        self._add_toc(doc)

        # 3. 正文内容
        self._add_body_content(doc, content)

        # 4. 附录
        self._add_appendix(doc, content)

        # 5. 签章页
        self._add_signature(doc)

        # 保存
        output_path = self._get_output_path("docx")
        doc.save(str(output_path))
        self.logger.info(f"Word报告已生成: {output_path}")
        return output_path

    def create_html_preview(self, content: Dict[str, Any]) -> Path:
        """生成HTML预览版本"""
        template = self.template_env.get_template("report_preview.html")
        html = template.render(**content)

        output_path = self._get_output_path("html")
        output_path.write_text(html, encoding="utf-8")
        return output_path

    def generate(self) -> Dict[str, Path]:
        """完整生成流程"""
        self.logger.info(f"开始生成{self.config.report_type.value}报告")

        # 1. 收集数据
        data = self.gather_data()

        # 2. 分析数据
        analysis = self.analyze_data(data)

        # 3. 生成内容
        content = self.generate_content(analysis)

        # 4. 生成文档
        results = {}
        results["word"] = self.create_word_report(content)
        results["html"] = self.create_html_preview(content)

        return results

    def _set_default_font(self, doc: Document):
        """设置文档默认字体"""
        style = doc.styles['Normal']
        font = style.font
        font.name = '宋体'
        font.size = Pt(12)
        style.element.rPr.rFonts.set(qn('w:eastAsia'), '宋体')

    def _add_cover(self, doc: Document, content: Dict[str, Any]):
        """添加封面"""
        # 公司logo
        if self.config.logo_path and self.config.logo_path.exists():
            doc.add_picture(str(self.config.logo_path), width=Inches(2.0))
            last_paragraph = doc.paragraphs[-1]
            last_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER

        # 报告标题
        title = doc.add_paragraph()
        run = title.add_run(content.get('title', '监测报告'))
        run.font.size = Pt(22)
        run.font.bold = True
        run.font.name = '黑体'
        run.element.rPr.rFonts.set(qn('w:eastAsia'), '黑体')
        title.alignment = WD_ALIGN_PARAGRAPH.CENTER

        # 项目信息
        doc.add_paragraph()
        info = doc.add_paragraph()
        info.add_run(f"项目名称：{self.config.project_name}").font.size = Pt(14)
        info.alignment = WD_ALIGN_PARAGRAPH.CENTER

        info = doc.add_paragraph()
        info.add_run(f"报告日期：{self.config.report_date.strftime('%Y年%m月%d日')}").font.size = Pt(14)
        info.alignment = WD_ALIGN_PARAGRAPH.CENTER

        info = doc.add_paragraph()
        info.add_run(f"编制单位：{self.config.company_name}").font.size = Pt(14)
        info.alignment = WD_ALIGN_PARAGRAPH.CENTER

        # 分页
        doc.add_page_break()

    def _add_toc(self, doc: Document):
        """添加目录"""
        heading = doc.add_heading('目  录', level=1)
        heading.alignment = WD_ALIGN_PARAGRAPH.CENTER

        # 添加目录条目（简化版，实际可使用docx的TOC域）
        toc_items = [
            "一、工程概况",
            "二、监测工作概述",
            "三、监测数据汇总",
            "四、数据分析与评价",
            "五、结论与建议",
            "附录"
        ]
        for item in toc_items:
            p = doc.add_paragraph(item)
            p.paragraph_format.left_indent = Inches(0.5)

        doc.add_page_break()

    def _add_body_content(self, doc: Document, content: Dict[str, Any]):
        """添加正文内容"""
        sections = content.get('sections', [])

        for section in sections:
            # 章节标题
            heading = doc.add_heading(section['title'], level=section.get('level', 1))

            # 章节内容
            for paragraph in section.get('paragraphs', []):
                p = doc.add_paragraph(paragraph)
                p.paragraph_format.first_line_indent = Inches(0.5)

            # 表格
            for table_data in section.get('tables', []):
                self._add_table(doc, table_data)

            # 图片
            for image_path in section.get('images', []):
                if Path(image_path).exists():
                    doc.add_picture(image_path, width=Inches(6.0))

    def _add_table(self, doc: Document, table_data: Dict):
        """添加表格"""
        headers = table_data.get('headers', [])
        rows = table_data.get('rows', [])

        if not headers or not rows:
            return

        table = doc.add_table(rows=1 + len(rows), cols=len(headers))
        table.style = 'Table Grid'

        # 表头
        for i, header in enumerate(headers):
            cell = table.rows[0].cells[i]
            cell.text = header
            cell.paragraphs[0].runs[0].font.bold = True

        # 数据行
        for row_idx, row_data in enumerate(rows):
            for col_idx, value in enumerate(row_data):
                table.rows[row_idx + 1].cells[col_idx].text = str(value)

    def _add_appendix(self, doc: Document, content: Dict[str, Any]):
        """添加附录"""
        doc.add_page_break()
        doc.add_heading('附录', level=1)

        # 检查清单
        doc.add_heading('附录A：数据完整性检查清单', level=2)
        checklist = content.get('checklist', [])
        for item in checklist:
            doc.add_paragraph(f"□ {item}")

    def _add_signature(self, doc: Document):
        """添加签章页"""
        doc.add_page_break()
        doc.add_heading('签章页', level=1)

        table = doc.add_table(rows=3, cols=2)
        table.style = 'Table Grid'

        signatures = [
            ("编制", self.config.author),
            ("审核", self.config.reviewer),
            ("批准", self.config.approver)
        ]

        for i, (role, name) in enumerate(signatures):
            table.rows[i].cells[0].text = f"{role}："
            table.rows[i].cells[1].text = name

    def _get_output_path(self, ext: str) -> Path:
        """获取输出文件路径"""
        date_str = self.config.report_date.strftime("%Y%m%d")
        report_type_str = self.config.report_type.value
        filename = f"{self.config.project_id}_{report_type_str}_{date_str}.{ext}"

        output_path = self.config.output_dir / filename
        output_path.parent.mkdir(parents=True, exist_ok=True)
        return output_path


class DailyReportGenerator(ReportGenerator):
    """监测日报生成器"""

    def gather_data(self) -> Dict[str, Any]:
        """收集日报数据"""
        # 获取当日数据
        today = self.config.report_date
        yesterday = today - timedelta(days=1)

        # 模拟从数据库查询（实际项目中替换为真实查询）
        data = {
            "report_date": today,
            "period_start": yesterday,
            "period_end": today,
            "total_points": 45,
            "active_points": 43,
            "observation_count": 86,
            "alert_count": 2,
            "points_data": self._mock_daily_data()
        }
        return data

    def analyze_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """分析日报数据"""
        points = data.get('points_data', [])

        summary = MonitoringSummary()
        summary.total_points = data['total_points']
        summary.active_points = data['active_points']
        summary.alert_points = data['alert_count']
        summary.normal_points = summary.active_points - summary.alert_points

        if points:
            displacements = [p['cumulative'] for p in points]
            rates = [p['rate'] for p in points]
            summary.max_displacement = max(displacements)
            summary.max_rate = max(rates)
            summary.avg_displacement = sum(displacements) / len(displacements)
            summary.data_completeness = summary.active_points / summary.total_points * 100

        return {
            "summary": summary,
            "points": points,
            "alerts": [p for p in points if p['alert_level'] != '正常']
        }

    def generate_content(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
        """生成日报内容"""
        summary = analysis['summary']
        points = analysis['points']
        alerts = analysis['alerts']

        content = {
            "title": f"{self.config.project_name}监测日报",
            "sections": [
                {
                    "title": "一、工程概况",
                    "level": 1,
                    "paragraphs": [
                        f"本项目为{self.config.project_name}，监测工作由{self.config.company_name}承担。",
                        f"本次报告统计时段：{(self.config.report_date - timedelta(days=1)).strftime('%Y年%m月%d日')}至{self.config.report_date.strftime('%Y年%m月%d日')}。"
                    ]
                },
                {
                    "title": "二、监测工作概述",
                    "level": 1,
                    "paragraphs": [
                        f"本日共完成{summary.active_points}个监测点的观测工作，",
                        f"数据完整率{summary.data_completeness:.1f}%。",
                        f"其中正常测点{summary.normal_points}个，预警测点{summary.alert_points}个。"
                    ]
                },
                {
                    "title": "三、监测数据汇总",
                    "level": 1,
                    "tables": [
                        {
                            "headers": ["测点编号", "测点名称", "本次变化(mm)", "累计变化(mm)", "变化速率(mm/d)", "预警状态"],
                            "rows": [
                                [p['point_id'], p['point_name'],
                                 f"{p['change']:.2f}", f"{p['cumulative']:.2f}",
                                 f"{p['rate']:.2f}", p['alert_level']]
                                for p in points
                            ]
                        }
                    ]
                },
                {
                    "title": "四、预警情况分析",
                    "level": 1,
                    "paragraphs": alerts and [
                        f"本日共发现{len(alerts)}个预警测点，需重点关注："
                    ] or ["本日无预警测点，监测数据正常。"]
                },
                {
                    "title": "五、结论与建议",
                    "level": 1,
                    "paragraphs": [
                        "1. 监测数据整体正常，各测点变化趋势稳定。",
                        "2. 建议继续按现有频率进行监测。",
                        "3. 预警测点需加强巡查，必要时加密观测频率。"
                    ]
                }
            ],
            "checklist": [
                "数据完整性检查通过",
                "阈值比对完成",
                "趋势分析完成",
                "预警状态确认",
                "报告审核完成"
            ]
        }
        return content

    def _mock_daily_data(self) -> List[Dict]:
        """生成模拟日报数据（实际项目中从数据库查询）"""
        return [
            {
                "point_id": "JC-01",
                "point_name": "3号线K12+350左线",
                "change": 0.15,
                "cumulative": 2.35,
                "rate": 0.15,
                "alert_level": "正常"
            },
            {
                "point_id": "JC-02",
                "point_name": "3号线K12+350右线",
                "change": 0.22,
                "cumulative": 3.12,
                "rate": 0.22,
                "alert_level": "正常"
            },
            {
                "point_id": "JC-03",
                "point_name": "3号线K12+400左线",
                "change": 1.85,
                "cumulative": 8.56,
                "rate": 1.85,
                "alert_level": "黄色预警"
            }
        ]


class AlertReportGenerator(ReportGenerator):
    """预警快报生成器"""

    def __init__(self, config: ReportConfig, alert_points: List[PointStatus]):
        super().__init__(config)
        self.alert_points = alert_points

    def gather_data(self) -> Dict[str, Any]:
        """收集预警数据"""
        return {
            "alert_time": datetime.now(timezone.utc),
            "alert_points": [p.__dict__ for p in self.alert_points]
        }

    def analyze_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """分析预警数据"""
        points = data['alert_points']

        # 按预警等级分组
        alert_groups = {}
        for p in points:
            level = p['alert_level']
            alert_groups[level] = alert_groups.get(level, []) + [p]

        return {
            "alert_groups": alert_groups,
            "total_alerts": len(points),
            "max_level": max(points, key=lambda x: ['正常', '黄色预警', '橙色预警', '红色预警'].index(x['alert_level']))['alert_level']
        }

    def generate_content(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
        """生成预警快报内容"""
        content = {
            "title": f"【预警快报】{self.config.project_name}",
            "sections": [
                {
                    "title": "一、预警概述",
                    "level": 1,
                    "paragraphs": [
                        f"监测时间：{datetime.now().strftime('%Y年%m月%d日 %H:%M')}",
                        f"预警等级：{analysis['max_level']}",
                        f"预警测点数量：{analysis['total_alerts']}个"
                    ]
                },
                {
                    "title": "二、预警测点详情",
                    "level": 1,
                    "tables": [
                        {
                            "headers": ["测点编号", "测点名称", "当前值(mm)", "累计值(mm)", "变化速率(mm/d)", "预警等级", "控制值(mm)"],
                            "rows": [
                                [p['point_id'], p['point_name'],
                                 f"{p['current_value']:.2f}", f"{p['cumulative_value']:.2f}",
                                 f"{p['change_rate']:.2f}", p['alert_level'],
                                 f"{p['threshold']:.2f}"]
                                for p in self.alert_points
                            ]
                        }
                    ]
                },
                {
                    "title": "三、应急处置建议",
                    "level": 1,
                    "paragraphs": [
                        "1. 立即对预警测点进行现场巡查，确认数据真实性。",
                        "2. 加密观测频率至每2小时一次。",
                        "3. 通知施工单位和监理单位，采取应急措施。",
                        "4. 持续跟踪监测数据变化趋势。"
                    ]
                }
            ],
            "checklist": [
                "预警信息已通知相关人员",
                "现场巡查已安排",
                "加密观测已启动",
                "应急措施已落实"
            ]
        }
        return content


# 报告生成工厂
def create_report_generator(
    report_type: ReportType,
    config: ReportConfig,
    **kwargs
) -> ReportGenerator:
    """创建报告生成器"""
    generators = {
        ReportType.DAILY: DailyReportGenerator,
        ReportType.WEEKLY: DailyReportGenerator,  # 可扩展为WeeklyReportGenerator
        ReportType.MONTHLY: DailyReportGenerator,  # 可扩展为MonthlyReportGenerator
        ReportType.ALERT: AlertReportGenerator,
    }

    generator_class = generators.get(report_type)
    if not generator_class:
        raise ValueError(f"不支持的报告类型: {report_type}")

    if report_type == ReportType.ALERT:
        return generator_class(config, kwargs.get('alert_points', []))

    return generator_class(config)

2.2 数据可视化模块

"""数据可视化模块 - 生成监测趋势图、分布图等"""

import matplotlib
matplotlib.use('Agg')  # 非交互式后端
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.patches import Rectangle
import numpy as np
from pathlib import Path
from typing import List, Dict, Optional, Tuple


class MonitoringVisualizer:
    """监测数据可视化器"""

    def __init__(self, output_dir: Path = Path("./charts")):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

        # 设置中文字体
        plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS']
        plt.rcParams['axes.unicode_minus'] = False

    def plot_time_series(
        self,
        data: List[Dict],
        point_id: str,
        title: str = "",
        threshold: Optional[float] = None,
        control_value: Optional[float] = None,
        figsize: Tuple[int, int] = (12, 6)
    ) -> Path:
        """
        绘制时间序列趋势图

        Args:
            data: 数据列表，每项包含'time'和'value'
            point_id: 测点编号
            title: 图表标题
            threshold: 预警阈值
            control_value: 控制值
            figsize: 图像尺寸

        Returns:
            保存的图像路径
        """
        fig, ax = plt.subplots(figsize=figsize)

        times = [d['time'] for d in data]
        values = [d['value'] for d in data]

        # 绘制数据线
        ax.plot(times, values, 'b-', linewidth=1.5, label='监测值')
        ax.scatter(times, values, c='blue', s=20, zorder=5)

        # 添加阈值线
        if threshold:
            ax.axhline(y=threshold, color='orange', linestyle='--',
                      linewidth=1.5, label=f'预警阈值: {threshold}mm')

        if control_value:
            ax.axhline(y=control_value, color='red', linestyle='--',
                      linewidth=1.5, label=f'控制值: {control_value}mm')
            ax.axhline(y=-control_value, color='red', linestyle='--',
                      linewidth=1.5)

        # 标注预警区域
        if threshold:
            for i, (t, v) in enumerate(zip(times, values)):
                if abs(v) > threshold:
                    ax.annotate(f'{v:.2f}', xy=(t, v),
                               xytext=(5, 5), textcoords='offset points',
                               fontsize=8, color='red')

        # 设置图表属性
        ax.set_xlabel('时间', fontsize=12)
        ax.set_ylabel('位移 (mm)', fontsize=12)
        ax.set_title(title or f'{point_id} 监测时间序列', fontsize=14)
        ax.legend(loc='upper left')
        ax.grid(True, alpha=0.3)

        # 格式化x轴日期
        ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))
        ax.xaxis.set_major_locator(mdates.DayLocator(interval=7))
        plt.xticks(rotation=45)

        plt.tight_layout()

        # 保存
        output_path = self.output_dir / f"{point_id}_timeseries.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close()

        return output_path

    def plot_multi_point_comparison(
        self,
        data: Dict[str, List[Dict]],
        title: str = "多测点对比",
        figsize: Tuple[int, int] = (14, 8)
    ) -> Path:
        """绘制多测点对比图"""
        fig, ax = plt.subplots(figsize=figsize)

        colors = plt.cm.tab10(np.linspace(0, 1, len(data)))

        for idx, (point_id, point_data) in enumerate(data.items()):
            times = [d['time'] for d in point_data]
            values = [d['value'] for d in point_data]
            ax.plot(times, values, color=colors[idx], linewidth=1.5,
                   label=point_id, marker='o', markersize=3)

        ax.set_xlabel('时间', fontsize=12)
        ax.set_ylabel('位移 (mm)', fontsize=12)
        ax.set_title(title, fontsize=14)
        ax.legend(loc='upper left', ncol=2)
        ax.grid(True, alpha=0.3)

        ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d'))
        plt.xticks(rotation=45)

        plt.tight_layout()

        output_path = self.output_dir / "multi_point_comparison.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close()

        return output_path

    def plot_alert_distribution(
        self,
        alert_summary: Dict[str, int],
        title: str = "预警分布统计"
    ) -> Path:
        """绘制预警分布饼图"""
        fig, ax = plt.subplots(figsize=(8, 8))

        labels = list(alert_summary.keys())
        sizes = list(alert_summary.values())
        colors = ['#52c41a', '#faad14', '#fa8c16', '#f5222d']

        wedges, texts, autotexts = ax.pie(
            sizes, labels=labels, colors=colors[:len(labels)],
            autopct='%1.1f%%', startangle=90,
            textprops={'fontsize': 12}
        )

        ax.set_title(title, fontsize=14)

        output_path = self.output_dir / "alert_distribution.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close()

        return output_path

    def plot_heatmap(
        self,
        data: pd.DataFrame,
        title: str = "监测数据热力图"
    ) -> Path:
        """绘制监测数据热力图"""
        fig, ax = plt.subplots(figsize=(14, 8))

        im = ax.imshow(data.values, cmap='RdYlGn_r', aspect='auto')

        # 设置坐标轴
        ax.set_xticks(range(len(data.columns)))
        ax.set_xticklabels(data.columns, rotation=45, ha='right')
        ax.set_yticks(range(len(data.index)))
        ax.set_yticklabels(data.index)

        ax.set_title(title, fontsize=14)
        ax.set_xlabel('时间', fontsize=12)
        ax.set_ylabel('测点编号', fontsize=12)

        # 添加颜色条
        cbar = plt.colorbar(im, ax=ax)
        cbar.set_label('位移 (mm)', fontsize=12)

        plt.tight_layout()

        output_path = self.output_dir / "monitoring_heatmap.png"
        plt.savefig(output_path, dpi=150, bbox_inches='tight')
        plt.close()

        return output_path


# 辅助函数：生成模拟数据用于测试
def generate_mock_timeseries(
    days: int = 30,
    base_value: float = 0.0,
    trend: float = 0.1,
    noise: float = 0.5
) -> List[Dict]:
    """生成模拟时间序列数据"""
    from datetime import timedelta

    base_time = datetime.now(timezone.utc) - timedelta(days=days)
    data = []

    for i in range(days * 4):  # 每天4次观测
        time = base_time + timedelta(hours=i * 6)
        value = base_value + trend * i + np.random.normal(0, noise)
        data.append({'time': time, 'value': round(value, 2)})

    return data

3. 参数说明

3.1 ReportConfig 配置参数

参数名	类型	必填	默认值	说明
`project_id`	str	是	-	项目唯一标识
`project_name`	str	是	-	项目名称（显示用）
`report_type`	ReportType	是	-	报告类型枚举
`report_date`	datetime	是	-	报告日期
`author`	str	否	“RailWise监测系统”	编制人
`reviewer`	str	否	“”	审核人
`approver`	str	否	“”	批准人
`template_dir`	Path	否	`./templates`	模板文件目录
`output_dir`	Path	否	`./reports`	输出文件目录
`logo_path`	Path	否	None	公司logo图片路径
`company_name`	str	否	“宁波睿威工程技术有限公司”	编制单位名称

3.2 MonitoringSummary 统计参数

参数名	类型	说明
`total_points`	int	监测点总数
`active_points`	int	活跃监测点数
`alert_points`	int	预警测点数
`normal_points`	int	正常测点数
`max_displacement`	float	最大累计位移(mm)
`max_rate`	float	最大变化速率(mm/d)
`avg_displacement`	float	平均累计位移(mm)
`data_completeness`	float	数据完整率(%)

3.3 PointStatus 测点状态参数

参数名	类型	说明
`point_id`	str	测点编号
`point_name`	str	测点名称
`point_type`	str	测点类型（水平/竖向/收敛等）
`current_value`	float	本次变化值(mm)
`cumulative_value`	float	累计变化值(mm)
`change_rate`	float	变化速率(mm/d)
`alert_level`	AlertLevel	预警等级
`threshold`	float	预警阈值(mm)
`trend`	str	变化趋势（稳定/上升/下降/波动）

4. 使用示例

4.1 生成监测日报

from datetime import datetime, timezone
from pathlib import Path
from template_report_generation import (
    ReportConfig, ReportType, create_report_generator
)

# 1. 配置报告参数
config = ReportConfig(
    project_id="RW-2024-001",
    project_name="绕城高速管廊工程1标段",
    report_type=ReportType.DAILY,
    report_date=datetime.now(timezone.utc),
    author="张三",
    reviewer="李四",
    approver="王五",
    template_dir=Path("./templates"),
    output_dir=Path("./reports"),
    logo_path=Path("./assets/logo.png")
)

# 2. 创建生成器并生成报告
generator = create_report_generator(ReportType.DAILY, config)
results = generator.generate()

print(f"Word报告: {results['word']}")
print(f"HTML预览: {results['html']}")

4.2 生成预警快报

from template_report_generation import (
    ReportConfig, ReportType, AlertReportGenerator,
    PointStatus, AlertLevel
)

# 准备预警测点数据
alert_points = [
    PointStatus(
        point_id="JC-03",
        point_name="3号线K12+400左线",
        point_type="水平位移",
        current_value=1.85,
        cumulative_value=8.56,
        change_rate=1.85,
        alert_level=AlertLevel.YELLOW,
        threshold=10.0,
        trend="上升"
    ),
    PointStatus(
        point_id="JC-05",
        point_name="3号线K12+450右线",
        point_type="竖向位移",
        current_value=2.12,
        cumulative_value=12.35,
        change_rate=2.12,
        alert_level=AlertLevel.ORANGE,
        threshold=15.0,
        trend="快速上升"
    )
]

# 配置并生成预警快报
config = ReportConfig(
    project_id="RW-2024-001",
    project_name="绕城高速管廊工程1标段",
    report_type=ReportType.ALERT,
    report_date=datetime.now(timezone.utc),
    author="自动监测系统"
)

generator = AlertReportGenerator(config, alert_points)
results = generator.generate()

print(f"预警快报已生成: {results['word']}")

4.3 批量生成周报

from datetime import timedelta

# 生成过去4周的周报
base_date = datetime.now(timezone.utc)

for week in range(4):
    report_date = base_date - timedelta(weeks=week)
    config = ReportConfig(
        project_id="RW-2024-001",
        project_name="绕城高速管廊工程1标段",
        report_type=ReportType.WEEKLY,
        report_date=report_date
    )

    generator = create_report_generator(ReportType.WEEKLY, config)
    results = generator.generate()

    print(f"第{week+1}周报告已生成: {results['word']}")

4.4 生成趋势图

from template_report_generation import MonitoringVisualizer, generate_mock_timeseries

# 创建可视化器
viz = MonitoringVisualizer(output_dir=Path("./charts"))

# 生成模拟数据
data = generate_mock_timeseries(days=30, base_value=0.0, trend=0.2, noise=0.3)

# 绘制时间序列图
chart_path = viz.plot_time_series(
    data=data,
    point_id="JC-01",
    title="JC-01 水平位移监测趋势",
    threshold=5.0,
    control_value=10.0
)

print(f"趋势图已生成: {chart_path}")

# 绘制多测点对比图
multi_data = {
    "JC-01": generate_mock_timeseries(days=30, base_value=0.0, trend=0.2),
    "JC-02": generate_mock_timeseries(days=30, base_value=0.0, trend=0.15),
    "JC-03": generate_mock_timeseries(days=30, base_value=0.0, trend=0.35),
}

comparison_path = viz.plot_multi_point_comparison(
    data=multi_data,
    title="关键测点水平位移对比"
)

print(f"对比图已生成: {comparison_path}")

5. 输出示例

5.1 日报内容结构

【封面】
- 公司Logo
- 报告标题：绕城高速管廊工程1标段监测日报
- 报告日期：2024年07月08日
- 编制单位：宁波睿威工程技术有限公司

【目录】
一、工程概况
二、监测工作概述
三、监测数据汇总
四、数据分析与评价
五、结论与建议
附录

【正文】
一、工程概况
  本项目为绕城高速管廊工程1标段施工期间轨道交通3号线控制保护区监测项目，
  监测工作由宁波睿威工程技术有限公司承担。
  本次报告统计时段：2024年07月07日至2024年07月08日。

二、监测工作概述
  本日共完成43个监测点的观测工作，数据完整率95.6%。
  其中正常测点41个，预警测点2个。

三、监测数据汇总
  [表格：测点编号 | 测点名称 | 本次变化 | 累计变化 | 变化速率 | 预警状态]

四、预警情况分析
  本日共发现2个预警测点，需重点关注：
  - JC-03: 黄色预警，累计位移8.56mm
  - JC-05: 橙色预警，累计位移12.35mm

五、结论与建议
  1. 监测数据整体正常，各测点变化趋势稳定。
  2. 建议继续按现有频率进行监测。
  3. 预警测点需加强巡查，必要时加密观测频率。

【附录】
  □ 数据完整性检查通过
  □ 阈值比对完成
  □ 趋势分析完成
  □ 预警状态确认
  □ 报告审核完成

5.2 生成的文件

reports/
├── RW-2024-001_daily_20240708.docx    # Word报告
├── RW-2024-001_daily_20240708.html    # HTML预览
└── RW-2024-001_alert_202407081400.docx  # 预警快报

charts/
├── JC-01_timeseries.png              # 单测点趋势图
├── multi_point_comparison.png        # 多测点对比图
├── alert_distribution.png            # 预警分布图
└── monitoring_heatmap.png            # 数据热力图

6. 扩展指南

6.1 自定义报告模板

创建Jinja2模板：

<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
    <style>
        body { font-family: 'SimHei', sans-serif; }
        .alert-yellow { background: #fffbe6; }
        .alert-orange { background: #fff7e6; }
        .alert-red { background: #fff1f0; }
    </style>
</head>
<body>
    <h1>{{ title }}</h1>
    {% for section in sections %}
    <h2>{{ section.title }}</h2>
    {% for p in section.paragraphs %}
    <p>{{ p }}</p>
    {% endfor %}
    {% endfor %}
</body>
</html>

在生成器中使用：

template = self.template_env.get_template("custom_report.html")
html = template.render(**content)

6.2 添加新的报告类型

class MonthlyReportGenerator(ReportGenerator):
    """月报生成器"""

    def gather_data(self) -> Dict[str, Any]:
        # 获取月度数据
        pass

    def analyze_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
        # 月度统计分析
        pass

    def generate_content(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
        # 生成月报内容
        pass

# 注册到工厂
generators[ReportType.MONTHLY] = MonthlyReportGenerator

6.3 集成邮件发送

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

def send_report_email(
    report_path: Path,
    recipient: str,
    subject: str = "监测报告"
):
    """发送报告邮件"""
    msg = MIMEMultipart()
    msg['From'] = 'monitoring@railwise.com'
    msg['To'] = recipient
    msg['Subject'] = subject

    # 添加附件
    with open(report_path, 'rb') as f:
        attachment = MIMEBase('application', 'vnd.openxmlformats-officedocument.wordprocessingml.document')
        attachment.set_payload(f.read())
        encoders.encode_base64(attachment)
        attachment.add_header('Content-Disposition', f'attachment; filename={report_path.name}')
        msg.attach(attachment)

    # 发送邮件
    with smtplib.SMTP('smtp.railwise.com') as server:
        server.send_message(msg)

6.4 定时自动生成

from apscheduler.schedulers.background import BackgroundScheduler

def scheduled_daily_report():
    """定时生成日报"""
    config = ReportConfig(
        project_id="RW-2024-001",
        project_name="绕城高速管廊工程1标段",
        report_type=ReportType.DAILY,
        report_date=datetime.now(timezone.utc)
    )

    generator = create_report_generator(ReportType.DAILY, config)
    results = generator.generate()

    # 发送邮件
    send_report_email(results['word'], 'manager@railwise.com')

# 设置定时任务
scheduler = BackgroundScheduler()
scheduler.add_job(scheduled_daily_report, 'cron', hour=8, minute=0)
scheduler.start()

7. 常见问题

Q1: 中文字体显示乱码怎么办？

A: 确保系统安装了中文字体（如SimHei、Microsoft YaHei），或在代码中指定字体路径：

import matplotlib
matplotlib.rcParams['font.family'] = ['SimHei', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False

Q2: 如何自定义Word文档样式？

A: 修改_set_default_font和_add_cover方法，或使用预定义模板：

doc = Document('template.docx')  # 使用预定义模板

Q3: 报告生成速度太慢怎么办？

使用异步IO处理数据查询
缓存频繁访问的数据
使用多线程并行生成图表
考虑使用Celery等任务队列

Q4: 如何支持PDF输出？

A: 使用docx2pdf库转换：

from docx2pdf import convert

convert("report.docx", "report.pdf")

Q5: 如何与WorkWise平台集成？

A: 使用WorkWise API上传报告：

import requests

def upload_to_workwise(report_path: Path, project_id: str):
    url = f"https://api.workwise.railwise.com/v1/projects/{project_id}/reports"
    with open(report_path, 'rb') as f:
        response = requests.post(url, files={'file': f})
    return response.json()

元数据标签

ai_tags:
  domain: ["工程监测", "报告编制", "数据可视化"]
  technology: ["Python", "python-docx", "Jinja2", "Matplotlib", "Pandas"]
  standard: ["GB 50497", "CJJ/T 202", "JGJ 8"]
  product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"]
  difficulty: "intermediate"
  role: ["监测工程师", "报告编制员", "项目经理"]
  task_type: ["报告生成", "数据可视化", "自动化办公"]

监测报告自动生成模板

监测报告自动生成模板

1. 模板概述

1.1 用途

1.2 适用场景

1.3 输入输出

2. 代码示例

2.1 核心报告生成引擎

2.2 数据可视化模块

3. 参数说明

3.1 ReportConfig 配置参数

3.2 MonitoringSummary 统计参数

3.3 PointStatus 测点状态参数

4. 使用示例

4.1 生成监测日报

4.2 生成预警快报

4.3 批量生成周报

4.4 生成趋势图

5. 输出示例

5.1 日报内容结构

5.2 生成的文件

6. 扩展指南

6.1 自定义报告模板

6.2 添加新的报告类型

6.3 集成邮件发送

6.4 定时自动生成

7. 常见问题

Q1: 中文字体显示乱码怎么办？

Q2: 如何自定义Word文档样式？

Q3: 报告生成速度太慢怎么办？

Q4: 如何支持PDF输出？

Q5: 如何与WorkWise平台集成？

相关文档链接

元数据标签