Coze 电商商品图自动文案生成工具（实时监控版）- 部署与使用文档

一、应用场景说明

本工具专为 跨境电商、国内电商商家及电商运营人员 设计，核心用于解决“商品图片批量处理+自动生成商品文案”的高效工作需求。适用于以下场景：

• 批量处理商品图片，自动调用Coze API生成符合平台要求的商品描述文案；

• 电商商品上架前，快速完成图片预处理（压缩、编码）与文案生成，提升上架效率；

• 需要实时监控新增商品图片，无需手动重复运行脚本，新增图片自动触发处理流程；

• 电商素材批量生产工作流，实现“图片上传→自动处理→文案输出→文件归档”全自动化。

工具无需人工干预（除首次配置外），仅在人为终止程序时退出，适配电商日常批量处理、高频新增图片的工作场景。

二、工具核心功能

1. 图片自动预处理：自动压缩图片（控制尺寸与质量）、转换为带标准前缀的Base64编码，完全适配Coze API接口要求；

2. Coze API无缝集成：修复Token编码、参数格式、字段类型等常见报错，确保API调用稳定，成功生成商品文案；

3. 实时文件夹监控：持续监控input文件夹，新增图片（复制、粘贴、拖拽导入）自动触发处理流程，无需手动重启脚本；

4. 文件自动归档：处理完成的图片自动移入processed文件夹，文案结果以JSON格式保存至output文件夹，便于后续查阅与使用；

5. 防重复处理机制：记录已处理的图片文件，重启程序也会自动识别已处理文件，避免重复调用API造成资源浪费；

6. 多线程高效处理：支持3线程并行处理，批量图片处理速度更快，提升工作效率。

三、环境准备与依赖安装

3.1 环境要求

• Python版本：3.8及以上（推荐Python 3.10-3.12）；

• 运行环境：Windows、Mac、Linux均可；

• 网络要求：可正常访问Coze API（确保网络通畅，无防火墙拦截）。

3.2 依赖安装

首次运行前，需执行以下命令安装所需依赖（复制到终端/命令提示符执行）：

pip install pillow requests watchdog

依赖说明：

• pillow：用于图片压缩、格式转换与Base64编码；

• requests：用于调用Coze API，发送HTTP请求；

• watchdog：用于实时监控input文件夹的文件变化。

四、完整Python脚本（实时监控优化版）

将以下脚本复制保存为 control.py 文件，无需修改原有接口逻辑，仅需配置个人Coze API密钥即可使用。

import os
import time
import base64
import json
import shutil
import requests
import threading
from concurrent.futures import ThreadPoolExecutor
from PIL import Image
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# -------------------------- 配置项（务必替换为你的有效Token） --------------------------
# 替换为个人Coze API密钥（仅填纯Token，不要添加Bearer前缀）
COZE_API_KEY = "本人的api密钥"  # 固定API地址，无需修改
MAX_THREADS = 3  # 多线程数量，默认3线程，可根据电脑性能调整
COMPRESS_QUALITY = 85  # 图片压缩质量（1-100），85为兼顾清晰度与大小的最优值
RESIZE_MAX_SIZE = (1024, 1024)  # 图片最大尺寸（宽高），避免图片过大导致API调用失败
MONITOR_INTERVAL = 2  # 文件夹监控间隔（秒），默认2秒，可调整
processed_files = set()  # 记录已处理的图片，避免重复处理

# -------------------------- 工具核心函数（无需修改） --------------------------
def init_folders():
    """初始化文件夹（input、output、processed，不存在则自动创建）"""
    for folder in ["input", "output", "processed"]:
        if not os.path.exists(folder):
            os.makedirs(folder)
            print(f"创建文件夹: {folder}")
    # 初始化已处理文件列表（重启程序时，识别已处理的图片）
    for root, dirs, files in os.walk("input"):
        for file in files:
            image_extensions = (".jpg", ".jpeg", ".png", ".bmp", ".webp")
            if file.lower().endswith(image_extensions):
                # 检查output文件夹是否有对应JSON文案文件，有则标记为已处理
                json_file = os.path.splitext(file)[0] + ".json"
                if os.path.exists(os.path.join("output", json_file)):
                    processed_files.add(file)

def compress_and_convert_to_base64(image_path):
    """压缩图片并转换为带标准前缀的Base64编码（适配Coze API要求）"""
    try:
        with Image.open(image_path) as img:
            # 处理透明通道（避免JPEG格式报错）
            if img.mode in ("RGBA", "P"):
                img = img.convert("RGB")
            # 按比例缩放图片（不拉伸，最大尺寸不超过RESIZE_MAX_SIZE）
            img.thumbnail(RESIZE_MAX_SIZE, Image.Resampling.LANCZOS)
            # 压缩图片并保存到内存
            import io
            buffer = io.BytesIO()
            img.save(buffer, format="JPEG", quality=COMPRESS_QUALITY)
            # 生成带标准前缀的Base64编码（解决API识别失败问题）
            base64_str = base64.b64encode(buffer.getvalue()).decode("utf-8")
            base64_with_prefix = f"data:image/jpeg;base64,{base64_str}"
            return base64_with_prefix
    except Exception as e:
        print(f"❌ 图片处理失败 {image_path}: {str(e)}")
        return None

def call_coze_workflow(base64_str, image_name):
    """调用Coze API生成商品文案（修复Token编码、参数格式等问题）"""
    try:
        print(f"🔄 正在调用Coze API处理: {image_name}")

        # 修复Token编码问题：避免latin-1编码报错
        headers = {"Content-Type": "application/json; charset=utf-8"}
        auth_token = f"Bearer {COZE_API_KEY}".encode("utf-8").decode("latin-1", errors="ignore")
        headers["Authorization"] = auth_token

        # 构造API请求参数（完全适配Coze API要求）
        data = {
            "product_image": {
                "url": base64_str,  # 带前缀的Base64编码
                "file_type": "image"  # 固定值，符合API字段要求
            }
        }

        # 发送请求，强制UTF-8编码，避免中文/特殊字符报错
        response = requests.post(
            COZE_API_URL,
            headers=headers,
            json=data,
            timeout=60,
            verify=False  # 临时关闭SSL验证，避免环境兼容问题
        )
        response.raise_for_status()  # 捕获HTTP错误
        result = response.json()
        print(f"✅ Coze API调用成功: {image_name}")
        return result

    except requests.exceptions.HTTPError as e:
        print(f"❌ Coze API HTTP错误 {image_name}: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ 调用Coze API失败 {image_name}: {str(e)}")
    return None

def process_single_image(image_path):
    """单张图片完整处理流程：预处理→API调用→结果保存→文件归档"""
    image_name = os.path.basename(image_path)
    
    # 跳过已处理的图片，避免重复调用API
    if image_name in processed_files:
        return
    
    # 定义文案保存路径和图片归档路径
    json_filename = os.path.splitext(image_name)[0] + ".json"
    json_path = os.path.join("output", json_filename)
    processed_path = os.path.join("processed", image_name)

    # 1. 图片预处理：压缩+Base64编码
    base64_str = compress_and_convert_to_base64(image_path)
    if not base64_str:
        return

    # 2. 调用Coze API生成商品文案
    coze_result = call_coze_workflow(base64_str, image_name)
    if not coze_result:
        return

    # 3. 保存文案结果（JSON格式），归档处理完成的图片
    try:
        with open(json_path, "w", encoding="utf-8") as f:
            json.dump(coze_result, f, ensure_ascii=False, indent=4)
        print(f"📁 图片 {image_name} 处理完成，JSON文案已保存至: {json_path}")

        # 处理图片重名问题，避免归档时覆盖
        if os.path.exists(processed_path):
            processed_path = os.path.join(
                "processed",
                f"{os.path.splitext(image_name)[0]}_{datetime.now().strftime('%Y%m%d%H%M%S')}{os.path.splitext(image_name)[1]}"
            )
        shutil.move(image_path, processed_path)
        
        # 标记为已处理，加入已处理集合
        processed_files.add(image_name)
        
    except Exception as e:
        print(f"❌ 保存JSON文案失败 {json_path}: {str(e)}")

def process_existing_images():
    """处理input文件夹中已存在的图片（程序启动时执行）"""
    print("📌 开始处理input文件夹中已存在的图片...")
    image_extensions = (".jpg", ".jpeg", ".png", ".bmp", ".webp")
    image_files = [
        os.path.join("input", f) for f in os.listdir("input")
        if f.lower().endswith(image_extensions) and os.path.isfile(os.path.join("input", f))
    ]

    if not image_files:
        print("ℹ️ input文件夹中暂无待处理图片")
        return

    print(f"📌 共检测到 {len(image_files)} 张待处理图片，启动{MAX_THREADS}线程处理...")
    # 多线程并行处理，提升效率
    with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
        executor.map(process_single_image, image_files)

class ImageFileHandler(FileSystemEventHandler):
    """文件夹监控处理器：检测到新增图片自动触发处理"""
    def on_created(self, event):
        # 仅处理文件，不处理文件夹
        if not event.is_directory:
            file_ext = os.path.splitext(event.src_path)[1].lower()
            image_extensions = (".jpg", ".jpeg", ".png", ".bmp", ".webp")
            # 仅处理图片文件
            if file_ext in image_extensions:
                time.sleep(0.5)  # 等待文件完全写入，避免读取不完整
                print(f"\n🔍 检测到新图片: {os.path.basename(event.src_path)}")
                process_single_image(event.src_path)

def start_folder_monitor():
    """启动input文件夹实时监控，持续检测新增图片"""
    event_handler = ImageFileHandler()
    observer = Observer()
    # 监控input文件夹的文件创建事件（不递归监控子文件夹）
    observer.schedule(event_handler, path="input", recursive=False)
    observer.start()
    print(f"\n🚀 文件夹监控已启动，实时检测input文件夹新增图片（监控间隔{MONITOR_INTERVAL}秒）")
    print("⚠️  操作提示：按 Ctrl+C 可终止程序，终止前请确保所有图片处理完成\n")
    
    # 持续运行，仅在接收Ctrl+C终止信号时退出
    try:
        while True:
            time.sleep(MONITOR_INTERVAL)
    except KeyboardInterrupt:
        print("\n🛑 接收到终止信号，正在停止监控...")
        observer.stop()
    observer.join()
    print("✅ 程序已正常退出")

def main():
    """主函数：初始化→处理存量图片→启动实时监控"""
    init_folders()  # 初始化文件夹和已处理文件列表
    process_existing_images()  # 处理input文件夹中已有的图片
    start_folder_monitor()  # 启动实时监控，等待新增图片

if __name__ == "__main__":
    print("=" * 50)
    print("📢 首次运行请先安装依赖：pip install pillow requests watchdog")
    print("=" * 50 + "\n")
    
    # 检查依赖是否安装，未安装则提示并退出
    try:
        from watchdog.observers import Observer
        from watchdog.events import FileSystemEventHandler
    except ImportError:
        print("❌ 缺少watchdog库，请先执行命令安装：pip install watchdog")
        exit(1)
    
    main()
    

五、使用步骤（详细版）

1. 配置API密钥：

   • 登录Coze官网（https://www.coze.com），进入「开发者中心」→「API密钥管理」；

   • 生成新的API密钥（确保拥有「工作流调用」「图片识别」权限）；

   • 将生成的纯Token复制，替换脚本中 COZE_API_KEY 后的字符串（不要添加任何前缀，如Bearer）。

2. 准备图片文件夹：

   • 将 control.py 脚本保存到任意文件夹（如：D:\跨境优选\跨境工作流\自动生成电商文案）；

   • 运行脚本一次，会自动生成3个文件夹：input（存放待处理图片）、output（存放文案JSON文件）、processed（存放处理完成的图片）。

3. 运行脚本：

   • 打开终端/命令提示符，进入脚本所在文件夹（如：cd D:\跨境优选\跨境工作流\自动生成电商文案）；

   • 执行命令：python control.py；

   • 脚本启动后，会先处理 input 文件夹中已有的图片，处理完成后进入实时监控模式。

4. 实时处理新增图片：

   • 将需要处理的商品图片（支持jpg、jpeg、png、bmp、webp格式）复制/粘贴/拖拽到 input 文件夹；

   • 脚本会自动检测到新增图片，触发处理流程，处理完成后：

     • 图片会自动移入 processed 文件夹；

     • 生成的商品文案以JSON格式保存到 output 文件夹（文件名与图片一致，后缀为.json）。

5. 终止程序：

   • 在终端/命令提示符中，按 Ctrl+C 即可终止程序；

   • 终止前请确保所有图片处理完成（终端显示“✅ Coze API调用成功”或“📁 图片处理完成”即为处理完毕）。

六、常见问题与解决方案

七、注意事项

• 务必替换 COZE_API_KEY 为个人有效Token，否则无法调用Coze API；

• 图片格式仅支持jpg、jpeg、png、bmp、webp，其他格式会导致处理失败；

• 避免在图片处理过程中删除/移动 input 文件夹中的图片，否则会导致程序报错；

• 脚本运行期间，不要修改 output 和 processed 文件夹中的文件，避免影响防重复处理机制；

• 若网络不稳定导致API调用失败，脚本会提示报错，可重新将图片放入 input文件夹重新处理；

• 多线程数量（MAX_THREADS）可根据电脑性能调整，建议不超过5线程，避免占用过多资源。

八、工具优势

• 全自动化：从图片预处理、API调用、文案保存到文件归档，无需人工干预；

• 实时高效：新增图片自动处理，无需手动重复运行脚本，提升电商上架效率；

• 兼容性强：修复所有常见报错（编码、参数、Token等），适配Coze API最新要求；

• 易操作：无需专业编程知识，配置Token后即可运行，适合电商运营人员使用。