markitdown自动化脚本：定时批量文档转换任务

凌朦慧Richard

847人浏览 · 2025-09-18 07:09:25

凌朦慧Richard · 2025-09-18 07:09:25 发布

markitdown自动化脚本：定时批量文档转换任务

【免费下载链接】markitdown 将文件和办公文档转换为 Markdown 的 Python 工具项目地址: https://gitcode.com/GitHub_Trending/ma/markitdown

痛点直击：你还在手动处理文档转换吗？

企业日常运营中，大量办公文档需要转换为Markdown（标记语言）格式以便于版本控制、内容分析或LLM（大语言模型）处理。手动逐个转换不仅耗时耗力，还容易出现遗漏和格式不一致问题。本文将详细介绍如何使用markitdown结合Python定时任务框架，构建自动化批量文档转换系统，彻底解放双手。

读完本文你将获得：

从零搭建定时文档转换服务的完整方案
支持15+文件格式的批量转换脚本
错误处理与日志监控的实现方法
资源占用优化的实战技巧

技术架构概览

mermaid

核心组件包括：

文件监控模块：定时扫描指定目录
任务调度器：管理转换队列和并发控制
转换引擎：基于markitdown的多格式处理核心
结果处理：输出管理与错误反馈
配置中心：统一参数管理

环境准备与安装

系统要求

Python 3.10+
2GB以上内存（处理图片和PDF时建议4GB+）
10GB以上空闲磁盘空间

安装步骤

创建并激活虚拟环境：

python -m venv .venv
source .venv/bin/activate  # Linux/Mac
.venv\Scripts\activate     # Windows

克隆项目并安装依赖：

git clone https://gitcode.com/GitHub_Trending/ma/markitdown
cd markitdown
pip install '.[all]'
pip install schedule python-dotenv tqdm

验证安装：

markitdown --version

核心实现：自动化转换脚本开发

1. 配置文件设计

创建.env配置文件：

# 源目录配置
SOURCE_DIR=/data/documents/to_convert
# 目标目录配置
DEST_DIR=/data/documents/converted_md
# 支持的文件格式 (逗号分隔)
SUPPORTED_FORMATS=pdf,docx,pptx,xlsx,jpg,png,mp3,csv,json
# 排除文件模式 (通配符)
EXCLUDE_PATTERNS=*~,*.tmp,*.backup
# 定时任务配置 (分钟)
SCHEDULE_INTERVAL=60
# 并发任务数
MAX_WORKERS=4
# 日志文件路径
LOG_FILE=/var/log/markitdown/auto_convert.log
# 转换超时时间(秒)
CONVERT_TIMEOUT=300

2. 文件扫描模块

import os
import fnmatch
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

class FileScanner:
    def __init__(self):
        self.source_dir = os.getenv('SOURCE_DIR')
        self.supported_formats = os.getenv('SUPPORTED_FORMATS').split(',')
        self.exclude_patterns = os.getenv('EXCLUDE_PATTERNS').split(',')
        self.processed_records = set()
        
    def is_supported(self, filename):
        """检查文件是否支持转换"""
        ext = Path(filename).suffix.lower()[1:]
        return ext in self.supported_formats
        
    def is_excluded(self, filename):
        """检查文件是否需要排除"""
        for pattern in self.exclude_patterns:
            if fnmatch.fnmatch(filename, pattern):
                return True
        return False
        
    def scan_new_files(self):
        """扫描目录获取新文件列表"""
        new_files = []
        for root, _, files in os.walk(self.source_dir):
            for file in files:
                file_path = os.path.join(root, file)
                if self.is_supported(file) and not self.is_excluded(file):
                    # 使用文件路径+修改时间作为唯一标识
                    file_id = f"{file_path}_{os.path.getmtime(file_path)}"
                    if file_id not in self.processed_records:
                        new_files.append(file_path)
                        self.processed_records.add(file_id)
        return new_files

3. 转换任务实现

import os
import logging
from pathlib import Path
from markitdown import MarkItDown
from tqdm import tqdm

class DocumentConverter:
    def __init__(self, dest_dir, timeout=300):
        self.dest_dir = dest_dir
        self.timeout = timeout
        self.converter = MarkItDown(enable_plugins=True)
        # 创建目标目录
        Path(dest_dir).mkdir(parents=True, exist_ok=True)
        
    def convert_file(self, source_path):
        """转换单个文件"""
        try:
            # 获取相对路径用于保持目录结构
            rel_path = os.path.relpath(source_path, os.getenv('SOURCE_DIR'))
            # 构建目标路径
            dest_path = os.path.join(
                self.dest_dir, 
                os.path.splitext(rel_path)[0] + '.md'
            )
            # 创建目标目录
            Path(os.path.dirname(dest_path)).mkdir(parents=True, exist_ok=True)
            
            # 执行转换
            result = self.converter.convert(source_path)
            
            # 保存结果
            with open(dest_path, 'w', encoding='utf-8') as f:
                f.write(result.text_content)
                
            return {
                'status': 'success',
                'source': source_path,
                'dest': dest_path,
                'message': '转换成功'
            }
            
        except Exception as e:
            logging.error(f"转换失败 {source_path}: {str(e)}")
            return {
                'status': 'error',
                'source': source_path,
                'dest': None,
                'message': str(e)
            }
    
    def batch_convert(self, file_list):
        """批量转换文件"""
        results = []
        for file_path in tqdm(file_list, desc="转换进度"):
            results.append(self.convert_file(file_path))
        return results

4. 定时任务调度

import schedule
import time
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

class Scheduler:
    def __init__(self, scanner, converter, interval=60, max_workers=4):
        self.scanner = scanner
        self.converter = converter
        self.interval = interval  # 分钟
        self.max_workers = max_workers
        self.logger = self._setup_logger()
        
    def _setup_logger(self):
        """配置日志"""
        logger = logging.getLogger('markitdown-scheduler')
        logger.setLevel(logging.INFO)
        
        # 文件处理器
        fh = logging.FileHandler(os.getenv('LOG_FILE'))
        fh.setLevel(logging.INFO)
        
        # 控制台处理器
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        
        # 格式器
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        
        logger.addHandler(fh)
        logger.addHandler(ch)
        
        return logger
        
    def run_task(self):
        """执行扫描和转换任务"""
        self.logger.info("开始定时扫描任务")
        new_files = self.scanner.scan_new_files()
        
        if not new_files:
            self.logger.info("未发现新文件")
            return
            
        self.logger.info(f"发现{len(new_files)}个待转换文件")
        
        # 并发处理文件转换
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {executor.submit(self.converter.convert_file, f): f for f in new_files}
            
            for future in as_completed(futures):
                result = future.result()
                if result['status'] == 'success':
                    self.logger.info(f"成功: {result['source']} -> {result['dest']}")
                else:
                    self.logger.error(f"失败: {result['source']} - {result['message']}")
        
    def start(self):
        """启动调度器"""
        self.logger.info(f"定时任务启动，每{self.interval}分钟执行一次")
        schedule.every(self.interval).minutes.do(self.run_task)
        
        while True:
            schedule.run_pending()
            time.sleep(60)

5. 主程序整合

创建auto_convert.py：

import os
from dotenv import load_dotenv
from file_scanner import FileScanner
from document_converter import DocumentConverter
from scheduler import Scheduler

def main():
    # 加载环境变量
    load_dotenv()
    
    # 初始化组件
    scanner = FileScanner()
    converter = DocumentConverter(
        dest_dir=os.getenv('DEST_DIR'),
        timeout=int(os.getenv('CONVERT_TIMEOUT'))
    )
    scheduler = Scheduler(
        scanner=scanner,
        converter=converter,
        interval=int(os.getenv('SCHEDULE_INTERVAL')),
        max_workers=int(os.getenv('MAX_WORKERS'))
    )
    
    # 启动服务
    scheduler.start()

if __name__ == "__main__":
    main()

高级配置与优化

支持格式与性能对比

文件类型	转换速度	内存占用	依赖组件
TXT/CSV	极快(ms级)	低	内置
PDF(文本)	快(1-3s)	中	[pdf]
DOCX	中(2-5s)	中	[docx]
PPTX	中(3-8s)	中高	[pptx]
XLSX	中(2-6s)	中	[xlsx]
图片(OCR)	慢(5-15s)	高	[image]
音频	很慢(30s+)	高	[audio-transcription]
ZIP	取决于内容	中高	内置

资源优化策略

内存控制：处理大文件时设置内存限制

# 在DocumentConverter初始化时添加
import resource
resource.setrlimit(resource.RLIMIT_AS, (4 * 1024 * 1024 * 1024, 4 * 1024 * 1024 * 1024))  # 4GB限制

优先级调度：按文件大小动态调整处理顺序

# 在scan_new_files方法中添加
new_files.sort(key=lambda x: os.path.getsize(x))  # 从小到大排序

缓存机制：避免重复转换相同文件

# 在FileScanner中实现基于文件哈希的缓存
import hashlib

def _file_hash(self, file_path):
    """计算文件内容哈希"""
    hasher = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(4096):
            hasher.update(chunk)
    return hasher.hexdigest()

部署与监控

系统服务配置

创建systemd服务文件/etc/systemd/system/markitdown-convert.service：

[Unit]
Description=Markitdown Auto Convert Service
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/opt/markitdown-auto
ExecStart=/opt/markitdown-auto/.venv/bin/python auto_convert.py
Restart=always
RestartSec=30

[Install]
WantedBy=multi-user.target

启用并启动服务：

sudo systemctl daemon-reload
sudo systemctl enable markitdown-convert
sudo systemctl start markitdown-convert

日志轮转配置

创建/etc/logrotate.d/markitdown：

/var/log/markitdown/*.log {
    daily
    missingok
    rotate 14
    compress
    delaycompress
    notifempty
    create 0640 www-data www-data
}

常见问题与解决方案

1. 转换中文乱码

问题：Word文档转换后中文显示乱码
解决：确保系统安装中文字体，在转换前设置环境变量：

export PYTHONUTF8=1

2. 大文件转换超时

问题：大型PDF转换经常超时
解决：修改配置文件，针对大文件单独设置超时：

# .env中添加
LARGE_FILE_THRESHOLD=10485760  # 10MB
LARGE_FILE_TIMEOUT=600  # 10分钟

3. 内存占用过高

问题：同时转换多个图片文件导致内存溢出
解决：限制图片并发数，修改调度器配置：

# 在Scheduler中添加图片文件单独处理逻辑
image_extensions = {'jpg', 'jpeg', 'png', 'gif', 'bmp'}
image_files = [f for f in new_files if Path(f).suffix[1:].lower() in image_extensions]
other_files = [f for f in new_files if Path(f).suffix[1:].lower() not in image_extensions]

# 图片文件使用较低并发
with ThreadPoolExecutor(max_workers=1) as img_executor:
    # 处理图片...
    
# 其他文件使用正常并发
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
    # 处理其他文件...