Python自动化办公全攻略之Excel/Word/PDF/邮件批量处理

# 批量安装核心依赖库（指定稳定版本，避免版本兼容问题）
# pip install openpyxl==3.1.2 pandas==2.0.3 python-docx==0.8.16 pypdf2==2.12.1 pdfplumber==0.9.0

# 环境验证代码
import openpyxl
import pandas as pd
import docx
import pypdf2
import pdfplumber
import smtplib

print("环境验证通过：")
print(f"openpyxl版本：{openpyxl.__version__}")
print(f"pandas版本：{pd.__version__}")
print(f"python-docx版本：{docx.__version__}")
print(f"pypdf2版本：{pypdf2.__version__}")
print(f"pdfplumber版本：{pdfplumber.__version__}")
print(f"smtplib版本：{smtplib.__version__}")  # python内置，版本与python一致
    

from openpyxl import workbook
from openpyxl.styles import font, patternfill, alignment
from openpyxl.utils.dataframe import dataframe_to_rows

def batch_generate_excel_reports(data_list, save_dir):
    """
    批量生成带格式的excel报表
    :param data_list: 数据列表，每个元素是字典（含报表数据）
    :param save_dir: 保存目录
    """
    for idx, data in enumerate(data_list):
        # 1. 创建工作簿
        wb = workbook()
        ws = wb.active
        ws.title = "月度报表"
        
        # 2. 写入标题并设置样式
        ws["a1"] = f"{data['部门']}2024年{data['月份']}度报表"
        ws["a1"].font = font(name="微软雅黑", size=16, bold=true)
        ws["a1"].fill = patternfill(start_color="e6f3ff", end_color="e6f3ff", fill_type="solid")
        ws["a1"].alignment = alignment(horizontal="center")
        ws.merge_cells("a1:d1")  # 合并单元格
        
        # 3. 写入表头
        headers = ["序号", "项目", "金额（元）", "备注"]
        for col, header in enumerate(headers, 1):
            cell = ws.cell(row=3, column=col, value=header)
            cell.font = font(name="微软雅黑", size=12, bold=true)
            cell.fill = patternfill(start_color="d9e2f3", end_color="d9e2f3", fill_type="solid")
        
        # 4. 写入数据
        for row, item in enumerate(data["details"], 4):
            ws.cell(row=row, column=1, value=item["序号"])
            ws.cell(row=row, column=2, value=item["项目"])
            ws.cell(row=row, column=3, value=item["金额"])
            ws.cell(row=row, column=4, value=item["备注"])
        
        # 5. 调整列宽
        ws.column_dimensions["a"].width = 8
        ws.column_dimensions["b"].width = 20
        ws.column_dimensions["c"].width = 15
        ws.column_dimensions["d"].width = 30
        
        # 6. 保存文件
        save_path = f"{save_dir}/{data['部门']}_{data['月份']}月报表.xlsx"
        wb.save(save_path)
        print(f"已生成报表：{save_path}")

# 测试数据
test_data = [
    {
        "部门": "技术部",
        "月份": 10,
        "details": [
            {"序号": 1, "项目": "服务器租赁", "金额": 5000, "备注": "季度付"},
            {"序号": 2, "项目": "软件订阅", "金额": 3000, "备注": "月度付"}
        ]
    },
    {
        "部门": "市场部",
        "月份": 10,
        "details": [
            {"序号": 1, "项目": "广告投放", "金额": 20000, "备注": "线上推广"},
            {"序号": 2, "项目": "活动策划", "金额": 8000, "备注": "线下活动"}
        ]
    }
]

# 调用函数（需提前创建save_dir目录）
batch_generate_excel_reports(test_data, "./monthly_reports")
    

import pandas as pd
import os

def batch_merge_excel_files(file_dir, save_path):
    """
    批量读取指定目录下的所有excel文件，合并数据后保存
    :param file_dir: excel文件目录
    :param save_path: 合并后保存路径
    """
    # 1. 遍历目录，获取所有excel文件路径
    excel_files = [f for f in os.listdir(file_dir) if f.endswith(".xlsx") and not f.startswith("~$")]
    if not excel_files:
        print("目录下无有效excel文件")
        return
    
    # 2. 批量读取并合并
    all_data = []
    for file in excel_files:
        file_path = os.path.join(file_dir, file)
        # 读取excel的"数据"工作表（假设所有文件结构一致）
        df = pd.read_excel(file_path, sheet_name="数据")
        # 添加"来源文件"列，便于追溯数据
        df["来源文件"] = file
        all_data.append(df)
    
    # 3. 合并所有数据（忽略索引，重新生成）
    merged_df = pd.concat(all_data, ignore_index=true)
    
    # 4. 数据清洗（示例：去除空行、重复行）
    merged_df = merged_df.dropna(how="all")  # 去除全空行
    merged_df = merged_df.drop_duplicates()  # 去除重复行
    
    # 5. 保存合并后的数据
    merged_df.to_excel(save_path, index=false, sheet_name="合并数据")
    print(f"合并完成，共 {len(merged_df)} 条数据，保存至：{save_path}")

# 调用函数
batch_merge_excel_files("./excel_files", "./merged_data.xlsx")
    

from docx import document
from docx.shared import inches, pt
from docx.enum.text import wd_paragraph_alignment

def batch_generate_contracts(template_path, data_list, save_dir):
    """
    基于word模板批量生成合同
    :param template_path: 合同模板路径（.docx）
    :param data_list: 合同数据列表，每个元素是字典
    :param save_dir: 保存目录
    """
    for idx, data in enumerate(data_list):
        # 1. 加载模板
        doc = document(template_path)
        
        # 2. 替换文档中的变量（模板中用{{变量名}}标记）
        # 遍历所有段落
        for para in doc.paragraphs:
            for key, value in data.items():
                if f"{{{{{key}}}}}" in para.text:
                    # 替换变量
                    para.text = para.text.replace(f"{{{{{key}}}}}", str(value))
                    # 设置段落样式（可选）
                    para.alignment = wd_paragraph_alignment.justify  # 两端对齐
                    for run in para.runs:
                        run.font.name = "微软雅黑"
                        run.font.size = pt(11)
        
        # 3. 替换表格中的变量（如果模板中有表格）
        for table in doc.tables:
            for row in table.rows:
                for cell in row.cells:
                    for key, value in data.items():
                        if f"{{{{{key}}}}}" in cell.text:
                            cell.text = cell.text.replace(f"{{{{{key}}}}}", str(value))
                            # 设置表格文本样式
                            for para in cell.paragraphs:
                                para.alignment = wd_paragraph_alignment.center
                                for run in para.runs:
                                    run.font.name = "微软雅黑"
                                    run.font.size = pt(10)
        
        # 4. 插入附件说明（可选，在文档末尾添加）
        para = doc.add_paragraph("附件说明：")
        para.font.name = "微软雅黑"
        para.font.size = pt(11)
        para.font.bold = true
        for attach in data["attachments"]:
            doc.add_paragraph(f"1. {attach}", style="list bullet")
        
        # 5. 保存文件
        save_path = f"{save_dir}/合同_{data['甲方名称']}_{data['合同编号']}.docx"
        doc.save(save_path)
        print(f"已生成合同：{save_path}")

# 测试数据
test_contract_data = [
    {
        "合同编号": "ht202410001",
        "甲方名称": "xx科技有限公司",
        "乙方名称": "yy软件有限公司",
        "项目名称": "企业oa系统开发",
        "合同金额": "500000元",
        "签订日期": "2024年10月20日",
        "有效期": "2年",
        "attachments": ["技术需求说明书", "报价单"]
    },
    {
        "合同编号": "ht202410002",
        "甲方名称": "zz贸易有限公司",
        "乙方名称": "yy软件有限公司",
        "项目名称": "客户管理系统升级",
        "合同金额": "200000元",
        "签订日期": "2024年10月21日",
        "有效期": "1年",
        "attachments": ["升级需求清单", "维护协议"]
    }
]

# 调用函数（需提前准备模板文件，模板中用{{合同编号}}等变量标记）
batch_generate_contracts("./contract_template.docx", test_contract_data, "./contracts")

from pypdf2 import pdfmerger
import os

def batch_merge_pdfs(pdf_dir, save_path):
    """
    批量合并指定目录下的所有pdf文件
    :param pdf_dir: pdf文件目录
    :param save_path: 合并后保存路径
    """
    # 1. 初始化合并器
    merger = pdfmerger()
    
    # 2. 遍历目录，添加所有pdf文件
    pdf_files = [f for f in os.listdir(pdf_dir) if f.endswith(".pdf")]
    if not pdf_files:
        print("目录下无有效pdf文件")
        return
    
    # 按文件名排序（确保合并顺序正确）
    pdf_files.sort()
    for file in pdf_files:
        file_path = os.path.join(pdf_dir, file)
        try:
            merger.append(file_path)  # 添加pdf文件
            print(f"已添加：{file}")
        except exception as e:
            print(f"添加{file}失败：{e}")
    
    # 3. 合并并保存
    merger.write(save_path)
    merger.close()
    print(f"合并完成，保存至：{save_path}")

# 调用函数
batch_merge_pdfs("./pdf_files", "./merged_pdf.pdf")
    

import pdfplumber
import pandas as pd
import os

def batch_extract_pdf_tables(pdf_dir, save_dir):
    """
    批量提取多个pdf中的表格数据，保存为excel
    :param pdf_dir: pdf文件目录
    :param save_dir: 保存目录
    """
    # 确保保存目录存在
    os.makedirs(save_dir, exist_ok=true)
    
    pdf_files = [f for f in os.listdir(pdf_dir) if f.endswith(".pdf")]
    if not pdf_files:
        print("目录下无有效pdf文件")
        return
    
    for file in pdf_files:
        file_path = os.path.join(pdf_dir, file)
        save_path = f"{save_dir}/{os.path.splitext(file)[0]}_表格数据.xlsx"
        
        # 打开pdf文件
        with pdfplumber.open(file_path) as pdf:
            # 初始化excel写入器
            with pd.excelwriter(save_path, engine="openpyxl") as writer:
                table_count = 0
                # 遍历所有页面
                for page_idx, page in enumerate(pdf.pages, 1):
                    # 提取当前页面的所有表格
                    tables = page.extract_tables()
                    if not tables:
                        continue
                    # 遍历当前页面的表格
                    for table_idx, table in enumerate(tables, 1):
                        table_count += 1
                        # 转换为dataframe
                        df = pd.dataframe(table[1:], columns=table[0])  # table[0]是表头
                        # 保存到excel的不同工作表
                        sheet_name = f"第{page_idx}页_表格{table_idx}"
                        df.to_excel(writer, sheet_name=sheet_name, index=false)
        
        if table_count > 0:
            print(f"已提取{file}中的{table_count}个表格，保存至：{save_path}")
        else:
            print(f"{file}中未发现表格")

# 调用函数
batch_extract_pdf_tables("./pdf_files", "./pdf_tables")
    

import smtplib
from email.mime.text import mimetext
from email.mime.multipart import mimemultipart
from email.header import header
import os

def batch_send_emails(smtp_info, email_data_list):
    """
    批量发送带附件的邮件
    :param smtp_info: smtp服务器信息，字典格式
    :param email_data_list: 邮件数据列表，每个元素是字典（含收件人、主题、正文、附件）
    """
    try:
        # 1. 连接smtp服务器
        smtp = smtplib.smtp_ssl(smtp_info["smtp_server"], smtp_info["smtp_port"])
        # 2. 登录验证（注意：qq邮箱用授权码，企业邮箱用密码）
        smtp.login(smtp_info["sender"], smtp_info["password"])
        print("smtp服务器连接成功")
        
        # 3. 批量发送邮件
        for email_data in email_data_list:
            # 构造邮件对象（带附件）
            msg = mimemultipart()
            msg["from"] = header(smtp_info["sender_name"], "utf-8")  # 发件人名称
            msg["to"] = header(";".join(email_data["receivers"]), "utf-8")  # 收件人列表
            msg["subject"] = header(email_data["subject"], "utf-8")  # 邮件主题
            
            # 添加正文
            msg.attach(mimetext(email_data["content"], "html", "utf-8"))  # 支持html格式正文
            
            # 添加附件
            for attach_path in email_data["attachments"]:
                if not os.path.exists(attach_path):
                    print(f"附件{attach_path}不存在，跳过")
                    continue
                # 读取附件文件
                with open(attach_path, "rb") as f:
                    attach = mimetext(f.read(), "base64", "utf-8")
                    attach["content-type"] = "application/octet-stream"
                    # 设置附件名称（避免中文乱码）
                    attach["content-disposition"] = f'attachment; filename="{header(os.path.basename(attach_path), "utf-8").encode()}"'
                    msg.attach(attach)
            
            # 发送邮件
            smtp.sendmail(smtp_info["sender"], email_data["receivers"], msg.as_string())
            print(f"已发送邮件至：{email_data['receivers']}，主题：{email_data['subject']}")
        
        # 4. 关闭连接
        smtp.quit()
        print("所有邮件发送完成")
    
    except exception as e:
        print(f"邮件发送失败：{e}")

# 配置smtp信息（以企业邮箱为例，qq邮箱需替换为smtp.qq.com，端口465，密码用授权码）
smtp_config = {
    "smtp_server": "smtp.exmail.qq.com",
    "smtp_port": 465,
    "sender": "admin@company.com",
    "sender_name": "行政部",
    "password": "your_password"  # 企业邮箱密码，qq邮箱用授权码
}

# 邮件数据（批量发送的邮件列表）
email_data = [
    {
        "receivers": ["tech@company.com", "manager1@company.com"],
        "subject": "技术部10月报表",
        "content": "<p>各位领导、同事：</p><p>附件为技术部2024年10月报表，请查收。</p>",
        "attachments": ["./monthly_reports/技术部_10月报表.xlsx"]
    },
    {
        "receivers": ["market@company.com", "manager2@company.com"],
        "subject": "市场部10月报表",
        "content": "<p>各位领导、同事：</p><p>附件为市场部2024年10月报表，请查收。</p>",
        "attachments": ["./monthly_reports/市场部_10月报表.xlsx"]
    }
]

# 调用函数
batch_send_emails(smtp_config, email_data)
    

import pandas as pd
import openpyxl
from docx import document
import pypdf2
import smtplib
from email.mime.text import mimetext
from email.mime.multipart import mimemultipart
from email.header import header
import os
import win32com.client  # 需安装pywin32：pip install pywin32
from datetime import datetime

def excel_to_pdf(excel_path, pdf_path):
    """
    将excel文件转换为pdf（调用excel进程，保证格式一致）
    :param excel_path: excel文件路径
    :param pdf_path: pdf保存路径
    """
    excel = win32com.client.dispatchex("excel.application")
    excel.visible = false
    excel.displayalerts = false  # 禁用警告
    try:
        workbook = excel.workbooks.open(excel_path)
        # 转换为pdf（quality=17表示高质量）
        workbook.exportasfixedformat(0, pdf_path, quality=17)
        print(f"已将{excel_path}转换为pdf：{pdf_path}")
    except exception as e:
        print(f"excel转pdf失败：{e}")
    finally:
        workbook.close(savechanges=false)
        excel.quit()

def monthly_report_automation(config):
    """
    月度报表全链路自动化
    :param config: 配置字典，含数据目录、模板路径、smtp信息等
    """
    # 1. 初始化目录（按年份/月份/部门）
    now = datetime.now()
    year = now.year
    month = now.month
    base_dir = f"{config['base_dir']}/{year}/{month}"
    excel_dir = f"{base_dir}/excel"
    pdf_dir = f"{base_dir}/pdf"
    archive_dir = f"{base_dir}/归档"
    for dir_path in [excel_dir, pdf_dir, archive_dir]:
        os.makedirs(dir_path, exist_ok=true)
    
    # 2. 读取各部门提交的原始数据（标准化模板）
    raw_data_dir = config["raw_data_dir"]
    raw_files = [f for f in os.listdir(raw_data_dir) if f.endswith(".xlsx") and not f.startswith("~$")]
    if not raw_files:
        print("原始数据目录下无有效文件")
        return
    
    # 3. 批量生成标准化excel报表
    report_template = config["report_template"]
    email_data_list = []
    for file in raw_files:
        # 读取原始数据
        raw_path = os.path.join(raw_data_dir, file)
        raw_df = pd.read_excel(raw_path, sheet_name="原始数据")
        
        # 数据校验（示例：检查是否有缺失字段）
        required_fields = ["项目", "金额（元）", "备注"]
        if not all(field in raw_df.columns for field in required_fields):
            print(f"{file}缺失必要字段，跳过")
            continue
        
        # 提取部门名称（假设文件名格式：部门_原始数据.xlsx）
        dept_name = os.path.splitext(file)[0].replace("_原始数据", "")
        
        # 加载报表模板，生成标准化报表
        wb = openpyxl.load_workbook(report_template)
        ws = wb.active
        ws["a1"] = f"{dept_name}{year}年{month}月报表"
        ws["a1"].font = openpyxl.styles.font(name="微软雅黑", size=16, bold=true)
        ws.merge_cells("a1:d1")
        
        # 写入数据
        for row_idx, (_, row) in enumerate(raw_df.iterrows(), 4):
            ws.cell(row=row_idx, column=1, value=row_idx-3)  # 序号
            ws.cell(row=row_idx, column=2, value=row["项目"])
            ws.cell(row=row_idx, column=3, value=row["金额（元）"])
            ws.cell(row=row_idx, column=4, value=row["备注"])
        
        # 保存excel报表
        excel_path = f"{excel_dir}/{dept_name}_{year}年{month}月报表.xlsx"
        wb.save(excel_path)
        
        # 4. 转换为pdf
        pdf_path = f"{pdf_dir}/{dept_name}_{year}年{month}月报表.pdf"
        excel_to_pdf(excel_path, pdf_path)
        
        # 5. 整理邮件数据
        # 读取收件人配置（假设config["receivers_config"]是收件人字典）
        receivers = config["receivers_config"].get(dept_name, config["default_receivers"])
        email_data = {
            "receivers": receivers,
            "subject": f"{dept_name}{year}年{month}月报表",
            "content": f"<p>各位领导、同事：</p><p>附件为{dept_name}{year}年{month}月报表（excel+pdf），请查收。</p>",
            "attachments": [excel_path, pdf_path]
        }
        email_data_list.append(email_data)
        
        # 6. 归档文件
        archive_excel_path = f"{archive_dir}/{dept_name}_{year}年{month}月报表.xlsx"
        archive_pdf_path = f"{archive_dir}/{dept_name}_{year}年{month}月报表.pdf"
        os.copy(excel_path, archive_excel_path)
        os.copy(pdf_path, archive_pdf_path)
    
    # 7. 批量发送邮件
    smtp_info = config["smtp_info"]
    smtp = smtplib.smtp_ssl(smtp_info["smtp_server"], smtp_info["smtp_port"])
    smtp.login(smtp_info["sender"], smtp_info["password"])
    for email_data in email_data_list:
        msg = mimemultipart()
        msg["from"] = header(smtp_info["sender_name"], "utf-8")
        msg["to"] = header(";".join(email_data["receivers"]), "utf-8")
        msg["subject"] = header(email_data["subject"], "utf-8")
        msg.attach(mimetext(email_data["content"], "html", "utf-8"))
        
        # 添加附件
        for attach_path in email_data["attachments"]:
            if os.path.exists(attach_path):
                with open(attach_path, "rb") as f:
                    attach = mimetext(f.read(), "base64", "utf-8")
                    attach["content-type"] = "application/octet-stream"
                    attach["content-disposition"] = f'attachment; filename="{header(os.path.basename(attach_path), "utf-8").encode()}"'
                    msg.attach(attach)
        
        smtp.sendmail(smtp_info["sender"], email_data["receivers"], msg.as_string())
        print(f"已发送邮件至：{email_data['receivers']}")
    smtp.quit()
    print("全链路自动化完成！")

# 配置信息
config = {
    "base_dir": "./月度报表自动化",
    "raw_data_dir": "./原始数据",
    "report_template": "./报表模板.xlsx",
    "smtp_info": {
        "smtp_server": "smtp.exmail.qq.com",
        "smtp_port": 465,
        "sender": "admin@company.com",
        "sender_name": "行政部",
        "password": "your_password"
    },
    "receivers_config": {
        "技术部": ["tech_manager@company.com", "ceo@company.com"],
        "市场部": ["market_manager@company.com", "ceo@company.com"],
        "财务部": ["finance_manager@company.com", "ceo@company.com"]
    },
    "default_receivers": ["admin@company.com", "ceo@company.com"]
}

# 执行全链路自动化
monthly_report_automation(config)
    

`def get_merged_cell_value(ws, row, col):
“”“获取合并单元格的值”“”
for merged_range in ws.merged_cells.ranges:
if row in merged_range.min_row:merged_range.max_row and col in merged_range.min_col:merged_range.max_col:
# 返回合并范围的起始单元格值
return ws.cell(row=merged_range.min_row, column=merged_range.min_col).value

`import fitz # pymupdf
​​​​​​​def merge_pdfs_with_chinese(pdf_paths, save_path):
“”“合并包含中文的pdf文件”“”
doc = fitz.open()
for pdf_path in pdf_paths:
with fitz.open(pdf_path) as sub_doc:
doc.insert_pdf(sub_doc)
doc.save(save_path)
doc.close()
print(f"合并完成：{save_path}")`

`msg = mimemultipart()
msg[“from”] = header(“行政部admin@company.com”, “utf-8”) # 规范发件人名称
msg[“to”] = header(“技术部经理tech_manager@company.com”, “utf-8”)
msg[“subject”] = header(“技术部10月报表”, “utf-8”)

def safe_excel_to_pdf(excel_path, pdf_path): excel = none workbook = none try: excel = win32com.client.dispatchex("excel.application") excel.visible = false excel.displayalerts = false workbook = excel.workbooks.open(excel_path) workbook.exportasfixedformat(0, pdf_path, quality=17) except exception as e: print(f"转换失败：{e}") finally: if workbook: workbook.close(savechanges=false) if excel: excel.quit() # 强制释放com对象（可选，进一步避免残留） import pythoncom pythoncom.couninitialize()