logo

开发者热搜

Python PIL批量添加文字水印全攻略：从基础到进阶

作者：搬砖的石头2025.10.10 17:03浏览量：1

简介：本文详细介绍如何使用Python PIL库批量为图片添加文字水印，涵盖基础操作、进阶技巧及性能优化，适合开发者及企业用户快速实现自动化水印处理。

PIL如何批量给图片添加文字水印？

一、PIL库基础与水印原理

Python Imaging Library（PIL）是Python生态中最经典的图像处理库之一，其分支Pillow提供了更完善的维护和扩展功能。文字水印的本质是通过在图片指定位置绘制半透明文字，实现版权标识或品牌宣传的目的。

1.1 PIL核心组件

  • Image：基础图像对象，支持加载、保存和转换图像格式
  • ImageDraw：提供2D绘图功能，包括文字、几何图形绘制
  • ImageFont：控制文字字体、大小和样式

1.2 水印关键参数

  • 文字内容：可动态配置（如添加时间戳）
  • 位置坐标：绝对定位或相对定位
  • 透明度：通过RGBA颜色通道的Alpha值控制
  • 字体样式：影响水印可读性和美观度

二、基础批量处理实现

2.1 单张图片水印添加

  1. from PIL import Image, ImageDraw, ImageFont
  3. def add_text_watermark(input_path, output_path, text, position, font_size=30, opacity=0.5):
  4.     # 加载图片
  5.     img = Image.open(input_path).convert("RGBA")
  7.     # 创建透明图层用于水印
  8.     txt = Image.new("RGBA", img.size, (255, 255, 255, 0))
  9.     draw = ImageDraw.Draw(txt)
  11.     # 加载字体（需系统存在或指定路径）
  12.     try:
  13.         font = ImageFont.truetype("arial.ttf", font_size)
  14.     except:
  15.         font = ImageFont.load_default()
  17.     # 计算文字尺寸
  18.     text_width, text_height = draw.textsize(text, font=font)
  20.     # 计算居中位置（示例为右下角定位）
  21.     x, y = position if position else (img.width - text_width - 10, img.height - text_height - 10)
  23.     # 绘制半透明文字
  24.     draw.text((x, y), text, font=font, fill=(255, 255, 255, int(255 * opacity)))
  26.     # 合并图层
  27.     out = Image.alpha_composite(img, txt)
  28.     out.convert("RGB").save(output_path)
  30. # 使用示例
  31. add_text_watermark("input.jpg", "output.jpg", "Sample Watermark", (10, 10))

2.2 批量处理框架

  1. import os
  2. from glob import glob
  4. def batch_watermark(input_dir, output_dir, text, **kwargs):
  5.     # 创建输出目录
  6.     os.makedirs(output_dir, exist_ok=True)
  8.     # 获取所有图片文件
  9.     image_paths = glob(f"{input_dir}/*.jpg") + glob(f"{input_dir}/*.png")
  11.     for input_path in image_paths:
  12.         # 构造输出路径
  13.         filename = os.path.basename(input_path)
  14.         output_path = os.path.join(output_dir, filename)
  16.         # 添加水印
  17.         add_text_watermark(input_path, output_path, text, **kwargs)
  18.         print(f"Processed: {filename}")
  20. # 使用示例
  21. batch_watermark("input_images", "output_images", "© 2023 MyCompany")

三、进阶优化技巧

3.1 动态水印内容

  1. import datetime
  3. def get_dynamic_text(base_text):
  4.     now = datetime.datetime.now()
  5.     return f"{base_text} | {now.strftime('%Y-%m-%d')}"
  7. # 修改调用方式
  8. dynamic_text = get_dynamic_text("Batch Process")
  9. add_text_watermark("input.jpg", "output.jpg", dynamic_text)

3.2 多位置水印

  1. def add_multi_position_watermark(img_path, output_path, text, positions, **kwargs):
  2.     img = Image.open(img_path).convert("RGBA")
  3.     txt = Image.new("RGBA", img.size, (255, 255, 255, 0))
  4.     draw = ImageDraw.Draw(txt)
  6.     try:
  7.         font = ImageFont.truetype("arial.ttf", kwargs.get("font_size", 30))
  8.     except:
  9.         font = ImageFont.load_default()
  11.     for pos in positions:
  12.         draw.text(pos, text, font=font, fill=(255, 255, 255, int(255 * kwargs.get("opacity", 0.5))))
  14.     out = Image.alpha_composite(img, txt)
  15.     out.convert("RGB").save(output_path)
  17. # 使用示例（四角水印）
  18. positions = [(10,10), (10,390), (390,10), (390,390)]  # 假设400x400图片
  19. add_multi_position_watermark("input.jpg", "output.jpg", "WATERMARK", positions)

3.3 性能优化策略

  1. 字体缓存：重复使用字体对象避免频繁加载
    ```python
    font_cache = {}

def get_cached_font(font_path, size):
key = (font_path, size)
if key not in font_cache:
try:
font_cache[key] = ImageFont.truetype(font_path, size)
except:
font_cache[key] = ImageFont.load_default()
return font_cache[key]

  2. 2. **多线程处理**：使用`concurrent.futures`加速批量处理
  3. ```python
  4. from concurrent.futures import ThreadPoolExecutor
  6. def parallel_batch(input_dir, output_dir, text, workers=4):
  7.     os.makedirs(output_dir, exist_ok=True)
  8.     image_paths = glob(f"{input_dir}/*.jpg")
  10.     def process_single(input_path):
  11.         output_path = os.path.join(output_dir, os.path.basename(input_path))
  12.         add_text_watermark(input_path, output_path, text)
  13.         return output_path
  15.     with ThreadPoolExecutor(max_workers=workers) as executor:
  16.         results = list(executor.map(process_single, image_paths))
  18.     return results

四、常见问题解决方案

4.1 中文乱码问题

  1. # 使用支持中文的字体文件
  2. def add_chinese_watermark(input_path, output_path, text):
  3.     try:
  4.         font = ImageFont.truetype("simhei.ttf", 40)  # 黑体
  5.     except:
  6.         font = ImageFont.load_default()
  8.     img = Image.open(input_path).convert("RGBA")
  9.     txt = Image.new("RGBA", img.size, (255,255,255,0))
  10.     draw = ImageDraw.Draw(txt)
  11.     draw.text((10,10), text, font=font, fill=(255,255,255,128))
  13.     out = Image.alpha_composite(img, txt)
  14.     out.convert("RGB").save(output_path)

4.2 水印位置计算

  1. def calculate_position(img_width, img_height, text_width, text_height, align="center"):
  2.     if align == "center":
  3.         return (img_width//2 - text_width//2, img_height//2 - text_height//2)
  4.     elif align == "top_right":
  5.         return (img_width - text_width - 10, 10)
  6.     # 可扩展其他对齐方式

五、企业级应用建议

  1. 配置化管理：将水印参数（文字、位置、透明度）存储在JSON/YAML配置文件中
  2. 日志系统：记录处理过程和异常信息
  3. 异常处理：添加图片加载失败、字体缺失等异常捕获
  4. 格式支持：扩展支持WEBP、TIFF等更多格式
  5. 质量保证：添加MD5校验确保处理前后图片完整性

六、完整企业级实现示例

  1. import json
  2. import logging
  3. from pathlib import Path
  4. from PIL import Image, ImageDraw, ImageFont
  5. from concurrent.futures import ThreadPoolExecutor
  7. # 配置日志
  8. logging.basicConfig(
  9.     level=logging.INFO,
  10.     format='%(asctime)s - %(levelname)s - %(message)s',
  11.     handlers=[logging.FileHandler('watermark.log'), logging.StreamHandler()]
  12. )
  14. class WatermarkProcessor:
  15.     def __init__(self, config_path):
  16.         with open(config_path) as f:
  17.             self.config = json.load(f)
  18.         self.font_cache = {}
  20.     def get_font(self, size):
  21.         font_path = self.config.get("font_path", "arial.ttf")
  22.         key = (font_path, size)
  23.         if key not in self.font_cache:
  24.             try:
  25.                 self.font_cache[key] = ImageFont.truetype(font_path, size)
  26.             except:
  27.                 self.font_cache[key] = ImageFont.load_default()
  28.                 logging.warning(f"Fallback to default font for size {size}")
  29.         return self.font_cache[key]
  31.     def process_image(self, input_path, output_path):
  32.         try:
  33.             img = Image.open(input_path).convert("RGBA")
  34.             txt = Image.new("RGBA", img.size, (255,255,255,0))
  35.             draw = ImageDraw.Draw(txt)
  37.             font = self.get_font(self.config["font_size"])
  38.             text = self.config["text"]
  39.             text_width, text_height = draw.textsize(text, font=font)
  41.             # 位置计算
  42.             x, y = self.config["position"]
  43.             if x == "center":
  44.                 x = (img.width - text_width) // 2
  45.             if y == "center":
  46.                 y = (img.height - text_height) // 2
  48.             draw.text((x, y), text, font=font, 
  49.                      fill=(255,255,255, int(255 * self.config["opacity"])))
  51.             out = Image.alpha_composite(img, txt)
  52.             out.convert("RGB").save(output_path, quality=95)
  53.             logging.info(f"Successfully processed: {input_path}")
  54.             return True
  55.         except Exception as e:
  56.             logging.error(f"Error processing {input_path}: {str(e)}")
  57.             return False
  59.     def batch_process(self, input_dir, output_dir):
  60.         input_dir = Path(input_dir)
  61.         output_dir = Path(output_dir)
  62.         output_dir.mkdir(exist_ok=True)
  64.         image_files = list(input_dir.glob("*." + self.config.get("extension", "jpg")))
  65.         image_files += list(input_dir.glob("*." + self.config.get("extension2", "png")))
  67.         with ThreadPoolExecutor(max_workers=self.config.get("workers", 4)) as executor:
  68.             futures = []
  69.             for img_file in image_files:
  70.                 output_path = output_dir / img_file.name
  71.                 futures.append(executor.submit(self.process_image, str(img_file), str(output_path)))
  73.             success_count = sum(f.result() for f in futures)
  74.             logging.info(f"Batch processing completed. Success: {success_count}/{len(image_files)}")
  76. # 配置文件示例 (config.json)
  77. """
  78. {
  79.     "text": "© 2023 Company",
  80.     "font_size": 36,
  81.     "opacity": 0.7,
  82.     "position": ["center", "bottom"],  # 或具体坐标如 [100, 100]
  83.     "workers": 8,
  84.     "extension": "jpg",
  85.     "extension2": "png"
  86. }
  87. """
  89. # 使用示例
  90. if __name__ == "__main__":
  91.     processor = WatermarkProcessor("config.json")
  92.     processor.batch_process("input_images", "output_images")

七、总结与最佳实践

  1. 字体管理：始终处理字体加载异常，提供备用方案
  2. 内存优化：批量处理大图片时注意内存使用，可分批处理
  3. 格式一致性：确保输出格式与输入格式兼容，避免质量损失
  4. 测试验证：先在小批量图片上测试效果，再大规模应用
  5. 自动化集成：可将脚本集成到CI/CD流程中，实现自动化水印添加

通过以上方法，开发者可以高效、稳定地实现图片批量水印添加功能，满足从个人到企业级的不同需求。实际开发中，建议根据具体场景调整参数，并添加完善的错误处理和日志记录机制。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数
活动
咨询