python+豆包seed大模型api实现ai电影电视剧视频无限生成Bfwshoter代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python+豆包seed大模型api实现ai电影电视剧视频无限生成Bfwshoter代码，支持剧本分镜头、角色、道具、片段九宫格分镜头及视频生成，支持角色音色生成，可后期等seedance2 api开放后作为参考生成视频，一个python代码就能拥有多项目管理的ai电影电视剧无限生成框架。

代码标签： python 豆包seed 大模型 api ai 电影电视剧视频无限生成 Bfwshoter

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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

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

ARK_API_KEY = os.environ.get("ARK_API_KEY", "")
ARK_BASE_URL = os.environ.get("ARK_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
MODEL_CHAT = "doubao-1-5-pro-32k-250115"
MODEL_IMAGE = "doubao-seedream-5-0-260128"
MODEL_VIDEO = "doubao-seedance-1-5-pro-251215"

WORK_DIR = Path("workspace")
WORK_DIR.mkdir(exist_ok=True)
PROJECTS_FILE = WORK_DIR / "projects.json"
app = Flask(__name__)

def log_api_call(endpoint, req_data, res_data):
    print(f"\n{'='*20}[API CALL] {endpoint} {'='*20}")
    print(">>> REQUEST PARAMS:")
    print(json.dumps(req_data, ensure_ascii=False, indent=2) if req_data else "None")
    print("<<< RESPONSE PARAMS:")
    res_str = json.dumps(res_data, ensure_ascii=False, indent=2) if res_data else "None"
    if len(res_str) > 3000: res_str = res_str[:3000] + "\n...[TRUNCATED]..."
    print(res_str)
    print("="*60 + "\n")

def _get_target_resolution(aspect_ratio, target_pixels=4687500):
    mapping = {"16:9":16/9,"9:16":9/16,"4:3":4/3,"3:4":3/4,"1:1":1.0,"3:2":3/2}
    r = mapping.get(aspect_ratio, 16/9)
    h = int(math.sqrt(target_pixels / r)); w = int(r * h)
    ar = w/h if h else r
    if ar < 1/16: w = int(h/16)
    elif ar > 16: w = int(h*16)
    w = (w//16)*16; h = (h//16)*16
    if w == 0: w = 16
    if h == 0: h = 16
    return f"{w}x{h}"

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, (1280,720))

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

def _normalize_scene_characters(scene: dict):
    chars = scene.get("characters")
    if isinstance(chars, list) and chars: scene["character_ids"] = chars[:]
    elif scene.get("speaking_character"): scene["characters"]=[scene["speaking_character"]]; scene["character_ids"]=[scene["speaking_character"]]
    else: scene["characters"]=[]; scene["character_ids"]=[]
    if not isinstance(scene.get("props"), list): scene["props"]=[]

def _normalize_script_structure(script: dict):
    script.setdefault("title",""); script.setdefault("synopsis","")
    script.setdefault("characters",[]); script.setdefault("props",[]); script.setdefault("scenes",[])
    for scene in script.get("scenes",[]): _normalize_scene_characters(scene)
    return script

def _collect_scene_character_ids(scene):
    ids = scene.get("characters",[])[:]
    if scene.get("speaking_character"): ids.append(scene["speaking_character"])
    return list(dict.fromkeys([x for x in ids if x]))

def _collect_scene_prop_ids(scene):
    return list(dict.fromkeys([x for x in scene.get("props",[]) if x]))

def _public_or_base64_image(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
    return url

def _get_grid_label(gc):
    m={1:"单格",2:"两宫格",3:"三宫格",4:"四宫格",6:"六宫格",9:"九宫格",12:"十二宫格",16:"十六宫格"}
    return m.get(gc, f"{gc}宫格")

def call_doubao_chat(messages, model=MODEL_CHAT, temperature=0.8, response_format=None):
    url = f"{ARK_BASE_URL}/chat/completions"
    payload = {"model":model,"messages":messages,"temperature":temperature}
    if response_format: payload["response_format"]=response_format
    req = urllib.request.Request(url, data=json.dumps(payload).encode("utf-8"), method="POST")
    req.add_header("Content-Type","application/json")
    req.add_header("Authorization",f"Bearer {ARK_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 urllib.error.HTTPError as e: print(f"Chat API Error {e.code}: {e.read().decode('utf-8')}")
    except Exception as e: print(f"Chat API Error: {e}")
    return None

def _call_doubao_image(prompt, ref_images=None, size="2160x2160"):
    url = f"{ARK_BASE_URL}/images/generations"
    payload = {"model":MODEL_IMAGE,"prompt":prompt,"size":size,"output_format":"png","watermark":False}
    if ref_images and isinstance(ref_images, list) and len(ref_images)>0:
        payload["image"]=[img for img in ref_images if img]
    req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'), method="POST")
    req.add_header("Content-Type","application/json"); req.add_header("Authorization",f"Bearer {ARK_API_KEY}")
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            data = json.loads(resp.read().decode('utf-8'))
            if "data" in data and len(data["data"])>0: return data["data"][0]["url"]
    except urllib.error.HTTPError as e: print(f"Image API Error {e.code}: {e.read().decode('utf-8')}")
    except Exception as e: print(f"Image API Error: {e}")
    return None

class AIMediaAPI:
    @staticmethod
    def generate_character_image(desc, style, out_dir):
        p = out_dir / f"char_{uuid.uuid4().hex[:8]}.png"
        prompt = f"整体画风必须严格保持为：{style}。角色描述：{desc}。请生成一张角色全身正面设定图。必须是全身，正面站立。背景必须是纯白色背景，干净，无场景。"
        u = _call_doubao_image(prompt, size="2048x2048")
        if u:
            try: urllib.request.urlretrieve(u, str(p)); return str(p)
            except: pass
        _create_placeholder_image(p, f"角色图\n{desc}", 1024, 1024); return str(p)

    @staticmethod
    def generate_prop_image(desc, style, out_dir):
        p = out_dir / f"prop_{uuid.uuid4().hex[:8]}.png"
        prompt = f"整体画风必须严格保持为：{style}。道具描述：{desc}。请生成一张专业道具设定图，主体居中。背景纯白色。"
        u = _call_doubao_image(prompt, size="2048x2048")
        if u:
            try: urllib.request.urlretrieve(u, str(p)); return str(p)
            except: pass
        _create_placeholder_image(p, f"道具图\n{desc}", 1024, 1024); return str(p)

    @staticmethod
    def generate_voice_sample(voice_desc, sample_text, out_dir):
        ap = out_dir / f"voice_{uuid.uuid4().hex[:8]}.wav"
        url = f"{ARK_BASE_URL}/audio/speech"
        vd = voice_desc.strip() if voice_desc and voice_desc.strip() else "沉稳的中年男性播音员"
        st = sample_text.strip() if sample_text and sample_text.strip() else "各位听众朋友大家好这是一个很长的测试句子来满足字数要求。"
        payload = {"model":"doubao-tts","input":st,"voice":"zh_female_vv_jupiter_bigtts","response_format":"wav"}
        req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'), method="POST")
        req.add_header("Content-Type","application/json"); req.add_header("Authorization",f"Bearer {ARK_API_KEY}")
        try:
            with urllib.request.urlopen(req, timeout=60) as response:
                with open(ap,'wb') as f: f.write(response.read())
                return str(ap), f"doubao_voice_{uuid.uuid4().hex[:8]}"
        except Exception as e:
            print(f"TTS failed: {e}"); _create_placeholder_audio(ap)
            return str(ap), f"mock_voice_{uuid.uuid4().hex[:8]}"

    @staticmethod
    def generate_segment_storyboard_image(scene_text, style, ref_images, ref_hint, out_dir, aspect_ratio="16:9", grid_count=9, custom_prompt=None):
        p = out_dir / f"seg_shots_{uuid.uuid4().hex[:8]}.png"
        size_str = _get_target_resolution(aspect_ratio, 4687500)
        gl = _get_grid_label(grid_count)
        if custom_prompt: text_prompt = custom_prompt
        else:
            text_prompt = (f"请根据上传的多张参考图生成一张完整的剧情分镜排版图。\n整体画风必须严格保持为：{style}。\n"
                f"【核心排版要求】：请将以下所有分镜画面按剧情顺序，一次性排列生成在这一张图片中，"
                f"必须采用{gl}的形式来连贯将所有片段分镜放在一个图片中(确保宫格数量={grid_count})，"
                f"并在每个分镜画面的角落明显标上对应的编号(1, 2, 3...)。\n\n"
                f"【各分镜画面极其详细描述】：\n{scene_text}\n\n【角色与道具参考要求】：\n{ref_hint}\n"
                "必须极度忠实地参考上传的参考图，确保角色外貌、服装、发型一致，道具特征一致。")
        u = _call_doubao_image(text_prompt, ref_images=ref_images, size=size_str)
        if u:
            try: urllib.request.urlretrieve(u, str(p)); return str(p), text_prompt
            except: pass
        w,h = _get_resolution(aspect_ratio); _create_placeholder_image(p, text_prompt, w, h)
        return str(p), text_prompt

    @staticmethod
    def generate_segment_video_clip(img_url, video_prompt, dur, out_dir, aspect_ratio="16:9"):
        vp = out_dir / f"seg_vid_{uuid.uuid4().hex[:8]}.mp4"
        url = f"{ARK_BASE_URL}/contents/generations/tasks"
        pwp = f"{video_prompt.strip()} --duration {dur} --camerafixed false --watermark false"
        payload = {"model":MODEL_VIDEO,"content":[{"type":"text","text":pwp},{"type":"image_url","image_url":{"url":img_url.strip()}}]}
        req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'), method="POST")
        req.add_header("Content-Type","application/json"); req.add_header("Authorization",f"Bearer {ARK_API_KEY}")
        task_id = None
        try:
            with urllib.request.urlopen(req, timeout=60) as resp: task_id = json.loads(resp.read().decode('utf-8')).get("id")
        except Exception as e: print(f"Video submit error: {e}")
        if task_id:
            poll_url = f"{ARK_BASE_URL}/contents/generations/tasks/{task_id}"
            pr = urllib.request.Request(poll_url, headers={"Authorization":f"Bearer {ARK_API_KEY}"}, method="GET")
            st = time.time(); vu = None
            while time.time()-st < 900:
                try:
                    with urllib.request.urlopen(pr, timeout=30) as resp:
                        pd = json.loads(resp.read().decode('utf-8')); s = pd.get("status")
                        if s=="succeeded": vu=pd.get("content",{}).get("video_url"); break
                        elif s in["failed","canceled"]: break
                except: pass
                time.sleep(10)
            if vu:
                try: urllib.request.urlretrieve(vu, str(vp)); return str(vp)
                except: pass
        w,h = _get_resolution(aspect_ratio); _create_placeholder_video(vp, dur, w, h); return str(vp)

    @staticmethod
    def merge_videos(clips, .........完整代码请登录后点击上方下载按钮下载查看