批量处理商品图：调用豆包的视觉理解，生成标题、卖点和图中文字，写入 CSV，标记为“待人工审核”后再导入店铺后台

批量处理商品图：调用豆包的视觉理解，生成标题、卖点和图中文字，写入 CSV，标记为“待人工审核”后再导入店铺后台。重点是发送前先按方舟文档的限制处理图片：用 Pillow 识别真实格式（不看扩展名），像素超标就等比缩小，体积超过 10MB 就转成 JPEG 压缩。

先看清成品

• 测试： pytest 2 项通过。
• 在模拟服务器上跑完 4 张测试图：
  • 识别出一张扩展名是 .png、实际是 JPEG 的图片。
  • 一张 8000×5000 的图被缩小到 1900×1187。
• 注意： 标题和卖点是模拟服务器返回的固定假数据，不代表豆包的真实效果。

没有在 box 上跑的部分： box 上没有方舟 API Key，没有调用过豆包 / 火山方舟的真实接口。代码按官方文档的 Chat API（OpenAI 兼容，base_url https://ark.cn-beijing.volces.com/api/v3）编写，测试时把 ARK_BASE_URL 指向两种替身：

• 本地 Ollama。
• 我自己写的模拟服务器：它只照着文档检查了请求格式，返回的内容是固定的假数据。

合规提醒：方舟 API Key（ARK_API_KEY）只放在环境变量里，不要写进网页前端。商品图、用户评论和聊天记录都可能含有个人信息，上传前要确认有权使用、做好脱敏。AI 生成的商品描述和客服回答必须经过人工审核才能发布，不能写夸大或无法证实的功效（《广告法》）。对公众提供生成式 AI 服务，需要按《生成式人工智能服务管理暂行办法》《人工智能生成合成内容标识办法》等规定履行相应义务。价格以方舟控制台的“模型定价”为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文档里的限制（截至本文写作时，“图片理解”页）：

• 用 base64 传入时，单张图片小于 10MB，整个请求体不能超过 64MB；用 Files API 上传，最大 512MB。
• data URL 里的 mime_type 要和真实格式对应。
• 像素数和 detail 档位的具体数值见文档，代码里的常量按文档填写。

文件结构

43-doubao-A-src/
├── ark_vision.py
├── fake_ark.py
└── test_vision.py

第 1 步：代码

ark_vision.py：

"""商品图批量生成标题和卖点（豆包视觉理解，火山方舟 Chat API）。结果写入 CSV，方便导入店铺后台前人工审核。
图片预处理（按方舟“图片理解”文档的限制）：
  - 像素数必须在 [196, 36,000,000] 之间，否则接口直接报错 → 过大的先等比缩小
  - base64 传入时单张图 < 10 MB → 超过就转成 JPEG 压缩
  - data URL 里的 mime_type 必须和真实格式一致 → 用 Pillow 识别真实格式，不看扩展名
环境变量：ARK_API_KEY、ARK_BASE_URL（默认 https://ark.cn-beijing.volces.com/api/v3）、ARK_MODEL（默认 doubao-seed-2-1-lite-260915）
用法：python ark_vision.py 图片目录 输出.csv"""
import base64, csv, io, json, os, sys
from pathlib import Path
from openai import OpenAI
from PIL import Image

client = OpenAI(base_url=os.environ.get("ARK_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"),
                api_key=os.environ.get("ARK_API_KEY") or "missing-key")
MODEL = os.environ.get("ARK_MODEL", "doubao-seed-2-1-lite-260915")
MAX_PIXELS, MIN_PIXELS, MAX_BYTES = 2_257_920, 196, 10 * 1024 * 1024   # 2257920 = detail=high 的像素上限，再大也只是多传数据
PROMPT = ('这是一张商品图。输出 json：{"title": "不超过 20 字的商品标题", "points": ["卖点1", "卖点2", "卖点3"], '
          '"text_in_image": "图中能看到的文字，没有就填空字符串"}。只描述图里看得到的东西，看不出来的材质、尺寸不要编。')

def to_data_url(path):
    im = Image.open(path); fmt = im.format; w, h = im.size; notes = []
    if Path(path).suffix.lower().lstrip(".").replace("jpg", "jpeg") != fmt.lower(): notes.append(f"扩展名是 {Path(path).suffix}，真实格式是 {fmt}")
    if w * h < MIN_PIXELS: raise ValueError(f"像素太少（{w}x{h}）")
    if w * h > MAX_PIXELS:
        s = (MAX_PIXELS / (w * h)) ** 0.5; im = im.resize((int(w * s), int(h * s))); notes.append(f"缩小 {w}x{h}→{im.size[0]}x{im.size[1]}")
    buf = io.BytesIO()
    if fmt not in ("JPEG", "PNG", "WEBP") or notes: fmt = "JPEG"; im.convert("RGB").save(buf, "JPEG", quality=88)
    else: buf.write(Path(path).read_bytes())
    if buf.tell() > MAX_BYTES:
        buf = io.BytesIO(); im.convert("RGB").save(buf, "JPEG", quality=75); fmt = "JPEG"; notes.append("压缩为 JPEG")
    return f"data:image/{fmt.lower()};base64,{base64.b64encode(buf.getvalue()).decode()}", notes

def describe(path):
    url, notes = to_data_url(path)
    r = client.chat.completions.create(model=MODEL, response_format={"type": "json_object"},
        extra_body={"thinking": {"type": "disabled"}},                    # 简单描述任务关掉深度思考，省时间和 tokens
        messages=[{"role": "user", "content": [{"type": "image_url", "image_url": {"url": url, "detail": "low"}},
                                               {"type": "text", "text": PROMPT}]}])
    return json.loads(r.choices[0].message.content), notes, r.usage

if __name__ == "__main__":
    files = sorted(p for p in Path(sys.argv[1]).iterdir() if p.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp"))
    with open(sys.argv[2], "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f); w.writerow(["文件", "标题", "卖点", "图中文字", "预处理", "审核状态"])
        for p in files:
            try:
                d, notes, u = describe(p)
                w.writerow([p.name, d.get("title", ""), " / ".join(d.get("points", [])), d.get("text_in_image", ""), "；".join(notes), "待人工审核"])
                print(f"✓ {p.name}  {d.get('title')}  {'；'.join(notes)}  （tokens {u.prompt_tokens}+{u.completion_tokens}）")
            except Exception as e:
                w.writerow([p.name, "", "", "", "", f"失败：{e}"]); print(f"✗ {p.name}  {e}")
    print(f"已写入 {sys.argv[2]}")

fake_ark.py（仅用于离线测试）：

"""本地模拟方舟 Chat API（只用于没有 API key 时测试流程）：按方舟文档的限制检查图片
（data URL 的 mime 与真实格式一致、单图 < 10 MB、像素在 [196, 36000000]），通过后返回固定的示例 JSON。
注意：这里的报错文字是我写的，不是方舟的原文；它也不会真的“看图”。
用法：python fake_ark.py   （监听 127.0.0.1:18200）"""
import base64, io, json
from flask import Flask, jsonify, request
from PIL import Image

app = Flask(__name__)

def err(msg): return jsonify(error={"code": "FakeInvalidParameter", "message": msg}), 400

@app.post("/api/v3/chat/completions")
def chat():
    if not request.headers.get("Authorization", "").startswith("Bearer "): return jsonify(error={"message": "missing key"}), 401
    for m in request.get_json()["messages"]:
        for part in m["content"] if isinstance(m["content"], list) else []:
            if part["type"] != "image_url": continue
            head, b64 = part["image_url"]["url"].split(",", 1)
            raw = base64.b64decode(b64); im = Image.open(io.BytesIO(raw))
            if head != f"data:image/{im.format.lower()};base64": return err(f"mime 与真实格式不符：{head} vs {im.format}")
            if len(raw) >= 10 * 1024 * 1024: return err("图片超过 10 MB")
            if not 196 <= im.size[0] * im.size[1] <= 36_000_000: return err(f"像素数 {im.size[0]*im.size[1]} 超出范围")
    content = json.dumps({"title": "（模拟）示例商品", "points": ["模拟卖点1", "模拟卖点2", "模拟卖点3"], "text_in_image": ""}, ensure_ascii=False)
    return jsonify(id="fake", object="chat.completion", model="fake", choices=[{"index": 0, "finish_reason": "stop",
                   "message": {"role": "assistant", "content": content}}], usage={"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2})

if __name__ == "__main__":
    app.run(port=18200)

test_vision.py：

import base64, io, os
os.environ.setdefault("ARK_API_KEY", "test")
from PIL import Image
from ark_vision import to_data_url, MAX_PIXELS

def test_mime_follows_real_format_not_extension(tmp_path):
    p = tmp_path / "x.png"; Image.new("RGB", (100, 100)).save(p, "JPEG")
    url, notes = to_data_url(p)
    assert url.startswith("data:image/jpeg;base64,") and "真实格式是 JPEG" in notes[0]

def test_big_image_resized(tmp_path):
    p = tmp_path / "big.png"; Image.new("RGB", (8000, 5000)).save(p)
    url, _ = to_data_url(p)
    im = Image.open(io.BytesIO(base64.b64decode(url.split(",", 1)[1])))
    assert im.size[0] * im.size[1] <= MAX_PIXELS

第 2 步：运行

$ export ARK_API_KEY=...            # 真实使用
$ python ark_vision.py 商品图/ 结果.csv
# box 上用模拟服务器：python fake_ark.py 后设置 ARK_BASE_URL=http://127.0.0.1:18200/api/v3
✓ bag.jpg  （模拟）示例商品    （tokens 1+1）
✓ cup_wrong_ext.png  （模拟）示例商品  扩展名是 .png，真实格式是 JPEG  （tokens 1+1）
✓ lamp.png  （模拟）示例商品  缩小 8000x5000→1900x1187  （tokens 1+1）
✓ mug.jpg  （模拟）示例商品    （tokens 1+1）
已写入 结果.csv

结果.csv 的最后一列都是“待人工审核”，“预处理”列记录了程序对每张图做过的处理。

调试示例：扩展名和真实格式不一致

第一版按扩展名决定 mime 类型。cup_wrong_ext.png 实际上是一张 JPEG，于是请求里写的是 data:image/png。我的模拟服务器按文档“mime_type 需要与文件格式对应”的要求做了检查，返回了 400。改成用 Pillow 的 Image.open().format 识别真实格式后就正常了。

说明：真实的方舟接口遇到格式不一致时会怎么处理，我没有验证过，因为没有 Key。按文档的要求发送就不会有问题。

常见问题

模板规则

• 发送前按文档的限制处理图片。
• 格式以文件内容为准，不看扩展名。
• 生成的文案一律经过人工审核才能发布。

项目资源

先把图片处理到符合文档要求，再交给模型，批量处理时就不会因为一张图卡住整批任务。