Video

#!/usr/bin/env python3
"""Video generation script - one-stop submit → poll → download"""

import requests
import json
import time
import os
from datetime import datetime
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

PROXY_URL = 'http://sc-proxy.internal:8080'
PROXIES = {'http': PROXY_URL, 'https': PROXY_URL}

def generate_video(prompt, model="alibaba/happy-horse/text-to-video", duration=5, resolution="720p", image_url=None):
    """Generate video end-to-end. Returns dict with success/error/paths."""
    
    caller_id = f"video:{int(time.time())}"
    headers = {
        'Authorization': 'Key fake-falai-key-12345',
        'Content-Type': 'application/json',
        'SC-CALLER-ID': caller_id
    }
    
    body = {'prompt': prompt, 'duration': duration, 'aspect_ratio': "16:9"}
    if 'happy-horse' in model or 'kling' in model:
        body['resolution'] = resolution
    
    # Handle image input — must be a public https URL.
    # Recommended: publish via skills/video/publish_asset.py + community preview slug `fal-assets`.
    if image_url:
        if image_url.startswith('data:') or not image_url.startswith(('http://', 'https://')):
            return {"success": False, "error": "image_url must be a public HTTP(S) URL. Use publish_asset.py + fal-assets preview to expose local files."}
        if not model.endswith('/image-to-video'):
            model = model.replace('/text-to-video', '/image-to-video')
        body['image_url'] = image_url
    
    # Submit
    submit_url = f'https://queue.fal.run/{model}'
    response = requests.post(submit_url, headers=headers, json=body, proxies=PROXIES, verify=False, timeout=90)
    
    if response.status_code != 200:
        return {"success": False, "error": f"Submit failed: {response.status_code} - {response.text[:200]}"}
    
    data = response.json()
    request_id = data['request_id']
    status_url = data['status_url']
    result_url = data.get('response_url', data.get('result_url'))
    cost = float(response.headers.get('X-Credits-Used', 0))
    
    print(f"✅ Submitted: {request_id}, cost=${cost:.2f}")
    
    # Poll
    deadline = time.time() + 900  # 15min timeout
    while time.time() < deadline:
        poll_resp = requests.get(status_url, headers={'Authorization': 'Key fake-falai-key-12345'}, proxies=PROXIES, verify=False, timeout=60)
        status = poll_resp.json().get('status')
        
        if status == 'COMPLETED':
            break
        elif status in ('FAILED', 'CANCELLED'):
            return {"success": False, "request_id": request_id, "cost": cost, "error": f"Generation {status}"}
        
        time.sleep(5)
    else:
        return {"success": False, "request_id": request_id, "cost": cost, "error": "Timeout"}
    
    # Get result & download
    result_resp = requests.get(result_url, headers={'Authorization': 'Key fake-falai-key-12345'}, proxies=PROXIES, verify=False, timeout=90)
    video_url = result_resp.json()['video']['url']
    
    os.makedirs('output/videos', exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    model_short = model.split('/')[-1]
    local_path = f"output/videos/{timestamp}_{model_short}_{duration}s_{resolution}.mp4"
    
    video_data = requests.get(video_url, timeout=120).content
    open(local_path, 'wb').write(video_data)
    
    return {
        "success": True,
        "request_id": request_id,
        "cost": cost,
        "video_url": video_url,
        "local_path": local_path,
        "file_size_mb": len(video_data) / 1024 / 1024
    }

def estimate_cost(model, duration, resolution="720p"):
    """Estimate generation cost in USD"""
    prices = {
        "alibaba/happy-horse/text-to-video": 0.14,
        "fal-ai/wan/v2.5/text-to-video": 0.05,
        "fal-ai/kling-video/v2.6/pro/text-to-video": 0.07,
        "bytedance/seedance-2.0/fast/text-to-video": 0.2419,
        "fal-ai/hunyuanvideo": 0.40,  # flat rate
    }
    
    if model == "fal-ai/hunyuanvideo":
        return 0.40
    
    unit_price = prices.get(model, 0.10)  # default fallback
    if 'happy-horse' in model and resolution == "1080p":
        unit_price *= 2
    
    return round(unit_price * duration, 4)

QUICK_MODELS = {
    "budget": "fal-ai/wan/v2.5/text-to-video",
    "balanced": "alibaba/happy-horse/text-to-video", 
    "premium": "bytedance/seedance-2.0/fast/text-to-video"
}

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python generate_video.py 'prompt' [model|tier] [duration]")
        print("Tiers: budget, balanced, premium")
        sys.exit(1)
    
    prompt = sys.argv[1]
    model_or_tier = sys.argv[2] if len(sys.argv) > 2 else "balanced"
    duration = int(sys.argv[3]) if len(sys.argv) > 3 else 5
    
    model = QUICK_MODELS.get(model_or_tier, model_or_tier)
    print(f"Model: {model}, Est cost: ${estimate_cost(model, duration)}")
    
    result = generate_video(prompt, model, duration)
    print(json.dumps(result, indent=2))

#!/usr/bin/env python3
"""Poll existing video by request_id"""

import requests
import time
import os
from datetime import datetime
import urllib3
urllib3.disable_warnings()

def poll_video(request_id, download=True):
    """Poll video status and download if completed"""
    
    # Try common URL patterns
    patterns = [
        f"https://queue.fal.run/requests/{request_id}",
        f"https://queue.fal.run/alibaba/happy-horse/requests/{request_id}",
    ]
    
    headers = {'Authorization': 'Key fake-falai-key-12345'}
    proxies = {'http': 'http://sc-proxy.internal:8080', 'https': 'http://sc-proxy.internal:8080'}
    
    status_url = None
    for pattern in patterns:
        try:
            test_resp = requests.get(f"{pattern}/status", headers=headers, proxies=proxies, verify=False, timeout=10)
            if test_resp.status_code == 200:
                status_url = f"{pattern}/status"
                result_url = pattern
                break
        except:
            continue
    
    if not status_url:
        return {"success": False, "error": f"Invalid request_id: {request_id}"}
    
    # Poll until complete
    for _ in range(180):  # 15min max
        resp = requests.get(status_url, headers=headers, proxies=proxies, verify=False, timeout=30)
        status = resp.json().get('status')
        
        if status == 'COMPLETED':
            break
        elif status in ('FAILED', 'CANCELLED'):
            return {"success": False, "status": status, "error": "Generation failed"}
        
        time.sleep(5)
    else:
        return {"success": False, "error": "Timeout"}
    
    if not download:
        return {"success": True, "status": "COMPLETED", "request_id": request_id}
    
    # Download result
    result_resp = requests.get(result_url, headers=headers, proxies=proxies, verify=False, timeout=60)
    video_url = result_resp.json()['video']['url']
    
    os.makedirs('output/videos', exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    local_path = f"output/videos/{timestamp}_{request_id}_result.mp4"
    
    video_data = requests.get(video_url, timeout=120).content
    open(local_path, 'wb').write(video_data)
    
    return {
        "success": True,
        "request_id": request_id,
        "video_url": video_url,
        "local_path": local_path,
        "file_size_mb": len(video_data) / 1024 / 1024
    }

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python poll_status.py <request_id>")
        sys.exit(1)
    
    result = poll_video(sys.argv[1])
    print(result)

#!/usr/bin/env python3
"""Publish a local image/video to output/fal_assets/ for fal.ai reference inputs.

Workflow:
  1. Drop file into output/fal_assets/
  2. Combine with the public preview base URL (see SKILL.md) to form the URL fal needs.

If the file is already a URL, it is downloaded first.
"""
from __future__ import annotations
import os, shutil, sys, mimetypes
from pathlib import Path
import requests

ASSETS_DIR = Path('output/fal_assets')
IMAGE_EXTS = {'.jpg', '.jpeg', '.png', '.webp', '.gif', '.bmp'}
VIDEO_EXTS = {'.mp4', '.mov', '.webm', '.mkv', '.m4v'}
ALLOWED_EXTS = IMAGE_EXTS | VIDEO_EXTS

MAX_IMAGE_BYTES = 10 * 1024 * 1024
MAX_VIDEO_BYTES = 100 * 1024 * 1024


def publish_local(src_path: str, rename: str | None = None) -> dict:
    p = Path(src_path)
    if not p.exists() or not p.is_file():
        return {"success": False, "error": f"file not found: {src_path}"}

    ext = p.suffix.lower()
    if ext not in ALLOWED_EXTS:
        return {"success": False, "error": f"unsupported extension: {ext}"}

    size = p.stat().st_size
    limit = MAX_IMAGE_BYTES if ext in IMAGE_EXTS else MAX_VIDEO_BYTES
    if size > limit:
        return {"success": False, "error": f"file too large: {size} > {limit} bytes"}

    ASSETS_DIR.mkdir(parents=True, exist_ok=True)
    target_name = rename or p.name
    dst = ASSETS_DIR / target_name
    shutil.copy(p, dst)

    return {
        "success": True,
        "local_path": str(dst),
        "filename": target_name,
        "kind": "image" if ext in IMAGE_EXTS else "video",
        "size_bytes": size,
        "hint": "Combine with public preview base URL: <preview_base>/<filename>",
    }


def publish_from_url(src_url: str, rename: str | None = None) -> dict:
    if not src_url.startswith(('http://', 'https://')):
        return {"success": False, "error": "src_url must be http(s)"}

    try:
        r = requests.get(src_url, timeout=60)
        r.raise_for_status()
    except Exception as e:
        return {"success": False, "error": f"download failed: {e}"}

    name = rename or src_url.rstrip('/').split('/')[-1].split('?')[0]
    if '.' not in name:
        ct = (r.headers.get('Content-Type') or '').split(';')[0].strip()
        ext_guess = mimetypes.guess_extension(ct) or '.bin'
        name = f"{name}{ext_guess}"

    ext = Path(name).suffix.lower()
    if ext not in ALLOWED_EXTS:
        return {"success": False, "error": f"unsupported extension: {ext}"}

    size = len(r.content)
    limit = MAX_IMAGE_BYTES if ext in IMAGE_EXTS else MAX_VIDEO_BYTES
    if size > limit:
        return {"success": False, "error": f"file too large: {size} > {limit} bytes"}

    ASSETS_DIR.mkdir(parents=True, exist_ok=True)
    dst = ASSETS_DIR / name
    dst.write_bytes(r.content)

    return {
        "success": True,
        "local_path": str(dst),
        "filename": name,
        "kind": "image" if ext in IMAGE_EXTS else "video",
        "size_bytes": size,
        "hint": "Combine with public preview base URL: <preview_base>/<filename>",
    }


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python skills/video/publish_asset.py <local_path|url> [rename]")
        sys.exit(1)
    src = sys.argv[1]
    rename = sys.argv[2] if len(sys.argv) > 2 else None
    fn = publish_from_url if src.startswith('http') else publish_local
    print(fn(src, rename))