#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 带货短视频工作流 —— AutoDL ComfyUI API 客户端 流程: 1. 提交产品图(ref_image_*) + 带货提示词 -> MiniMax H3 图生视频 2. 轮询异步任务, 成功后立即下载视频(结果URL短时效, 必须马上保存) 3. 可选: 调 indextts2 生成带货配音 -> ffmpeg 合成进视频 关键约束(实测确认): - 图片必须是可公网访问的URL(jpg/png/webp)。 本地图片无法直接上传: /api/v1/file 需要浏览器登录态, token 调用返回"登录超时"。 本地图片需先传到图床/COS/OSS 拿到公开 URL。 - 结果 video URL 有效期短, 拿到立刻下载。 - 计费: 视频按 minimax_h3 档位(20 credits, 闲时 10); TTS 20 credits/次。 用法: export AUTODL_TOKEN=xxx python3 douyin_video.py --image "URL" --prompt "镜头描述" \ --duration 5 --resolution 480p竖 --narration "文案" --voice "音色URL" --out ./run """ import argparse, json, os, sys, time, pathlib, subprocess, urllib.request try: import requests except ImportError: requests = None BASE = "https://autodl.art/api/v1/comfyui" WF_IMAGE_AUDIO_VIDEO = "minimax_h3_image_audio_to_video_v2_15s" WF_TTS = "indextts2-v1" def headers(): token = os.environ.get("AUTODL_TOKEN") if not token: sys.exit("请先设置环境变量 AUTODL_TOKEN (令牌管理->创建令牌->分组选 ComfyUI)") return {"Authorization": token, "Content-Type": "application/json"} def _post(path, body): if requests is None: req = urllib.request.Request(BASE + path, data=json.dumps(body).encode(), headers=headers(), method="POST") return json.load(urllib.request.urlopen(req, timeout=60)) r = requests.post(BASE + path, json=body, headers=headers(), timeout=60) r.raise_for_status() return r.json() def _get(path): if requests is None: req = urllib.request.Request(BASE + path, headers=headers(), method="GET") return json.load(urllib.request.urlopen(req, timeout=60)) r = requests.get(BASE + path, headers=headers(), timeout=60) r.raise_for_status() return r.json() def submit(workflow_id, body): j = _post("/comfyui_workflow/" + workflow_id, body) if j.get("code") != "Success": raise RuntimeError("提交失败: %s - %s" % (j.get("code"), j.get("msg"))) tid = j["data"]["task_id"] print("[submit] workflow=%s task_id=%s" % (j["data"]["workflow"], tid), flush=True) return tid def wait(task_id, poll_sec=10, timeout=1800): url = "/comfyui_workflow/result/" + task_id t0 = time.time() while True: j = _get(url) d = j.get("data", {}) st = d.get("status") print("[poll] status=%s duration=%ss" % (st, d.get("duration")), flush=True) if st == "SUCCESS": return d if st in ("FAILED", "ERROR"): raise RuntimeError("任务失败: " + json.dumps(j, ensure_ascii=False)) if time.time() - t0 > timeout: raise TimeoutError("超时 %ss: task_id=%s" % (timeout, task_id)) time.sleep(poll_sec) def download(url, dest): pathlib.Path(dest).parent.mkdir(parents=True, exist_ok=True) print("[download] -> " + dest, flush=True) if requests is None: req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=300) as r, open(dest, "wb") as f: f.write(r.read()) else: with requests.get(url, stream=True, timeout=300) as r: r.raise_for_status() with open(dest, "wb") as f: for chunk in r.iter_content(1 << 16): f.write(chunk) return dest def ensure_public_url(u, out_dir="run"): """本地图片自动传 COS 拿公网URL; 已是https的URL原样返回。""" u = str(u) if u.startswith("http://") or u.startswith("https://"): return u import os, time, uuid as _uuid if not os.path.exists(u): raise SystemExit("图片不存在且不是URL: " + u) import cos_upload cfg = cos_upload.load_config() base = os.path.basename(u) key = "douyin/" + time.strftime("%Y%m%d") + "/" + _uuid.uuid4().hex[:8] + "_" + base cos_upload.put_object(cfg, u, key) public = "https://" + cfg["bucket"] + ".cos." + cfg["region"] + ".myqcloud.com/" + key print("[cos] 本地图片已上传: %s" % public) return public def gen_video(image_urls, prompt, duration=5, resolution="480p竖", out_dir="run"): body = {"prompt": prompt, "duration": int(duration), "resolution": resolution} for i, u in enumerate(image_urls[:9]): body["ref_image_%d" % i] = ensure_public_url(u, out_dir) tid = submit(WF_IMAGE_AUDIO_VIDEO, body) d = wait(tid) video = d["results"][0]["url"] dest = str(pathlib.Path(out_dir) / "product_video.mp4") download(video, dest) print("[done] video saved: " + dest) return dest def gen_narration(text, voice_ref_url, out_dir="run", emo_happy=0.5): body = {"prompt_text": text, "prompt_simple": voice_ref_url, "emo_control_method": "与音色参考音频相同", "emo_happy": emo_happy} tid = submit(WF_TTS, body) d = wait(tid) for res in d.get("results", []): if res.get("type") in ("audio", "video"): dest = str(pathlib.Path(out_dir) / "narration." + res.get("file_type", "mp3")) download(res["url"], dest) print("[done] narration saved: " + dest) return dest print("[warn] 未找到音频结果: " + json.dumps(d, ensure_ascii=False)) return None def mux(video_path, audio_path, bg_volume=0.4, voice_volume=1.6): out = str(pathlib.Path(video_path).with_name("final.mp4")) cmd = ["ffmpeg", "-y", "-i", video_path, "-i", audio_path, "-filter_complex", "[0:a]volume=%s[bg];[1:a]volume=%s[vo]" % (bg_volume, voice_volume), "-map", "0:v", "-map", "[bg]", "-map", "[vo]", "-c:v", "copy", "-shortest", out] subprocess.run(cmd, check=True, capture_output=True) print("[done] final: " + out) return out def main(): ap = argparse.ArgumentParser(description="带货短视频工作流") ap.add_argument("--image", action="append", required=True, help="产品图URL, 可多次(最多9张)") ap.add_argument("--prompt", required=True, help="带货提示词(主体/动作/场景/镜头)") ap.add_argument("--duration", type=int, default=5, help="1-15秒") ap.add_argument("--resolution", default="480p竖", choices=["480p竖", "768p竖", "480p横", "768p横"]) ap.add_argument("--narration", default=None, help="配音文案") ap.add_argument("--voice", default=None, help="音色参考音频URL") ap.add_argument("--out", default="run", help="输出目录") ap.add_argument("--no-mux", action="store_true", help="不合成配音") a = ap.parse_args() os.makedirs(a.out, exist_ok=True) video = gen_video(a.image, a.prompt, a.duration, a.resolution, a.out) if a.narration and a.voice: narration = gen_narration(a.narration, a.voice, a.out) if narration and not a.no_mux: mux(video, narration) else: print("[ok] video ready: " + video) if __name__ == "__main__": main()