报表自动化开发

中际旭创的岗位描述："设计并实现定期自动推送的Excel报表、PDF报告或邮件报表，减少人工统计工作量"

翻译成程序员语言就是：

• 从 MES / ERP / SQL Server 自动查询生产数据
• 用 Python 自动生成 格式化的 Excel（带公式、图表、条件格式）
• 自动生成 PDF 报告（标题、表格、图表）
• 定时自动通过 邮件推送给相关人员
• 整个过程 无人值守，设置好定时任务就行

# Excel 操作库
pip install openpyxl

# 数据库连接（SQL Server）
pip install pymssql

# 如果用 xlsxwriter（备选）
pip install XlsxWriter

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.formatting.rule import CellIsRule
from openpyxl.chart import BarChart, Reference
from openpyxl.utils import get_column_letter
from datetime import datetime, timedelta

def generate_oee_excel(report_date=None):
    """生成每日 OEE 日报 Excel 文件"""

    # ========== 1. 准备数据（实际场景从数据库查询）==========
    if report_date is None:
        report_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

    # 模拟各产线 OEE 数据
    production_data = [
        {"line": "Line-01", "product": "100G光模块", "plan_qty": 1200, "actual_qty": 1150,
         "defect_qty": 23, "run_hours": 22.5, "plan_hours": 24},
        {"line": "Line-02", "product": "400G光模块", "plan_qty": 800, "actual_qty": 720,
         "defect_qty": 15, "run_hours": 20.0, "plan_hours": 24},
        {"line": "Line-03", "product": "800G光模块", "plan_qty": 500, "actual_qty": 485,
         "defect_qty": 8, "run_hours": 21.0, "plan_hours": 24},
        {"line": "Line-04", "product": "200G光模块", "plan_qty": 1000, "actual_qty": 780,
         "defect_qty": 45, "run_hours": 18.5, "plan_hours": 24},
        {"line": "Line-05", "product": "100G光模块", "plan_qty": 1200, "actual_qty": 1180,
         "defect_qty": 12, "run_hours": 23.0, "plan_hours": 24},
    ]

    # 计算 OEE 三要素
    for d in production_data:
        d["availability"] = d["run_hours"] / d["plan_hours"] * 100     # 可用率
        d["performance"]  = d["actual_qty"] / d["plan_qty"] * 100   # 表现指数
        d["quality"]      = (d["actual_qty"] - d["defect_qty"]) / d["actual_qty"] * 100
        d["oee"]          = d["availability"] * d["performance"] * d["quality"] / 10000

    # ========== 2. 创建 Excel 工作簿 ==========
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "OEE日报"

    # ========== 3. 定义样式 ==========
    header_font = Font(name="微软雅黑", size=12, bold=True, color="FFFFFF")
    header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
    title_font  = Font(name="微软雅黑", size=18, bold=True, color="1F4E79")
    data_font   = Font(name="微软雅黑", size=11)
    center_align = Alignment(horizontal="center", vertical="center")
    thin_border  = Border(
        left=Side(style="thin"), right=Side(style="thin"),
        top=Side(style="thin"), bottom=Side(style="thin")
    )

    # ========== 4. 写入标题行 ==========
    ws.merge_cells("A1:K1")
    title_cell = ws["A1"]
    title_cell.value = f"中际旭创 生产OEE日报 - {report_date}"
    title_cell.font = title_font
    title_cell.alignment = center_align

    # ========== 5. 写入表头 ==========
    headers = ["产线", "产品", "计划产量", "实际产量", "不良数",
               "运行时间(h)", "计划时间(h)", "可用率%",
               "表现指数%", "质量指数%", "OEE%"]

    for col, h in enumerate(headers, 1):
        cell = ws.cell(row=3, column=col, value=h)
        cell.font = header_font
        cell.fill = header_fill
        cell.alignment = center_align
        cell.border = thin_border

    # ========== 6. 写入数据行 ==========
    for i, d in enumerate(production_data):
        row = i + 4
        values = [d["line"], d["product"], d["plan_qty"], d["actual_qty"],
                  d["defect_qty"], d["run_hours"], d["plan_hours"],
                  round(d["availability"], 1), round(d["performance"], 1),
                  round(d["quality"], 1), round(d["oee"], 1)]
        for col, val in enumerate(values, 1):
            cell = ws.cell(row=row, column=col, value=val)
            cell.font = data_font
            cell.alignment = center_align
            cell.border = thin_border

    # ========== 7. 写入汇总行（使用 Excel 公式）==========
    summary_row = 4 + len(production_data)
    ws.cell(row=summary_row, column=1, value="合计/平均").font = Font(bold=True, size=11)
    ws.cell(row=summary_row, column=1).alignment = center_align

    # SUM 公式
    ws.cell(row=summary_row, column=3).value = "=SUM(C4:C8)"    # 计划产量合计
    ws.cell(row=summary_row, column=4).value = "=SUM(D4:D8)"    # 实际产量合计
    ws.cell(row=summary_row, column=5).value = "=SUM(E4:E8)"    # 不良数合计

    # AVERAGE 公式
    ws.cell(row=summary_row, column=8).value = "=AVERAGE(H4:H8)"  # 平均可用率
    ws.cell(row=summary_row, column=9).value = "=AVERAGE(I4:I8)"  # 平均表现
    ws.cell(row=summary_row, column=10).value = "=AVERAGE(J4:J8)" # 平均质量
    ws.cell(row=summary_row, column=11).value = "=AVERAGE(K4:K8)" # 平均 OEE

    for col in range(1, 12):
        ws.cell(row=summary_row, column=col).font = Font(bold=True, size=11)
        ws.cell(row=summary_row, column=col).alignment = center_align
        ws.cell(row=summary_row, column=col).border = thin_border

    # ========== 8. 条件格式：OEE < 80% 标红 ==========
    red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
    red_font = Font(color="9C0006", bold=True)
    ws.conditional_formatting.add(
        "K4:K8",
        CellIsRule(operator="lessThan", formula=["80"], fill=red_fill, font=red_font)
    )

    # ========== 9. 调整列宽 ==========
    col_widths = [10, 14, 12, 12, 10, 14, 14, 10, 12, 12, 10]
    for i, w in enumerate(col_widths, 1):
        ws.column_dimensions[get_column_letter(i)].width = w

    # ========== 10. 创建柱状图 ==========
    chart = BarChart()
    chart.type = "col"
    chart.title = f"各产线 OEE 对比 - {report_date}"
    chart.y_axis.title = "百分比 (%)"
    chart.x_axis.title = "产线"
    chart.style = 10

    # 图表数据：OEE 列（K列 = 第11列）
    data_ref = Reference(ws, min_col=11, min_row=3, max_row=8)
    cats_ref = Reference(ws, min_col=1, min_row=4, max_row=8)
    chart.add_data(data_ref, titles_from_data=True)
    chart.set_categories(cats_ref)
    chart.shape = 4
    chart.width = 20
    chart.height = 12

    ws.add_chart(chart, "A11")

    # ========== 11. 保存文件 ==========
    filename = f"OEE日报_{report_date}.xlsx"
    wb.save(filename)
    print(f"[OK] Excel 已生成: {filename}")
    return filename


# 运行
if __name__ == "__main__":
    generate_oee_excel()

import pymssql
from datetime import datetime, timedelta

def query_production_data(report_date=None):
    """从 SQL Server 查询生产数据"""
    if report_date is None:
        report_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

    conn = pymssql.connect(
        server="192.168.1.100",     # MES 数据库地址
        user="report_user",
        password="your_password",
        database="MES_Production"
    )
    cursor = conn.cursor()

    sql = """
    SELECT
        line_name,
        product_model,
        planned_qty,
        actual_qty,
        defect_qty,
        run_hours,
        planned_hours
    FROM production_daily
    WHERE report_date = %s
    ORDER BY line_name
    """

    cursor.execute(sql, (report_date,))
    rows = cursor.fetchall()

    # 转换为字典列表
    columns = ["line", "product", "plan_qty", "actual_qty",
               "defect_qty", "run_hours", "plan_hours"]
    data = [dict(zip(columns, row)) for row in rows]

    conn.close()
    return data

pip install reportlab
# 中文字体支持（reportlab 默认不支持中文）
pip install reportlab
# 需要下载中文字体文件：SimHei.ttf 或 微软雅黑.ttf

from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import mm, cm
from reportlab.platypus import (
    SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, Image, PageBreak
)
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from datetime import datetime, timedelta
import os

# ========== 注册中文字体 ==========
FONT_PATH = "/usr/share/fonts/SimHei.ttf"  # Linux 服务器路径
# FONT_PATH = "C:/Windows/Fonts/simhei.ttf"  # Windows 路径
if os.path.exists(FONT_PATH):
    pdfmetrics.registerFont(TTFont("SimHei", FONT_PATH))
    CN_FONT = "SimHei"
else:
    CN_FONT = "Helvetica"  # fallback


def generate_pdf_report(report_date=None):
    """生成 PDF 生产日报"""
    if report_date is None:
        report_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

    filename = f"生产日报_{report_date}.pdf"
    doc = SimpleDocTemplate(
        filename,
        pagesize=A4,
        rightMargin=2*cm, leftMargin=2*cm,
        topMargin=2*cm, bottomMargin=2*cm
    )

    styles = getSampleStyleSheet()

    # 自定义中文样式
    styles.add(ParagraphStyle(
        name="CNTitle",
        fontName=CN_FONT,
        fontSize=22,
        alignment=1,  # 居中
        spaceAfter=20,
        textColor=colors.HexColor("#7EC8F0")
    ))
    styles.add(ParagraphStyle(
        name="CNSubtitle",
        fontName=CN_FONT,
        fontSize=14,
        alignment=1,
        spaceAfter=10,
        textColor=colors.grey
    ))
    styles.add(ParagraphStyle(
        name="CNBody",
        fontName=CN_FONT,
        fontSize=11,
        leading=18,
        spaceAfter=6
    ))
    styles.add(ParagraphStyle(
        name="CNHeading",
        fontName=CN_FONT,
        fontSize=14,
        spaceAfter=8,
        textColor=colors.HexColor("#7EC8F0")
    ))

    elements = []

    # ========== 1. 标题页 ==========
    elements.append(Spacer(1, 80*mm))
    elements.append(Paragraph("中际旭创股份有限公司", styles["CNTitle"]))
    elements.append(Spacer(1, 10*mm))
    elements.append(Paragraph(f"生产日报 - {report_date}", styles["CNTitle"]))
    elements.append(Spacer(1, 15*mm))
    elements.append(Paragraph("报告类型：每日生产OEE汇总", styles["CNSubtitle"]))
    elements.append(Paragraph(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", styles["CNSubtitle"]))
    elements.append(Paragraph("数据来源：MES 生产管理系统", styles["CNSubtitle"]))
    elements.append(PageBreak())

    # ========== 2. 概要信息 ==========
    elements.append(Paragraph("一、生产概要", styles["CNHeading"]))
    summary_data = [
        ["指标", "数值"],
        ["报告日期", report_date],
        ["产线数量", "5 条"],
        ["总计划产量", "4,700 件"],
        ["总实际产量", "4,315 件"],
        ["达成率", "91.8%"],
        ["平均 OEE", "80.4%"],
        ["异常产线", "Line-04 (OEE: 56.7%, 需关注)"],
    ]
    summary_table = Table(summary_data, colWidths=[80*mm, 90*mm])
    summary_table.setStyle(TableStyle([
        ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#7EC8F0")),
        ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
        ("FONTSIZE", (0, 0), (-1), -1, 11),
        ("ALIGN", (0, 0), (-1), -1, "CENTER"),
        ("GRID", (0, 0), (-1), -1, 0.5, colors.grey),
        ("ROWBACKGROUNDS", (0, 1), (-1), -1),
        ("FONTNAME", (0, 0), (-1), -1, CN_FONT),
    ]))
    elements.append(summary_table)
    elements.append(Spacer(1, 15*mm))

    # ========== 3. 详细数据表 ==========
    elements.append(Paragraph("二、各产线 OEE 详情", styles["CNHeading"]))
    detail_data = [
        ["产线", "产品", "计划", "实际", "不良",
         "可用率", "表现", "质量", "OEE"],
        ["Line-01", "100G", 1200, 1150, 23, 93.8, 95.8, 98.0, 88.0],
        ["Line-02", "400G", 800, 720, 15, 83.3, 90.0, 97.9, 80.7],
        ["Line-03", "800G", 500, 485, 8, 87.5, 97.0, 98.4, 83.6],
        ["Line-04", "200G", 1000, 780, 45, 77.1, 78.0, 94.2, 56.7],
        ["Line-05", "100G", 1200, 1180, 12, 95.8, 98.3, 99.0, 93.2],
    ]
    detail_table = Table(detail_data, colWidths=[45]*9)
    detail_table.setStyle(TableStyle([
        ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#7EC8F0")),
        ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
        # Line-04 OEE 标红
        ("TEXTCOLOR", (8, 4), (8, 4), colors.red),
        ("FONTSIZE", (0, 0), (-1), -1, 10),
        ("ALIGN", (0, 0), (-1), -1, "CENTER"),
        ("GRID", (0, 0), (-1), -1, 0.5, colors.grey),
        ("ROWBACKGROUNDS", (0, 1), (-1), -1),
    ]))
    elements.append(detail_table)
    elements.append(Spacer(1, 15*mm))

    # ========== 4. 嵌入图表（如果有 ECharts 导出的图片）==========
    chart_image = f"oee_chart_{report_date}.png"
    if os.path.exists(chart_image):
        elements.append(Paragraph("三、OEE 对比图", styles["CNHeading"]))
        img = Image(chart_image, width=150*mm, height=90*mm)
        elements.append(img)

    # ========== 5. 构建 PDF ==========
    doc.build(elements)
    print(f"[OK] PDF 已生成: {filename}")
    return filename


if __name__ == "__main__":
    generate_pdf_report()

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
from datetime import datetime, timedelta
import os


# ========== SMTP 邮件配置 ==========
SMTP_CONFIG = {
    "host": "smtp.company.com",   # 企业邮箱 SMTP 地址
    "port": 465,                    # SSL 端口（TLS 用 587）
    "user": "report@company.com",
    "password": "your_app_password",  # 授权码，不是登录密码
    "use_ssl": True,
}

# 收件人列表
RECIPIENTS = {
    "to": [
        "manager@company.com",      # 生产经理
        "director@company.com",     # 生产总监
    ],
    "cc": [
        "engineer1@company.com",    # 工程师
        "engineer2@company.com",    # 工程师
    ],
}


def create_html_body(report_date, summary_data):
    """生成 HTML 邮件正文"""
    return f"""
    <html>
    <body style="font-family: Microsoft YaHei, sans-serif; color:#CCColor: #CCC;">
      <h2 style="color: #7EC8F0;">
        中际旭创 生产OEE日报 - {report_date}
      </h2>

      <table style="border-collapse: collapse; width: 100%; margin: 20px 0;">
        <tr style="background: #7EC8F0; color: white;">
          <th style="padding: 10px; border: 1px solid #ddd;">指标</th>
          <th style="padding: 10px; border: 1px solid #ddd;">数值</th>
        </tr>
        <tr>
          <td style="padding: 8px; border: 1px solid #ddd;">总计划产量</td>
          <td style="padding: 8px; border: 1px solid #ddd;">
            {summary_data['total_plan']:,}
          </td>
        </tr>
        <tr style="background: #f5f5f5;">
          <td style="padding: 8px; border: 1px solid #ddd;">总实际产量</td>
          <td style="padding: 8px; border: 1px solid #ddd;">
            {summary_data['total_actual']:,}
          </td>
        </tr>
        <tr>
          <td style="padding: 8px; border: 1px solid #ddd;">平均 OEE</td>
          <td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;
             color: {'red' if summary_data['avg_oee'] < 80 else 'green'};">
            {summary_data['avg_oee']:.1f}%
          </td>
        </tr>
        <tr style="background: #fff3cd;">
          <td style="padding: 8px; border: 1px solid #ddd;">异常产线</td>
          <td style="padding: 8px; border: 1px solid #ddd; color: red;">
            {summary_data.get('alert_lines', '无')}
          </td>
        </tr>
      </table>

      <p style="color:#CCColor: #CCC; font-size: 12px;">
        本邮件由报表自动推送系统发送，请勿直接回复。
        详细数据请查看附件中的 Excel 报表。
      </p>
    </body>
    </html>
    """


def send_report_email(report_date, excel_file, pdf_file, summary_data):
    """发送报表邮件"""
    try:
        # 1. 构建邮件
        msg = MIMEMultipart()
        msg["From"] = SMTP_CONFIG["user"]
        msg["To"] = ",".join(RECIPIENTS["to"])
        msg["Cc"] = ",".join(RECIPIENTS["cc"])
        msg["Subject"] = f"[自动报表] 生产OEE日报 - {report_date}"

        # 2. HTML 正文
        html_body = create_html_body(report_date, summary_data)
        msg.attach(MIMEText(html_body, "html", "utf-8"))

        # 3. 添加 Excel 附件
        if excel_file and os.path.exists(excel_file):
            with open(excel_file, "rb") as f:
                excel_attach = MIMEApplication(f.read())
                excel_attach.add_header(
                    "Content-Disposition", "attachment",
                    filename=os.path.basename(excel_file)
                )
                msg.attach(excel_attach)

        # 4. 添加 PDF 附件
        if pdf_file and os.path.exists(pdf_file):
            with open(pdf_file, "rb") as f:
                pdf_attach = MIMEApplication(f.read())
                pdf_attach.add_header(
                    "Content-Disposition", "attachment",
                    filename=os.path.basename(pdf_file)
                )
                msg.attach(pdf_attach)

        # 5. 发送邮件
        all_recipients = RECIPIENTS["to"] + RECIPIENTS["cc"]

        if SMTP_CONFIG["use_ssl"]:
            server = smtplib.SMTP_SSL(SMTP_CONFIG["host"], SMTP_CONFIG["port"])
        else:
            server = smtplib.SMTP(SMTP_CONFIG["host"], SMTP_CONFIG["port"])
            server.starttls()

        server.login(SMTP_CONFIG["user"], SMTP_CONFIG["password"])
        server.sendmail(SMTP_CONFIG["user"], all_recipients, msg.as_string())
        server.quit()

        print(f"[OK] 邮件已发送给 {len(all_recipients)} 人")
        return True

    except Exception as e:
        print(f"[ERROR] 邮件发送失败: {e}")
        return False


# 常见邮箱 SMTP 配置参考
# QQ邮箱:     smtp.qq.com,     465(SSL) / 587(TLS)
# 163邮箱:    smtp.163.com,    465(SSL)
# 企业微信:   smtp.exmail.qq.com, 465(SSL)
# Outlook:    smtp.office365.com, 587(TLS)
# Gmail:      smtp.gmail.com,  465(SSL) / 587(TLS)

# 编辑定时任务
crontab -e

# 每天早上 6:00 执行日报脚本
0 6 * * * /usr/bin/python3 /opt/reports/daily_report.py >> /var/log/report.log 2>&1

# 每周一早上 7:00 执行周报脚本
0 7 * * 1 /usr/bin/python3 /opt/reports/weekly_report.py >> /var/log/report.log 2>&1

# 每月1号早上 8:00 执行月报脚本
0 8 1 * * /usr/bin/python3 /opt/reports/monthly_report.py >> /var/log/report.log 2>&1

# 工作日（周一到周五）早上 6:30 执行
30 6 * * 1-5 /usr/bin/python3 /opt/reports/daily_report.py

# 方法1：通过命令行创建计划任务
schtasks /create /tn "DailyOEEReport" /tr "python C:\reports\daily_report.py" /sc daily /st 06:00

# 方法2：通过 GUI 操作
# 1. Win+R → taskschd.msc → 打开任务计划程序
# 2. 创建基本任务 → 名称: "每日OEE报表"
# 3. 触发器: 每天 → 时间: 06:00
# 4. 操作: 启动程序 → python.exe
# 5. 参数: C:\reports\daily_report.py

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

scheduler = BlockingScheduler()

# 每天 6:00 执行日报
@scheduler.scheduled_job("cron", hour=6, minute=0)
def daily_report_job():
    logger.info("[START] 开始生成日报...")
    try:
        # 导入并执行日报脚本
        from daily_report import run_daily_report
        run_daily_report()
        logger.info("[DONE] 日报生成并发送成功")
    except Exception as e:
        logger.error(f"[ERROR] 日报生成失败: {e}")

# 每周一 7:00 执行周报
@scheduler.scheduled_job("cron", day_of_week="mon", hour=7, minute=0)
def weekly_report_job():
    logger.info("[START] 开始生成周报...")
    from weekly_report import run_weekly_report
    run_weekly_report()

# 每10分钟检查一次数据异常
@scheduler.scheduled_job("interval", minutes=10)
def monitor_job():
    from monitor import check_anomalies
    check_anomalies()

if __name__ == "__main__":
    logger.info("报表调度器已启动，等待执行...")
    scheduler.start()

const cron = require('node-cron');
const { exec } = require('child_process');

// 每天 6:00 执行 Python 日报脚本
cron.schedule('0 6 * * *', () => {
  console.log('[START] 开始生成日报...');
  exec('python3 /opt/reports/daily_report.py', (error, stdout, stderr) => {
    if (error) {
      console.error(`[ERROR] ${error.message}`);
      return;
    }
    console.log(`[DONE] ${stdout}`);
  });
});

console.log('报表调度器已启动...');

"""
日报自动推送系统 - 完整脚本
功能：每天早上6点自动生成 OEE 日报并推送给管理层
"""

import os
import sys
import logging
from datetime import datetime, timedelta

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("/var/log/daily_report.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 输出目录
OUTPUT_DIR = "/opt/reports/output"
os.makedirs(OUTPUT_DIR, exist_ok=True)


# =====================================================
# 第一步：从数据库查询生产数据
# =====================================================
def fetch_production_data(report_date):
    """从 MES 数据库查询昨日生产数据"""
    import pymssql
    logger.info(f"正在查询 {report_date} 的生产数据...")

    try:
        conn = pymssql.connect(
            server=os.environ.get("DB_HOST", "192.168.1.100"),
            user=os.environ.get("DB_USER", "report_user"),
            password=os.environ.get("DB_PASSWORD"),
            database="MES_Production"
        )
        cursor = conn.cursor()

        sql = """
        SELECT line_name, product_model, planned_qty, actual_qty,
               defect_qty, run_hours, planned_hours
        FROM production_daily
        WHERE report_date = %s
        ORDER BY line_name
        """
        cursor.execute(sql, (report_date,))
        rows = cursor.fetchall()

        columns = ["line", "product", "plan_qty", "actual_qty",
                   "defect_qty", "run_hours", "plan_hours"]
        data = [dict(zip(columns, row)) for row in rows]

        conn.close()
        logger.info(f"查询完成，共 {len(data)} 条记录")
        return data

    except Exception as e:
        logger.error(f"数据库查询失败: {e}")
        raise


# =====================================================
# 第二步：生成 Excel 报表
# =====================================================
def create_excel_report(data, report_date):
    """生成带格式和图表的 Excel"""
    import openpyxl
    from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
    from openpyxl.formatting.rule import CellIsRule
    from openpyxl.chart import BarChart, Reference

    logger.info("正在生成 Excel 报表...")

    # 计算 OEE
    for d in data:
        d["availability"] = d["run_hours"] / d["plan_hours"] * 100
        d["performance"]  = d["actual_qty"] / d["plan_qty"] * 100
        d["quality"]      = (d["actual_qty"] - d["defect_qty"]) / d["actual_qty"] * 100
        d["oee"]          = d["availability"] * d["performance"] * d["quality"] / 10000

    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "OEE日报"

    # 样式定义（同 Section 2，此处省略重复代码）
    header_font = Font(name="微软雅黑", size=12, bold=True, color="FFFFFF")
    header_fill = PatternFill(start_color="1F4E79", fill_type="solid")
    center = Alignment(horizontal="center", vertical="center")
    border = Border(
        left=Side("thin"), right=Side("thin"),
        top=Side("thin"), bottom=Side("thin")
    )

    # 标题
    ws.merge_cells("A1:K1")
    ws["A1"].value = f"中际旭创 生产OEE日报 - {report_date}"
    ws["A1"].font = Font(name="微软雅黑", size=18, bold=True, color="1F4E79")
    ws["A1"].alignment = center

    # 表头
    headers = ["产线", "产品", "计划产量", "实际产量", "不良数",
               "运行时间", "计划时间", "可用率%", "表现%", "质量%", "OEE%"]
    for col, h in enumerate(headers, 1):
        c = ws.cell(row=3, column=col, value=h)
        c.font, c.fill, c.alignment, c.border = header_font, header_fill, center, border

    # 数据行
    for i, d in enumerate(data):
        row = i + 4
        vals = [d["line"], d["product"], d["plan_qty"], d["actual_qty"],
                d["defect_qty"], d["run_hours"], d["plan_hours"],
                round(d["availability"],1), round(d["performance"],1),
                round(d["quality"],1), round(d["oee"],1)]
        for col, val in enumerate(vals, 1):
            c = ws.cell(row=row, column=col, value=val)
            c.font = Font(name="微软雅黑", size=11)
            c.alignment, c.border = center, border

    # 条件格式：OEE < 80% 标红
    end_row = 3 + len(data)
    red_fill = PatternFill(start_color="FFC7CE", fill_type="solid")
    red_font = Font(color="9C0006", bold=True)
    ws.conditional_formatting.add(
        f"K4:K{end_row}",
        CellIsRule(operator="lessThan", formula=["80"], fill=red_fill, font=red_font)
    )

    # 柱状图
    chart = BarChart()
    chart.title = f"各产线 OEE 对比 - {report_date}"
    chart.y_axis.title = "OEE%"
    data_ref = Reference(ws, min_col=11, min_row=3, max_row=end_row)
    cats_ref = Reference(ws, min_col=1, min_row=4, max_row=end_row)
    chart.add_data(data_ref, titles_from_data=True)
    chart.set_categories(cats_ref)
    chart.width, chart.height = 20, 12
    ws.add_chart(chart, f"A{end_row + 2}")

    # 保存
    filepath = os.path.join(OUTPUT_DIR, f"OEE日报_{report_date}.xlsx")
    wb.save(filepath)
    logger.info(f"Excel 已生成: {filepath}")
    return filepath


# =====================================================
# 第三步：生成 PDF 报告
# =====================================================
def create_pdf_report(data, report_date):
    """生成 PDF 报告"""
    # （代码同 Section 3，此处简化）
    from report_generator import generate_pdf_report # 导入 Section 3 的函数

    filepath = generate_pdf_report(report_date)
    new_path = os.path.join(OUTPUT_DIR, os.path.basename(filepath))
    os.rename(filepath, new_path)
    logger.info(f"PDF 已生成: {new_path}")
    return new_path


# =====================================================
# 第四步：发送邮件
# =====================================================
def send_email(report_date, excel_file, pdf_file, data):
    """发送报表邮件"""
    import smtplib
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    from email.mime.application import MIMEApplication

    logger.info("正在发送邮件...")

    # 汇总数据
    total_plan = sum(d["plan_qty"] for d in data)
    total_actual = sum(d["actual_qty"] for d in data)
    avg_oee = sum(d["oee"] for d in data) / len(data)
    alert_lines = [d["line"] for d in data if d["oee"] < 80]

    summary = {
        "total_plan": total_plan,
        "total_actual": total_actual,
        "avg_oee": avg_oee,
        "alert_lines": ", ".join(alert_lines) if alert_lines else "无",
    }

    # 构建邮件（复用 Section 4 的代码）
    msg = MIMEMultipart()
    msg["From"] = os.environ["SMTP_USER"]
    msg["To"] = os.environ.get("REPORT_TO", "manager@company.com")
    msg["Subject"] = f"[自动报表] 生产OEE日报 - {report_date}"

    # HTML 正文
    html = f"""
    <h2>中际旭创 生产OEE日报 - {report_date}</h2>
    <table border="1" cellpadding="8" style="border-collapse:collapse;">
      <tr><td>总计划产量</td><td>{total_plan:,}</td></tr>
      <tr><td>总实际产量</td><td>{total_actual:,}</td></tr>
      <tr><td>平均 OEE</td><td style="color:{'red' if avg_oee<80 else 'green'};font-weight:bold">
        {avg_oee:.1f}%</td></tr>
      <tr><td>异常产线</td><td style="color:red">{summary['alert_lines']}</td></tr>
    </table>
    <p style="color:#CCC;font-size:12px;">本邮件由报表系统自动发送</p>
    """
    msg.attach(MIMEText(html, "html", "utf-8"))

    # 添加附件
    for filepath in [excel_file, pdf_file]:
        if filepath and os.path.exists(filepath):
            with open(filepath, "rb") as f:
                attach = MIMEApplication(f.read())
                attach.add_header("Content-Disposition", "attachment",
                                   filename=os.path.basename(filepath))
                msg.attach(attach)

    # 发送
    with smtplib.SMTP_SSL(os.environ["SMTP_HOST"], 465) as server:
        server.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
        server.sendmail(msg["From"], msg["To"].split(","), msg.as_string())

    logger.info("邮件发送成功！")


# =====================================================
# 主流程：串联所有步骤
# =====================================================
def run_daily_report():
    """日报主流程"""
    report_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    logger.info(f"========== 开始生成 {report_date} 日报 ==========")

    try:
        # Step 1: 查询数据
        data = fetch_production_data(report_date)
        if not data:
            logger.warning("无数据，跳过今日报表")
            return

        # Step 2: 生成 Excel
        excel_file = create_excel_report(data, report_date)

        # Step 3: 生成 PDF
        pdf_file = create_pdf_report(data, report_date)

        # Step 4: 发送邮件
        send_email(report_date, excel_file, pdf_file, data)

        logger.info(f"========== {report_date} 日报完成 ==========")

    except Exception as e:
        logger.error(f"日报生成失败: {e}", exc_info=True)
        # 发送失败通知
        try:
            send_error_notification(report_date, str(e))
        except:
            pass


def send_error_notification(report_date, error_msg):
    """发送失败通知（简化版）"""
    import smtplib
    from email.mime.text import MIMEText
    msg = MIMEText(f"日报生成失败\n日期: {report_date}\n错误: {error_msg}")
    msg["Subject"] = f"[ERROR] 日报生成失败 - {report_date}"
    with smtplib.SMTP_SSL(os.environ["SMTP_HOST"], 465) as s:
        s.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"])
        s.sendmail(os.environ["SMTP_USER"], "admin@company.com", msg.as_string())


# 入口
if __name__ == "__main__":
    run_daily_report()

# 中际旭创 日报自动推送
# 每天早上 6:00 执行
0 6 * * * report_user /usr/bin/python3 /opt/reports/daily_report.py >> /var/log/daily_report.log 2>&1

# 环境变量在 /opt/reports/.env 中配置
# DB_HOST=192.168.1.100
# DB_USER=report_user
# DB_PASSWORD=xxx
# SMTP_HOST=smtp.company.com
# SMTP_USER=report@company.com
# SMTP_PASSWORD=xxx
# REPORT_TO=manager@company.com

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
from enum import Enum


class ReportType(Enum):
    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"


class ReportFormat(Enum):
    EXCEL = "excel"
    PDF = "pdf"
    BOTH = "both"


@dataclass
class ReportConfig:
    """报表参数配置"""
    report_type: ReportType = ReportType.DAILY
    report_format: ReportFormat = ReportFormat.BOTH
    start_date: Optional[str] = None       # 自定义开始日期
    end_date: Optional[str] = None         # 自定义结束日期
    lines: Optional[List[str]] = None     # 指定产线，None=全部
    shift: Optional[str] = None           # 班次: day/night/None=全部
    recipients: List[str] = None       # 邮件收件人
    send_email: bool = True


class ReportEngine:
    """动态报表引擎"""

    def __init__(self, config: ReportConfig):
        self.config = config
        self._set_date_defaults()

    def _set_date_defaults(self):
        """根据报表类型自动设置日期范围"""
        if self.config.start_date is None:
            today = datetime.now()
            if self.config.report_type == ReportType.DAILY:
                self.config.start_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
                self.config.end_date = self.config.start_date
            elif self.config.report_type == ReportType.WEEKLY:
                self.config.start_date = (today - timedelta(days=7)).strftime("%Y-%m-%d")
                self.config.end_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
            elif self.config.report_type == ReportType.MONTHLY:
                self.config.start_date = (today.replace(day=1) - timedelta(days=1)).replace(day=1).strftime("%Y-%m-%d")
                self.config.end_date = (today.replace(day=1) - timedelta(days=1)).strftime("%Y-%m-%d")

    def fetch_data(self):
        """根据参数查询数据"""
        import pymssql
        conn = pymssql.connect(server="192.168.1.100", ...)

        # 动态构建 SQL
        conditions = ["report_date BETWEEN %s AND %s"]
        params = [self.config.start_date, self.config.end_date]

        if self.config.lines:
            placeholders = ",".join(["%s"] * len(self.config.lines))
            conditions.append(f"line_name IN ({placeholders})")
            params.extend(self.config.lines)

        if self.config.shift:
            conditions.append("shift = %s")
            params.append(self.config.shift)

        where = " AND ".join(conditions)
        sql = f"SELECT * FROM production_daily WHERE {where} ORDER BY line_name"
        # ... 执行查询

    def generate(self):
        """生成报表"""
        data = self.fetch_data()
        files = []

        if self.config.report_format in (ReportFormat.EXCEL, ReportFormat.BOTH):
            files.append(self._gen_excel(data))

        if self.config.report_format in (ReportFormat.PDF, ReportFormat.BOTH):
            files.append(self._gen_pdf(data))

        if self.config.send_email:
            self._send_email(files, data)

        return files


# ========== 使用示例 ==========

# 1. 标准日报：全部产线，昨天数据
config1 = ReportConfig(report_type=ReportType.DAILY)
engine1 = ReportEngine(config1)
engine1.generate()

# 2. 自定义：只看 Line-01 和 Line-03，最近7天
config2 = ReportConfig(
    start_date="2026-06-03",
    end_date="2026-06-09",
    lines=["Line-01", "Line-03"],
    report_format=ReportFormat.EXCEL,
)
engine2 = ReportEngine(config2)
engine2.generate()

# 3. 夜班报表
config3 = ReportConfig(
    report_type=ReportType.DAILY,
    shift="night",
    recipients=["night_manager@company.com"],
)
engine3 = ReportEngine(config3)
engine3.generate()

<!-- Jinja2 邮件模板 -->
<html>
<body style="font-family: 'Microsoft YaHei', sans-serif;">
  <h2 style="color: #7EC8F0;">
    {{ company_name }} 生产{{ report_type_name }} - {{ report_date }}
  </h2>

  <table border="1" cellpadding="8" style="border-collapse: collapse; width: 100%;">
    <tr style="background: #7EC8F0; color: white;">
      <th>产线</th>
      <th>产品</th>
      <th>OEE</th>
      <th>状态</th>
    </tr>
    {% for line in lines %}
    <tr>
      <td>{{ line.name }}</td>
      <td>{{ line.product }}</td>
      <td style="color: {{ 'red' if line.oee < 80 else 'green' }}">
        {{ "%.1f"|format(line.oee) }}%
      </td>
      <td>
        {% if line.oee >= 85 %} 正常
        {% elif line.oee >= 80 %} 关注
        {% else %} 异常 {% endif %}
      </td>
    </tr>
    {% endfor %}
  </table>

  <p style="color:#CCColor: #999; font-size: 12px;">
    本邮件由报表自动推送系统发送 | 生成时间: {{ generated_at }}
  </p>
</body>
</html>

import os
from datetime import datetime

def organize_report_files(base_dir="/opt/reports/output"):
    """按年/月组织报表文件"""
    now = datetime.now()
    year_month = now.strftime("%Y-%m")
    target_dir = os.path.join(base_dir, year_month)
    os.makedirs(target_dir, exist_ok=True)
    return target_dir

def cleanup_old_reports(base_dir, keep_months=6):
    """清理超过 keep_months 个月的历史报表"""
    import shutil
    cutoff = datetime.now()
    for dirname in os.listdir(base_dir):
        dirpath = os.path.join(base_dir, dirname)
        if not os.path.isdir(dirpath):
            continue
        try:
            dir_date = datetime.strptime(dirname, "%Y-%m")
            months_diff = (cutoff.year - dir_date.year) * 12 + cutoff.month - dir_date.month
            if months_diff > keep_months:
                shutil.rmtree(dirpath)
                print(f"已清理: {dirname}")
        except ValueError:
            continue