# report_template.ipynb 的第一个单元格（参数单元格） # 标记为 "parameters" 标签 date_range = "2025-01-01至2025-12-31" region = "华东区" output_file = "report_east.html" top_n = 10 threshold = 0.05

# 使用 Papermill 批量运行参数化 Notebook import papermill as pm import os from datetime import datetime # 参数组合 params_list = [ {"date_range": "2025-Q1", "region": "华东区", "output_file": "report_q1_east.html", "top_n": 5, "threshold": 0.05}, {"date_range": "2025-Q1", "region": "华南区", "output_file": "report_q1_south.html", "top_n": 5, "threshold": 0.05}, {"date_range": "2025-Q2", "region": "华东区", "output_file": "report_q2_east.html", "top_n": 10, "threshold": 0.03}, ] for params in params_list: output_path = f"output/{params['output_file']}" pm.execute_notebook( input_path='templates/report_template.ipynb', output_path=output_path, parameters=params, kernel_name='python3', progress_bar=False ) print(f"[{datetime.now().strftime('%H:%M:%S')}] 生成: {output_path}") print("全部报告生成完毕")

# 使用 nbconvert 命令行导出多种格式 # HTML 导出（保留代码和输出） !jupyter nbconvert --to html --template classic output/report_q1_east.ipynb # HTML 无代码版（仅展示输出结果，适合管理层） !jupyter nbconvert --to html --template lab --no-input output/report_q1_east.ipynb # Markdown 导出 !jupyter nbconvert --to markdown output/report_q1_east.ipynb # Python 脚本导出 !jupyter nbconvert --to script output/report_q1_east.ipynb # PDF 导出（需要 LaTeX 环境） !jupyter nbconvert --to pdf output/report_q1_east.ipynb

# 方案一：使用 pdfkit（基于 wkhtmltopdf） import pdfkit options = { 'page-size': 'A4', 'margin-top': '15mm', 'margin-right': '15mm', 'margin-bottom': '15mm', 'margin-left': '15mm', 'encoding': 'UTF-8', 'no-outline': None, 'enable-local-file-access': None, } pdfkit.from_file('output/report_q1_east.html', 'output/report_q1_east.pdf', options=options) print("pdfkit PDF 导出成功") # 方案二：使用 WeasyPrint（纯Python方案，无需额外安装wkhtmltopdf） from weasyprint import HTML HTML('output/report_q1_east.html').write_pdf('output/report_q1_east_v2.pdf') print("WeasyPrint PDF 导出成功")

# 安装 # pip install ydata-profiling import pandas as pd from ydata_profiling import ProfileReport # 加载数据 df = pd.read_csv('sales_data_2025.csv') print(f"数据集: {df.shape[0]} 行, {df.shape[1]} 列") # 生成基础报告 profile = ProfileReport(df, title='2025年销售数据EDA报告') profile.to_file('output/eda_report.html') print("EDA报告已生成: output/eda_report.html")

# 高级配置 profile = ProfileReport(df, title='2025年销售数据 - 详细EDA报告', # 数据集描述 dataset={ "description": "2025年度各区域销售数据，包含订单、客户、产品信息", "creator": "数据分析团队", "copyright_holder": "XX科技有限公司", "url": "https://internal.data/corp-sales" }, # 变量分析配置 variables={ "descriptions": { "order_id": "订单唯一编号", "amount": "订单金额（元）", "region": "所属区域", "category": "产品品类", "order_date": "下单日期" } }, # 排序与筛选 sort="ascending", # 交互图表开关 interactions={"continuous": True, "target": "amount"}, # 相关性配置 correlations={ "auto": {"calculate": True}, "pearson": {"calculate": True}, "spearman": {"calculate": True}, "kendall": {"calculate": True}, "phi_k": {"calculate": True}, "cramers": {"calculate": True}, }, # 缺失值可视化 missing_diagrams={ "bar": True, "matrix": True, "heatmap": False, "dendrogram": False }, # 采样设置（大数据集时减少计算量） sampe=None, # 不采样，使用全部数据 # 进度条 progress_bar=True, # 最小观测数 minimal=True, # 快速模式，跳过耗时计算 ) # 保存为HTML profile.to_file('output/eda_report_detailed.html') # 保存为JSON（供后续程序处理） profile.to_file('output/eda_report.json') # 在Jupyter中直接显示 # profile

# 安装 # pip install sweetviz import sweetviz as sv import pandas as pd # 加载数据 train_df = pd.read_csv('train_data.csv') test_df = pd.read_csv('test_data.csv') print(f"训练集: {train_df.shape}, 测试集: {test_df.shape}") # 分析整个数据集（无对比） report = sv.analyze(train_df, target_feat='target') report.show_html('output/sweetviz_analysis.html') # 训练集 vs 测试集 对比 compare_report = sv.compare([train_df, '训练集'], [test_df, '测试集'], target_feat='target') compare_report.show_html('output/sweetviz_compare.html') print("对比报告已生成")

# 特征关联分析 # 分析训练集中所有特征与目标的关联 feature_report = sv.analyze(train_df, target_feat='target', feat_cfg=sv.FeatureConfig( # 跳过某些特征 skip=['id', 'timestamp'], # 强制指定某些特征为分类变量 force_cat=['region_code', 'channel_type'], # 强制指定某些特征为数值变量 force_num=['member_days'], # 强制指定某些特征为文本变量 force_text=['remark'], )) feature_report.show_html('output/sweetviz_feature_analysis.html') # 对比关联 compare_feat_report = sv.compare( [train_df, '训练集'], [test_df, '测试集'], target_feat='target', feat_cfg=sv.FeatureConfig( force_cat=['region', 'channel', 'category'] )) compare_feat_report.show_html('output/sweetviz_feature_compare.html') print("特征关联分析完成")

import matplotlib.pyplot as plt import seaborn as sns import pandas as pd import numpy as np # 设置全局样式 plt.style.use('seaborn-v0_8-whitegrid') plt.rcParams.update({ 'font.family': 'Microsoft YaHei', 'font.size': 12, 'axes.titlesize': 16, 'axes.labelsize': 14, 'figure.dpi': 150, 'savefig.dpi': 300, 'savefig.bbox': 'tight', 'savefig.pad_inches': 0.1, }) # 生成示例数据 np.random.seed(42) df = pd.DataFrame({ '月份': pd.date_range('2025-01-01', periods=12, freq='M').strftime('%Y-%m'), '销售额': np.random.randint(800000, 1500000, 12), '成本': np.random.randint(500000, 900000, 12), '利润': np.random.randint(200000, 600000, 12), }) # 绘制图表 fig, ax = plt.subplots(figsize=(12, 6)) ax.plot(df['月份'], df['销售额'], marker='o', linewidth=2, label='销售额', color='#3498db') ax.plot(df['月份'], df['利润'], marker='s', linewidth=2, label='利润', color='#2ecc71') ax.fill_between(df['月份'], df['成本'], alpha=0.3, label='成本区间', color='#e74c3c') ax.set_xlabel('月份') ax.set_ylabel('金额（元）') ax.set_title('2025年度销售趋势分析') ax.legend(loc='upper left') plt.xticks(rotation=45) plt.tight_layout() # 导出为PNG（高分辨率适合打印） plt.savefig('output/sales_trend.png', dpi=300, bbox_inches='tight') # 导出为PDF（矢量图，适合嵌入LaTeX/Word） plt.savefig('output/sales_trend.pdf', bbox_inches='tight') # 导出为SVG（矢量图，适合网页嵌入） plt.savefig('output/sales_trend.svg', format='svg', bbox_inches='tight') print("图表导出完成") plt.close()

# 多图合并（适合报告中的仪表板页面） fig = plt.figure(figsize=(16, 12)) # 设置子图布局 gs = fig.add_gridspec(3, 3, hspace=0.3, wspace=0.3) # 左上：销售趋势（占2列1行） ax1 = fig.add_subplot(gs[0, :2]) ax1.plot(df['月份'], df['销售额'], color='#3498db', linewidth=2, marker='o') ax1.set_title('月度销售额趋势', fontsize=14, fontweight='bold') ax1.tick_params(axis='x', rotation=45) # 右上：分布箱线图 ax2 = fig.add_subplot(gs[0, 2]) sns.boxplot(data=df[['销售额', '利润']], ax=ax2, palette=['#3498db', '#2ecc71']) ax2.set_title('销售额与利润分布') # 中排：相关性热图 ax3 = fig.add_subplot(gs[1, :]) corr_data = df[['销售额', '成本', '利润']].corr() sns.heatmap(corr_data, annot=True, fmt='.2f', cmap='RdBu_r', center=0, ax=ax3, cbar_kws={'shrink': 0.8}) ax3.set_title('核心指标相关性矩阵') # 下排：各月构成对比 ax4 = fig.add_subplot(gs[2, :]) df_plot = df.set_index('月份')[['销售额', '成本']] df_plot.plot(kind='bar', ax=ax4, color=['#3498db', '#e74c3c'], width=0.7) ax4.set_title('月度销售额与成本对比', fontsize=14) ax4.set_xlabel('') ax4.legend(loc='upper right') plt.xticks(rotation=45) plt.suptitle('2025年度业务全景仪表板', fontsize=20, fontweight='bold', y=1.02) plt.savefig('output/dashboard.png', dpi=300, bbox_inches='tight') print("仪表板已导出") plt.close()

# 定义可复用的图表模板 import matplotlib.pyplot as plt from dataclasses import dataclass from typing import Optional @dataclass class ChartConfig: """图表配置模板""" figsize: tuple = (12, 6) dpi: int = 150 title_size: int = 16 label_size: int = 14 tick_size: int = 11 style: str = 'seaborn-v0_8-whitegrid' font: str = 'Microsoft YaHei' color_palette: list = None def __post_init__(self): if self.color_palette is None: self.color_palette = ['#3498db', '#2ecc71', '#e74c3c', '#f39c12', '#9b59b6', '#1abc9c'] def apply(self): """应用配置到全局 matplotlib 设置""" plt.style.use(self.style) plt.rcParams.update({ 'font.family': self.font, 'font.size': self.tick_size, 'axes.titlesize': self.title_size, 'axes.labelsize': self.label_size, 'figure.dpi': self.dpi, 'savefig.dpi': self.dpi * 2, 'savefig.bbox': 'tight', }) def create_line_chart(df, x, y, title, config: Optional[ChartConfig] = None): """创建折线图""" if config is None: config = ChartConfig() config.apply() fig, ax = plt.subplots(figsize=config.figsize) ax.plot(df[x], df[y], marker='o', linewidth=2, color=config.color_palette[0]) ax.set_title(title) ax.set_xlabel(x) ax.set_ylabel(y) plt.xticks(rotation=45) plt.tight_layout() return fig # 使用模板 config = ChartConfig(figsize=(10, 5), title_size=18) fig = create_line_chart(df, '月份', '销售额', '月度销售趋势（使用模板）') plt.savefig('output/template_chart.png') plt.close()

from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side, numbers from openpyxl.chart import BarChart, Reference from openpyxl.utils import get_column_letter wb = Workbook() ws = wb.active ws.title = "2025年销售报告" # 定义样式 title_font = Font(name='Microsoft YaHei', size=16, bold=True, color='FFFFFF') header_font = Font(name='Microsoft YaHei', size=11, bold=True, color='FFFFFF') data_font = Font(name='Microsoft YaHei', size=10) title_fill = PatternFill(start_color='2C3E50', end_color='2C3E50', fill_type='solid') header_fill = PatternFill(start_color='3498DB', end_color='3498DB', fill_type='solid') even_fill = PatternFill(start_color='EBF5FB', end_color='EBF5FB', fill_type='solid') thin_border = Border( left=Side(style='thin', color='BDC3C7'), right=Side(style='thin', color='BDC3C7'), bottom=Side(style='thin', color='BDC3C7'), top=Side(style='thin', color='BDC3C7'), ) center_align = Alignment(horizontal='center', vertical='center', wrap_text=True) # 写入标题（合并单元格） ws.merge_cells('A1:F1') ws['A1'] = '2025年度销售数据报告' ws['A1'].font = title_font ws['A1'].fill = title_fill ws['A1'].alignment = Alignment(horizontal='center', vertical='center') ws.row_dimensions[1].height = 40 # 写入表头 headers = ['月份', '销售额', '成本', '利润', '利润率', '达标'] for col, header in enumerate(headers, 1): cell = ws.cell(row=2, column=col, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = center_align cell.border = thin_border # 写入数据 data_rows = [ ('2025-01', 1200000, 700000, 500000, None, '是'), ('2025-02', 980000, 620000, 360000, None, '否'), ('2025-03', 1350000, 780000, 570000, None, '是'), ] for row_idx, (month, sales, cost, profit, _,达标) in enumerate(data_rows, 3): margin = round(profit / sales * 100, 1) row_data = [month, sales, cost, profit, f'{margin}%',达标] for col_idx, value in enumerate(row_data, 1): cell = ws.cell(row=row_idx, column=col_idx, value=value) cell.font = data_font cell.alignment = center_align cell.border = thin_border if row_idx % 2 == 0: cell.fill = even_fill # 利润率公式 ws.cell(row=row_idx, column=5).value = f'=ROUND(D{row_idx}/B{row_idx}*100,1)&"%"' # 设置列宽 col_widths = [15, 15, 15, 15, 12, 10] for i, w in enumerate(col_widths, 1): ws.column_dimensions[get_column_letter(i)].width = w # 生成图表 chart = BarChart() chart.type = "col" chart.title = "月度销售趋势" chart.y_axis.title = "金额（元）" chart.x_axis.title = "月份" chart.style = 10 data_ref = Reference(ws, min_col=2, min_row=2, max_col=4, max_row=5) cats_ref = Reference(ws, min_col=1, min_row=3, max_row=5) chart.add_data(data_ref, titles_from_data=True) chart.set_categories(cats_ref) chart.shape = 4 ws.add_chart(chart, "A8") wb.save('output/sales_report.xlsx') print("Excel报告已生成: output/sales_report.xlsx")

import xlsxwriter # 创建大文件（xlsxwriter 写入性能更优） workbook = xlsxwriter.Workbook('output/large_sales_report.xlsx') worksheet = workbook.add_worksheet('数据明细') # 定义格式 header_format = workbook.add_format({ 'bold': True, 'font_color': 'white', 'bg_color': '#3498DB', 'border': 1, 'text_wrap': True, 'align': 'center', 'valign': 'vcenter' }) money_format = workbook.add_format({'num_format': '#,##0', 'border': 1, 'align': 'center'}) pct_format = workbook.add_format({'num_format': '0.0%', 'border': 1, 'align': 'center'}) # 批量写入数据（xlsxwriter 适合大文件） import random headers = ['日期', '区域', '产品', '销量', '单价', '金额', '利润率'] for col, header in enumerate(headers): worksheet.write(0, col, header, header_format) # 模拟1000行数据 for row in range(1, 1001): worksheet.write(row, 0, f'2025-{random.randint(1,12):02d}-{random.randint(1,28):02d}') worksheet.write(row, 1, random.choice(['华东', '华南', '华北', '西南'])) worksheet.write(row, 2, random.choice(['产品A', '产品B', '产品C', '产品D'])) worksheet.write(row, 3, random.randint(10, 500), money_format) worksheet.write(row, 4, random.uniform(10.0, 500.0), money_format) worksheet.write(row, 5, random.randint(1000, 50000), money_format) worksheet.write(row, 6, random.uniform(0.05, 0.45), pct_format) # 设置列宽 worksheet.set_column('A:A', 14) worksheet.set_column('B:B', 10) worksheet.set_column('C:C', 10) worksheet.set_column('D:G', 12) # 冻结首行 worksheet.freeze_panes(1, 0) # 添加自动筛选 worksheet.autofilter(0, 0, 1000, 6) workbook.close() print("大文件Excel已生成（1000行数据）")

# 安装 # pip install python-docx 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 from docx.oxml.ns import qn import datetime doc = Document() # 设置默认字体 style = doc.styles['Normal'] font = style.font font.name = 'Microsoft YaHei' font.size = Pt(11) font.color.rgb = RGBColor(0x33, 0x33, 0x33) style.element.rPr.rFonts.set(qn('w:eastAsia'), 'Microsoft YaHei') # 标题 title = doc.add_heading('2025年度数据分析报告', level=0) title.alignment = WD_ALIGN_PARAGRAPH.CENTER # 基本信息段落 p = doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER run = p.add_run(f'生成日期: {datetime.date.today()}') run.font.size = Pt(10) run.font.color.rgb = RGBColor(0x7F, 0x8C, 0x8D) # 目录段落 doc.add_paragraph() doc.add_heading('目录', level=1) toc_items = ['一、数据概览', '二、销售趋势分析', '三、区域对比', '四、结论与建议'] for item in toc_items: doc.add_paragraph(item, style='List Number') # 章节一：数据概览 doc.add_page_break() doc.add_heading('一、数据概览', level=1) doc.add_paragraph( '本次分析覆盖2025年1月至12月的全量销售数据，' f'共计处理交易记录1,284,567条，涉及4个区域、12个产品品类。' f'整体数据完整率为98.7%，缺失值已通过均值填充处理。' ) # 插入图片 doc.add_heading('1.1 销售趋势图', level=2) doc.add_picture('output/sales_trend.png', width=Inches(6)) last_paragraph = doc.paragraphs[-1] last_paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER # 章节二：销售趋势分析 doc.add_heading('二、销售趋势分析', level=1) p = doc.add_paragraph() p.add_run('核心发现: ').bold = True p.add_run('全年销售额呈现稳步增长态势，Q4季度环比增长23.5%，' '主要受促销活动推动。7月和11月为两个销售高峰。') # 引用块 doc.add_paragraph('') quote = doc.add_paragraph() quote.alignment = WD_ALIGN_PARAGRAPH.LEFT run = quote.add_run('"2025年整体GMV同比增长18.7%，超额完成年度目标103.2%。"') run.italic = True run.font.color.rgb = RGBColor(0x55, 0x55, 0x55) # 表格插入 doc.add_heading('2.1 季度销售数据', level=2) table = doc.add_table(rows=5, cols=5) table.style = 'Light Grid Accent 1' table.alignment = WD_TABLE_ALIGNMENT.CENTER headers = ['季度', '销售额（万）', '成本（万）', '利润（万）', '利润率'] quarter_data = [ ['Q1', '3,530', '2,100', '1,430', '40.5%'], ['Q2', '4,120', '2,350', '1,770', '43.0%'], ['Q3', '4,580', '2,480', '2,100', '45.9%'], ['Q4', '5,660', '2,950', '2,710', '47.9%'], ] for col_idx, header in enumerate(headers): table.cell(0, col_idx).text = header for row_idx, row_data in enumerate(quarter_data, 1): for col_idx, value in enumerate(row_data): table.cell(row_idx, col_idx).text = value # 保存文档 doc.save('output/sales_report.docx') print("Word报告已生成: output/sales_report.docx")

# 基于模板的Word报告生成（使用Jinja2 + python-docx） # 先准备好带占位符的模板文件，然后替换 from docx import Document import re def fill_template(template_path, output_path, replacements): """替换Word模板中的占位符""" doc = Document(template_path) # 替换段落中的占位符 for paragraph in doc.paragraphs: for run in paragraph.runs: for key, value in replacements.items(): if key in run.text: run.text = run.text.replace(key, str(value)) # 替换表格中的占位符 for table in doc.tables: for row in table.rows: for cell in row.cells: for paragraph in cell.paragraphs: for run in paragraph.runs: for key, value in replacements.items(): if key in run.text: run.text = run.text.replace(key, str(value)) doc.save(output_path) print(f"模板替换完成: {output_path}") # 使用示例 replacements = { '{{报告日期}}': '2025年12月31日', '{{总销售额}}': '17,890万元', '{{同比增长}}': '18.7%', '{{总利润}}': '8,010万元', '{{利润率}}': '44.8%', '{{数据完整率}}': '98.7%', '{{客户总数}}': '256,891', '{{区域列表}}': '华东、华南、华北、西南', '{{分析师姓名}}': '数据分析团队', } fill_template('templates/report_template.docx', 'output/final_report.docx', replacements)

import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email.header import Header from email import encoders import os # 邮件配置 SMTP_HOST = 'smtp.company.com' SMTP_PORT = 587 SMTP_USER = 'report-bot@company.com' SMTP_PASS = 'your-password-here' # 生产环境应从密钥管理服务获取 def send_html_email(recipients, subject, html_content, attachments=None): """发送HTML格式邮件""" msg = MIMEMultipart('alternative') msg['From'] = SMTP_USER msg['To'] = ', '.join(recipients) msg['Subject'] = Header(subject, 'utf-8') # HTML正文 html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) # 添加附件 if attachments: for filepath in attachments: if os.path.exists(filepath): with open(filepath, 'rb') as f: attachment = MIMEBase('application', 'octet-stream') attachment.set_payload(f.read()) encoders.encode_base64(attachment) filename = os.path.basename(filepath) attachment.add_header( 'Content-Disposition', 'attachment', filename=('utf-8', '', filename) ) msg.attach(attachment) # 发送 with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server: server.starttls() server.login(SMTP_USER, SMTP_PASS) server.sendmail(SMTP_USER, recipients, msg.as_string()) print(f"邮件已发送至 {len(recipients)} 个收件人") # 构建HTML邮件正文 html_report = """ """ # 实际发送 send_html_email( recipients=['manager@company.com', 'director@company.com'], subject='2025年度销售数据分析报告', html_content=html_report, attachments=[ 'output/sales_report.xlsx', 'output/sales_trend.png', 'output/sales_report.docx' ] )

# 多区域批量邮件分发 import yaml import schedule import time # 区域配置（可从YAML文件读取） config = """ recipients: 华东区: - manager-east@company.com - director-east@company.com 华南区: - manager-south@company.com 华北区: - manager-north@company.com - analyst-north@company.com 西南区: - manager-west@company.com """ config_dict = yaml.safe_load(config) def generate_and_send_all(): """执行完整的数据拉取-分析-生成-发送流程""" print(f"[{datetime.now()}] 开始自动化报告流程") for region, emails in config_dict['recipients'].items(): # 1. 生成区域报告（调用Papermill） params = {"region": region, "date_range": "2025全年"} pm.execute_notebook( 'templates/region_template.ipynb', f'output/report_{region}.ipynb', parameters=params, kernel_name='python3' ) # 2. 转换为HTML import nbformat from nbconvert import HTMLExporter with open(f'output/report_{region}.ipynb', 'r') as f: nb = nbformat.read(f, as_version=4) html_exporter = HTMLExporter() body, _ = html_exporter.from_notebook_node(nb) with open(f'output/report_{region}.html', 'w') as f: f.write(body) # 3. 发送邮件 send_html_email( recipients=emails, subject=f'{region}2025年度销售报告', html_content=body, attachments=[f'output/report_{region}.html'] ) print(f" [{region}] 报告已发送至 {len(emails)} 人") print(f"[{datetime.now()}] 全部报告发送完成") # 定时任务（每周一早上9点执行） schedule.every().monday.at("09:00").do(generate_and_send_all) # 持续运行 while True: schedule.run_pending() time.sleep(60)

# 完整自动化报告 Pipeline import os import sys import logging from datetime import datetime from pathlib import Path # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler(f'logs/report_{datetime.now():%Y%m%d}.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class AutoReportPipeline: """自动化报告管线""" def __init__(self, config_path='config.yaml'): self.config = self._load_config(config_path) self.output_dir = Path(self.config.get('output_dir', 'output')) self.output_dir.mkdir(parents=True, exist_ok=True) def _load_config(self, path): with open(path, 'r') as f: return yaml.safe_load(f) def step_1_data_extraction(self): """步骤1: 数据提取""" logger.info("开始数据提取...") # 从数据库/API/文件读取数据 # df = pd.read_sql("SELECT * FROM sales WHERE date >= %s", conn, params=(start_date,)) logger.info("数据提取完成") def step_2_eda_analysis(self): """步骤2: 自动EDA""" logger.info("开始自动EDA...") # profile = ProfileReport(df, title='EDA报告') # profile.to_file(str(self.output_dir / 'eda_report.html')) logger.info("EDA报告生成完成") def step_3_visualization(self): """步骤3: 可视化""" logger.info("开始生成图表...") # 生成各类图表并保存 # plt.savefig(self.output_dir / 'chart.png') logger.info("图表生成完成") def step_4_report_generation(self): """步骤4: 报告文档生成""" logger.info("开始生成报告文档...") # 生成Excel/Word/PDF报告 logger.info("报告文档生成完成") def step_5_email_delivery(self): """步骤5: 邮件发送""" logger.info("开始发送邮件...") # send_html_email(...) logger.info("邮件发送完成") def run(self): """执行完整管线""" start_time = datetime.now() logger.info("=" * 50) logger.info("自动化报告管线启动") logger.info("=" * 50) try: self.step_1_data_extraction() self.step_2_eda_analysis() self.step_3_visualization() self.step_4_report_generation() self.step_5_email_delivery() elapsed = (datetime.now() - start_time).total_seconds() logger.info(f"管线执行完成，耗时 {elapsed:.2f} 秒") return True except Exception as e: logger.error(f"管线执行失败: {str(e)}") # 发送失败通知 return False if __name__ == '__main__': pipeline = AutoReportPipeline('config.yaml') success = pipeline.run() sys.exit(0 if success else 1)