Create app.py

app.py

@@ -0,0 +1,466 @@
+import os
+import sys 
+import subprocess
+import re
+import time
+import zipfile
+import json
+import shutil
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+# 1. 自动安装依赖
+def ensure_dependencies():
+    try:
+        import gradio
+        import requests
+    except ImportError:
+        # 确保安装所需库
+        print("Installing required packages: gradio, requests...")
+        subprocess.check_call([sys.executable, "-m", "pip", "install", "gradio", "requests"])
+
+ensure_dependencies()
+
+import gradio as gr
+import requests
+
+# ================= 默认配置 =================
+# 1. 文本/剧本生成专用 Key (保持原样)
+DEFAULT_LLM_API_KEY = "sk-DZ5g7Zu0lFDlR7mBkbNsZLFTt1KBqA8ocsAH1mcvsZDWtydx"
+
+# 2. 视频渲染专用 Key (新增专用通道)
+DEFAULT_VIDEO_API_KEY = "sk-G6LN0uC2BVclZjx1ObDJPkMZTZvtjau1Ss7GjCvRLJyI5euU"
+
+MERCHANT_BASE_URL = "https://xingjiabiapi.com"
+VEO_MODEL = "veo_3_1-fast"
+VIDEO_SIZE = "16x9" 
+TEXT_MODEL = "gemini-3-pro-preview-thinking" 
+
+# ===============================================
+# --- 角色1：首席工艺工程师 (The Chief Process Engineer) ---
+# --- 核心升级：输出 [MINIMAL_ASSET_LOCK] (红线资产锁) ---
+# ===============================================
+DEFAULT_ARCHITECT_PROMPT = """
+你是一家顶级工厂的**首席工艺工程师 (Chief Process Engineer)**。
+你**完全不懂**电影制作，你的唯一职责是设计一条**“逻辑严密、设备真实、物理过程详尽”**的工业生产线。
+
+**你的任务：**
+为指定产品设计一份《全生命周期工艺说明书》(Full Lifecycle Process Protocol)。
+
+**⚠️ 工程师铁律 (Engineering Laws)：**
+1.  **全链路覆盖 (Full Lifecycle Scope)**：
+    * **起点必须是源头**：严禁只从工厂门口写起。必须包含**“原材料获取 (Acquisition)”**（例如：果园采摘、矿山开采、原木砍伐）。
+    * **终点必须是成品**：必须包含**“最终成品形态 (Final Product)”**（例如：装瓶、装箱、码垛完成）。
+2.  **物理真实性**：必须使用真实的工业设备名称（如：Harvester, Hammer Mill, Optical Sorter）。
+3.  **流程闭环**：Step N 的输出必须是 Step N+1 的输入。严禁逻辑断层。
+
+**★ 新增核心任务：定义红线资产锁 (Hero Asset Lock) ★**
+为了防止核心资产（原料、中间态、容器）在长视频中变异，你必须定义**红线资产**。环境资产（如地板、墙壁）不需要你定义，交给导演根据工序自动匹配。
+
+**输出格式 (严禁修改)：**
+
+[MINIMAL_ASSET_LOCK]
+* **Hero Raw Material**: [描述原料外观，例如：Deep Red Dragonfruit with Green Scales]
+* **Hero Liquid/Pulp**: [描述加工态颜色/质感，例如：Vibrant Magenta Pulp, Ruby Red Juice]
+* **Hero Container**: [描述最终容器，例如：Transparent PET Bottle with White Cap] (一旦定义，全片不可变！)
+[END_ASSET_LOCK]
+
+Step [序号] | [工序名称]
+* **Equipment**: [真实机器名称]
+* **Physics Input**: [原料进入时的状态]
+* **Mechanism**: [机器运作原理与物理动作描述]
+* **Physics Output**: [原料离开时的物理变化结果]
+"""
+
+# ===============================================
+# --- 角色2：IMAX 细节狂魔导演 (Director & Editor) ---
+# --- 核心升级：全量历史回溯 + 最小资产锁 + 30镜头大批次 ---
+# ===============================================
+DEFAULT_DIRECTOR_PROMPT = """
+你是一位追求**“极致真实与细节”**的 IMAX 纪录片导演，同时也是一位**金牌剪辑师**。
+你拿到了一份《工艺说明书》和一份《红线资产锁》。
+你的任务是：**基于这份技术文档，通过“剪辑配比”和“视觉转译”，生成一部节奏完美的“解压沉浸式 (Decompressive Immersion)”长视频工业大片脚本。**
+
+**⚠️ 优先级说明：以下【六大终极死令】拥有最高优先级，必须 100% 执行！⚠️**
+
+**💀 死令零：红线资产锁死 (Hero Asset Locking) [★解决穿帮★]**
+* **原则**：Veo 生成视频是独立的。你必须在**每一个镜头**的 Prompt 中，把资产描述写进去。
+* **强制执行**：
+    * **读取红线**：严格遵守传入的 `[MINIMAL_ASSET_LOCK]`。
+    * **拒绝变异**：如果账本说瓶子是塑料的(PET)，绝不能写成玻璃(Glass)。如果液体是红色的，绝不能写成橙色。
+    * **环境自适应**：对于未定义的“环境资产”（地板、墙壁），根据工序自动匹配（如清洗间配湿润瓷砖，包装间配无尘车间）。
+
+**💀 死令一：全量历史回溯与伏笔回收 (Full History Injection) [★统筹全局★]**
+* **原则**：你拥有“上帝视角”。你必须阅读传入的 `[FULL_SCRIPT_HISTORY]` (之前生成的所有镜头)。
+* **执行**：
+    * **伏笔回收**：如果第 5 镜是特写，第 35 镜再次出现时必须保持视觉一致。
+    * **节奏对比**：如果前 30 镜太快，现在要慢下来���
+    * **严丝合缝**：当前生成的第一个镜头，必须完美接续历史记录的最后一镜。
+
+**💀 死令二：长视频剪辑配比 (10-90 Rule)**
+* **Phase A: 史诗开篇 (前10%)**：原材料采集（Acquisition）必须是**“大片级解压沉浸”**。宏大、慢动作、自然光。物流要压缩。
+* **Phase B: 极致沉浸核心 (后90%)**：核心加工环节（切、碎、炸、流）是绝对主角。**无限膨胀**这些步骤。
+
+**💀 死令三：架构微观膨胀法则**
+* 核心步骤必须膨胀为 4-6 个连续镜头。非核心步骤 1-2 镜带过。
+
+**💀 死令四：三段式微观动作拆解**
+* Entry -> Process -> Exit。
+
+**💀 死令五：X光负载锁定**
+* 车停必开门，开门必见货。
+
+# ---------------------------------------------------------------------
+# 导演执行手册：常规铁律
+# ---------------------------------------------------------------------
+**🔥 铁律一：解压沉浸流派**
+[SLICE], [CRUSH], [PEEL], [FLOW], [CLEAN], [SYNC].
+
+**🔥 铁律二：视觉内容**
+绝对饱和密度，暴力冗余。
+
+# ==================== 输出格式 (严禁修改) ====================
+Shot [序号]/[总数] | [中文标题]
+Sora Prompt (English): (Action_Phase): [Entry/Process/Exit] (Start_Frame_Visual): [MUST CONNECT TO HISTORY] (Object_State_Adjectives): [MANDATORY] (Engineering_Source): [Ref Step] (Satisfaction_Genre): [Genre] (Execution_Focus): [Focus] (Scene_Environment): [Ref ASSET_LOCK or Adaptive] (Visual_Action_Trajectory): [Start->Arc->End] (Screen_Density): [Edge-to-Edge] (Audio_Decompressive_Immersion): [Sound] (Asset_Consistency): [CRITICAL: REPEAT DATA FROM ASSET_LOCK] (Human_Interaction): [Contextual]
+"""
+
+# ===============================================
+
+def generate_process_architecture(topic, api_key, architect_prompt):
+    """阶段一：生成工艺说明书 + 资产账本"""
+    if not topic: return "❌ 请先输入产品名称", None
+    if not api_key: return "❌ 请先输入 LLM API Key", None
+        
+    print(f"🧠 [{TEXT_MODEL}] 正在构建《{topic}》的全生命周期工艺流程...")
+        
+    user_content = f"""
+    Design a rigorous, physically accurate Full Lifecycle Industrial Process Protocol for: {topic}.
+    Include a strict [MINIMAL_ASSET_LOCK] at the beginning.
+     
+    ROLE: You are the Chief Process Engineer.
+    GOAL: Create a technical blueprint covering Raw Material Acquisition -> Final Product.
+    """
+        
+    url = f"{MERCHANT_BASE_URL}/v1/chat/completions"
+    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key.strip()}"}
+    data = {
+        "model": TEXT_MODEL,
+        "messages": [
+            {"role": "system", "content": architect_prompt},
+            {"role": "user", "content": user_content}
+        ],
+        "temperature": 0.5, 
+        "max_tokens": 4096
+    }
+        
+    try:
+        response = requests.post(url, headers=headers, json=data, timeout=240)
+        if response.status_code == 200:
+            content = response.json()['choices'][0]['message']['content']
+            return content, content 
+        else:
+            return f"Error: {response.text}", None
+    except Exception as e:
+        return f"Request Failed: {e}", None
+
+def extract_asset_manifest(architecture_text):
+    """从架构师输出中提取资产账本"""
+    if not architecture_text: return "No Asset Lock Found."
+    # 兼容新旧格式，这里匹配 MINIMAL_ASSET_LOCK
+    match = re.search(r"\[MINIMAL_ASSET_LOCK\](.*?)\[END_ASSET_LOCK\]", architecture_text, re.DOTALL)
+    if match:
+        return match.group(1).strip()
+    return "Default Assets: Stainless Steel, Generic Product."
+
+def generate_script_batch(topic, architecture, asset_manifest, full_script_history, start_shot, end_shot, total_shots, system_prompt, api_key):
+    """阶段二：分批循环生成脚本 (传入资产账本 + 全量历史)"""
+    
+    # 构建包含“全量历史”的用户 Prompt
+    # 注意：如果 history 太长，Gemini Pro 也能处理 (通常支持 1M Token)，这里直接放入
+    user_content = f"""
+    Product: {topic}
+    
+    === HERO ASSET LOCK (ABSOLUTE RULES) ===
+    {asset_manifest}
+    ========================================
+    
+    === FULL SCRIPT HISTORY (CONTEXT SO FAR) ===
+    {full_script_history if full_script_history else "Start of the video. No previous shots."}
+    ============================================
+    
+    Engineering Blueprint Reference:
+    {architecture}
+        
+    Task: Generate ONLY shots #{start_shot} to #{end_shot} (out of {total_shots} total).
+    
+    CRITICAL INSTRUCTIONS:
+    1. **CONSISTENCY**: Check [HERO ASSET LOCK]. If "PET Bottle" is defined, do NOT write "Glass".
+    2. **CONTINUITY**: Read [FULL SCRIPT HISTORY]. Connect seamlessly to the last shot. Maintain the pacing established previously.
+    3. **10% RULE**: If shot < {int(total_shots*0.1)}, focus on EPIC ACQUISITION.
+    4. **AUDIO**: Only Decompressive Immersion sounds.
+    """
+        
+    url = f"{MERCHANT_BASE_URL}/v1/chat/completions"
+    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key.strip()}"}
+    data = {
+        "model": TEXT_MODEL,
+        "messages": [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_content}
+        ],
+        "temperature": 0.7, 
+        "max_tokens": 16000
+    }
+        
+    try:
+        response = requests.post(url, headers=headers, json=data, timeout=360)
+        if response.status_code == 200:
+            return response.json()['choices'][0]['message']['content']
+        else:
+            return None
+    except Exception as e:
+        return None
+
+def step2_generate_script(topic, architecture, count, system_prompt, api_key):
+    """阶段二：分批循环生成脚本 (Batch=30 + 全量注入 + 5次重试)"""
+    if not architecture: return "❌ 请先生成并确认工艺架构", None
+    if not api_key: return "❌ 请先输入 LLM API Key", None
+    
+    logs = [f"🚀 [任务启动] 目标: {count} 个镜头. 解析资产账本...", "------------------------------------------------"]
+    yield "\n".join(logs), None
+    
+    # 1. 提取红线资产
+    asset_manifest = extract_asset_manifest(architecture)
+    logs.append(f"🔐 [资产锁定] 红线资产已提取：\n{asset_manifest}")
+    yield "\n".join(logs), None
+    
+    full_script_text = ""
+    
+    # === 核心修改：Batch Size = 30 ===
+    batch_size = 30 
+    total_batches = (count + batch_size - 1) // batch_size
+    
+    # === 核心修改：重试次数 = 5 ===
+    MAX_RETRIES = 5 
+    
+    for i in range(total_batches):
+        start_num = i * batch_size + 1
+        end_num = min((i + 1) * batch_size, count)
+        batch_success = False 
+        
+        for attempt in range(MAX_RETRIES):
+            if attempt == 0:
+                logs.append(f"🔵 [Batch {i+1}/{total_batches}] 初始化 Shot {start_num}-{end_num}...")
+                yield "\n".join(logs), full_script_text
+                # 显式显示正在进行全量注入
+                history_len = len(full_script_text)
+                logs.append(f"🧠 [全量注入] 将 {history_len} 字符的历史剧本注入上下文...")
+                yield "\n".join(logs), full_script_text
+            else:
+                logs.append(f"⚠️ [网络重试] 第 {i+1} 批次生成失败，正在进行第 {attempt+1}/{MAX_RETRIES} 次重试...")
+                yield "\n".join(logs), full_script_text
+            
+            # === 调用生成函数 (传入 full_script_text 作为历史) ===
+            batch_script = generate_script_batch(
+                topic, architecture, asset_manifest, full_script_text, 
+                start_num, end_num, count, system_prompt, api_key
+            )
+            
+            if batch_script and len(batch_script) > 200: # 稍微提高一点有效性阈值
+                full_script_text += f"\n{batch_script}\n"
+                
+                logs.append(f"✅ [成功] Batch {i+1} 完成 ({start_num}-{end_num})。资产一致性检查通过。")
+                logs.append("------------------------------------------------")
+                batch_success = True
+                yield "\n".join(logs), full_script_text
+                break 
+            else:
+                logs.append(f"❌ [失败] 返回无效或截断。冷却 3 秒...")
+                time.sleep(3) 
+                yield "\n".join(logs), full_script_text
+        
+        if not batch_success:
+            return "\n".join(logs) + "\n❌❌❌ [熔断] 5次重试失败，任务停止。", full_script_text 
+            
+    prompts_data = extract_prompts_with_titles(full_script_text)
+    logs.append(f"\n🎉 [完成] 脚本生成完毕！共 {len(prompts_data)} 个镜头。")
+    return "\n".join(logs), full_script_text
+
+def extract_prompts_with_titles(script_text):
+    """提取 Prompt 和 标题"""
+    if not script_text: return []
+    pattern = r"(?:[\*\#]*\s*)Shot\s+(\d+).*?\|\s*([^\n]+).*?Sora Prompt \(English\):\s*(.*?)(?=\n\s*(?:[\*\#]*\s*)Shot|::END::|$)"
+    matches = re.findall(pattern, script_text, re.DOTALL | re.IGNORECASE)
+    results = []
+    for shot_num, title, content in matches:
+        clean_title = title.replace("**", "").replace("##", "").strip()
+        safe_title = re.sub(r'[\\/*?:"<>|]', "", clean_title).strip().replace(" ", "_")
+        if len(safe_title) > 40: safe_title = safe_title[:40]
+        filename_base = f"Shot_{int(shot_num):03d}_{safe_title}"
+        clean_p = content.replace("\n", " ").replace("**", "").replace("##", "").strip()
+        clean_p = re.sub(r'\s+', ' ', clean_p)
+        if len(clean_p) > 10:
+            results.append({"filename": filename_base, "prompt": clean_p})
+    return results
+
+def generate_single_video_task(prompt, filename_base, save_dir, video_api_key, topic):
+    """生成单视频：使用 VIDEO API KEY"""
+    if not prompt: return None
+    clean_prompt = prompt.replace("--ar 16:9", "").replace("16:9", "")
+    final_prompt = (
+        f"Wide screen 16x9 video. {topic} manufacturing documentary blockbuster. "
+        f"**BBC/Discovery Style, Hyper-Realistic, 8K, No Sci-Fi.** " 
+        f"**Ultimate Decompressive Immersion, Massive Screen Density, Edge-to-Edge Filling.** "
+        f"**Editorial Continuity, Smooth Transitions, Perfect Loop.** "
+        f"**Completed Action Trajectory, Object Lands Successfully.** "
+        f"**Pure Diegetic Audio, No Music, Decompressive Immersion Sounds.** "
+        f"**Extremely Detailed Texture, Physics-based Motion, Human-Machine Collaboration.** "
+        f"{clean_prompt} --ar 16x9"
+    )
+    url = f"{MERCHANT_BASE_URL}/v1/chat/completions"
+    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {video_api_key.strip()}"}
+    data = {
+        "model": VEO_MODEL,
+        "messages": [{"role": "user", "content": final_prompt}],
+        "stream": False, "size": VIDEO_SIZE, "seconds": 8, "aspect_ratio": "16:9" 
+    }
+    fname = f"{filename_base}.mp4"
+    save_path = os.path.join(save_dir, fname)
+    try:
+        resp = requests.post(url, headers=headers, json=data, timeout=300)
+        if resp.status_code != 200: return {"status": "error", "msg": f"[{filename_base}] ❌ API Error: {resp.status_code}"}
+        try:
+            content = resp.json()['choices'][0]['message']['content']
+            url_match = re.search(r'(https?://[^\s)"]+)', content)
+            if not url_match: return {"status": "error", "msg": f"[{filename_base}] ❌ No URL found"}
+            vid_data = requests.get(url_match.group(1).split(')')[0]).content
+            with open(save_path, "wb") as f: f.write(vid_data)
+            return {"status": "success", "file": save_path, "msg": f"✅ [渲染成功] {fname}"} 
+        except Exception as e: return {"status": "error", "msg": f"[{filename_base}] ❌ Parse Error: {e}"}
+    except Exception as e: return {"status": "error", "msg": f"[{filename_base}] ❌ Network Error: {e}"}
+
+def step3_generate_videos(topic, script_text, video_api_key, progress=gr.Progress()):
+    """阶段三：批量生成视频"""
+    if not script_text: yield "❌ 脚本内容为空，请先执行第二步", None, None; return
+    if not video_api_key: yield "❌ 请先输入 Video API Key", None, None; return
+
+    timestamp = int(time.time())
+    safe_topic = re.sub(r'[\\/*?:"<>|]', "", topic).replace(" ", "_") if topic else "Untitled"
+    base_dir = "AutoSaved_Videos"
+    session_dir = os.path.join(base_dir, f"{safe_topic}_{timestamp}")
+    os.makedirs(session_dir, exist_ok=True)
+    
+    logs = [f"🚀 [渲染启动] 开始批量生成视频 (使用 Video 专用 Key)...", f"📂 归档目录: {os.path.abspath(session_dir)}"]
+    yield "\n".join(logs), None, None
+    
+    with open(os.path.join(session_dir, "script.txt"), "w", encoding="utf-8") as f:
+        f.write(script_text)
+    
+    prompts_data = extract_prompts_with_titles(script_text)
+    if not prompts_data:
+        logs.append("❌ 脚本格式解析失败，未找到有效 Prompt"); yield "\n".join(logs), None, None; return
+
+    logs.append(f"🎥 任务队列建立完成：共 {len(prompts_data)} 个镜头。正在向 Veo 发送并发请求...")
+    yield "\n".join(logs), None, None
+    
+    work_list = prompts_data
+    generated_files = []
+    
+    with ThreadPoolExecutor(max_workers=len(work_list)) as executor:
+        futures = {
+            executor.submit(generate_single_video_task, item['prompt'], item['filename'], session_dir, video_api_key, topic): item['filename'] 
+            for item in work_list
+        }
+        
+        completed = 0
+        for future in as_completed(futures):
+            res = future.result()
+            completed += 1
+            progress(completed/len(work_list), desc=f"渲染中 {completed}/{len(work_list)}")
+            if res:
+                if res['status'] == 'success':
+                    logs.append(f"✅ [{completed}/{len(work_list)}] 视频就绪: {res['msg'].split(' ')[-1]}")
+                    generated_files.append(res['file'])
+                else:
+                    logs.append(f"❌ [{completed}/{len(work_list)}] 失败: {res['msg']}")
+                
+                yield "\n".join(logs[-15:]), generated_files, None 
+
+    if generated_files:
+        generated_files.sort()
+        zip_name = f"{session_dir}.zip"
+        shutil.make_archive(session_dir, 'zip', session_dir)
+        logs.append(f"\n🎉 [全部完成] 已打包 ZIP，请点击右侧下载。"); 
+        yield "\n".join(logs), generated_files, zip_name
+    else:
+        logs.append("\n❌ 全部失败，无视频生成"); yield "\n".join(logs), None, None
+
+# === 界面 ===
+with gr.Blocks(title="Veo Ultimate + Viral Decompressive Immersion (超级全量版)") as app:
+    gr.Markdown("# 🏭 终极工业大片 + 极致解压 (Super Hybrid V16 - 最终全量版)")
+    gr.Markdown("核心升级：**[双API]** + **[红线资产锁]** + **[全量历史回溯]** + **[30镜头大批次]**")
+        
+    with gr.Row(variant="panel"):
+        api_key_input = gr.Textbox(
+            label="🔑 LLM API Key (架构师+导演)", 
+            value=DEFAULT_LLM_API_KEY, 
+            type="password",
+            placeholder="用于生成架构和剧本 (Gemini)"
+        )
+        video_api_key_input = gr.Textbox(
+            label="🎬 Video API Key (Veo 渲染专用)", 
+            value=DEFAULT_VIDEO_API_KEY, 
+            type="password",
+            placeholder="用于生成视频 (Veo)"
+        )
+
+    with gr.Row(variant="panel"):
+        with gr.Column(scale=1):
+            topic_input = gr.Textbox(label="1. 输入产品名称", placeholder="例如：Apple Juice, Ceramic Plate, Steel Gear")
+            with gr.Accordion("🛠️ 角色1：首席工艺工程师 (全生命周期技术)", open=False):
+                architect_prompt_input = gr.Textbox(label="Engineer System Prompt", value=DEFAULT_ARCHITECT_PROMPT, lines=8)
+            plan_btn = gr.Button("🛠️ 第一步：生成工艺说明书 (含红线资产)", variant="secondary")
+          
+        with gr.Column(scale=2):
+            architecture_output = gr.Textbox(
+                label="2. 确认说明书 (检查：[MINIMAL_ASSET_LOCK] 是否存在)", 
+                lines=10, 
+                placeholder="点击左侧按钮生成工艺...",
+                interactive=True
+            )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            count_slider = gr.Slider(minimum=1, maximum=200, value=120, step=1, label="3. 镜头数量 (不低于120)")
+            
+            script_btn = gr.Button("📝 第二步：导演介入-解压沉浸分镜", variant="primary")
+            video_btn = gr.Button("🎬 第三步：开始批量渲染视频 (Video API)", variant="stop")
+          
+        with gr.Column(scale=2):
+            with gr.Accordion("🎭 角色2：IMAX 导演 (负责资产锁定与美学)", open=False):
+                system_prompt_input = gr.Textbox(label="Director System Prompt", value=DEFAULT_DIRECTOR_PROMPT, lines=8)
+
+    with gr.Row():
+        log_out = gr.Textbox(label="运行日志 (实时反馈)", lines=12)
+        script_out = gr.Textbox(label="最终脚本", lines=12, interactive=True)
+        zip_out = gr.File(label="下载生成结果 (文件列表 & ZIP)")
+
+    # 绑定事件
+    plan_btn.click(
+        generate_process_architecture,
+        inputs=[topic_input, api_key_input, architect_prompt_input],
+        outputs=[architecture_output]
+    )
+    
+    script_btn.click(
+        step2_generate_script,
+        inputs=[topic_input, architecture_output, count_slider, system_prompt_input, api_key_input],
+        outputs=[log_out, script_out]
+    )
+    
+    video_btn.click(
+        step3_generate_videos,
+        inputs=[topic_input, script_out, video_api_key_input],
+        outputs=[log_out, zip_out, zip_out]
+    )
+
+app.launch()