Create ups.py

ups.py

@@ -0,0 +1,450 @@
+import os
+import time
+import gradio as gr
+from byteplussdkarkruntime import Ark
+import requests
+from datetime import datetime
+
+# Get API key from Hugging Face secret "Key" and CLEAN it
+API_KEY = os.environ.get("Key", "").strip()
+print(f"✅ Key loaded, length: {len(API_KEY)}")
+
+# Initialize client with proxy
+client = Ark(
+    base_url="https://1hit.no/proxy/proxy.php",
+    api_key=API_KEY,
+    timeout=30.0,
+    max_retries=3,
+)
+
+# Fresh new prompts
+DEFAULT_PROMPTS = {
+    "Cinematic Nature": "Majestic drone shot soaring above misty mountains at golden hour, sunlight breaking through clouds, cinematic 4k quality, smooth motion --duration 5 --camerafixed false",
+    "Urban Exploration": "Dynamic drone flight through narrow alleyways of a neon-lit Tokyo street at night, rain-slicked surfaces reflecting lights, cyberpunk aesthetic --duration 5 --camerafixed false",
+    "Action Sequence": "High-speed drone chasing a sports car through a winding coastal road, dramatic cliffside views, action movie style --duration 5 --camerafixed false",
+    "Abstract Art": "Surreal drone flight through floating geometric shapes in a dreamlike void, pastel colors, ethereal atmosphere --duration 5 --camerafixed false",
+    "Wildlife Documentary": "Drone following a herd of elephants across the African savanna at sunset, National Geographic style, majestic wildlife cinematography --duration 5 --camerafixed false",
+    "Sports Highlight": "Drone racing alongside professional skiers carving through fresh powder in the Alps, high-energy sports broadcast style --duration 5 --camerafixed false",
+    "Beach Paradise": "Drone gliding over turquoise waters and white sand beaches, palm trees swaying, tropical paradise aerial view --duration 5 --camerafixed false",
+    "Ancient Temple": "Drone circling around ancient Angkor Wat temple at dawn, mystical atmosphere, historical documentary style --duration 5 --camerafixed false"
+}
+
+def poll_via_json(task_id):
+    """Check io.json for task status"""
+    json_url = "https://1hit.no/proxy/io.json"
+    try:
+        response = requests.get(json_url)
+        data = response.json()
+        if task_id in data:
+            task_data = data[task_id]
+            status = task_data.get('status')
+            if status == 'succeeded':
+                return task_data.get('video_url')
+            elif status == 'failed':
+                return "FAILED"
+            else:
+                return "PROCESSING"
+    except:
+        pass
+    return None
+
+def generate_video(image_path, manual_url, prompt_text, progress=gr.Progress()):
+    """Generate video with proper polling and URL override"""
+    
+    if not API_KEY:
+        yield "❌ API Key not configured. Please add 'Key' secret.", None
+        return
+    
+    # Determine which URL to use
+    if manual_url and manual_url.strip():
+        image_url = manual_url.strip()
+        source = "manual URL"
+    elif image_path:
+        image_url = f"/file={image_path}"
+        source = "uploaded image"
+    else:
+        yield "⚠️ Please upload an image or enter a URL", None
+        return
+    
+    try:
+        progress(0, desc="Preparing image...")
+        print(f"Using {source}: {image_url}")
+        yield f"✅ Using {source}...", 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
+
+def update_prompt(choice):
+    return DEFAULT_PROMPTS.get(choice, "")
+
+# MANUAL POLLING FUNCTIONS
+def manual_poll(task_id):
+    """Manually poll a specific task and update io.json"""
+    if not task_id:
+        return "❌ Please enter a task ID"
+    
+    try:
+        # Call proxy to poll this specific task
+        url = f"https://1hit.no/proxy/proxy.php?task_id={task_id}"
+        headers = {"Authorization": f"Bearer {API_KEY}"}
+        
+        response = requests.get(url, headers=headers)
+        
+        # Also fetch updated io.json
+        io_response = requests.get("https://1hit.no/proxy/io.json")
+        io_data = io_response.json()
+        
+        if task_id in io_data:
+            status = io_data[task_id].get('status')
+            video_url = io_data[task_id].get('video_url')
+            
+            result = f"✅ Task {task_id}\n"
+            result += f"Status: {status}\n"
+            if video_url:
+                result += f"Video URL: {video_url}\n"
+            if 'response' in io_data[task_id]:
+                result += f"Full response: {io_data[task_id]['response']}\n"
+            return result
+        else:
+            return f"❌ Task {task_id} not found in io.json"
+            
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
+
+def get_raw_json():
+    """Fetch raw io.json"""
+    try:
+        r = requests.get("https://1hit.no/proxy/io.json")
+        return r.json()
+    except:
+        return {"error": "Could not fetch io.json"}
+
+# WATCH & DOWNLOAD FUNCTIONS
+def get_task_list():
+    """Get list of all tasks from io.json"""
+    try:
+        r = requests.get("https://1hit.no/proxy/io.json")
+        data = r.json()
+        tasks = []
+        for task_id, task_data in data.items():
+            status = task_data.get('status', 'unknown')
+            created = task_data.get('created', 0)
+            # Format time
+            time_str = datetime.fromtimestamp(created).strftime('%Y-%m-%d %H:%M:%S')
+            
+            # Get prompt preview
+            prompt = task_data.get('request', {}).get('content', [{}])[0].get('text', 'No prompt')
+            prompt_preview = prompt[:50] + '...' if len(prompt) > 50 else prompt
+            
+            tasks.append({
+                "id": task_id,
+                "status": status,
+                "created": time_str,
+                "created_ts": created,
+                "video_url": task_data.get('video_url', ''),
+                "prompt": prompt_preview
+            })
+        # Sort by created time, newest first
+        tasks.sort(key=lambda x: x['created_ts'], reverse=True)
+        return tasks
+    except Exception as e:
+        print(f"Error getting task list: {e}")
+        return []
+
+def refresh_task_list():
+    """Refresh the task list dropdown"""
+    tasks = get_task_list()
+    choices = [f"{t['id']} | {t['status']} | {t['created']}" for t in tasks]
+    return gr.Dropdown(choices=choices, value=choices[0] if choices else None)
+
+def load_task(selection):
+    """Load selected task details"""
+    if not selection:
+        return "No task selected", None, None
+    
+    task_id = selection.split(' | ')[0]
+    try:
+        r = requests.get("https://1hit.no/proxy/io.json")
+        data = r.json()
+        task = data.get(task_id, {})
+        
+        video_url = task.get('video_url', '')
+        status = task.get('status', 'unknown')
+        
+        # Get full prompt
+        prompt = task.get('request', {}).get('content', [{}])[0].get('text', 'No prompt')
+        
+        # Get created time
+        created = task.get('created', 0)
+        created_str = datetime.fromtimestamp(created).strftime('%Y-%m-%d %H:%M:%S')
+        
+        # Get response data if available
+        response = task.get('response', {})
+        
+        info = f"### Task Details\n\n"
+        info += f"**Task ID:** `{task_id}`\n"
+        info += f"**Status:** `{status}`\n"
+        info += f"**Created:** {created_str}\n"
+        info += f"**Prompt:** {prompt}\n\n"
+        
+        if status == 'succeeded' and video_url:
+            info += f"✅ **Video ready!**\n"
+            info += f"**Download:** [Click here]({video_url})\n"
+        elif status == 'failed':
+            error = task.get('response', {}).get('error', 'Unknown error')
+            info += f"❌ **Failed:** {error}\n"
+        
+        return info, video_url, video_url
+    except Exception as e:
+        return f"Error loading task: {e}", None, None
+
+def update_url_display(image_path, manual_url):
+    """Update the URL display based on inputs"""
+    if manual_url and manual_url.strip():
+        return manual_url.strip()
+    elif image_path:
+        return f"/file={image_path}"
+    else:
+        return "No image selected"
+
+# Create the interface
+with gr.Blocks(title="BytePlus Video Generator", theme=gr.themes.Soft()) as demo:
+    
+    gr.Markdown("# 🎥 BytePlus Video Generator")
+    
+    with gr.Row():
+        if API_KEY:
+            gr.Markdown("✅ **API Key:** Configured")
+        else:
+            gr.Markdown("❌ **API Key:** Not configured - please add 'Key' secret")
+    
+    # Create tabs
+    with gr.Tabs():
+        # Tab 1: Generate Video (UPDATED with URL display)
+        with gr.TabItem("🎬 Generate Video"):
+            with gr.Row():
+                with gr.Column():
+                    image_input = gr.Image(
+                        label="Upload Starting Image",
+                        type="filepath",
+                        height=300
+                    )
+                    
+                    # URL override section
+                    with gr.Accordion("🔗 Image URL Override", open=False):
+                        image_url_input = gr.Textbox(
+                            label="Enter image URL directly",
+                            placeholder="https://example.com/image.jpg",
+                            info="Leave empty to use uploaded image"
+                        )
+                        gr.Markdown("💡 *Using a URL will override the uploaded image*")
+                    
+                    # Current URL display
+                    with gr.Row():
+                        current_url_display = gr.Textbox(
+                            label="Current Image URL",
+                            value="No image selected",
+                            interactive=False,
+                            lines=2
+                        )
+                    
+                    shot_type = gr.Dropdown(
+                        choices=list(DEFAULT_PROMPTS.keys()),
+                        label="🎬 Shot Type",
+                        value="Cinematic Nature"
+                    )
+                    
+                    prompt = gr.Textbox(
+                        label="📝 Custom Prompt",
+                        lines=3,
+                        value=DEFAULT_PROMPTS["Cinematic Nature"]
+                    )
+                    
+                    shot_type.change(fn=update_prompt, inputs=shot_type, outputs=prompt)
+                    
+                    generate_btn = gr.Button("🚀 Generate Video", variant="primary", size="lg")
+                
+                with gr.Column():
+                    status = gr.Textbox(label="Status", lines=6)
+                    video_output = gr.Video(label="Generated Video")
+            
+            # Quick shot buttons
+            gr.Markdown("### Quick Shot Select")
+            with gr.Row():
+                gr.Button("🏔️ Nature").click(fn=lambda: "Cinematic Nature", outputs=shot_type)
+                gr.Button("🌃 City").click(fn=lambda: "Urban Exploration", outputs=shot_type)
+                gr.Button("🏎️ Action").click(fn=lambda: "Action Sequence", outputs=shot_type)
+                gr.Button("🎨 Abstract").click(fn=lambda: "Abstract Art", outputs=shot_type)
+                gr.Button("🦁 Wildlife").click(fn=lambda: "Wildlife Documentary", outputs=shot_type)
+                gr.Button("⛷️ Sports").click(fn=lambda: "Sports Highlight", outputs=shot_type)
+            
+            # Update URL display when inputs change
+            image_input.change(
+                fn=update_url_display,
+                inputs=[image_input, image_url_input],
+                outputs=current_url_display
+            )
+            
+            image_url_input.change(
+                fn=update_url_display,
+                inputs=[image_input, image_url_input],
+                outputs=current_url_display
+            )
+        
+        # Tab 2: Manual Polling
+        with gr.TabItem("🔧 Manual Polling"):
+            gr.Markdown("### 🔧 Manual Polling & Debug")
+            
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Poll Specific Task")
+                    task_id_input = gr.Textbox(
+                        label="Task ID", 
+                        placeholder="Enter task ID (cgt-...)"
+                    )
+                    poll_btn = gr.Button("🔄 Poll Now", variant="primary")
+                    poll_result = gr.Textbox(label="Poll Result", lines=10)
+                    
+                    poll_btn.click(
+                        fn=manual_poll, 
+                        inputs=task_id_input, 
+                        outputs=poll_result
+                    )
+                
+                with gr.Column():
+                    gr.Markdown("#### Current io.json")
+                    refresh_btn = gr.Button("🔄 Refresh io.json", variant="secondary")
+                    raw_json = gr.JSON(label="io.json Contents")
+                    
+                    refresh_btn.click(
+                        fn=get_raw_json, 
+                        outputs=raw_json
+                    )
+            
+            gr.Markdown("""
+            ### 📝 Instructions
+            1. Copy a task ID from above (like `cgt-20260217072358-hszt9`)
+            2. Paste it in the Task ID field
+            3. Click "Poll Now" to force an update
+            4. Check io.json to see if status changed
+            """)
+        
+        # Tab 3: Watch & Download
+        with gr.TabItem("📺 Watch & Download"):
+            gr.Markdown("### 📺 Browse and Download Generated Videos")
+            
+            with gr.Row():
+                with gr.Column(scale=1):
+                    refresh_list_btn = gr.Button("🔄 Refresh Task List", variant="primary")
+                    task_list = gr.Dropdown(
+                        label="Select Task",
+                        choices=[],
+                        interactive=True
+                    )
+                    
+                with gr.Column(scale=2):
+                    task_info = gr.Markdown("Select a task to view details")
+            
+            with gr.Row():
+                with gr.Column():
+                    video_player = gr.Video(label="Video Player")
+                    download_link = gr.File(label="Download Video")
+            
+            # Load task list on button click
+            refresh_list_btn.click(
+                fn=refresh_task_list,
+                outputs=task_list
+            )
+            
+            # Load task when selected
+            task_list.change(
+                fn=load_task,
+                inputs=task_list,
+                outputs=[task_info, video_player, download_link]
+            )
+            
+            # Auto-load task list when tab is opened
+            demo.load(
+                fn=refresh_task_list,
+                outputs=task_list
+            )
+    
+    # Connect generate button with updated inputs
+    generate_btn.click(
+        fn=generate_video,
+        inputs=[image_input, image_url_input, prompt],
+        outputs=[status, video_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0")