代码语言：python

所属分类：其他

代码描述：python+千问大模型api实现ai电商带货视频生成工具Bfwprodshoter代码上传一张商品图片，选择或生成场景及主播，ai排版和台词自动生成或选择，最后生成该商品的带货视频，支持手持 商品或穿在身上进行展示，支持多项目。

代码标签： python 千问大模型 api ai 电商 带货 视频 生成 工具 Bfwprodshoter 代

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI电商带货视频生成工具Bfwprodshoter (异步轮询增强版)
"""

import os
import math
import json
import uuid
import time
import base64
import urllib.request
import urllib.error
from datetime import datetime
from pathlib import Path
from flask import Flask, request, jsonify, send_from_directory

# ============================================================
# 配置 - 阿里云 DashScope
# ============================================================
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "sk-")
DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", OPENAI_API_KEY)
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")

WORK_DIR = Path("ecom_workspace")
LIBRARY_DIR = WORK_DIR / "library_assets1"
WORK_DIR.mkdir(exist_ok=True)
LIBRARY_DIR.mkdir(exist_ok=True)
PROJECTS_FILE = WORK_DIR / "projects1.json"
LIBRARY_FILE = WORK_DIR / "library1.json"
app = Flask(__name__)

# ============================================================
# 辅助工具
# ============================================================
def _get_resolution(aspect_ratio):
    mapping = {"16:9": (1280, 720), "9:16": (720, 1280), "4:3": (1024, 768), "3:4": (768, 1024), "1:1": (1024, 1024)}
    return mapping.get(aspect_ratio, (720, 1280))

def get_base64_image_uri(local_filepath: Path) -> str:
    with open(local_filepath, "rb") as f: b64_data = base64.b64encode(f.read()).decode("utf-8")
    ext = local_filepath.suffix.lower().replace(".", "")
    if ext in ["jpg", "jpeg"]: ext = "jpeg"
    elif ext == "webp": ext = "webp"
    else: ext = "png"
    return f"data:image/{ext};base64,{b64_data}"

def _public_or_base64(project_id, url):
    if not url: return None
    if url.startswith("/workspace/"):
        lp = WORK_DIR / project_id / os.path.basename(url)
        return get_base64_image_uri(lp) if lp.exists() else None
    if url.startswith("/library_assets/"):
        lp = LIBRARY_DIR / os.path.basename(url)
        return get_base64_image_uri(lp) if lp.exists() else None
    if url.startswith("data:image/"): return url
    return url

def combine_images_to_base64(image_list, output_path: Path):
    try:
        from PIL import Image
        from io import BytesIO
        imgs =[]
        for item in image_list:
            try:
                if item.startswith("data:image"):
                    b64_data = item.split(",", 1)[1]
                    imgs.append(Image.open(BytesIO(base64.b64decode(b64_data))).convert("RGB"))
                elif os.path.exists(item):
                    imgs.append(Image.open(item).convert("RGB"))
            except: continue
        if not imgs: return None
        widths, heights = zip(*(i.size for i in imgs))
        total_width, max_height = sum(widths), max(heights)
        scale = min(2048.0 / total_width, 2048.0 / max_height) if total_width > 2048 or max_height > 2048 else 1.0
        new_width, new_height = int(total_width * scale), int(max_height * scale)
        combined = Image.new('RGB', (new_width, new_height), (255, 255, 255))
        x_offset = 0
        for img in imgs:
            nw, nh = int(img.size[0] * scale), int(img.size[1] * scale)
            combined.paste(img.resize((nw, nh), Image.Resampling.LANCZOS), (x_offset, 0))
            x_offset += nw
        combined.save(str(output_path), "JPEG", quality=85)
        return get_base64_image_uri(output_path)
    except Exception as e:
        print(f"Combine images error: {e}")
        return None

# ============================================================
# 阿里云 DashScope API
# ============================================================
def call_openai_chat(messages, model="qwen-plus"):
    req = urllib.request.Request(f"{OPENAI_BASE_URL}/chat/completions", data=json.dumps({"model": model, "messages": messages, "temperature": 0.8}).encode("utf-8"), method="POST")
    req.add_header("Content-Type", "application/json"); req.add_header("Authorization", f"Bearer {OPENAI_API_KEY}")
    try:
        with urllib.request.urlopen(req, timeout=120) as resp: return json.loads(resp.read().decode("utf-8"))["choices"][0]["message"]["content"]
    except Exception as e: 
        print(f"Chat error: {e}")
        return None

def dashscope_generate_image(prompt, size="1024*1024"):
    url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis"
    payload = {"model": "wanx2.1-t2i-turbo", "input": {"prompt": prompt[:1000]}, "parameters": {"size": size, "n": 1}}
    headers = {"Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "application/json", "X-DashScope-Async": "enable"}
    try:
        req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'), headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=30) as resp:
            task_id = json.loads(resp.read().decode('utf-8')).get("output", {}).get("task_id")
    except: return None
    if not task_id: return None
    
    start_time = time.time()
    while time.time() - start_time < 300: # Wait up to 5 mins for image
        time.sleep(5)
        try:
            req = urllib.request.Request(f"https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}", headers={"Authorization": f"Bearer {DASHSCOPE_API_KEY}"}, method="GET")
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read().decode('utf-8'))
                st = data.get("output", {}).get("task_status")
                if st == "SUCCEEDED": return data["output"]["results"][0]["url"]
                elif st in ["FAILED", "CANCELED"]: return None
        except: pass
    return None

def dashscope_multimodal_image(contents, size="1024*1024"):
    payload = {"model": "qwen-image-2.0-pro", "input": {"messages": [{"role": "user", "content": contents}]}, "parameters": {"n": 1, "prompt_extend": True, "size": size}}
    try:
        req = urllib.request.Request("https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation", data=json.dumps(payload).encode('utf-8'), headers={"Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "application/json"}, method="POST")
        with urllib.request.urlopen(req, timeout=240) as response:
            choices = json.loads(response.read().decode('utf-8')).get("output", {}).get("choices", [])
            for item in choices[0].get("message", {}).get("content", []) if choices else[]:
                if "image" in item: return item["image"]
    except: pass
    return None

# --- 新增：独立的视频异步提交与查询方法 ---
def submit_i2v_task(img_url, prompt, model="wan2.5-i2v-preview"):
    url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/video-synthesis"
    payload = {
        "model": model,
        "input": {"prompt": prompt, "img_url": img_url},
        "parameters": {"resolution": "480P", "duration": 10, "prompt_extend": True}
    }
    headers = {"Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "applic.........完整代码请登录后点击上方下载按钮下载查看

网友评论0

取消

评论提交中...

提交评论