Create a.py

a.py

@@ -0,0 +1,181 @@
+import gradio as gr
+import os
+import time
+import logging
+import socket
+import requests
+from byteplussdkarkruntime import Ark
+import byteplussdkcore
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Get API key from environment variable "key"
+API_KEY = os.environ.get("key", "")
+logger.info(f"API Key loaded: {'Yes' if API_KEY else 'No'}")
+logger.info(f"Key length: {len(API_KEY)}")
+if API_KEY:
+    logger.info(f"First 4 chars: {API_KEY[:4]}")
+
+# Test basic connectivity first
+def test_connectivity():
+    """Test if we can reach the API server"""
+    try:
+        # Test DNS resolution
+        ip = socket.gethostbyname('ark.ap-southeast.bytepluses.com')
+        logger.info(f"✅ DNS resolved: {ip}")
+        
+        # Test basic HTTPS connection
+        r = requests.get("https://ark.ap-southeast.bytepluses.com", timeout=5)
+        logger.info(f"✅ Base URL reachable (status: {r.status_code})")
+        return True
+    except Exception as e:
+        logger.error(f"❌ Connection test failed: {e}")
+        return False
+
+# Run connection test on startup
+connectivity_ok = test_connectivity()
+
+# Configure SDK
+try:
+    configuration = byteplussdkcore.Configuration()
+    configuration.client_side_validation = True
+    configuration.schema = "http"
+    configuration.debug = True  # Enable debug logging
+    configuration.logger_file = "sdk.log"
+    byteplussdkcore.Configuration.set_default(configuration)
+    logger.info("✅ SDK configured")
+except Exception as e:
+    logger.error(f"❌ SDK config failed: {e}")
+
+# Initialize client WITH THE KEY
+try:
+    client = Ark(
+        base_url="https://ark.ap-southeast.bytepluses.com/api/v3",
+        api_key=API_KEY,  # The key is here!
+    )
+    logger.info("✅ Client initialized with API key")
+except Exception as e:
+    logger.error(f"❌ Client init failed: {e}")
+    client = None
+
+def generate_video(prompt_text, image_url, progress=gr.Progress()):
+    """Generate video with detailed debugging"""
+    
+    if not API_KEY:
+        yield "❌ No API key found in environment variable 'key'", None
+        return
+    
+    if not client:
+        yield "❌ Client not initialized", None
+        return
+    
+    if not connectivity_ok:
+        yield "❌ Cannot reach API server", None
+        return
+    
+    try:
+        progress(0, desc="Creating task...")
+        logger.info("Creating task...")
+        
+        # Log the request details (without exposing full key)
+        logger.info(f"Prompt: {prompt_text[:50]}...")
+        logger.info(f"Image URL: {image_url}")
+        logger.info(f"Using API key: {API_KEY[:4]}...{API_KEY[-4:]}")
+        
+        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
+                    }
+                }
+            ]
+        )
+        
+        task_id = create_result.id
+        logger.info(f"✅ Task created: {task_id}")
+        yield f"Task created: {task_id}", None
+        
+        progress(0.3, desc="Polling...")
+        
+        # Poll for results
+        attempts = 0
+        while attempts < 60:
+            try:
+                get_result = client.content_generation.tasks.get(task_id=task_id)
+                status = get_result.status
+                logger.info(f"Status: {status}")
+                
+                if status == "succeeded":
+                    progress(1.0, desc="Complete!")
+                    video_url = get_result.output[0].get('video_url') if get_result.output else None
+                    yield "Success!", 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/60)*0.7, desc=f"Status: {status}")
+                    yield f"Status: {status}...", None
+                    time.sleep(1)
+                    attempts += 1
+            except Exception as e:
+                logger.error(f"Polling error: {e}")
+                yield f"Polling error: {e}", None
+                return
+                
+        yield "Timeout", None
+        
+    except Exception as e:
+        logger.error(f"❌ Error: {e}", exc_info=True)
+        yield f"Error: {str(e)}", None
+
+# Simple interface
+with gr.Blocks() as demo:
+    gr.Markdown("# BytePlus Video Generator")
+    
+    # Show connection and key status
+    status_lines = []
+    status_lines.append(f"🔑 API Key: {'✅ Loaded' if API_KEY else '❌ Missing'}")
+    if API_KEY:
+        status_lines.append(f"   Length: {len(API_KEY)} chars")
+        status_lines.append(f"   Starts with: {API_KEY[:4]}")
+    status_lines.append(f"🌐 API Server: {'✅ Reachable' if connectivity_ok else '❌ Unreachable'}")
+    status_lines.append(f"📦 SDK Client: {'✅ Initialized' if client else '❌ Failed'}")
+    
+    gr.Markdown("\n".join(status_lines))
+    
+    with gr.Row():
+        with gr.Column():
+            prompt = gr.Textbox(
+                label="Prompt",
+                lines=3,
+                value="At breakneck speed, drones fly through obstacles --duration 5 --camerafixed false"
+            )
+            image_url = gr.Textbox(
+                label="Image URL",
+                value="https://ark-doc.tos-ap-southeast-1.bytepluses.com/seepro_i2v%20.png"
+            )
+            btn = gr.Button("Generate", variant="primary")
+        
+        with gr.Column():
+            status = gr.Textbox(label="Status", lines=8)
+            video = gr.Video(label="Generated Video")
+    
+    btn.click(
+        fn=generate_video,
+        inputs=[prompt, image_url],
+        outputs=[status, video]
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0")