ab.py

import os
import time
import gradio as gr
from byteplussdkarkruntime import Ark
import httpx
import socket
import requests

# Get API key from Hugging Face secret "Key"
API_KEY = os.environ.get("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 test_connection():
    """Test if we can reach the API before trying"""
    try:
        socket.gethostbyname('ark.ap-southeast.bytepluses.com')
        return True, "DNS resolved"
    except Exception as e:
        return False, f"DNS failed: {e}"

def poll_via_json(task_id):
    """Fallback method: poll io.json for results"""
    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]
            if task_data.get('status') == 'succeeded':
                return task_data.get('video_url')
            elif task_data.get('status') == 'failed':
                return None
    except:
        pass
    return None

def generate_video(image_path, prompt_text, progress=gr.Progress()):
    """Generate video with robust error handling"""
    
    # Test connection first
    conn_ok, conn_msg = test_connection()
    if not conn_ok:
        yield f"❌ Connection Error: {conn_msg}. This might be a network restriction from Hugging Face.", None
        return
    
    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 HF temp URL
        image_url = f"/file={image_path}"
        print(f"Using image URL: {image_url}")
        
        yield "✅ Image ready! Creating video request...", None
        
        progress(0.2, desc="Creating request...")
        
        # Create task with timeout
        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:
            error_str = str(e)
            if "Connection error" in error_str or "202602" in error_str:
                yield f"❌ Network Error: Cannot reach BytePlus API. This is likely a Hugging Face network restriction.", None
            else:
                yield f"❌ API Error: {error_str}", 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
                    print(f"Video URL: {video_url}")
                    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(1)
                    attempts += 1
                    
            except Exception as e:
                # If SDK fails, try JSON fallback
                if not used_fallback:
                    print("SDK polling failed, trying JSON fallback...")
                    used_fallback = True
                
                video_url = poll_via_json(task_id)
                if video_url:
                    yield "✅ Video generated successfully! (via JSON)", video_url
                    return
                elif video_url is None:
                    yield "❌ Task failed (via JSON)", None
                    return
                else:
                    yield f"⏳ Status: processing... (JSON fallback)", None
                    time.sleep(5)
                    attempts += 1
        
        yield "⏰ Timeout after 2 minutes", None
        
    except Exception as e:
        print(f"Error: {e}")
        if "Connection error" in str(e):
            yield "❌ Connection Error: Cannot reach BytePlus API. This is likely a Hugging Face network restriction. Try running locally instead.", None
        else:
            yield f"❌ Error: {str(e)}", None

def update_prompt(choice):
    """Update prompt when shot type changes"""
    return DEFAULT_PROMPTS.get(choice, "")

# Interface
with gr.Blocks(title="BytePlus Video Generator", theme=gr.themes.Soft()) as demo:
    
    gr.Markdown("""
    # 🎥 BytePlus Video Generator
    Upload an image and choose a shot type to generate a stunning video.
    """)
    
    # Status row
    with gr.Row():
        if API_KEY:
            gr.Markdown("✅ **API Key:** Configured")
        else:
            gr.Markdown("❌ **API Key:** Not configured - please add 'Key' secret")
    
    # Connection test button
    with gr.Row():
        test_btn = gr.Button("🔌 Test Connection", size="sm")
        test_result = gr.Textbox(label="Connection Test", lines=1)
    
    def run_connection_test():
        ok, msg = test_connection()
        return f"{'✅' if ok else '❌'} {msg}"
    
    test_btn.click(fn=run_connection_test, outputs=test_result)
    
    gr.Markdown("---")
    
    with gr.Row():
        with gr.Column():
            # Image upload
            image_input = gr.Image(
                label="Upload Starting Image",
                type="filepath",
                height=300
            )
            
            # Shot type dropdown
            shot_type = gr.Dropdown(
                choices=list(DEFAULT_PROMPTS.keys()),
                label="🎬 Shot Type",
                value="Cinematic Nature"
            )
            
            # Custom prompt
            prompt = gr.Textbox(
                label="📝 Custom Prompt (override)",
                lines=3,
                placeholder="Or write your own prompt here...",
                value=DEFAULT_PROMPTS["Cinematic Nature"]
            )
            
            # Update prompt when shot type changes
            shot_type.change(
                fn=update_prompt,
                inputs=shot_type,
                outputs=prompt
            )
            
            # Generate button
            generate_btn = gr.Button("🚀 Generate Video", variant="primary", size="lg")
        
        with gr.Column():
            # Status
            status = gr.Textbox(
                label="Status",
                lines=5,
                interactive=False
            )
            
            # Video output
            video_output = gr.Video(
                label="Generated Video",
                interactive=False
            )
    
    # Example shots as quick buttons
    gr.Markdown("---")
    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
        )
    
    # Connect generate button
    generate_event = 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")