店群运营中，图片处理往往是被低估的耗时环节。
一个商品需要5-8张图。从供应商拿到原始图片，要裁剪、压缩、加水印、调色、转格式，然后上传到各个店铺。人工处理一张图平均2-3分钟，一个店铺每天上架50个商品，光图片就要耗费几个小时。
更麻烦的是，不同平台对图片的要求不同：拼多多要求主图800x800，TEMU要求1000x1000且不能有文字，TikTok Shop要求1:1且不能超过2MB。
我们以前是靠美工手动批量处理，但店群规模大了以后，美工团队扩到5人还是忙不过来。后来我们用Python+影刀RPA搭建了一套批量图片智能处理与优化流水线，将图片处理的效率提升了10倍，人力成本降低80%。
这篇文章不讲调度也不讲商品上架。专门聊聊店群场景下图片处理的自动化：如何从素材源自动抓取、如何按平台规范批量转换、如何与上架脚本无缝集成，以及如何优化存储和CDN加速。
适用场景：图片数量大、多平台、需要批量标准化的店群项目。 技术栈：Pillow、OpenCV、FFmpeg、阿里云OSS、影刀RPA。
一、店群图片处理的痛点
先还原一个真实场景。
运营从供应商那里拿到100套商品的图片素材，每个商品10张图，共1000张。原始图片各种尺寸：有的是2000x2000，有的是1024x768，有的是长方形。格式有JPG、PNG、甚至WEBP。
需要做的事情：
- 将所有图片统一尺寸（比如正方形1:1）
- 压缩到平台要求的文件大小（如<2MB）
- 添加店铺水印（防止盗图）
- 重命名按规范（SKU_序号.jpg）
- 上传到OSS，生成URL
- 最后把URL填入上架脚本的图片列表
手工做的话，至少两天。而且容易出错：漏掉某张图、水印位置不对、尺寸不符合平台规范被驳回。
自动化的目标是：上传原始图片到系统，一键触发处理流程，直接得到符合各平台规范的图片URL列表，供上架脚本直接使用。
二、整体架构
我们设计了一条图片处理流水线：
输入层：运营上传原始图片（ZIP包或文件夹）到NAS或OSS。
处理层：Python编排任务，调用Pillow/OpenCV进行批量处理：尺寸裁剪/缩放、压缩、水印、格式转换、重命名。
存储层：处理后的图片上传到云对象存储（OSS/S3），CDN加速。
输出层：生成每个商品对应的图片URL列表JSON，写入数据库，供影刀上架脚本读取。
流水线完全自动化，运营只需点击“开始处理”，等待完成通知。
三、原始图片的规范化采集
首先，我们需要一套规范的图片命名和目录结构，否则程序无法识别哪个文件属于哪个SKU。
运营上传的ZIP包结构示例：
batch_20250606/
├── SKU_A001/
│ ├── raw_1.jpg
│ ├── raw_2.png
│ └── raw_3.jpg
├── SKU_A002/
│ ├── raw_1.jpg
│ └── raw_2.jpg
└── manifest.csv (包含SKU、平台、需要几张图等元信息)
Python脚本扫描目录，读取manifest.csv，为每个SKU生成处理任务。
# image_pipeline/orchestrator.py
import os
import csv
from pathlib import Path
class ImagePipeline:
def __init__(self, batch_path):
self.batch_path = Path(batch_path)
self.manifest = self.load_manifest()
def load_manifest(self):
with open(self.batch_path / "manifest.csv") as f:
reader = csv.DictReader(f)
return list(reader)
def run(self):
for sku_info in self.manifest:
sku_id = sku_info["sku_id"]
platform = sku_info["platform"] # pdd / temu / tiktok
required_count = int(sku_info["image_count"])
raw_dir = self.batch_path / sku_id
raw_files = sorted(raw_dir.glob("raw_*"))
if len(raw_files) < required_count:
raise ValueError(f"{sku_id} 图片不足，需要{required_count}张，实际{len(raw_files)}")
self.process_sku(sku_id, platform, raw_files[:required_count])
四、核心图片处理模块
图片处理的核心函数封装在ImageProcessor类中，支持多种操作。
# image_processor.py
from PIL import Image
import io
import os
class ImageProcessor:
@staticmethod
def resize_to_square(img, target_size):
"""将图片缩放并裁剪成正方形"""
width, height = img.size
if width == height:
img_resized = img.resize((target_size, target_size), Image.LANCZOS)
else:
# 短边填充，长边裁剪
scale = target_size / min(width, height)
new_width = int(width * scale)
new_height = int(height * scale)
img_resized = img.resize((new_width, new_height), Image.LANCZOS)
left = (new_width - target_size) // 2
top = (new_height - target_size) // 2
img_resized = img_resized.crop((left, top, left + target_size, top + target_size))
return img_resized
@staticmethod
def compress_image(img, max_size_kb=2000, initial_quality=85):
"""压缩图片到指定大小以内（KB）"""
buf = io.BytesIO()
quality = initial_quality
img.save(buf, format='JPEG', quality=quality, optimize=True)
while buf.tell() / 1024 > max_size_kb and quality > 30:
buf = io.BytesIO()
quality -= 5
img.save(buf, format='JPEG', quality=quality, optimize=True)
return buf.getvalue()
@staticmethod
def add_watermark(img, watermark_path, position='bottom_right', opacity=0.7):
"""添加水印"""
watermark = Image.open(watermark_path).convert('RGBA')
# 调整水印大小（占原图宽度的15%）
wm_width = int(img.width * 0.15)
wm_height = int(watermark.height * wm_width / watermark.width)
watermark = watermark.resize((wm_width, wm_height), Image.LANCZOS)
# 透明度混合
if watermark.mode != 'RGBA':
watermark = watermark.convert('RGBA')
# 设置位置
if position == 'bottom_right':
x = img.width - wm_width - 10
y = img.height - wm_height - 10
elif position == 'center':
x = (img.width - wm_width) // 2
y = (img.height - wm_height) // 2
# 合并
img.paste(watermark, (x, y), watermark)
return img
不同平台的规格配置存储在JSON中：
{
"pdd": {
"size": 800,
"max_kb": 500,
"watermark": true,
"format": "JPEG",
"watermark_position": "bottom_right"
},
"temu": {
"size": 1000,
"max_kb": 2000,
"watermark": false,
"format": "JPEG"
},
"tiktok": {
"size": 1080,
"max_kb": 2000,
"watermark": true,
"format": "JPEG",
"watermark_position": "center"
}
}
处理一个SKU的函数：
def process_sku(sku_id, platform, raw_files):
spec = platform_specs[platform]
processor = ImageProcessor()
output_urls = []
for idx, raw_path in enumerate(raw_files, start=1):
img = Image.open(raw_path)
# 转换颜色模式（如果需要）
if img.mode == 'RGBA' and spec['format'] == 'JPEG':
img = img.convert('RGB')
# 缩放到正方形
img = processor.resize_to_square(img, spec['size'])
# 加水印
if spec.get('watermark'):
img = processor.add_watermark(img, 'watermark.png', spec.get('watermark_position', 'bottom_right'))
# 压缩
img_bytes = processor.compress_image(img, spec['max_kb'])
# 上传OSS
oss_key = f"images/{platform}/{sku_id}/{idx:02d}.jpg"
oss_url = upload_to_oss(img_bytes, oss_key)
output_urls.append(oss_url)
# 保存URL列表到数据库
save_image_urls(sku_id, platform, output_urls)
return output_urls
处理完成后，运营可以在后台看到每个SKU的图片URL列表，并支持预览。
五、与影刀上架脚本的集成
上架脚本需要从数据库读取图片URL列表。影刀脚本调用Python接口，传入SKU ID和平台，获取预生成的图片URL数组。
# image_api.py
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/api/get_product_images', methods=['POST'])
def get_product_images():
data = request.json
sku_id = data['sku_id']
platform = data['platform']
urls = db.query("SELECT image_url FROM product_images WHERE sku_id=%s AND platform=%s ORDER BY seq", (sku_id, platform))
return jsonify({"urls": [u[0] for u in urls]})
影刀脚本中，发送HTTP请求获取图片列表，然后遍历上传或直接使用（如果是第三方图床）。
这样，上架脚本完全不需要关心图片处理细节，图片URL已经是现成的。
六、性能优化：多线程并发处理
批量处理几百个SKU，几千张图片，串行太慢。我们使用线程池并发处理。
from concurrent.futures import ThreadPoolExecutor, as_completed
def run_batch_parallel(batch_path, max_workers=8):
pipeline = ImagePipeline(batch_path)
skus = pipeline.manifest
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_sku = {}
for sku_info in skus:
sku_id = sku_info['sku_id']
platform = sku_info['platform']
raw_files = pipeline.get_raw_files(sku_id)
future = executor.submit(process_sku, sku_id, platform, raw_files)
future_to_sku[future] = sku_id
for future in as_completed(future_to_sku):
sku_id = future_to_sku[future]
try:
urls = future.result()
results[sku_id] = urls
except Exception as e:
results[sku_id] = {'error': str(e)}
return results
在一台8核16GB的服务器上，处理1000张图片（包含缩放、水印、压缩、上传）从原来的45分钟缩短到6分钟。
七、格式兼容与异常处理
供应商提供的图片格式多种多样，有些是CMYK颜色模式的JPG，有些是带透明通道的PNG。处理时会遇到各种坑。
坑1：CMYK图片无法直接压缩
Pillow打开CMYK图片，转换为RGB再处理。
if img.mode == 'CMYK':
img = img.convert('RGB')
坑2：PNG透明背景变黑色
缩放时透明区域会变成黑色。需要在缩放前新建白色背景图层合并。
if img.mode in ('RGBA', 'LA', 'P'):
background = Image.new('RGB', img.size, (255,255,255))
background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
img = background
坑3：图片方向错误（EXIF信息）
手机拍摄的照片带有方向标签，需要根据EXIF旋转。
from PIL import ImageOps
img = ImageOps.exif_transpose(img)
我们将这些异常处理全部封装到ImageProcessor中，避免因个别图片导致整个批次失败。
八、存储与CDN加速
处理后的图片上传到云OSS，并开启CDN加速。图片URL是长期有效的，上架后无需重新上传。
为了节省成本，我们设置了OSS生命周期规则：30天后自动转低频存储，180天后删除。
图片命名规范：/{platform}/{sku_id}/{seq}.jpg，方便检索和批量删除。
当商品下架时，可以调用API批量删除OSS中的对应图片目录，节省存储费用。
九、运营反馈与迭代
这套系统上线后，运营不再需要手动处理图片。他们只需要：
- 从供应商处下载原始图片
- 按SKU分文件夹，放在指定目录
- 在后台点一下“开始处理”
半小时后收到完成通知，上架脚本可以直接使用图片URL。
运营还提出了新需求：增加“智能裁剪”功能，自动识别图片主体，裁剪时保留主体区域。我们接入了OpenCV的显著性检测，实现了简单的智能裁剪，进一步减少了手动调整的工作量。
十、总结：让图片处理不再成为瓶颈
店群自动化，不只是商品上架和订单处理。图片处理这个看似边缘的环节，实则占据了大量人力。
通过搭建批量图片智能处理流水线，我们实现了：
- 图片处理效率提升10倍
- 人力成本节省80%
- 图片格式、尺寸、水印100%合规，降低平台驳回率
- 与上架脚本无缝集成，实现从素材到上架的全自动化
如果你也深受图片处理之苦，建议从最简单的“统一尺寸+压缩”开始，逐步加入水印、格式转换等功能。不需要一次性做得很复杂，每增加一个自动化环节，都能看到明显的人效提升。
作者：林焱