以土豆之名,行学习之实

Word文档批量处理(占位符填充) - docxtpl


Word文档批量处理工具 (docxtpl库)- 完整技术文档

📋 项目概述

基于Python和docxtpl库的Word文档批量处理工具,通过CSV映射关系自动替换Word模板中的占位符,高效生成个性化文档。

🎯 核心功能

1. CSV映射处理

  • 智能读取: 自动解析CSV中的占位符-值映射

  • 配置集成: 支持在CSV中配置输入/输出目录、处理延迟

  • 注释支持: 忽略#开头的注释行和空行

  • 编码兼容: 默认utf-8-sig编码,完美支持中文

2. Word模板处理

  • 专业渲染: 使用docxtpl库进行模板替换

  • 自动扩展: 默认添加Today日期字段

  • 批量处理: 支持目录内所有Word文档

  • 临时文件过滤: 自动跳过~$开头的临时文件

3. 配置管理

配置项包括:
input_dir: 模板文件目录
output_dir: 输出文件目录  
delay_seconds: 文件处理间隔时间


4. 文件系统管理

  • 智能目录创建: 自动创建不存在的输出目录

  • 文件名清理: 移除Windows非法字符

  • 路径长度控制: 自动截断过长文件夹名(>100字符)

🔄 工作流程

环境准备

项目根目录/
├── 程序文件.py    # 主程序
├── 信息表.csv     # 占位符映射配置
└── 模板/          # Word模板文件
    ├── 模板1.docx
    └── 模板2.docx


CSV配置示例

# 基本信息
ProjectName,我的项目
# 配置项
input_dir,模板路径
output_dir,输出路径
delay_seconds,0.5
# 占位符
ClientName,客户名称
UserName,张三


Word模板语法

项目名称:{{ProjectName}}
客户名称:{{ClientName}}  
处理人员:{{UserName}}
日期:{{Today}}


⚙️ 核心代码实现

主处理器类

class DocxPlaceholderProcessor:
    def read_csv_mapping(self, file_path: str, encoding: str = 'utf-8-sig')
    def process_documents(self, input_dir: str, csv_file: str, 
                         output_dir: Optional[str] = None, 
                         delay_seconds: float = 0)
    def process_single_document(self, input_path: str, output_path: str, 
                               mapping: Dict[str, str])


关键方法

  • read_csv_mapping(): 读取CSV映射关系

  • extract_config_from_mapping(): 提取配置参数

  • validate_mapping(): 验证必要字段

  • extend_mapping_with_defaults(): 添加默认值

🛡️ 错误处理与验证

输入验证

  • ✅ CSV文件存在性检查

  • ✅ 模板目录存在性检查

  • ✅ ProjectName字段验证

  • ✅ Word文档存在性检查

异常处理

  • ❌ 文件不存在明确提示

  • ❌ 编码问题处理

  • ❌ 单个文档失败不影响批量作业

📊 输出结果

输出目录/
├── 模板1_processed.docx
├── 模板2_processed.docx
└── ...


🎁 特色功能

智能日期处理

自动添加Today字段,格式为"YYYY年MM月DD日"

灵活目录管理

  • 支持相对/绝对路径

  • 自动创建多级目录

  • 智能处理特殊字符

可调节处理速度

通过delay_seconds控制处理间隔,适合大量文档场景

💡 使用场景

企业文档生成

  • 合同文档、报告文件

  • 证书制作、通知函件

教育机构应用

  • 成绩单、录取通知书

  • 毕业证书、各类证明

政府部门

  • 行政文书、公示文件

  • 审批文档、通知公告

📋 依赖环境

requirements.txt

docxtpl==0.20.1
Jinja2==3.1.6
python-docx==1.2.0
lxml==6.0.2


🚀 快速开始

  1. 安装依赖: pip install -r requirements.txt

  2. 准备CSV: 配置占位符映射关系

  3. 制作模板: 在Word中使用{{placeholder}}语法

  4. 运行程序: 执行主程序自动批量处理


该工具特别适合需要基于模板生成大量个性化文档的场景,大幅提升文档处理效率和准确性。


🔧 附:代码

import time
import csv
import re
from datetime import datetime
from docxtpl import DocxTemplate
from typing import Dict, Optional
from pathlib import Path


class DocxPlaceholderProcessor:
    """
    Word文档占位符处理器 - 使用docxtpl库进行Word文档模板处理
    主要功能:读取CSV中的占位符映射,批量替换Word模板中的占位符
    """

    def __init__(self):
        """初始化处理器,暂无特殊配置需要初始化"""
        pass

    @staticmethod
    def read_csv_mapping(file_path: str, encoding: str = 'utf-8-sig') -> Dict[str, str]:
        """
        从CSV文件中读取占位符和值的映射关系

        Args:
            file_path: CSV文件路径
            encoding: 文件编码,默认使用utf-8-sig处理BOM

        Returns:
            Dict[str, str]: 占位符到值的映射字典,包含配置项

        Raises:
            Exception: 文件不存在或读取失败时抛出异常
        """
        mapping = {}  # 存储占位符映射
        config = {}  # 存储配置参数

        try:
            # 打开CSV文件并读取内容
            with open(file_path, 'r', encoding=encoding) as csvfile:
                reader = csv.reader(csvfile)

                # 逐行处理CSV内容
                for row in reader:
                    # 跳过空行、注释行(以#开头)
                    if not row or not any(row) or row[0].startswith('#'):
                        continue

                    # 确保行至少有2列(键和值)
                    if len(row) >= 2:
                        key = row[0].strip()  # 占位符键
                        value = row[1].strip() if row[1] is not None else ''  # 对应的值

                        if key:
                            # 检查是否为配置项(输入目录、输出目录、延迟时间)
                            if key in ['input_dir', 'output_dir', 'delay_seconds']:
                                config[key] = value
                            else:
                                # 普通占位符添加到映射字典
                                mapping[key] = value

        except FileNotFoundError:
            raise Exception(f"CSV文件不存在: {file_path}")
        except Exception as e:
            raise Exception(f"读取CSV文件失败: {e}")

        # 将配置信息添加到映射中,使用特殊前缀避免与普通占位符冲突
        for key, value in config.items():
            mapping[f'__config_{key}'] = value

        return mapping

    @staticmethod
    def extract_config_from_mapping(mapping: Dict[str, str]) -> Dict[str, str]:
        """
        从映射字典中提取配置参数

        Args:
            mapping: 包含占位符和配置的完整映射字典

        Returns:
            Dict[str, str]: 提取出的配置参数字典
        """
        config = {}
        config_keys = ['input_dir', 'output_dir', 'delay_seconds']

        # 遍历配置键,从映射中提取配置值
        for key in config_keys:
            config_key = f'__config_{key}'
            if config_key in mapping:
                config[key] = mapping[config_key]
                # 从映射中移除配置项,避免影响普通占位符替换
                del mapping[config_key]

        return config

    @staticmethod
    def validate_mapping(mapping: Dict[str, str]) -> bool:
        """
        验证映射数据的完整性 - 只验证ProjectName字段是否存在

        Args:
            mapping: 占位符映射字典

        Returns:
            bool: 验证通过返回True,否则返回False
        """
        # 检查必要的ProjectName字段是否存在
        if 'ProjectName' not in mapping:
            return False
        return True

    @staticmethod
    def create_output_directory(base_dir: str, folder_name: str) -> str:
        """
        创建输出目录 - 如果目录不存在则自动创建

        Args:
            base_dir: 基础目录路径
            folder_name: 要创建的文件夹名称

        Returns:
            str: 创建的完整目录路径
        """
        # 定义Windows文件名中的非法字符
        invalid_chars = r'[<>:"/\|?*]'
        # 替换非法字符为下划线
        folder_name = re.sub(invalid_chars, '_', folder_name)

        # 限制文件夹名称长度,避免路径过长问题
        if len(folder_name) > 100:
            folder_name = folder_name[:100]

        # 构建完整输出路径
        output_dir = Path(base_dir) / folder_name
        # 确保目录存在,不存在则创建(包括父目录)
        output_dir.mkdir(parents=True, exist_ok=True)

        return str(output_dir)

    @staticmethod
    def get_current_date(format_str: str = "%Y年%m月%d日") -> str:
        """
        获取当前日期字符串

        Args:
            format_str: 日期格式字符串

        Returns:
            str: 格式化后的当前日期字符串
        """
        return datetime.now().strftime(format_str)

    def extend_mapping_with_defaults(self, mapping: Dict[str, str]) -> Dict[str, str]:
        """
        使用默认值扩展映射 - 主要添加Today日期字段

        Args:
            mapping: 原始占位符映射

        Returns:
            Dict[str, str]: 扩展后的映射字典
        """
        extended = mapping.copy()

        # 只添加Today日期字段,如果不存在则添加当前日期
        if 'Today' not in extended:
            extended['Today'] = self.get_current_date()

        return extended

    def process_single_document(self, input_path: str, output_path: str, mapping: Dict[str, str]):
        """
        处理单个Word文档 - 使用docxtpl进行模板渲染

        Args:
            input_path: 输入模板文件路径
            output_path: 输出文件路径
            mapping: 占位符映射字典

        Returns:
            bool: 处理成功返回True,失败返回False
        """
        try:
            # 使用docxtpl加载Word模板
            doc = DocxTemplate(input_path)

            # 扩展映射,添加默认值
            extended_mapping = self.extend_mapping_with_defaults(mapping)

            # 渲染模板(自动处理所有占位符替换)
            doc.render(extended_mapping)

            # 保存渲染后的文档
            doc.save(output_path)
            return True

        except Exception as e:
            # 打印错误信息但不中断整个批处理过程
            print(f"处理文档 {input_path} 时出错: {e}")
            return False

    def process_documents(self,
                          input_dir: str,
                          csv_file: str,
                          output_dir: Optional[str] = None,
                          delay_seconds: float = 0) -> str:
        """
        批量处理Word文档

        Args:
            input_dir: 输入目录,包含Word模板文件
            csv_file: CSV映射文件路径
            output_dir: 输出目录,如果为None则自动创建
            delay_seconds: 文件处理间隔延迟(秒),用于避免系统资源冲突

        Returns:
            str: 输出目录路径

        Raises:
            Exception: 当必要的文件或目录不存在时抛出异常
        """
        # 读取CSV映射文件和配置
        mapping = self.read_csv_mapping(csv_file)
        config = self.extract_config_from_mapping(mapping)

        # 使用CSV中的配置覆盖参数(如果存在)
        if 'input_dir' in config and config['input_dir']:
            input_dir = config['input_dir']

        if 'output_dir' in config and config['output_dir']:
            output_dir = config['output_dir']

        if 'delay_seconds' in config and config['delay_seconds']:
            try:
                delay_seconds = float(config['delay_seconds'])
            except ValueError:
                print(f"警告: 无效的delay_seconds值 '{config['delay_seconds']}',使用默认值 {delay_seconds}")

        # 打印处理信息
        print(f"读取到 {len(mapping)} 个占位符映射")
        print(f"配置参数: input_dir={input_dir}, output_dir={output_dir}, delay_seconds={delay_seconds}")

        # 验证必要的ProjectName字段
        if not self.validate_mapping(mapping):
            raise Exception("缺少必要的ProjectName字段")

        # 设置输出目录
        if not output_dir:
            # 默认输出目录设为项目根目录,使用ProjectName作为文件夹名
            project_name = mapping.get('ProjectName', 'processed_documents')
            output_dir = self.create_output_directory(str(Path.cwd()), project_name)
        else:
            # 确保自定义的输出目录存在
            output_path = Path(output_dir)
            output_path.mkdir(parents=True, exist_ok=True)

        # 获取输入目录中的所有Word文档
        input_path = Path(input_dir)
        docx_files = list(input_path.glob("*.docx"))
        # 过滤掉Word临时文件(以~$开头的文件)
        docx_files = [f for f in docx_files if not f.name.startswith('~$')]

        if not docx_files:
            raise Exception(f"在目录 {input_dir} 中未找到Word文档")

        print(f"找到 {len(docx_files)} 个Word文档,开始处理...")

        # 批量处理文档
        success_count = 0
        for i, file_path in enumerate(docx_files, 1):
            print(f"正在处理 ({i}/{len(docx_files)}): {file_path.name}")

            # 构建输出文件路径(保持原文件名)
            output_path = Path(output_dir) / file_path.name

            # 处理单个文档
            if self.process_single_document(str(file_path), str(output_path), mapping):
                success_count += 1
                print(f"  ✓ 完成: {output_path.name}")
            else:
                print(f"  ✗ 失败: {file_path.name}")

            # 在文件处理之间添加延迟(如果设置了延迟时间)
            if i < len(docx_files) and delay_seconds > 0:
                time.sleep(delay_seconds)

        # 输出处理结果统计
        print(f"处理完成! 成功: {success_count}/{len(docx_files)}")
        return output_dir


def main():
    """
    主函数 - 配置并运行文档处理流程
    处理流程:
    1. 检查必要的文件和目录
    2. 读取配置
    3. 初始化处理器
    4. 执行批量处理
    5. 输出结果
    """
    # 获取当前工作目录
    current_dir = Path.cwd()

    # 定义必要的文件路径
    csv_file_path = current_dir / "信息表.csv"  # CSV映射文件
    template_dir = current_dir / "模板"  # 模板文件目录

    # 检查CSV文件是否存在
    if not csv_file_path.exists():
        print("错误: 在当前目录找不到 '信息表.csv' 文件")
        print("请确保CSV文件与程序文件在同一目录下")
        input("按回车键退出...")
        return

    # 检查模板目录是否存在,不存在则创建
    if not template_dir.exists():
        print("警告: 在当前目录找不到 '模板' 文件夹")
        print("将尝试创建模板目录...")
        template_dir.mkdir(exist_ok=True)
        print("请在 '模板' 文件夹中放置Word文档模板")
        input("按回车键继续...")

    # 默认配置参数
    config = {
        'input_directory': str(template_dir),  # 输入目录
        'csv_mapping_file': str(csv_file_path),  # CSV映射文件路径
        'output_directory': None,  # 输出目录(自动生成)
        'delay_between_files': 0.5,  # 文件处理间隔(秒)
    }

    try:
        # 初始化文档处理器
        processor = DocxPlaceholderProcessor()

        # 读取CSV文件中的配置信息
        mapping = processor.read_csv_mapping(config['csv_mapping_file'])
        csv_config = processor.extract_config_from_mapping(mapping)

        # 使用CSV中的配置覆盖默认配置
        if 'input_dir' in csv_config and csv_config['input_dir']:
            config['input_directory'] = csv_config['input_dir']

        if 'output_dir' in csv_config and csv_config['output_dir']:
            config['output_directory'] = csv_config['output_dir']

        if 'delay_seconds' in csv_config and csv_config['delay_seconds']:
            try:
                config['delay_between_files'] = float(csv_config['delay_seconds'])
            except ValueError:
                print(
                    f"警告: 无效的delay_seconds值 '{csv_config['delay_seconds']}',使用默认值 {config['delay_between_files']}")

        # 执行批量文档处理
        output_dir = processor.process_documents(
            input_dir=config['input_directory'],
            csv_file=config['csv_mapping_file'],
            output_dir=config['output_directory'],
            delay_seconds=config['delay_between_files']
        )

        # 输出处理完成信息
        print(f"
所有文档处理完成!")
        print(f"输出目录: {output_dir}")

    except Exception as e:
        # 捕获并显示处理过程中的错误
        print(f"处理过程中发生错误: {e}")

    # 等待用户确认后退出
    input("按回车键退出...")


if __name__ == "__main__":
    """
    程序入口点
    当直接运行此脚本时执行main函数   
    """
    main()