Create final.py

final.py

@@ -0,0 +1,92 @@
+def generate_video(image_path, prompt_text, progress=gr.Progress()):
+    """Generate video with proper polling"""
+    
+    if not API_KEY:
+        yield "❌ API Key not configured. Please add 'Key' secret.", None
+        return
+    
+    if image_path is None:
+        yield "⚠️ Please upload an image first", None
+        return
+    
+    try:
+        progress(0, desc="Preparing image...")
+        
+        # Get the Hugging Face temp URL for uploaded image
+        image_url = f"/file={image_path}"
+        print(f"Using uploaded image URL: {image_url}")
+        
+        yield "✅ Image uploaded! Creating video request...", None
+        
+        progress(0.2, desc="Creating request...")
+        
+        # Create task
+        try:
+            create_result = client.content_generation.tasks.create(
+                model="seedance-1-5-pro-251215",
+                content=[
+                    {"type": "text", "text": prompt_text},
+                    {"type": "image_url", "image_url": {"url": image_url}}
+                ]
+            )
+        except Exception as e:
+            yield f"❌ API Error: {str(e)}", None
+            return
+        
+        task_id = create_result.id
+        print(f"Task created: {task_id}")
+        yield f"✅ Task created: {task_id}", None
+        
+        progress(0.3, desc="Polling for results...")
+        
+        # Poll for results - TRY SDK FIRST, then fallback to JSON
+        attempts = 0
+        max_attempts = 120
+        used_fallback = False
+        
+        while attempts < max_attempts:
+            try:
+                # Try SDK method first
+                get_result = client.content_generation.tasks.get(task_id=task_id)
+                status = get_result.status
+                
+                if status == "succeeded":
+                    progress(1.0, desc="Complete!")
+                    video_url = get_result.content.video_url if hasattr(get_result, 'content') else None
+                    yield "✅ Video generated successfully!", video_url
+                    return
+                    
+                elif status == "failed":
+                    error_msg = get_result.error if hasattr(get_result, 'error') else "Unknown error"
+                    yield f"❌ Failed: {error_msg}", None
+                    return
+                else:
+                    progress(0.3 + (attempts/max_attempts)*0.7, desc=f"Status: {status}")
+                    yield f"⏳ Status: {status}... (attempt {attempts + 1})", None
+                    time.sleep(2)
+                    attempts += 1
+                    
+            except Exception as e:
+                # If SDK fails, use JSON fallback
+                if not used_fallback:
+                    print("SDK polling failed, using JSON fallback...")
+                    used_fallback = True
+                
+                # Poll JSON every 5 seconds
+                result = poll_via_json(task_id)
+                if result == "FAILED":
+                    yield "❌ Task failed (via JSON)", None
+                    return
+                elif result and result != "PROCESSING":
+                    yield "✅ Video generated successfully! (via JSON)", result
+                    return
+                else:
+                    progress(0.3 + (attempts/max_attempts)*0.7, desc="Status: processing (JSON)")
+                    yield f"⏳ Status: processing... (JSON fallback, attempt {attempts + 1})", None
+                    time.sleep(5)
+                    attempts += 1
+        
+        yield "⏰ Timeout after 2 minutes", None
+        
+    except Exception as e:
+        yield f"❌ Error: {str(e)}", None