Python 自动化之批量图片处理——水印、压缩、格式转换

日常工作中经常要批量处理图片——给产品图加水印、压缩图片大小以便上传网站、批量转换格式。用 Python 的 Pillow 库，几百张图片几秒钟搞定。

一、批量添加水印

1. 文字水印

fromPILimportImage,ImageDraw,ImageFontimportosclassWatermarkProcessor:"""批量水印处理器"""def__init__(self,input_dir,output_dir):self.input_dir=input_dir self.output_dir=output_dir os.makedirs(output_dir,exist_ok=True)defadd_text_watermark(self,text="版权所有",opacity=80):"""添加文字水印"""forfinos.listdir(self.input_dir):ifnotf.lower().endswith((".jpg",".jpeg",".png")):continueimg=Image.open(os.path.join(self.input_dir,f)).convert("RGBA")# 创建水印层watermark=Image.new("RGBA",img.size,(0,0,0,0))draw=ImageDraw.Draw(watermark)# 字体大小 = 图片宽度的 1/20font_size=max(img.width//20,20)try:font=ImageFont.truetype("C:/Windows/Fonts/msyh.ttc",font_size)except:font=ImageFont.load_default()# 计算文字尺寸bbox=draw.textbbox((0,0),text,font=font)text_w,text_h=bbox[2]-bbox[0],bbox[3]-bbox[1]# 在右下角添加margin=20x=img.width-text_w-margin y=img.height-text_h-margin# 绘制半透明文字draw.text((x,y),text,font=font,fill=(255,255,255,opacity))# 合并原图和水印result=Image.alpha_composite(img,watermark).convert("RGB")out_path=os.path.join(self.output_dir,f"wm_{f}")result.save(out_path,quality=95)print(f"已添加水印:{f}")defadd_image_watermark(self,watermark_file,position="右下角"):"""添加图片水印（如 Logo）"""wm=Image.open(watermark_file).convert("RGBA")forfinos.listdir(self.input_dir):ifnotf.lower().endswith((".jpg",".jpeg",".png")):continueimg=Image.open(os.path.join(self.input_dir,f)).convert("RGBA")# 水印缩放为图片宽度的 1/5wm_resized=wm.resize((img.width//5,int(wm.height*img.width//5/wm.width)),Image.LANCZOS)# 定位margin=20ifposition=="右下角":x=img.width-wm_resized.width-margin y=img.height-wm_resized.height-marginelifposition=="左上角":x=y=marginelifposition=="居中":x=(img.width-wm_resized.width)//2y=(img.height-wm_resized.height)//2# 粘贴水印img.paste(wm_resized,(x,y),wm_resized)out_path=os.path.join(self.output_dir,f"logo_{f}")img.convert("RGB").save(out_path,quality=95)print(f"已添加 Logo:{f}")# 使用processor=WatermarkProcessor("原始图片","加水印")processor.add_text_watermark("张老师技术栈")

二、批量压缩图片

1. 压缩到指定质量

defbatch_compress(input_dir,output_dir,quality=60,max_width=1920):"""批量压缩图片"""os.makedirs(output_dir,exist_ok=True)forfinos.listdir(input_dir):ifnotf.lower().endswith((".jpg",".jpeg",".png")):continuefilepath=os.path.join(input_dir,f)img=Image.open(filepath)# 限制最大宽度ifimg.width>max_width:ratio=max_width/img.width new_size=(max_width,int(img.height*ratio))img=img.resize(new_size,Image.LANCZOS)# 保存（quality 越低文件越小）out_path=os.path.join(output_dir,f"compressed_{f}")img.save(out_path,quality=quality,optimize=True)original_size=os.path.getsize(filepath)compressed_size=os.path.getsize(out_path)ratio=(1-compressed_size/original_size)*100print(f"{f}:{original_size//1024}KB →{compressed_size//1024}KB (压缩{ratio:.0f}%)")# 使用batch_compress("产品图片","压缩后",quality=60,max_width=1200)

2. 批量转 WebP 格式（更小的体积）

defconvert_to_webp(input_dir,output_dir,quality=80):"""批量转换为 WebP 格式"""os.makedirs(output_dir,exist_ok=True)forfinos.listdir(input_dir):ifnotf.lower().endswith((".jpg",".jpeg",".png")):continueimg=Image.open(os.path.join(input_dir,f))out_name=os.path.splitext(f)[0]+".webp"out_path=os.path.join(output_dir,out_name)img.save(out_path,"webp",quality=quality)original_size=os.path.getsize(os.path.join(input_dir,f))webp_size=os.path.getsize(out_path)print(f"{f}:{original_size//1024}KB → webp({webp_size//1024}KB)")

三、批量格式转换

defbatch_convert_format(input_dir,output_dir,target_format="png"):"""批量转换图片格式"""os.makedirs(output_dir,exist_ok=True)count=0forfinos.listdir(input_dir):name,ext=os.path.splitext(f)ifext.lower()notin(".jpg",".jpeg",".png",".bmp",".webp"):continuetry:img=Image.open(os.path.join(input_dir,f))out_path=os.path.join(output_dir,f"{name}.{target_format}")img.save(out_path)count+=1exceptExceptionase:print(f"转换失败{f}:{e}")print(f"已完成{count}张图片 →{target_format}格式")

四、批量创建缩略图

defbatch_create_thumbnails(input_dir,output_dir,size=(300,300)):"""批量创建缩略图"""os.makedirs(output_dir,exist_ok=True)forfinos.listdir(input_dir):ifnotf.lower().endswith((".jpg",".jpeg",".png")):continueimg=Image.open(os.path.join(input_dir,f))img.thumbnail(size,Image.LANCZOS)out_path=os.path.join(output_dir,f"thumb_{f}")img.save(out_path,quality=85)print(f"缩略图已生成，尺寸:{size[0]}×{size[1]}")

五、自动化工作流

defproduct_image_pipeline(input_dir,output_base):"""商品图片处理流水线"""# 1. 创建输出目录steps=["水印","压缩","缩略图","WebP"]dirs={s:os.path.join(output_base,s)forsinsteps}fordindirs.values():os.makedirs(d,exist_ok=True)# 2. 批量加水印print("步骤 1/4: 添加水印...")wm=WatermarkProcessor(input_dir,dirs["水印"])wm.add_text_watermark("版权所有",opacity=60)# 3. 批量压缩print("步骤 2/4: 压缩图片...")batch_compress(dirs["水印"],dirs["压缩"],quality=65)# 4. 创建缩略图print("步骤 3/4: 生成缩略图...")batch_create_thumbnails(dirs["水印"],dirs["缩略图"],(300,300))# 5. 转 WebPprint("步骤 4/4: 转换 WebP...")convert_to_webp(dirs["压缩"],dirs["WebP"])print(f"\n全部完成！输出目录:{output_base}")# 使用product_image_pipeline("原始图片","成品输出")

六、不同场景的压缩建议

七、处理速度对比

importtimedefbenchmark():"""对比不同处理方式的速度"""input_dir="测试图片"files=[fforfinos.listdir(input_dir)iff.endswith(".jpg")]# 单线程start=time.time()forfinfiles:img=Image.open(os.path.join(input_dir,f))img.thumbnail((800,800))print(f"单线程:{time.time()-start:.1f}s")# 多线程fromconcurrent.futuresimportThreadPoolExecutor start=time.time()withThreadPoolExecutor(max_workers=8)asexecutor:defprocess(f):img=Image.open(os.path.join(input_dir,f))img.thumbnail((800,800))executor.map(process,files)print(f"多线程:{time.time()-start:.1f}s")

八、完整工具类

classImageBatchProcessor:"""图片批量处理工具箱"""def__init__(self,input_dir,output_dir):self.input_dir=input_dir self.output_dir=output_dir os.makedirs(output_dir,exist_ok=True)defget_images(self):return[fforfinos.listdir(self.input_dir)iff.lower().endswith((".jpg",".jpeg",".png",".webp"))]defprocess(self,operations):""" 按顺序执行多个操作 operations: [("resize", {"width": 800}), ("watermark", {"text": "版权"})] """forfinself.get_images():img=Image.open(os.path.join(self.input_dir,f))forop_name,paramsinoperations:ifop_name=="resize":img.thumbnail((params["width"],params["height"]orparams["width"]))elifop_name=="watermark":# 添加水印逻辑passelifop_name=="compress":img.save(os.path.join(self.output_dir,f),quality=params.get("quality",85))continueimg.save(os.path.join(self.output_dir,f))print(f"处理完成:{f}")# 使用processor=ImageBatchProcessor("输入","输出")processor.process([("resize",{"width":1200,"height":1200}),("compress",{"quality":80}),])

💡 觉得有用的话，点赞 + 关注【张老师技术栈】吧！每周更新 Java/Python/爬虫 实战干货，不让你白来。