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

日常工作中，批量处理图片几乎是个绕不开的活儿——给产品图打水印、把图片压缩到适合上传的大小、批量换个格式。要是靠手动一张张来，几百张图够折腾一整天。但用 Python 的 Pillow 库，也就是几秒钟的事，几百张图“唰”一下搞定。

一、批量添加水印

先聊聊最常用的水印功能。不管是给原创图片打上版权声明，还是给公司产品图加个品牌标识，都能用代码一次性完成。

1. 文字水印

文字水印的实现其实不复杂。核心思路是生成一个半透明文字图层，然后叠加到原图上。代码里的WatermarkProcessor类封装了整个流程，你只需要指定输入和输出目录就行。

from PIL import Image, ImageDraw, ImageFont
import os

class WatermarkProcessor:
    """批量水印处理器"""

    def __init__(self, input_dir, output_dir):
        self.input_dir = input_dir
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def add_text_watermark(self, text="版权所有", opacity=80):
        """添加文字水印"""
        for f in os.listdir(self.input_dir):
            if not f.lower().endswith((".jpg", ".jpeg", ".png")):
                continue

            img = 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/20
            font_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 = 20
            x = 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.sa ve(out_path, quality=95)
            print(f"已添加水印: {f}")

    def add_image_watermark(self, watermark_file, position="右下角"):
        """添加图片水印（如 Logo）"""
        wm = Image.open(watermark_file).convert("RGBA")

        for f in os.listdir(self.input_dir):
            if not f.lower().endswith((".jpg", ".jpeg", ".png")):
                continue

            img = Image.open(os.path.join(self.input_dir, f)).convert("RGBA")

            # 水印缩放为图片宽度的 1/5
            wm_resized = wm.resize((img.width // 5, int(wm.height * img.width // 5 / wm.width)),
                                    Image.LANCZOS)

            # 定位
            margin = 20
            if position == "右下角":
                x = img.width - wm_resized.width - margin
                y = img.height - wm_resized.height - margin
            elif position == "左上角":
                x = y = margin
            elif position == "居中":
                x = (img.width - wm_resized.width) // 2
                y = (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").sa ve(out_path, quality=95)
            print(f"已添加 Logo: {f}")

# 使用
processor = WatermarkProcessor("原始图片", "加水印")
processor.add_text_watermark("张老师技术栈")

这里有个小细节值得注意：水印文字的透明度（opacity参数）建议不要设得太高，否则会遮挡原图内容。通常60-80比较合适，既能看清版权信息，又不会喧宾夺主。

二、批量压缩图片

说到图片压缩，很多人第一反应是“质量调低就行了”。但实际上，体积和画质之间如何平衡，得看具体场景。

1. 压缩到指定质量

下面这个batch_compress函数，可以一次性处理整个目录的图片。除了设置quality参数，还可以限制最大宽度——对网页端展示来说，一个1920px的图往往就够了。

def batch_compress(input_dir, output_dir, quality=60, max_width=1920):
    """批量压缩图片"""
    os.makedirs(output_dir, exist_ok=True)

    for f in os.listdir(input_dir):
        if not f.lower().endswith((".jpg", ".jpeg", ".png")):
            continue

        filepath = os.path.join(input_dir, f)
        img = Image.open(filepath)

        # 限制最大宽度
        if img.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.sa ve(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) * 100
        print(f"{f}: {original_size//1024}KB → {compressed_size//1024}KB (压缩 {ratio:.0f}%)")

# 使用
batch_compress("产品图片", "压缩后", quality=60, max_width=1200)

从输出信息里能看到每次压缩的比例，方便你根据效果调整参数。

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

WebP 格式的好处是体积比 JPG 小很多，但画质几乎无损。如果你的目标平台支持 WebP（现在主流浏览器基本都支持了），那用它来替代 JPG 是降低带宽成本的利器。

def convert_to_webp(input_dir, output_dir, quality=80):
    """批量转换为 WebP 格式"""
    os.makedirs(output_dir, exist_ok=True)

    for f in os.listdir(input_dir):
        if not f.lower().endswith((".jpg", ".jpeg", ".png")):
            continue

        img = 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.sa ve(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)")

三、批量格式转换

有时候从设计部门拿到的图是.bmp或者.webp，但项目需要的是.png。一个脚本就能解决所有格式统一的问题。

def batch_convert_format(input_dir, output_dir, target_format="png"):
    """批量转换图片格式"""
    os.makedirs(output_dir, exist_ok=True)
    count = 0

    for f in os.listdir(input_dir):
        name, ext = os.path.splitext(f)
        if ext.lower() not in (".jpg", ".jpeg", ".png", ".bmp", ".webp"):
            continue

        try:
            img = Image.open(os.path.join(input_dir, f))
            out_path = os.path.join(output_dir, f"{name}.{target_format}")
            img.sa ve(out_path)
            count += 1
        except Exception as e:
            print(f"转换失败 {f}: {e}")

    print(f"已完成 {count} 张图片 → {target_format} 格式")

四、批量创建缩略图

缩略图是列表页、首页等场景的刚需。不需要每张图都加载原尺寸，一张300x300的缩略图就能提供足够的信息量。Pillow 自带的thumbnail方法很聪明，它会保持原图比例，不会把图片拉伸变形。

def batch_create_thumbnails(input_dir, output_dir, size=(300, 300)):
    """批量创建缩略图"""
    os.makedirs(output_dir, exist_ok=True)

    for f in os.listdir(input_dir):
        if not f.lower().endswith((".jpg", ".jpeg", ".png")):
            continue

        img = Image.open(os.path.join(input_dir, f))
        img.thumbnail(size, Image.LANCZOS)

        out_path = os.path.join(output_dir, f"thumb_{f}")
        img.sa ve(out_path, quality=85)

    print(f"缩略图已生成，尺寸: {size[0]}×{size[1]}")

五、自动化工作流

如果只做单个操作，那还算不上“批量处理”。真正的高效，是把这些步骤串成一个流水线。比如商品图片的处理流程通常会是：先加水印 → 压缩 → 生成缩略图 → 转成 WebP。每个步骤的输出目录相互独立，方便检查中间结果。

def product_image_pipeline(input_dir, output_base):
    """商品图片处理流水线"""
    # 1. 创建输出目录
    steps = ["水印", "压缩", "缩略图", "WebP"]
    dirs = {s: os.path.join(output_base, s) for s in steps}
    for d in dirs.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. 转 WebP
    print("步骤 4/4: 转换 WebP...")
    convert_to_webp(dirs["压缩"], dirs["WebP"])

    print(f"n全部完成！输出目录: {output_base}")

# 使用
product_image_pipeline("原始图片", "成品输出")

这样做的好处是：一旦流水线搭建好，往后每次只要把原始图片丢进输入目录，跑一下脚本，所有输出就自动分类保存了。

六、不同场景的压缩建议

很多人在压缩图片时犯的一个错误是“一刀切”。实际上，不同用途的图片对画质和体积的要求完全不同。这里整理了一份参考表，直接照着调参数就行。

七、处理速度对比

如果你的图片数量动辄上千，单线程处理可能有点慢。这时候可以考虑多线程加速。下面这个简单的基准对比，能让你直观看到两者差距。

import time

def benchmark():
    """对比不同处理方式的速度"""
    input_dir = "测试图片"
    files = [f for f in os.listdir(input_dir) if f.endswith(".jpg")]

    # 单线程
    start = time.time()
    for f in files:
        img = Image.open(os.path.join(input_dir, f))
        img.thumbnail((800, 800))
    print(f"单线程: {time.time() - start:.1f}s")

    # 多线程
    from concurrent.futures import ThreadPoolExecutor
    start = time.time()
    with ThreadPoolExecutor(max_workers=8) as executor:
        def process(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")

通常情况下，多线程能节省50%以上的时间。当然，max_workers可以根据你机器的CPU核心数适当调整。

八、完整工具类

最后，把前面所有功能整合到一个ImageBatchProcessor类中。这样平时用起来就更顺手了，只需要定义好操作链，一句代码就能跑完整套流程。

class ImageBatchProcessor:
    """图片批量处理工具箱"""

    def __init__(self, input_dir, output_dir):
        self.input_dir = input_dir
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def get_images(self):
        return [f for f in os.listdir(self.input_dir)
                if f.lower().endswith((".jpg", ".jpeg", ".png", ".webp"))]

    def process(self, operations):
        """
        按顺序执行多个操作
        operations: [("resize", {"width": 800}), ("watermark", {"text": "版权"})]
        """
        for f in self.get_images():
            img = Image.open(os.path.join(self.input_dir, f))

            for op_name, params in operations:
                if op_name == "resize":
                    img.thumbnail((params["width"], params["height"] or params["width"]))
                elif op_name == "watermark":
                    # 添加水印逻辑
                    pass
                elif op_name == "compress":
                    img.sa ve(os.path.join(self.output_dir, f),
                            quality=params.get("quality", 85))
                    continue

            img.sa ve(os.path.join(self.output_dir, f))
            print(f"处理完成: {f}")

# 使用
processor = ImageBatchProcessor("输入", "输出")
processor.process([
    ("resize", {"width": 1200, "height": 1200}),
    ("compress", {"quality": 80}),
])

从文字水印到格式转换，从单步操作到流水线，Pillow 这套工具链可以说覆盖了日常图片处理的绝大多数需求。代码量不大，但每段都能直接拿到项目里用。