Python从零构建一个自动化报表生成器

pip install pandas openpyxl PyPDF2 pywin32  # pywin32用于Windows COM操作
# 如需连接数据库，还需安装 sqlalchemy pymysql 等

import pandas as pd

def read_csv_data(file_path):
    df = pd.read_csv(file_path, parse_dates=['date'])
    print(f"从CSV读取数据，共 {len(df)} 行")
    return df

from sqlalchemy import create_engine

def read_db_data(db_url, table_name, start_date, end_date):
    engine = create_engine(db_url)
    query = f"""
        SELECT * FROM {table_name}
        WHERE date BETWEEN '{start_date}' AND '{end_date}'
    """
    df = pd.read_sql(query, engine, parse_dates=['date'])
    print(f"从数据库读取数据，共 {len(df)} 行")
    return df

def clean_data(df):
    # 删除重复行
    df = df.drop_duplicates()
    
    # 处理缺失值：这里假设total_amount不能为空，若为空则填充为0
    df['total_amount'] = df['total_amount'].fillna(0)
    
    # 确保数值列类型正确
    df['quantity'] = pd.to_numeric(df['quantity'], errors='coerce').fillna(0)
    df['unit_price'] = pd.to_numeric(df['unit_price'], errors='coerce').fillna(0)
    
    # 过滤：销量和单价不能为负数
    df = df[(df['quantity'] >= 0) & (df['unit_price'] >= 0)]
    
    # 重新计算total_amount（防止原始数据有误）
    df['total_amount'] = df['quantity'] * df['unit_price']
    
    return df

def analyze_data(df):
    # 按区域和门店汇总
    summary_by_store = df.groupby(['region', 'store']).agg(
        total_sales=('total_amount', 'sum'),
        order_count=('quantity', 'count'),  # 假设每条记录为一笔订单
        avg_order_value=('total_amount', 'mean')
    ).reset_index()
    
    # 按日期汇总每日销售额
    daily_sales = df.groupby(df['date'].dt.date)['total_amount'].sum().reset_index()
    daily_sales.columns = ['date', 'daily_total']
    
    # 计算全局指标
    total_sales = df['total_amount'].sum()
    total_orders = len(df)
    avg_customer_value = total_sales / total_orders if total_orders > 0 else 0
    
    metrics = {
        'total_sales': total_sales,
        'total_orders': total_orders,
        'avg_customer_value': avg_customer_value
    }
    
    return summary_by_store, daily_sales, metrics

from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Border, Side, Alignment, NamedStyle
from openpyxl.chart import LineChart, Reference
from openpyxl.utils.dataframe import dataframe_to_rows
import pandas as pd

def create_excel_report(df_raw, summary_df, daily_df, metrics, output_path):
    wb = Workbook()
    
    # 1. 原始数据 sheet
    ws_raw = wb.active
    ws_raw.title = "Raw Data"
    # 将DataFrame逐行写入
    for r in dataframe_to_rows(df_raw, index=False, header=True):
        ws_raw.append(r)
    
    # 2. 汇总表 sheet
    ws_summary = wb.create_sheet("Summary by Store")
    for r in dataframe_to_rows(summary_df, index=False, header=True):
        ws_summary.append(r)
    
    # 3. 每日趋势 sheet
    ws_trend = wb.create_sheet("Daily Trend")
    for r in dataframe_to_rows(daily_df, index=False, header=True):
        ws_trend.append(r)
    
    # 保存临时文件，后续还要进行格式化和图表插入
    wb.save(output_path)
    return wb  # 返回工作簿对象以便继续操作

def format_excel(wb):
    # 定义常用样式
    header_font = Font(bold=True, color="FFFFFF")
    header_fill = PatternFill(start_color="4F81BD", end_color="4F81BD", fill_type="solid")
    thin_border = Border(
        left=Side(style='thin'), 
        right=Side(style='thin'), 
        top=Side(style='thin'), 
        bottom=Side(style='thin')
    )
    currency_style = NamedStyle(name="currency", number_format='"¥"#,##0.00')
    
    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
        
        # 设置标题行样式（第一行）
        for cell in ws[1]:
            cell.font = header_font
            cell.fill = header_fill
            cell.border = thin_border
            cell.alignment = Alignment(horizontal='center')
        
        # 调整列宽
        for col in ws.columns:
            max_length = 0
            col_letter = col[0].column_letter
            for cell in col:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length + 2, 50)
            ws.column_dimensions[col_letter].width = adjusted_width
        
        # 为包含金额的列应用货币格式（假设列名包含'sales'或'amount'）
        for row in ws.iter_rows(min_row=2, max_row=ws.max_row):
            for cell in row:
                cell.border = thin_border
                if cell.column_letter in ['C','D']:  # 根据实际列调整
                    cell.number_format = currency_style.number_format
    
    return wb

from openpyxl.chart import LineChart, Reference

def add_chart(wb):
    ws = wb["Daily Trend"]
    # 数据范围：假设日期在A列，销售额在B列，从第2行开始到最后
    data = Reference(ws, min_col=2, min_row=1, max_row=ws.max_row, max_col=2)
    dates = Reference(ws, min_col=1, min_row=2, max_row=ws.max_row)
    
    chart = LineChart()
    chart.add_data(data, titles_from_data=True)
    chart.set_categories(dates)
    chart.title = "每日销售额趋势"
    chart.x_axis.title = "日期"
    chart.y_axis.title = "销售额"
    chart.style = 12  # 内置样式
    
    # 将图表放在E2单元格附近
    ws.add_chart(chart, "E2")
    return wb

def create_summary_sheet(wb, metrics):
    ws = wb.create_sheet("Summary", 0)  # 插入到第一个位置
    
    # 写入标题
    ws['A1'] = "销售周报总结"
    ws['A1'].font = Font(size=16, bold=True)
    ws.merge_cells('A1:C1')
    
    # 写入核心指标
    row = 3
    ws[f'A{row}'] = "总销售额："
    ws[f'B{row}'] = metrics['total_sales']
    ws[f'B{row}'].number_format = '"¥"#,##0.00'
    row += 1
    ws[f'A{row}'] = "总订单数："
    ws[f'B{row}'] = metrics['total_orders']
    row += 1
    ws[f'A{row}'] = "客单价："
    ws[f'B{row}'] = metrics['avg_customer_value']
    ws[f'B{row}'].number_format = '"¥"#,##0.00'
    
    # 添加结论性文字
    row += 2
    ws[f'A{row}'] = "结论："
    row += 1
    conclusion = f"上周总销售额为¥{metrics['total_sales']:,.0f}，共完成{metrics['total_orders']}笔订单，客单价为¥{metrics['avg_customer_value']:.2f}。"
    ws[f'A{row}'] = conclusion
    ws.merge_cells(f'A{row}:C{row}')
    
    # 调整列宽
    ws.column_dimensions['A'].width = 15
    ws.column_dimensions['B'].width = 20
    ws.column_dimensions['C'].width = 15
    
    return wb

import win32com.client
import os

def export_sheet_to_pdf(excel_path, sheet_name, pdf_path):
    """
    使用Excel COM对象将指定sheet导出为PDF
    """
    excel = win32com.client.Dispatch("Excel.Application")
    excel.Visible = False  # 后台运行
    
    try:
        wb = excel.Workbooks.Open(os.path.abspath(excel_path))
        ws = wb.Sheets(sheet_name)
        
        # 导出为PDF
        ws.ExportAsFixedFormat(0, pdf_path)  # 0代表xlTypePDF
        print(f"已导出 {sheet_name} 到 {pdf_path}")
    except Exception as e:
        print(f"导出PDF失败: {e}")
        raise
    finally:
        wb.Close(SaveChanges=False)
        excel.Quit()

import PyPDF2

def encrypt_pdf(input_pdf, output_pdf, password):
    with open(input_pdf, 'rb') as file:
        pdf_reader = PyPDF2.PdfReader(file)
        pdf_writer = PyPDF2.PdfWriter()
        
        for page_num in range(len(pdf_reader.pages)):
            pdf_writer.add_page(pdf_reader.pages[page_num])
        
        pdf_writer.encrypt(password)
        
        with open(output_pdf, 'wb') as out_file:
            pdf_writer.write(out_file)
    print(f"PDF已加密，保存至 {output_pdf}")

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
import os

def send_email(sender, password, receiver, subject, body, attachments):
    # 创建邮件对象
    msg = MIMEMultipart()
    msg['From'] = sender
    msg['To'] = receiver
    msg['Subject'] = subject
    
    # 添加正文
    msg.attach(MIMEText(body, 'plain', 'utf-8'))
    
    # 添加附件
    for file_path in attachments:
        with open(file_path, 'rb') as attachment:
            part = MIMEBase('application', 'octet-stream')
            part.set_payload(attachment.read())
            encoders.encode_base64(part)
            part.add_header(
                'Content-Disposition',
                f'attachment; filename= {os.path.basename(file_path)}'
            )
            msg.attach(part)
    
    # 发送邮件（以QQ邮箱为例）
    try:
        server = smtplib.SMTP_SSL('smtp.qq.com', 465)
        server.login(sender, password)
        server.send_message(msg)
        server.quit()
        print("邮件发送成功")
    except Exception as e:
        print(f"邮件发送失败: {e}")

def generate_report(csv_path, excel_output, pdf_output, email_config):
    # 1. 读取数据
    df_raw = read_csv_data(csv_path)
    
    # 2. 清洗与分析
    df_clean = clean_data(df_raw)
    summary_df, daily_df, metrics = analyze_data(df_clean)
    
    # 3. 生成Excel（先保存基本数据）
    wb = create_excel_report(df_clean, summary_df, daily_df, metrics, excel_output)
    wb = format_excel(wb)
    wb = add_chart(wb)
    wb = create_summary_sheet(wb, metrics)
    wb.save(excel_output)
    print(f"Excel报表已生成: {excel_output}")
    
    # 4. 将Summary sheet导出为PDF
    export_sheet_to_pdf(excel_output, "Summary", pdf_output)
    
    # 可选：用PyPDF2加密PDF
    # pdf_encrypted = pdf_output.replace('.pdf', '_encrypted.pdf')
    # encrypt_pdf(pdf_output, pdf_encrypted, 'weekly123')
    
    # 5. 发送邮件
    send_email(
        sender=email_config['sender'],
        password=email_config['password'],
        receiver=email_config['receiver'],
        subject=email_config['subject'],
        body=email_config['body'],
        attachments=[excel_output, pdf_output]  # 同时发送Excel和PDF
    )
    
    print("报表生成及发送流程全部完成！")

if __name__ == "__main__":
    csv_file = "data/sales_data.csv"
    excel_file = "output/sales_report.xlsx"
    pdf_file = "output/summary.pdf"
    
    email_cfg = {
        'sender': 'your_email@qq.com',
        'password': 'your_authorization_code',  # QQ邮箱授权码
        'receiver': 'boss@company.com',
        'subject': '上周销售报表',
        'body': '您好，附件是上周销售报表，请查收。\n\n数据分析团队'
    }
    
    generate_report(csv_file, excel_file, pdf_file, email_cfg)