python+千问api实现AI电商带货视频生成工具Bfwprodshoter 1.1代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python+千问api实现AI电商带货视频生成工具Bfwprodshoter 1.1代码，修复之前版本问题，实现了多项目管理、商品库、模特库、场景库的生成或上传或选择，支持语料的自动生成与选择，支持视频的使用标记等等功能。

代码标签： python 千问api AI 电商带货视频生成工具 Bfwprodshoter 1.1 代

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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI电商带货视频生成工具Bfwprodshoter 1.1
"""

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/"):
        # 【修复跨项目引用】：从 URL /workspace/pid/xxx.png 中提取真实的 pid
        parts = url.split('/')
        if len(parts) >= 4:
            lp = WORK_DIR / parts[2] / parts[3]
        else:
            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": "application/json", "X-DashScope-Async": "enable"}
    try:
        req = urllib.re.........完整代码请登录后点击上方下载按钮下载查看