目录
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_workersself.supported_formats = ['.jpg', '.jpeg', '.png', '.webp', '.bmp']self.processed_count = 0self.error_count = 0self.start_time = Nonedef get_image_files(self):"""获取所有图片文件路径"""image_files = []for ext in self.supported_formats:image_files.extend(self.source_dir.glob(f"**/*{ext}"))return image_filesdef 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 imgdef 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 imgdef 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 imgdef process_image(self, image_path):"""处理单张图片"""try:# 构建目标文件路径relative_path = image_path.relative_to(self.source_dir)target_path = self.target_dir / relative_pathtarget_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 += 1return Trueexcept Exception as e:logging.error(f"处理图片失败 {image_path}: {e}")self.error_count += 1return Falsedef 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_timelogging.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等参数 - 异常处理：需要处理文件权限、图片损坏、格式不支持等多种异常情况
- 性能优化：可通过调整线程数、使用生成器分批处理大文件等方式优化
)
