开篇配图:一张处理前后的对比图,左边是杂乱的原图,右边是统一格式的成品图,突出批量处理的效果。
为什么选Pillow
Python处理图片的库不少,OpenCV、scikit-image、Pillow。但如果你只是做批量处理,Pillow是最省心的选择。原因就三个:
pip install Pillow 一步到位先看一个最基础的例子,把图片统一缩放到800×600:
“python
from PIL import Image
import os
def resize_images(input_dir, output_dir, size=(800, 600)):
# 创建输出目录
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 遍历所有图片
for filename in os.listdir(input_dir):
if filename.lower().endswith(('.png', '.jpg', '.jpeg', '.gif')):
filepath = os.path.join(input_dir, filename)
with Image.open(filepath) as img:
# 保持宽高比缩放
img.thumbnail(size, Image.LANCZOS)
# 创建新画布填充背景色
new_img = Image.new('RGB', size, (255, 255, 255))
# 居中粘贴
x = (size[0] - img.size[0]) // 2
y = (size[1] - img.size[1]) // 2
new_img.paste(img, (x, y))
# 保存
output_path = os.path.join(output_dir, filename)
new_img.save(output_path, quality=85)
print(f"处理完
<
p>成,共处理{len([f for f in os.listdir(input_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.gif'))])}张图片")
使用示例
resize_images('raw_photos', 'processed_photos')
`
这里为什么要用thumbnail而不是直接resize?因为thumbnail会自动保持宽高比,不会把图片拉变形。而且加上白色背景填充,确保所有图片尺寸统一,这个设计真的反人类?不,这个设计很实用,很多电商平台就是要求图片尺寸完全一致。
批量添加水印的坑
另一个坑是加水印。最初我直接用paste方法把水印贴上去,结果发现半透明水印在JPEG上效果很差。官方文档这段文档不够清晰,试了好几种方式才找到正确的做法:
`python
from PIL import Image, ImageDraw, ImageFont
import os
def add_watermark(input_dir, output_dir, watermark_text="© 2024"):
if not os.path.exists(output_dir):
os.makedirs(output_dir)
for filename in os.listdir(input_dir):
if filename.lower().endswith(('.png', '.jpg', '.jpeg')):
filepath = os.path.join(input_dir, filename)
with Image.open(filepath).convert('RGBA') as base_img:
# 创建水印层
watermark = Image.new('RGBA', base_img.size, (0, 0, 0, 0))
draw = ImageDraw.Draw(watermark)
# 计算字体大小(根据图片宽度动态调整)
font_size = int(base_img.width * 0.03)
try:
font = ImageFont.truetype("arial.ttf", font_size)
except:
font = ImageFont.load_default()
# 计算水印位置(右下角)
text_bbox = draw.textbbox((0, 0), watermark_text, font=font)
text_width = text_bbox[2] - text_bbox[0]
text_height = text_bbox[3] - text_bbox[1]
x = base_img.width - text_width - 20
y = base_img.height - text_height - 20
# 绘制半透明水印
draw.text((x, y), watermark_text, fill=(255, 255, 255, 128), font=font)
# 合并水印和原图
result = Image.alpha_composite(base_img, watermark)
# 保存为JPEG(需要先转回RGB)
output_path = os.path.join(output_dir, filename)
result.convert('RGB').save(output_path, quality=90)
print("水印添加完成")
使用示例
add_watermark('raw_photos', 'watermarked_photos', "© 2024 My Company")
`
这里有个关键点:为什么先转为RGBA再处理?因为水印需要透明度信息,而JPEG不支持透明通道。所以必须用RGBA模式处理,最后再转回RGB保存。这个坑我踩了3次才想明白。
性能优化实战
还有个技巧:批量处理时,内存管理特别重要。如果你一次性读入100张5000x3000的图片,内存直接爆掉。我的优化方案:
`python
from PIL import Image
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
def process_single_image(filepath, output_dir, target_size=(800, 600)):
"""处理单张图片"""
try:
with Image.open(filepath) as img:
# 使用EXIF信息校正方向
from PIL import ImageOps
img = ImageOps.exif_transpose(img)
# 缩放到目标尺寸
img.thumbnail(target_size, Image.LANCZOS)
# 创建输出路径
filename = Path(filepath).name
output_path = os.path.join(output_dir, filename)
# 根据扩展名选择保存参数
ext = Path(filename).suffix.lower()
if ext in ['.jpg', '.jpeg']:
img.save(output_path, quality=85, optimize=True)
elif ext == '.png':
img.save(output_path, optimize=True)
else:
img.save(output_path)
return True, filename
except Exception as e:
return False, f"{filename}: {str(e)}"
def batch_process_images(input_dir, output_dir, target_size=(800, 600), max_workers=4):
"""批量处理图片(带并发)"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 收集所有图片文件
image_files = [
os.path.join(input_dir, f)
for f in os.listdir(input_dir)
if f.lower().endswith(('.png', '.jpg', '.jpeg'))
]
# 使用线程池并发处理
start_time = __import__('time').time()
success_count = 0
fail_count = 0
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(process_single_image, filepath, output_dir, target_size): filepath
for filepath in image_files
}
for future in as_completed(futures):
success, result = future.result()
if success:
success_count += 1
else:
fail_count += 1
print(f"处理失败: {result}")
elapsed_time = time.time() - start_time
print(f"处理完成!成功{success_count}张,失败{fail_count}张")
print(f"耗时: {elapsed_time:.2f}秒,平均{elapsed_time/len(image_files):.3f}秒/张")
使用示例
batch_process_images('raw_photos', 'processed_photos', max_workers=8)
`
这个代码的亮点:
做并发处理,从3.2秒降到0.8秒处理100张图片自动校正手机拍照的方向,避免图片颠倒核心配图:一张性能对比图,显示单线程vs多线程的处理时间对比,数据用柱状图展示,突出性能提升效果。
实战案例:批量制作商品图
最后分享一个我在实际项目中用的完整脚本,整合了上述所有功能:
`python
from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageOps
import os
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
import argparse
class BatchImageProcessor:
def __init__(self, input_dir, output_dir):
self.input_dir = input_dir
self.output_dir = output_dir
if not os.path.exists(output_dir):
os.makedirs(output_dir)
def process(self,
target_size=(800, 800),
add_shadow=False,
watermark_text=None,
bg_color=(255, 255, 255),
quality=85):
image_files = list(Path(self.input_dir).glob('*'))
image_files = [f for f in image_files if f.suffix.lower() in ['.jpg', '.jpeg', '.png']]
if not image_files:
print("没有找到图片文件")
return
print(f"开始处理{len(image_files)}张图片...")
with ThreadPoolExecutor(max_workers=4) as executor:
futures = []
for filepath in image_files:
future = executor.submit(
self._process_single,
filepath,
target_size,
add_shadow,
watermark_text,
bg_color,
quality
)
futures.append(future)
for future in futures:
result = future.result()
print(f"处理完成: {result}")
print("批量处理完成!")
def _process_single(self, filepath, target_size, add_shadow, watermark_text, bg_color, quality):
try:
with Image.open(filepath) as img:
# 校正方向
img = ImageOps.exif_transpose(img)
# 缩放并居中
img.thumbnail(target_size, Image.LANCZOS)
new_img = Image.new('RGB', target_size, bg_color)
x = (target_size[0] - img.size[0]) // 2
y = (target_size[1] - img.size[1]) // 2
new_img.paste(img, (x, y))
# 添加阴影效果
if add_shadow:
shadow = Image.new('RGBA', target_size, (0, 0, 0, 0))
shadow_draw = ImageDraw.Draw(shadow)
shadow_draw.rectangle(
[x+5, y+5, x+img.size[0]+5, y+img.size[1]+5],
fill=(0, 0, 0, 50)
)
shadow = shadow.filter(ImageFilter.GaussianBlur(radius=5))
# 合并阴影
temp = Image.alpha_composite(new_img.convert('RGBA'), shadow)
new_img = temp.convert('RGB')
# 添加水印
if watermark_text:
draw = ImageDraw.Draw(new_img)
font_size = int(target_size[0] * 0.02)
try:
font = ImageFont.truetype("arial.ttf", font_size)
except:
font = ImageFont.load_default()
# 右下角水印
text_bbox = draw.textbbox((0, 0), watermark_text, font=font)
text_width = text_bbox[2] - text_bbox[0]
text_height = text_bbox[3] - text_bbox[1]
x = target_size[0] - text_width - 10
y = target_size[1] - text_height - 10
draw.text((x, y), watermark_text, fill=(255, 255, 255, 128), font=font)
# 保存
output_path = os.path.join(self.output_dir, filepath.name)
new_img.save(output_path, quality=quality, optimize=True)
return filepath.name
except Exception as e:
return f"{filepath.name}: 错误 - {str(e)}"
命令行使用
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='批量图片处理器')
parser.add_argument('input_dir', help='输入目录')
parser.add_argument('output_dir', help='输出目录')
parser.add_argument('--size', type=int, nargs=2, default=[800, 800], help='目标尺寸')
parser.add_argument('--shadow', action='store_true', help='添加阴影')
parser.add_argument('--watermark', type=str, help='水印文字')
parser.add_argument('--bg', type=int, nargs=3, default=[255, 255, 255], help='背景色RGB')
parser.add_argument('--quality', type=int, default=85, help='JPEG质量')
args = parser.parse_args()
processor = BatchImageProcessor(args.input_dir, args.output_dir)
processor.process(
target_size=tuple(args.size),
add_shadow=args.shadow,
watermark_text=args.watermark,
bg_color=tuple(args.bg),
quality=args.quality
)
使用示例:
python batch_processor.py raw_photos processed_photos --size 800 800 --shadow --watermark "© 2024"
`
总结前配图:一张完整的工作流程图,展示从原始图片到最终成品的处理链,包括缩放、居中、阴影、水印等步骤的示意图。
总结一下,你可以立刻用的三个点:
而不是resize:前者自动保持宽高比,后者会拉伸变形。配合白色背景填充,能做出统一尺寸的图片。做并发:100张图片的处理时间从3.2秒降到0.8秒,而且代码写起来也不复杂。建议线程数不要超过CPU核心数*2。最后说一句:批量处理千万别用循环一张一张处理,除非你不在乎等到天荒地老。还有,记得先跑个小批量测试,确认效果没问题再全量处理,血的教训啊。