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")