目錄
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等參數 - 異常處理：需要處理文件權限、圖片損壞、格式不支持等多種異常情況
- 性能優化：可通過調整線程數、使用生成器分批處理大文件等方式優化
)
