Create app.py

app.py

@@ -0,0 +1,99 @@
+import gradio as gr
+import requests
+import base64
+import time
+
+# --- 核心配置 ---
+API_KEY = "zz2026022"
+# 根据截图，Base URL 是 http://<IP>:<端口>/flow/v1
+BASE_URL = "http://154.40.59.124:8000/flow/v1" 
+
+def encode_image(image_path):
+    """将本地图片路径转换为 Base64 字符串"""
+    if not image_path:
+        return None
+    with open(image_path, "rb") as f:
+        encoded = base64.b64encode(f.read()).decode('utf-8')
+        return f"data:image/png;base64,{encoded}"
+
+def generate_video_task(prompt, start_img_path, end_img_path):
+    if not start_img_path:
+        return None, "错误：请至少上传起始帧图片。"
+
+    headers = {
+        "Authorization": f"Bearer {API_KEY}", #
+        "Content-Type": "application/json"
+    }
+
+    # 1. 转换图片为 Base64
+    start_b64 = encode_image(start_img_path)
+    end_b64 = encode_image(end_img_path) if end_img_path else None
+
+    # 2. 构造符合 /v1/videos 接口要求的 payload
+    input_refs = [start_b64]
+    if end_b64:
+        input_refs.append(end_b64) # 传入 [起始帧, 结束帧] 列表
+
+    payload = {
+        "prompt": prompt if prompt else "Generate transition video", #
+        "model": "veo_3_1", # 使用文档指定的模型名称
+        "input_reference": input_refs # 接口参数名为 input_reference
+    }
+
+    try:
+        # 3. 提交生成任务 (POST /v1/videos)
+        response = requests.post(f"{BASE_URL}/videos", json=payload, headers=headers)
+        res_data = response.json()
+
+        if response.status_code != 200:
+            return None, f"任务提交失败: {res_data}"
+
+        video_id = res_data.get("id")
+        if not video_id:
+            return None, "未获取到任务 ID，请检查接口返回格式。"
+
+        # 4. 轮询视频生成状态 (GET /v1/videos/{video_id})
+        status_url = f"{BASE_URL}/videos/{video_id}"
+        for _ in range(60):  # 最多等待 300 秒
+            time.sleep(5)
+            status_res = requests.get(status_url, headers=headers)
+            status_data = status_res.json()
+            
+            # 这里的状态判断逻辑需根据您的 API 实际返回字段名进行适配（如 status 或 state）
+            status = status_data.get("status") 
+            if status == "completed" or status == "succeeded":
+                return status_data.get("url"), f"生成成功！ID: {video_id}"
+            elif status == "failed":
+                return None, f"视频生成失败: {status_data.get('error')}"
+            
+            print(f"正在生成中... 状态: {status}")
+
+        return None, "任务超时，请稍后重试。"
+
+    except Exception as e:
+        return None, f"程序异常: {str(e)}"
+
+# --- Gradio 交互界面 ---
+with gr.Blocks(title="首尾帧视频生成测试") as demo:
+    gr.Markdown("## 🎬 视频接口生成测试 (/v1/videos)")
+    gr.Markdown("当前使用模型：`veo_3_1`")
+    
+    with gr.Row():
+        with gr.Column():
+            prompt_input = gr.Textbox(label="视频提示词 (Prompt)", value="画小猫") #
+            start_img = gr.Image(label="起始帧 (Start Frame)", type="filepath")
+            end_img = gr.Image(label="结束帧 (End Frame)", type="filepath")
+            btn = gr.Button("提交视频生成任务", variant="primary")
+        
+        with gr.Column():
+            video_out = gr.Video(label="生成结果")
+            info_out = gr.Textbox(label="任务状态信息", interactive=False)
+
+    btn.click(
+        fn=generate_video_task,
+        inputs=[prompt_input, start_img, end_img],
+        outputs=[video_out, info_out]
+    )
+
+if __name__ == "__main__":
+    demo.launch()