Python办公自动化之Excel/Word/PDF实战完整教学

# 进入项目根目录（含requirements.txt的位置）
cd C:\path\to\your\project
# 创建名为venv的虚拟环境
python -m venv venv
# 激活虚拟环境（PowerShell）
venv\Scripts\Activate.ps1
# 若提示执行策略被禁止，运行以下命令后再激活
Set-ExecutionPolicy RemoteSigned -Scope CurrentUser

pip install --upgrade pip
pip install -r requirements.txt

# 在Python交互式环境中执行
import openpyxl
import docx
from pypdf import PdfReader

# 测试Excel读取
wb = openpyxl.load_workbook("samples/excel/sample_data.xlsx")
print(f"Excel加载成功，工作表数：{len(wb.sheetnames)}")

# 测试Word读取
from docx import Document
doc = Document("samples/word/sample_contract.docx")
print(f"Word加载成功，段落数：{len(doc.paragraphs)}")

# 测试PDF读取
reader = PdfReader("samples/pdf/sample_invoice.pdf")
print(f"PDF加载成功，页数：{len(reader.pages)}")

pip install pywin32
python Scripts/pywin32_postinstall.py -install

from docx import Document
import os

doc = Document()
# ... 添加内容
output_path = "output/report.docx"
doc.save(output_path)
os.chmod(output_path, 0o644)  # 设置为可读可写

from utils.pdf_scanner_handler import extract_text_from_scanned_pdf
text = extract_text_from_scanned_pdf("samples/pdf/scanned_receipt.pdf", ocr_lang="chi_sim")

# 读取公式本身（用于审计）
wb = openpyxl.load_workbook("formulas.xlsx", data_only=False)

# 读取公式计算结果（常用）
wb = openpyxl.load_workbook("formulas.xlsx", data_only=True)

from docx.shared import Inches
# 插入时指定尺寸，自动触发压缩
doc.add_picture("chart.png", width=Inches(6.0))
# 或手动压缩（高级用法）
from docx.oxml import parse_xml
from docx.oxml.ns import nsdecls
# （详见utils/docx_compressor.py）

# excel_uuid_injector.py 核心逻辑
import uuid
import base64
import time

def generate_short_uuid():
    """生成12位Base32短UUID"""
    uid = uuid.uuid4().bytes[:8]  # 取前8字节保证唯一性
    return base64.b32encode(uid).decode('ascii').replace('=', '')[:12]

def generate_timestamp_uuid(prefix="CUST"):
    """生成时间戳+随机码格式"""
    timestamp = time.strftime("%Y%m%d")
    random_part = str(uuid.uuid4().hex[:4]).upper()
    return f"{prefix}-{timestamp}-{random_part}"

# 应用到Excel指定列
from openpyxl import load_workbook

wb = load_workbook("input/sales_data.xlsx")
ws = wb.active

# 假设客户编号在B列，从第2行开始（跳过标题）
for row in range(2, ws.max_row + 1):
    if not ws[f"B{row}"].value:  # 仅填充空单元格
        ws[f"B{row}"].value = generate_short_uuid()

wb.save("output/sales_data_with_uuid.xlsx")

# excel_splitter.py
from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Border, Side

def split_by_column(input_path, output_dir, split_column="地区", header_row=1):
    wb = load_workbook(input_path)
    ws = wb.active
    
    # 获取所有唯一地区值
    regions = set()
    for row in range(header_row + 1, ws.max_row + 1):
        region = ws[f"{split_column}{row}"].value
        if region:
            regions.add(str(region))
    
    # 为每个地区创建新工作簿
    for region in regions:
        new_wb = load_workbook(input_path)  # 复制原始模板（含公式/样式）
        new_ws = new_wb.active
        
        # 删除非目标地区行（从底向上删，避免索引错乱）
        rows_to_delete = []
        for row in range(ws.max_row, header_row, -1):  # 从最后一行倒序
            if str(ws[f"{split_column}{row}"].value) != region:
                rows_to_delete.append(row)
        
        for row in rows_to_delete:
            new_ws.delete_rows(row)
        
        # 保存为独立文件
        output_path = f"{output_dir}/{region}_销售报表.xlsx"
        new_wb.save(output_path)
        print(f"已生成：{output_path}")

# 调用示例
split_by_column(
    input_path="samples/excel/quarterly_report.xlsx",
    output_dir="output/region_split",
    split_column="地区",  # 支持中文列名
    header_row=1
)

# budget_sync.py
from openpyxl import load_workbook

def sync_budget(master_path, sub_paths, master_total_cell="B2", sub_ratio_cell="C1"):
    """
    主表B2为总预算，子表C1为分配比例（如0.15表示15%）
    """
    # 读取主表总预算
    master_wb = load_workbook(master_path, data_only=True)
    master_ws = master_wb.active
    total_budget = master_ws[master_total_cell].value
    
    # 遍历每个子公司文件
    for sub_path in sub_paths:
        sub_wb = load_workbook(sub_path)
        sub_ws = sub_wb.active
        
        # 读取子表比例
        ratio = sub_ws[sub_ratio_cell].value
        if not isinstance(ratio, (int, float)) or ratio <= 0:
            print(f"警告：{sub_path}中{sub_ratio_cell}比例无效，跳过")
            continue
        
        # 计算子公司应得预算 = 总预算 × 比例
        allocated = total_budget * ratio
        
        # 写入子表指定单元格（如D2为“核定预算”）
        sub_ws["D2"] = allocated
        sub_wb.save(sub_path)
        print(f"已同步：{sub_path} → 核定预算={allocated:.2f}")

# 调用示例
sync_budget(
    master_path="samples/excel/master_budget.xlsx",
    sub_paths=[
        "samples/excel/sub_budget_beijing.xlsx",
        "samples/excel/sub_budget_shanghai.xlsx"
    ],
    master_total_cell="B2",
    sub_ratio_cell="C1"
)

<!-- contract_template.docx 中的XML片段（实际不可见，由docxtpl渲染） -->
<p>甲方：{{ party_a }}</p>
<p>乙方：{{ party_b }}</p>
<table>
  {% for item in service_items %}
  <tr>
    <td>{{ item.name }}</td>
    <td>{{ item.price | currency }}</td>
  </tr>
  {% endfor %}
</table>

# word_template_filler.py
from docxtpl import DocxTemplate
import json

def fill_contract(template_path, context_json, output_path):
    """
    context_json: {"party_a": "北京科技有限公司", "party_b": "上海咨询公司", 
                   "service_items": [{"name": "系统开发", "price": 150000}]}
    """
    template = DocxTemplate(template_path)
    with open(context_json, 'r', encoding='utf-8') as f:
        context = json.load(f)
    
    # 自定义过滤器：金额转为中文大写
    def to_currency(amount):
        # 实现人民币大写转换（简化版）
        digits = ["零", "壹", "贰", "叁", "肆", "伍", "陆", "柒", "捌", "玖"]
        units = ["", "拾", "佰", "仟", "万", "拾", "佰", "仟", "亿"]
        # （完整实现见utils/currency_converter.py）
        return f"¥{amount:,.2f}"
    
    template.register_filter('currency', to_currency)
    template.render(context)
    template.save(output_path)

# 调用示例
fill_contract(
    template_path="templates/contract_template.docx",
    context_json="data/contract_context.json",
    output_path="output/contract_signed.docx"
)

# word_formula_converter.py
from docx2python import docx2python
import re

def extract_formulas(docx_path):
    """提取Word中所有公式（OMML格式）"""
    docx_content = docx2python(docx_path)
    formulas = []
    
    # 遍历所有段落和表格单元格
    for paragraph in docx_content.body:
        if hasattr(paragraph, 'math') and paragraph.math:
            formulas.append(paragraph.math)
    
    for table in docx_content.tables:
        for row in table:
            for cell in row:
                if hasattr(cell, 'math') and cell.math:
                    formulas.append(cell.math)
    
    return formulas

def omml_to_latex(omml_str):
    """简化版OMML转LaTeX（真实项目用omml2mathml + mathml2latex）"""
    # 示例：将 <m:oMath><m:r><m:t>E=mc^2</m:t></m:r></m:oMath> → "E=mc^2"
    clean = re.sub(r'<[^>]+>', '', omml_str)  # 去除XML标签
    # 实际需解析OMML节点，此处仅示意
    return clean.replace('^', '**').replace('_', '\\_')  # 基础符号映射

# 调用流程
formulas = extract_formulas("samples/word/tech_proposal.docx")
for i, omml in enumerate(formulas):
    latex = omml_to_latex(omml)
    print(f"公式{i+1}: {latex}")

template = DocxTemplate(template_path)
try:
    template.render(context, autoescape=True)  # autoescape防止XML注入
except Exception as e:
    print(f"模板渲染失败：{e}")
    # 检查模板中所有{{ }}和{% %}是否配对

from docx.shared import Inches

table = doc.add_table(rows=1, cols=3)
# 设置列宽：第一列2英寸，第二列1.5英寸，第三列1英寸
table.columns[0].width = Inches(2.0)
table.columns[1].width = Inches(1.5)
table.columns[2].width = Inches(1.0)

try:
    import xlwings as xw
    app = xw.App(visible=False)
    # ... 执行操作
except Exception as e:
    print("Excel COM调用失败，请检查Office是否安装及宏设置")
    # 降级为openpyxl处理

# pdf_merger.py
from pypdf import PdfReader, PdfWriter
from pypdf.generic import OutlineItem

def merge_with_bookmarks(pdf_paths, output_path, bookmarks=None):
    """
    pdf_paths: ["report_ch1.pdf", "report_ch2.pdf"]
    bookmarks: [{"title": "第一章", "page": 0}, {"title": "第二章", "page": 12}]
    """
    writer = PdfWriter()
    
    # 记录每个文件的起始页码
    page_offset = 0
    outline_items = []
    
    for i, path in enumerate(pdf_paths):
        reader = PdfReader(path)
        
        # 添加所有页面
        for page in reader.pages:
            writer.add_page(page)
        
        # 为当前文件添加书签
        if bookmarks and i < len(bookmarks):
            title = bookmarks[i]["title"]
            # 书签指向当前文件第0页（在合并后文档中的绝对页码）
            outline_item = OutlineItem(title, page_offset)
            outline_items.append(outline_item)
        
        # 更新偏移量
        page_offset += len(reader.pages)
    
    # 写入书签
    writer.add_outline_item_dict({item.title: item.page for item in outline_items})
    
    # 保存
    with open(output_path, "wb") as f:
        writer.write(f)
    print(f"已合并并添加书签：{output_path}")

# 调用示例
merge_with_bookmarks(
    pdf_paths=["samples/pdf/ch1_intro.pdf", "samples/pdf/ch2_analysis.pdf"],
    output_path="output/annual_report.pdf",
    bookmarks=[{"title": "1. 引言", "page": 0}, {"title": "2. 分析", "page": 8}]
)

sub_item = OutlineItem("1.1 背景", 1, parent=outline_items[0])

# pdf_watermarker.py
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
from pypdf import PdfReader, PdfWriter

def create_watermark(text, width=612, height=792):
    """生成单页水印PDF（A4尺寸）"""
    watermark_path = "temp/watermark.pdf"
    c = canvas.Canvas(watermark_path, pagesize=(width, height))
    
    # 设置字体和透明度
    c.setFont("Helvetica", 40)
    c.setFillColorRGB(0.8, 0.8, 0.8, alpha=0.3)  # RGB+alpha
    
    # 斜向绘制文字
    c.saveState()
    c.translate(width / 2, height / 2)
    c.rotate(30)
    c.drawCentredString(0, 0, text)
    c.restoreState()
    
    c.save()
    return watermark_path

def add_watermark(input_pdf, output_pdf, watermark_text="CONFIDENTIAL"):
    watermark_path = create_watermark(watermark_text)
    watermark_reader = PdfReader(watermark_path)
    
    input_reader = PdfReader(input_pdf)
    writer = PdfWriter()
    
    for page in input_reader.pages:
        # 将水印叠加到当前页
        page.merge_page(watermark_reader.pages[0])
        writer.add_page(page)
    
    with open(output_pdf, "wb") as f:
        writer.write(f)
    print(f"已添加水印：{output_pdf}")

# 调用
add_watermark("samples/pdf/financial_report.pdf", "output/secure_report.pdf")

# pdf_ocr_enhancer.py
from pdf2image import convert_from_path
import pytesseract
from pypdf import PdfReader, PdfWriter

def ocr_pdf(input_path, output_path, lang="chi_sim+eng"):
    """
    lang: "chi_sim"中文简体, "eng"英文, "chi_sim+eng"中英混合
    """
    # 将PDF转为图像（每页一张PNG）
    images = convert_from_path(input_path, dpi=300)  # 高DPI提升OCR精度
    
    # 创建新PDF写入器
    writer = PdfWriter()
    
    for i, image in enumerate(images):
        # OCR识别文字
        text = pytesseract.image_to_string(image, lang=lang, config="--psm 6")
        
        # 创建空白PDF页（同原PDF尺寸）
        from reportlab.pdfgen import canvas
        from reportlab.lib.pagesizes import letter
        from io import BytesIO
        
        packet = BytesIO()
        can = canvas.Canvas(packet, pagesize=letter)
        can.drawString(10, 10, text)  # 文字层（实际需精确匹配图像位置）
        can.save()
        
        # 合并图像层与文字层（简化版，真实项目用pdfkit或pdfrw）
        # （完整实现见utils/ocr_merger.py）
        print(f"第{i+1}页OCR完成，识别字数：{len(text)}")
    
    # 保存（此为示意，实际需用pdfkit生成含文本层的PDF）
    # writer.write(output_path)

# 注意：需提前安装Tesseract
# Windows: https://github.com/UB-Mannheim/tesseract/wiki
# macOS: brew install tesseract
# Ubuntu: sudo apt install tesseract-ocr libtesseract-dev

# core/logger.py
import logging
from datetime import datetime

def setup_logger(name, level=logging.INFO):
    logger = logging.getLogger(name)
    logger.setLevel(level)
    
    # 文件处理器：按天滚动
    file_handler = logging.handlers.RotatingFileHandler(
        f"logs/{name}_{datetime.now().strftime('%Y%m%d')}.log",
        maxBytes=10*1024*1024,  # 10MB
        backupCount=5
    )
    file_handler.setLevel(logging.DEBUG)
    
    # 控制台处理器：仅INFO及以上
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    
    # 格式化
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

# 在excel_uuid_injector.py中使用
logger = setup_logger("excel_uuid")
logger.info("开始处理sales_data.xlsx")
logger.debug(f"生成UUID: {uuid_str}")
logger.error("列B为空，跳过第5行")

# utils/progress_bar.py
from tqdm import tqdm
import time

def process_with_progress(items, desc="Processing", unit="item"):
    """通用进度条包装器"""
    for item in tqdm(items, desc=desc, unit=unit, colour="green"):
        yield item
        time.sleep(0.1)  # 模拟处理耗时

# 在pdf_merger.py中应用
pdf_paths = ["ch1.pdf", "ch2.pdf", "ch3.pdf"]
for pdf_path in process_with_progress(pdf_paths, desc="Merging PDFs", unit="file"):
    reader = PdfReader(pdf_path)
    # ... 合并逻辑