Video

"""Cost tracking helper for skill subprocesses.

Skills that call sc-proxy via plain `requests` need to:
  1. Tag every paid call with a SC-CALLER-ID that ties it back to the user
     turn that triggered the skill (so the agent's per-turn cost summary
     shows the cost in the right cost card).
  2. After each call, parse the sc-proxy response headers
     (`X-Credits-Used`, `X-Credits-Api-Type`) and write a row to the cost
     ledger that the agent reads back when it builds the SSE
     `cost_summary` event.

This file is intentionally zero-dependency (stdlib only) so it can be
dropped into any skill folder without coupling to starchild-clawd internals.

Env vars consumed (set by the agent before dispatching the bash subprocess):
  - STARCHILD_TOOL_CALLER_ID  — opaque tag for the current tool call
  - STARCHILD_USER_TURN_ID    — uuid of the current user turn
  - STARCHILD_COST_LEDGER_DIR — optional override for ledger directory

When env vars are absent (e.g. running the script outside an agent), the
helpers degrade gracefully: caller-id falls back to a synthetic string so
the call still goes through, and ledger writes still happen for audit but
the user-turn reader will skip them.
"""
from __future__ import annotations

import fcntl
import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional
from urllib.parse import urlparse


_DEFAULT_LEDGER_DIR = "/data/.starchild/cost_ledger"

# Allowlisted request payload keys we forward into the ledger row's
# `details` field. MUST stay in sync with starchild-clawd's
# core/http_client._record_cost_to_ledger allowlist — anything not in
# that allowlist won't be picked up by the agent and won't render in
# the frontend cost card.
_PAYLOAD_ALLOWLIST = (
    # Identity
    "model", "provider",
    # Image geometry
    "aspect_ratio", "quality", "resolution", "image_size", "size",
    # Video / motion
    "duration", "duration_s", "fps", "motion_strength",
    # Quantity
    "n", "count",
    # Generation knobs
    "seed", "steps", "guidance_scale", "cfg_scale", "strength",
    "scheduler", "sampler",
    # Reference / mode hints
    "image_to_image", "image_to_video", "use_reference", "reference_count",
)


def caller_headers(extra: Optional[Dict[str, str]] = None,
                   tool_default: str = "skill") -> Dict[str, str]:
    """Return an HTTP-headers dict with SC-CALLER-ID filled in.

    Resolution order:
      1. `extra["SC-CALLER-ID"]` (case-insensitive) — caller wins.
      2. STARCHILD_TOOL_CALLER_ID env (set by the agent)
      3. Synthetic `f"{tool_default}:{int(time.time())}"` — tags the call so
         charges are attributable to *some* identifier even when the agent
         didn't inject one (standalone CLI runs, tests, cron).
    """
    merged: Dict[str, str] = dict(extra or {})
    has_caller = any(k.lower() == "sc-caller-id" for k in merged)
    if not has_caller:
        cid = os.environ.get("STARCHILD_TOOL_CALLER_ID") \
            or f"{tool_default}:{int(time.time())}"
        merged["SC-CALLER-ID"] = cid
    return merged


def record_response(response,
                    request_url: str,
                    request_payload: Optional[Dict[str, Any]] = None,
                    api_type_hint: Optional[str] = None) -> None:
    """Inspect a sc-proxy response and append a ledger row when paid.

    Best-effort. Silently no-ops when:
      - response carries no X-Credits-Used / X-Credits-Api-Type
      - cost is 0 or unparseable
      - file write fails
    Never raises — must not break a real request flow.
    """
    try:
        headers = getattr(response, "headers", None) or {}
        used = headers.get("X-Credits-Used") or headers.get("x-credits-used")
        api_type = (headers.get("X-Credits-Api-Type")
                    or headers.get("x-credits-api-type")
                    or api_type_hint)
        if not used or not api_type:
            return
        try:
            cost_f = float(used)
        except (TypeError, ValueError):
            return
        if cost_f <= 0:
            return

        turn_id = os.environ.get("STARCHILD_USER_TURN_ID") or ""
        caller_id = os.environ.get("STARCHILD_TOOL_CALLER_ID") or ""

        host = ""
        try:
            host = urlparse(request_url).netloc or ""
        except Exception:
            pass

        details: Dict[str, Any] = {}
        if isinstance(request_payload, dict):
            for k in _PAYLOAD_ALLOWLIST:
                v = request_payload.get(k)
                if v not in (None, "", []):
                    details[k] = v

        # fal.ai puts the model in the URL path, not the body.
        if "model" not in details and api_type == "falai":
            try:
                path = urlparse(request_url).path or ""
                model_path = path.lstrip("/")
                if "/requests/" in model_path:
                    model_path = model_path.split("/requests/", 1)[0]
                if model_path and not model_path.startswith("requests/"):
                    details["model"] = model_path
                    details["provider"] = "fal"
            except Exception:
                pass

        _append_ledger(
            turn_id=turn_id,
            caller_id=caller_id,
            api_type=api_type,
            cost_usd=cost_f,
            url_host=host,
            details=details or None,
        )
    except Exception:
        # Never let cost tracking break the actual request.
        pass


def _ledger_dir() -> Path:
    base = os.environ.get("STARCHILD_COST_LEDGER_DIR") or _DEFAULT_LEDGER_DIR
    p = Path(base)
    try:
        p.mkdir(parents=True, exist_ok=True)
    except OSError:
        p = Path("/tmp/starchild_cost_ledger")
        p.mkdir(parents=True, exist_ok=True)
    return p


def _today_path() -> Path:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    return _ledger_dir() / f"{today}.jsonl"


def _derive_tool(caller_id: str, api_type: str) -> str:
    """Match starchild-clawd's _derive_tool_from_caller fallback."""
    if not caller_id:
        return api_type or "unknown"
    # chat:{sid}/tool:{name} → name
    if "/tool:" in caller_id:
        return caller_id.rsplit("/tool:", 1)[-1] or api_type
    # skill:{name} | job:{id} | video:{ts}
    head = caller_id.split(":", 1)[0]
    return head or api_type or "unknown"


def _append_ledger(*, turn_id: str, caller_id: str, api_type: str,
                   cost_usd: float, url_host: str,
                   details: Optional[Dict[str, Any]]) -> None:
    row = {
        "ts": round(time.time(), 3),
        "turn_id": turn_id,
        "caller_id": caller_id,
        "tool": _derive_tool(caller_id, api_type),
        "api_type": api_type or "unknown",
        "cost_usd": round(cost_usd, 8),
        "url_host": url_host or "",
    }
    if details:
        row["details"] = details

    line = json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n"
    path = _today_path()
    try:
        with open(path, "ab") as f:
            try:
                fcntl.flock(f.fileno(), fcntl.LOCK_EX)
            except OSError:
                pass
            try:
                f.write(line.encode("utf-8"))
                f.flush()
                try:
                    os.fsync(f.fileno())
                except OSError:
                    pass
            finally:
                try:
                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
                except OSError:
                    pass
    except OSError:
        pass

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

Cost tracking: this script runs as a `bash` subprocess of the agent. The
agent injects `STARCHILD_TOOL_CALLER_ID` and `STARCHILD_USER_TURN_ID` into
the subprocess env. We pass them through to sc-proxy via the SC-CALLER-ID
header (caller_headers helper) and, after each paid call, write a ledger
row (record_response helper) so the agent can fold this skill's cost into
the per-user-turn `cost_summary` SSE event and persist it under the
assistant message's `metadata.cost_summary`.

Status polls and CDN downloads return zero cost from sc-proxy, so the
helper silently no-ops on them. Only the actual submit gets billed and
recorded.
"""

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

# Make _cost_track importable when this script is invoked from any CWD
# (e.g. python -c "from skills.video.generate_video import generate_video").
_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
    sys.path.insert(0, _HERE)
from _cost_track import caller_headers, record_response  # noqa: E402

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

    headers = caller_headers({
        'Authorization': 'Key fake-falai-key-12345',
        'Content-Type': 'application/json',
    }, tool_default='video')

    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)
    # Record the paid submit call to the cost ledger so the agent's
    # per-turn cost_summary picks up this video's cost (no-op if 0).
    record_response(response, request_url=submit_url, request_payload=body)

    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)
    try:
        result_json = result_resp.json()
    except Exception:
        return {"success": False, "request_id": request_id, "cost": cost,
                "error": f"Result endpoint returned non-JSON (HTTP {result_resp.status_code}): {result_resp.text[:200]}",
                "polls": poll_count}

    # fal model response shapes vary. Try known shapes; if none match,
    # surface the actual top-level keys so the caller can see why parsing
    # failed (instead of a raw KeyError 200 lines deep).
    video_url = _extract_video_url(result_json)
    if not video_url:
        return {"success": False, "request_id": request_id, "cost": cost,
                "error": (f"Could not locate video URL in fal response. "
                          f"Top-level keys: {list(result_json.keys())}. "
                          f"Sample: {str(result_json)[:300]}"),
                "polls": poll_count}
    
    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 _extract_video_url(result_json):
    """Recover the video URL from fal's response across model variants.

    Known shapes (as of 2026-05):
      - happy-horse / kling / seedance:   {"video":  {"url": "..."}}
      - wan / some pipelines:             {"videos": [{"url": "..."}]}
      - some upscale/pipeline outputs:    {"output": [{"url": "..."}]}
      - voiceover/audio variants:         {"output_video": {"url": "..."}}

    Returns the URL string, or None if no recognised shape matches.
    """
    if not isinstance(result_json, dict):
        return None

    # Single-video object shapes
    for key in ("video", "output_video"):
        node = result_json.get(key)
        if isinstance(node, dict) and isinstance(node.get("url"), str):
            return node["url"]
        if isinstance(node, str) and node.startswith("http"):
            return node

    # Array shapes
    for key in ("videos", "output", "outputs"):
        arr = result_json.get(key)
        if isinstance(arr, list) and arr:
            first = arr[0]
            if isinstance(first, dict) and isinstance(first.get("url"), str):
                return first["url"]
            if isinstance(first, str) and first.startswith("http"):
                return first

    # Last resort: anything in top-level that looks like {"url": "...mp4"}
    for v in result_json.values():
        if isinstance(v, dict) and isinstance(v.get("url"), str) and ".mp4" in v["url"]:
            return v["url"]
    return None


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)

# NOTE 2026-05-11: 'fal-ai/wan/v2.5/text-to-video' was removed from the
# fal proxy allowlist (returns 404). Until a cheap replacement is curated,
# 'budget' falls back to happy-horse (same as 'balanced'). Cost is still
# ~$0.14/s instead of $0.05/s — budget tier is effectively unavailable.
QUICK_MODELS = {
    "budget": "alibaba/happy-horse/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

Cost tracking: status polls and result fetches return zero cost from
sc-proxy (they're free for already-submitted jobs), so the
record_response helper silently no-ops here. We still attach SC-CALLER-ID
via caller_headers so any future billing change is correctly attributed.
"""

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

_HERE = os.path.dirname(os.path.abspath(__file__))
if _HERE not in sys.path:
    sys.path.insert(0, _HERE)
from _cost_track import caller_headers, record_response  # noqa: E402


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 = caller_headers({'Authorization': 'Key fake-falai-key-12345'}, tool_default='video')
    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)
    record_response(result_resp, request_url=result_url)
    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))