用 Python 做本地图片批量处理：加水印/压缩/格式转换，100 张图 30 秒搞定

pip install Pillow tqdm

#!/usr/bin/env python3
"""本地图片批量处理工具：压缩 / 加水印 / 格式转换"""

import argparse
import sys
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
from PIL import Image, ImageDraw, ImageFont
from tqdm import tqdm

# 支持的格式
SUPPORTED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.tiff', '.webp'}
OUTPUT_FORMATS = {'.jpg', '.jpeg', '.png', '.webp', '.bmp'}


def compress_image(input_path: Path, output_path: Path, quality: int = 85) -> bool:
    """压缩图片，支持按比例缩小"""
    try:
        img = Image.open(input_path)
        # 转 RGBA 或 RGB，去掉透明通道（JPEG 不支持透明）
        if img.mode in ('RGBA', 'LA', 'P'):
            background = Image.new('RGB', img.size, (255, 255, 255))
            if img.mode == 'P':
                img = img.convert('RGBA')
            background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
            img = background
        elif img.mode != 'RGB':
            img = img.convert('RGB')

        # 保存时压缩
        ext = output_path.suffix.lower()
        save_kwargs = {'quality': quality, 'optimize': True}
        if ext == '.webp':
            save_kwargs['quality'] = quality
        img.save(output_path, **save_kwargs)
        return True
    except Exception as e:
        print(f"  ⚠️  压缩失败 {input_path.name}: {e}")
        return False


def add_watermark(
    input_path: Path,
    output_path: Path,
    text: str = "",
    image_path: Path = None,
    position: str = "bottom-right",
    opacity: int = 128,
    font_size: int = 24,
    margin: int = 10,
) -> bool:
    """给图片加水印，支持文字或图片水印"""
    try:
        img = Image.open(input_path).convert('RGBA')
        overlay = Image.new('RGBA', img.size, (0, 0, 0, 0))
        draw = ImageDraw.Draw(overlay)

        w, h = img.size

        if text:
            # 文字水印
            try:
                font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", font_size)
            except:
                font = ImageFont.load_default()

            # 计算文字位置
            bbox = draw.textbbox((0, 0), text, font=font)
            tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]

            positions = {
                "top-left": (margin, margin),
                "top-right": (w - tw - margin, margin),
                "bottom-left": (margin, h - th - margin),
                "bottom-right": (w - tw - margin, h - th - margin),
                "center": ((w - tw) // 2, (h - th) // 2),
            }
            x, y = positions.get(position, positions["bottom-right"])

            # 半透明效果
            color = (40, 40, 40, opacity)
            draw.text((x, y), text, font=font, fill=color)

        elif image_path and image_path.exists():
            # 图片水印
            wm = Image.open(image_path).convert('RGBA')
            # 缩放水印图（不超过原图的 20%）
            max_wm_size = max(w, h) // 5
            if wm.size[0] > max_wm_size or wm.size[1] > max_wm_size:
                ratio = max_wm_size / max(wm.size[0], wm.size[1])
                wm = wm.resize((int(wm.size[0] * ratio), int(wm.size[1] * ratio)), Image.Resampling.LANCZOS)
            # 透明化
            alpha = wm.split()[3]
            alpha = alpha.point(lambda p: int(p * opacity / 255))
            wm.putalpha(alpha)

            positions = {
                "top-left": (margin, margin),
                "top-right": (w - wm.size[0] - margin, margin),
                "bottom-left": (margin, h - wm.size[1] - margin),
                "bottom-right": (w - wm.size[0] - margin, h - wm.size[1] - margin),
                "center": ((w - wm.size[0]) // 2, (h - wm.size[1]) // 2),
            }
            x, y = positions.get(position, positions["bottom-right"])
            overlay.paste(wm, (x, y), wm)

        # 合成
        result = Image.alpha_composite(img, overlay)

        # 转 RGB 保存（如果目标是 JPEG/WebP）
        ext = output_path.suffix.lower()
        if ext in ('.jpg', '.jpeg', '.webp', '.bmp'):
            result = result.convert('RGB')

        result.save(output_path, optimize=True)
        return True
    except Exception as e:
        print(f"  ⚠️  水印失败 {input_path.name}: {e}")
        return False


def convert_format(input_path: Path, output_path: Path, target_format: str) -> bool:
    """转换图片格式"""
    try:
        img = Image.open(input_path)
        ext = output_path.suffix.lower()
        if ext in ('.jpg', '.jpeg'):
            if img.mode in ('RGBA', 'LA', 'P'):
                background = Image.new('RGB', img.size, (255, 255, 255))
                if img.mode == 'P':
                    img = img.convert('RGBA')
                background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
                img = background
            elif img.mode != 'RGB':
                img = img.convert('RGB')
        img.save(output_path, optimize=True)
        return True
    except Exception as e:
        print(f"  ⚠️  格式转换失败 {input_path.name}: {e}")
        return False


def process_batch(
    input_dir: Path,
    output_dir: Path,
    operations: list,
    max_workers: int = 4,
):
    """批量处理图片目录"""
    input_dir = input_dir.resolve()
    output_dir = output_dir.resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    # 收集所有图片
    images = [p for p in input_dir.rglob('*') if p.suffix.lower() in SUPPORTED_EXTENSIONS]
    if not images:
        print(f"❌ 在 {input_dir} 中没有找到图片文件")
        sys.exit(1)

    print(f"📂 找到 {len(images)} 张图片")
    print(f"📁 输出目录: {output_dir}")
    print(f"🔧 操作: {', '.join(operations)}")
    print("-" * 40)

    def process_one(img_path: Path) -> tuple:
        rel = img_path.relative_to(input_dir)
        output_path = output_dir / rel

        # 根据操作序列生成输出文件名
        name Stem = output_path.stem
        suffix = output_path.suffix
        target_format = None
        for op in operations:
            if op.startswith('format:'):
                target_format = op.split(':')[1].lower()
                suffix = f'.{target_format}'
                output_path = output_path.with_suffix(suffix)

        output_path.parent.mkdir(parents=True, exist_ok=True)

        # 按顺序应用操作
        current_path = img_path
        for op in operations:
            if op == 'compress' and current_path != output_path:
                new_path = output_path.with_stem(output_path.stem + '_tmp')
                if compress_image(current_path, new_path, quality=85):
                    current_path = new_path
            elif op.startswith('watermark:text:'):
                text = op.split(':', 2)[2]
                pos = "bottom-right"
                if ':' in text:
                    text, pos = text.rsplit(':', 1)
                new_path = output_path.with_stem(output_path.stem + '_tmp')
                if add_watermark(current_path, new_path, text=text, position=pos):
                    current_path = new_path
            elif op.startswith('watermark:image:'):
                wm_path = op.split(':', 2)[2]
                new_path = output_path.with_stem(output_path.stem + '_tmp')
                if add_watermark(current_path, new_path, image_path=Path(wm_path)):
                    current_path = new_path
            elif op.startswith('format:'):
                target = op.split(':')[1].lower()
                new_path = output_path.with_suffix(f'.{target}')
                if convert_format(current_path, new_path, target):
                    current_path = new_path

        # 最终移动到输出位置
        if current_path != output_path and current_path.exists():
            if output_path.exists():
                output_path.unlink()
            current_path.rename(output_path)

        orig_size = img_path.stat().st_size
        final_size = output_path.stat().st_size
        ratio = (1 - final_size / orig_size) * 100 if orig_size > 0 else 0
        return (str(output_path), orig_size, final_size, ratio)

    # 多线程处理
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_one, p): p for p in images}
        for future in tqdm(as_completed(futures), total=len(images), desc="处理中"):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"  ⚠️  处理失败: {e}")

    # 汇总
    total_orig = sum(r[1] for r in results)
    total_final = sum(r[2] for r in results)
    saved = total_orig - total_final
    print("-" * 40)
    print(f"✅ 处理完成！共 {len(results)} 张")
    print(f"📦 原始大小: {total_orig / 1024 / 1024:.1f} MB")
    print(f"📦 处理后: {total_final / 1024 / 1024:.1f} MB")
    print(f"💾 节省: {saved / 1024 / 1024:.1f} MB ({saved / total_orig * 100:.1f}%)")


def main():
    parser = argparse.ArgumentParser(
        description="本地图片批量处理工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 压缩所有图片到 80% 质量
  python image_processor.py ./photos --compress

  # 添加文字水印（右下角）
  python image_processor.py ./photos --watermark "Copyright 2026"

  # 压缩 + 加水印 + 转 WebP
  python image_processor.py ./photos --compress --watermark "MySite" --format webp

  # 只转换格式，不改压缩
  python image_processor.py ./photos --format webp
        """,
    )
    parser.add_argument("input_dir", help="输入图片目录")
    parser.add_argument("-o", "--output", help="输出目录（默认: 输入目录/output）")
    parser.add_argument("--compress", action="store_true", help="压缩图片（质量 85%）")
    parser.add_argument("--quality", type=int, default=85, help="压缩质量 1-100，默认 85")
    parser.add_argument("--watermark", "-w", help="文字水印内容")
    parser.add_argument("--wm-position", choices=["top-left", "top-right", "bottom-left", "bottom-right", "center"],
                        default="bottom-right", help="水印位置，默认右下角")
    parser.add_argument("--format", "-f", choices=["jpg", "png", "webp", "bmp"],
                        help="输出格式（默认保持原格式）")
    parser.add_argument("--workers", "-j", type=int, default=4, help="并行线程数，默认 4")
    args = parser.parse_args()

    input_dir = Path(args.input_dir)
    if not input_dir.is_dir():
        print(f"❌ 目录不存在: {input_dir}")
        sys.exit(1)

    output_dir = Path(args.output) if args.output else input_dir / "output"

    operations = []
    if args.compress:
        operations.append("compress")
    if args.watermark:
        operations.append(f"watermark:text:{args.watermark}:{args.wm_position}")
    if args.format:
        operations.append(f"format:{args.format}")

    if not operations:
        print("❌ 请至少指定一个操作: --compress / --watermark / --format")
        sys.exit(1)

    process_batch(input_dir, output_dir, operations, args.workers)


if __name__ == "__main__":
    main()

# 创建测试目录
mkdir -p ~/test_images
# 把几张图放进去

# 1. 压缩图片
python image_processor.py ~/test_images --compress --quality 75

# 2. 加水印（右下角，半透明）
python image_processor.py ~/test_images -w "© 我的网站"

# 3. 压缩 + 转 WebP（推荐，文件最小）
python image_processor.py ~/test_images --compress --format webp

# 4. 一次性完成所有操作
python image_processor.py ~/test_images --compress --quality 80 -w "NavBox" -f webp