final.py

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, 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

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

# 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
        with gr.TabItem("🎬 Generate Video"):
            with gr.Row():
                with gr.Column():
                    image_input = gr.Image(
                        label="Upload Starting Image",
                        type="filepath",
                        height=300
                    )
                    
                    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)
        
        # 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
    generate_btn.click(
        fn=generate_video,
        inputs=[image_input, prompt],
        outputs=[status, video_output]
    )

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0")