Python实例题：图片批量处理工具

目录

Python实例题

题目

问题描述

解题思路

关键代码框架

难点分析

Python实例题

题目

图片批量处理工具

问题描述

开发一个 Python 工具，实现以下功能：

• 遍历指定文件夹下的所有图片文件（支持常见格式如 jpg、png、webp）
• 对每张图片执行以下处理流程：
  • 尺寸缩放（按比例或固定尺寸）
  • 添加水印（文字或图片水印）
  • 格式转换（如将所有图片转为 png）
• 使用多线程加速处理过程
• 显示处理进度和耗时统计
• 支持错误处理（如损坏图片跳过、权限问题提示）
• 生成处理日志文件

解题思路

• 使用os或pathlib进行文件遍历
• 利用PIL(Pillow)库进行图像处理
• 通过concurrent.futures.ThreadPoolExecutor实现多线程
• 用tqdm库显示进度条
• 设计日志系统记录处理详情
• 采用面向对象设计封装不同功能模块

关键代码框架

import os
import time
import logging
from PIL import Image, ImageDraw, ImageFont
from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
from pathlib import Path

# 配置日志系统
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("image_processor.log"),
        logging.StreamHandler()
    ]
)

class ImageProcessor:
    def __init__(self, source_dir, target_dir, max_workers=5):
        self.source_dir = Path(source_dir)
        self.target_dir = Path(target_dir)
        self.target_dir.mkdir(exist_ok=True, parents=True)
        self.max_workers = max_workers
        self.supported_formats = ['.jpg', '.jpeg', '.png', '.webp', '.bmp']
        self.processed_count = 0
        self.error_count = 0
        self.start_time = None
        
    def get_image_files(self):
        """获取所有图片文件路径"""
        image_files = []
        for ext in self.supported_formats:
            image_files.extend(self.source_dir.glob(f"**/*{ext}"))
        return image_files
    
    def resize_image(self, img, target_size=(800, 600), keep_ratio=True):
        """调整图片尺寸"""
        if keep_ratio:
            img.thumbnail(target_size, Image.LANCZOS)
        else:
            img = img.resize(target_size, Image.LANCZOS)
        return img
    
    def add_text_watermark(self, img, text, position=(10, 10), font_size=16):
        """添加文字水印"""
        draw = ImageDraw.Draw(img)
        font = ImageFont.truetype("simhei.ttf", font_size)  # 需要中文字体文件
        draw.text(position, text, fill=(255, 255, 255, 128), font=font)
        return img
    
    def add_image_watermark(self, img, watermark_path, opacity=0.5):
        """添加图片水印"""
        try:
            watermark = Image.open(watermark_path).convert("RGBA")
            watermark = watermark.resize((100, 100), Image.LANCZOS)
            img = img.convert("RGBA")
            
            # 水印位置（右下角）
            pos = (img.width - watermark.width - 10, img.height - watermark.height - 10)
            transparent = Image.new('RGBA', img.size, (0, 0, 0, 0))
            transparent.paste(img, (0, 0))
            transparent.paste(watermark, pos, watermark)
            
            # 调整透明度
            alpha = int(255 * opacity)
            watermark = watermark.copy()
            watermark.putalpha(alpha)
            transparent.paste(watermark, pos, watermark)
            return transparent.convert("RGB")
        except Exception as e:
            logging.warning(f"添加图片水印失败: {e}")
            return img
    
    def process_image(self, image_path):
        """处理单张图片"""
        try:
            # 构建目标文件路径
            relative_path = image_path.relative_to(self.source_dir)
            target_path = self.target_dir / relative_path
            target_path.parent.mkdir(exist_ok=True, parents=True)
            
            # 打开图片
            img = Image.open(image_path)
            logging.info(f"处理图片: {image_path}")
            
            # 执行处理流程
            img = self.resize_image(img, target_size=(1000, 800))
            img = self.add_text_watermark(img, "版权所有 © 2025")
            img = self.add_image_watermark(img, "watermark.png")  # 需要存在水印图片
            
            # 转换格式并保存
            target_ext = ".png"  # 统一转为png格式
            target_path = target_path.with_suffix(target_ext)
            img.save(target_path, quality=95)
            self.processed_count += 1
            return True
        except Exception as e:
            logging.error(f"处理图片失败 {image_path}: {e}")
            self.error_count += 1
            return False
    
    def run(self):
        """启动批量处理"""
        self.start_time = time.time()
        image_files = self.get_image_files()
        logging.info(f"发现 {len(image_files)} 张图片待处理")
        
        if not image_files:
            logging.warning("未找到可处理的图片文件")
            return
        
        # 使用线程池处理
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            results = list(tqdm(
                executor.map(self.process_image, image_files),
                total=len(image_files),
                desc="处理进度",
                unit="图片"
            ))
        
        # 统计结果
        end_time = time.time()
        elapsed = end_time - self.start_time
        logging.info(f"处理完成！共处理 {self.processed_count} 张图片，"
                    f"失败 {self.error_count} 张，耗时 {elapsed:.2f} 秒")
        print(f"处理完成！总耗时: {elapsed:.2f} 秒")
        print(f"成功处理: {self.processed_count} 张，失败: {self.error_count} 张")

# 使用示例
if __name__ == "__main__":
    processor = ImageProcessor(
        source_dir="input_images",
        target_dir="output_images",
        max_workers=8
    )
    processor.run()

难点分析

• 多线程同步问题：需要确保processed_count等共享变量的线程安全
• 中文字体支持：添加中文水印时需要指定正确的字体文件路径
• 内存管理：大尺寸图片处理时可能导致内存溢出，需配合Image.LOAD_TRUNCATED_IMAGES等参数
• 异常处理：需要处理文件权限、图片损坏、格式不支持等多种异常情况
• 性能优化：可通过调整线程数、使用生成器分批处理大文件等方式优化