python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python图片水印、压缩、格式转换自动化处理

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

作者：张老师技术栈

日常工作中经常要批量处理图片,比如给产品图加水印、压缩图片大小以便上传网站、批量转换格式,本文给大家介绍了用 Python 的 Pillow 库几秒钟搞定几百张图片,需要的朋友可以参考下

引言

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

一、批量添加水印

1. 文字水印

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.save(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").save(out_path, quality=95)
            print(f"已添加 Logo: {f}")

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

二、批量压缩图片

1. 压缩到指定质量

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.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) * 100
        print(f"{f}: {original_size//1024}KB → {compressed_size//1024}KB (压缩 {ratio:.0f}%)")

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

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

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.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)")

三、批量格式转换

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.save(out_path)
            count += 1
        except Exception as e:
            print(f"转换失败 {f}: {e}")

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

四、批量创建缩略图

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.save(out_path, quality=85)

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

五、自动化工作流

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("原始图片", "成品输出")

六、不同场景的压缩建议

用途格式质量最大宽说明
微信分享JPG80%1080压缩比高，加载快
商品详情页WebP85%1200体积比 JPG 小 30%
缩略图JPG60%300列表页快速展示
打印存档PNG95%原尺寸无损保存
朋友圈JPG85%1440画质和体积的平衡

七、处理速度对比

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")

八、完整工具类

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.save(os.path.join(self.output_dir, f),
                            quality=params.get("quality", 85))
                    continue

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

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

到此这篇关于Python自动化处理之批量图片水印、压缩、格式转换的文章就介绍到这了,更多相关Python图片水印、压缩、格式转换自动化处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文