app.py

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()