日常工作中，我们经常需要处理各种格式的文件转换，如图片格式转换、文档格式转换等。手动一个个转换既费时又容易出错。本文将介绍如何使用Python打造一个多功能的批量文件格式转换工具。

功能设计

一个完善的格式转换工具应该支持：

• 图片格式互转（JPG/PNG/WebP/GIF/BMP）
• 文档格式转换（Markdown/HTML/PDF）
• 音频格式转换（MP3/WAV/OGG）
• 视频格式转换（MP4/AVI/MKV）
• 批量处理和进度显示
• 转换选项自定义

核心实现：文件格式转换工具

import os
import sys
import json
import time
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import logging

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


class ImageFormat(Enum):
    """支持的图片格式"""
    JPEG = ('jpg', 'jpeg')
    PNG = ('png',)
    WEBP = ('webp',)
    GIF = ('gif',)
    BMP = ('bmp',)
    TIFF = ('tiff', 'tif')


@dataclass
class ConversionTask:
    """转换任务"""
    source: Path
    target: Path
    status: str = 'pending'  # pending, processing, completed, failed
    error: Optional[str] = None


@dataclass
class ConversionResult:
    """转换结果"""
    total: int
    success: int
    failed: int
    skipped: int
    duration: float
    tasks: List[ConversionTask]


class ImageConverter:
    """图片格式转换器"""
    
    def __init__(self):
        self.supported_formats = {fmt.value[0] for fmt in ImageFormat}
    
    def convert_with_pillow(self, source: Path, target: Path, quality: int = 95) -> bool:
        """使用Pillow转换图片"""
        try:
            from PIL import Image
            
            with Image.open(source) as img:
                # 转换RGBA到RGB（JPEG不支持透明通道）
                if target.suffix.lower() in ['.jpg', '.jpeg'] and img.mode == 'RGBA':
                    background = Image.new('RGB', img.size, (255, 255, 255))
                    background.paste(img, mask=img.split()[3])
                    img = background
                
                # 保存
                save_kwargs = {}
                if target.suffix.lower() in ['.jpg', '.jpeg', '.webp']:
                    save_kwargs['quality'] = quality
                elif target.suffix.lower() == '.png':
                    save_kwargs['optimize'] = True
                
                img.save(target, **save_kwargs)
            
            return True
        except Exception as e:
            logger.error(f"转换失败 {source}: {e}")
            return False
    
    def resize_image(self, source: Path, target: Path, width: int = None, 
                     height: int = None, keep_ratio: bool = True) -> bool:
        """调整图片大小"""
        try:
            from PIL import Image
            
            with Image.open(source) as img:
                original_width, original_height = img.size
                
                if width and height:
                    new_size = (width, height)
                elif width:
                    if keep_ratio:
                        ratio = width / original_width
                        new_size = (width, int(original_height * ratio))
                    else:
                        new_size = (width, original_height)
                elif height:
                    if keep_ratio:
                        ratio = height / original_height
                        new_size = (int(original_width * ratio), height)
                    else:
                        new_size = (original_width, height)
                else:
                    return False
                
                resized = img.resize(new_size, Image.Resampling.LANCZOS)
                resized.save(target)
            
            return True
        except Exception as e:
            logger.error(f"调整大小失败 {source}: {e}")
            return False
    
    def create_thumbnail(self, source: Path, target: Path, size: Tuple[int, int] = (200, 200)) -> bool:
        """创建缩略图"""
        try:
            from PIL import Image
            
            with Image.open(source) as img:
                img.thumbnail(size, Image.Resampling.LANCZOS)
                img.save(target)
            
            return True
        except Exception as e:
            logger.error(f"创建缩略图失败 {source}: {e}")
            return False


class MarkdownConverter:
    """Markdown转换器"""
    
    def to_html(self, source: Path, target: Path, template: str = None) -> bool:
        """Markdown转HTML"""
        try:
            import markdown
            
            with open(source, 'r', encoding='utf-8') as f:
                md_content = f.read()
            
            html_content = markdown.markdown(
                md_content,
                extensions=['tables', 'fenced_code', 'codehilite', 'toc']
            )
            
            # 默认模板
            html_template = template or '''<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>{title}</title>
    <style>
        body {{ font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }}
        pre {{ background: #f5f5f5; padding: 15px; overflow-x: auto; }}
        code {{ background: #f5f5f5; padding: 2px 5px; }}
        table {{ border-collapse: collapse; width: 100%; }}
        th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
    </style>
</head>
<body>
{content}
</body>
</html>'''
            
            title = source.stem
            final_html = html_template.format(title=title, content=html_content)
            
            with open(target, 'w', encoding='utf-8') as f:
                f.write(final_html)
            
            return True
        except Exception as e:
            logger.error(f"Markdown转换失败 {source}: {e}")
            return False
    
    def to_pdf(self, source: Path, target: Path) -> bool:
        """Markdown转PDF"""
        try:
            import markdown
            from weasyprint import HTML, CSS
            
            with open(source, 'r', encoding='utf-8') as f:
                md_content = f.read()
            
            html_content = markdown.markdown(md_content)
            
            html_string = f'''<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>{source.stem}</title>
</head>
<body>
{html_content}
</body>
</html>'''
            
            HTML(string=html_string).write_pdf(target)
            return True
        except ImportError:
            logger.error("请安装: pip install weasyprint")
            return False
        except Exception as e:
            logger.error(f"PDF转换失败 {source}: {e}")
            return False


class BatchConverter:
    """批量转换管理器"""
    
    def __init__(self):
        self.image_converter = ImageConverter()
        self.md_converter = MarkdownConverter()
        self.tasks: List[ConversionTask] = []
        self.results = ConversionResult(0, 0, 0, 0, 0, [])
    
    def add_image_task(self, source: Path, target_format: str, **options):
        """添加图片转换任务"""
        if not target_format.startswith('.'):
            target_format = '.' + target_format
        
        target = source.with_suffix(target_format)
        self.tasks.append(ConversionTask(source, target))
    
    def add_resize_task(self, source: Path, width: int = None, height: int = None):
        """添加图片缩放任务"""
        suffix = f"_resized_{width or 'auto'}x{height or 'auto'}"
        target = source.parent / f"{source.stem}{suffix}{source.suffix}"
        self.tasks.append(ConversionTask(source, target, 'pending'))
    
    def add_thumbnail_task(self, source: Path, size: Tuple[int, int] = (200, 200)):
        """添加缩略图任务"""
        target = source.parent / f"{source.stem}_thumb{source.suffix}"
        self.tasks.append(ConversionTask(source, target, 'pending'))
    
    def add_markdown_task(self, source: Path, target_format: str):
        """添加Markdown转换任务"""
        if target_format == 'html':
            target = source.with_suffix('.html')
            self.tasks.append(ConversionTask(source, target, 'pending'))
        elif target_format == 'pdf':
            target = source.with_suffix('.pdf')
            self.tasks.append(ConversionTask(source, target, 'pending'))
    
    def scan_directory(self, directory: Path, pattern: str = '*', recursive: bool = False):
        """扫描目录下所有匹配的文件"""
        if recursive:
            files = directory.rglob(pattern)
        else:
            files = directory.glob(pattern)
        return [f for f in files if f.is_file()]
    
    def execute(self, show_progress: bool = True) -> ConversionResult:
        """执行所有转换任务"""
        start_time = time.time()
        self.results = ConversionResult(len(self.tasks), 0, 0, 0, 0, self.tasks)
        
        for i, task in enumerate(self.tasks):
            if show_progress:
                self._print_progress(i, len(self.tasks), task.source.name)
            
            task.status = 'processing'
            
            try:
                # 图片转换
                if task.source.suffix.lower() in ['.jpg', '.jpeg', '.png', '.webp', '.gif', '.bmp', '.tiff']:
                    success = self.image_converter.convert_with_pillow(
                        task.source, task.target
                    )
                
                # Markdown转换
                elif task.source.suffix.lower() == '.md':
                    if task.target.suffix == '.html':
                        success = self.md_converter.to_html(task.source, task.target)
                    elif task.target.suffix == '.pdf':
                        success = self.md_converter.to_pdf(task.source, task.target)
                    else:
                        success = False
                
                else:
                    success = False
                    task.error = 'Unsupported format'
                
                if success:
                    task.status = 'completed'
                    self.results.success += 1
                else:
                    task.status = 'failed'
                    self.results.failed += 1
                    
            except Exception as e:
                task.status = 'failed'
                task.error = str(e)
                self.results.failed += 1
        
        self.results.duration = time.time() - start_time
        
        if show_progress:
            self._print_final_result()
        
        return self.results
    
    @staticmethod
    def _print_progress(current: int, total: int, filename: str):
        """打印进度"""
        percent = (current + 1) / total * 100
        bar = '█' * int(percent / 2) + '░' * (50 - int(percent / 2))
        print(f"\r[{bar}] {percent:5.1f}% {filename:<30}", end='', flush=True)
    
    def _print_final_result(self):
        """打印最终结果"""
        print("\n\n" + "=" * 50)
        print("转换完成!")
        print(f"  ✓ 成功: {self.results.success}")
        print(f"  ✗ 失败: {self.results.failed}")
        print(f"  ⏱ 耗时: {self.results.duration:.2f}秒")
        print("=" * 50)
        
        # 显示失败的任务
        if self.results.failed > 0:
            print("\n失败任务:")
            for task in self.tasks:
                if task.status == 'failed':
                    print(f"  - {task.source.name}: {task.error}")


def main():
    converter = BatchConverter()
    
    # 示例1: 批量转换图片格式
    print("模式1: 批量图片格式转换")
    images = converter.scan_directory(Path('./images'), '*.png')
    for img in images:
        converter.add_image_task(img, '.jpg')
    
    # 示例2: 创建缩略图
    print("\n模式2: 创建缩略图")
    images = converter.scan_directory(Path('./images'), '*')
    for img in images:
        converter.add_thumbnail_task(img, (300, 300))
    
    # 示例3: Markdown转HTML
    print("\n模式3: Markdown转HTML")
    md_files = converter.scan_directory(Path('./docs'), '*.md')
    for md in md_files:
        converter.add_markdown_task(md, 'html')
    
    # 执行
    results = converter.execute()
    
    # 保存结果
    with open('conversion_results.json', 'w') as f:
        json.dump({
            'total': results.total,
            'success': results.success,
            'failed': results.failed,
            'duration': results.duration
        }, f, indent=2)


if __name__ == '__main__':
    main()

扩展：视频格式转换

class VideoConverter:
    """视频格式转换器（需要ffmpeg）"""
    
    def convert(self, source: Path, target: Path, 
                codec: str = None, bitrate: str = None) -> bool:
        """使用ffmpeg转换视频"""
        try:
            cmd = ['ffmpeg', '-i', str(source)]
            
            if codec:
                cmd.extend(['-c:v', codec])
            if bitrate:
                cmd.extend(['-b:v', bitrate])
            
            cmd.extend(['-y', str(target)])  # -y 覆盖输出
            
            result = subprocess.run(cmd, capture_output=True, timeout=300)
            return result.returncode == 0
        except Exception as e:
            logger.error(f"视频转换失败: {e}")
            return False

使用示例

# 图片转WebP
converter = BatchConverter()
for f in Path('./images').glob('*.png'):
    converter.add_image_task(f, '.webp')
converter.execute()

# Markdown批量转HTML
converter = BatchConverter()
for f in Path('./docs').glob('**/*.md'):
    converter.add_markdown_task(f, 'html')
converter.execute()

总结

通过Python的Pillow库和markdown库，我们可以轻松实现各种文件格式的批量转换。这个工具可以根据实际需求继续扩展更多功能，如添加水印、批量压缩、格式校验等。🙃