Lipnardo

#!/usr/bin/env python3
"""Lipnardo -- HeyGen Asset Manager

Purpose:
    Upload, list, and inspect HeyGen assets (images, videos, audio files)
    used in avatar video generation.

Usage:
    # Upload an asset:
    python3 asset_manager.py upload --file /path/to/image.png [--type image] [--api-key KEY]

    # List assets:
    python3 asset_manager.py list [--type image|video|audio] [--limit 50] [--api-key KEY]

    # Get asset info:
    python3 asset_manager.py info --asset-id ID [--api-key KEY]

Dependencies: Python 3.8+ stdlib only
"""

import argparse
import os
import sys
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from heygen_client import (
    resolve_api_key,
    api_request,
    api_upload,
    output_json,
    log_json,
    error_exit,
    HeyGenAPIError,
)

# Extension-to-type mapping for auto-detection
EXTENSION_TYPE_MAP = {
    ".jpg": "image",
    ".jpeg": "image",
    ".png": "image",
    ".webp": "image",
    ".gif": "image",
    ".bmp": "image",
    ".mp4": "video",
    ".mov": "video",
    ".avi": "video",
    ".mkv": "video",
    ".webm": "video",
    ".mp3": "audio",
    ".wav": "audio",
    ".aac": "audio",
    ".ogg": "audio",
    ".flac": "audio",
    ".m4a": "audio",
}


def detect_asset_type(file_path: str) -> str:
    """Auto-detect asset type from file extension.

    Returns:
        Asset type string: 'image', 'video', or 'audio'
    """
    ext = Path(file_path).suffix.lower()
    asset_type = EXTENSION_TYPE_MAP.get(ext)
    if not asset_type:
        error_exit(
            f"Cannot auto-detect asset type for extension '{ext}'. "
            f"Supported: {', '.join(sorted(EXTENSION_TYPE_MAP.keys()))}. "
            "Use --type to specify explicitly."
        )
    return asset_type


# ---------------------------------------------------------------------------
# Upload subcommand
# ---------------------------------------------------------------------------

def cmd_upload(args):
    """Upload a file to HeyGen as an asset."""
    api_key = resolve_api_key(args.api_key)

    file_path = args.file
    if not os.path.isfile(file_path):
        error_exit(f"File not found: {file_path}")

    # Determine asset type
    asset_type = args.type if args.type else detect_asset_type(file_path)
    filename = os.path.basename(file_path)

    log_json({
        "action": "upload_asset",
        "file": file_path,
        "type": asset_type,
        "filename": filename,
    })

    try:
        resp = api_upload(
            path="/v1/asset",
            api_key=api_key,
            file_path=file_path,
            fields={"type": asset_type},
        )
    except HeyGenAPIError as e:
        output_json({
            "error": True,
            "status_code": e.status_code,
            "error_code": e.error_code,
            "message": e.message,
            "endpoint": e.endpoint,
        })
        sys.exit(1)

    data = resp.get("data", resp)

    output_json({
        "success": True,
        "asset_id": data.get("asset_id", data.get("id", "")),
        "url": data.get("url", ""),
        "type": asset_type,
        "filename": filename,
    })


# ---------------------------------------------------------------------------
# List subcommand
# ---------------------------------------------------------------------------

def cmd_list(args):
    """List HeyGen assets with optional type filter."""
    api_key = resolve_api_key(args.api_key)
    limit = args.limit if args.limit else 50

    # Build query path with parameters
    path = f"/v1/asset/list?limit={limit}"
    if args.type:
        path += f"&type={args.type}"

    log_json({
        "action": "list_assets",
        "type_filter": args.type,
        "limit": limit,
    })

    try:
        resp = api_request("GET", path, api_key)
    except HeyGenAPIError as e:
        output_json({
            "error": True,
            "status_code": e.status_code,
            "error_code": e.error_code,
            "message": e.message,
            "endpoint": e.endpoint,
        })
        sys.exit(1)

    data = resp.get("data", resp)

    # Extract asset list — API may nest under different keys
    if isinstance(data, list):
        assets_raw = data
    elif isinstance(data, dict):
        assets_raw = data.get("assets", data.get("list", data.get("items", [])))
    else:
        assets_raw = []

    # Normalize asset entries
    assets = []
    for a in assets_raw:
        assets.append({
            "asset_id": a.get("asset_id", a.get("id", "")),
            "url": a.get("url", ""),
            "type": a.get("type", ""),
            "created_at": a.get("created_at", a.get("create_time", "")),
        })

    output_json({
        "success": True,
        "total": len(assets),
        "type_filter": args.type,
        "assets": assets,
    })


# ---------------------------------------------------------------------------
# Info subcommand
# ---------------------------------------------------------------------------

def cmd_info(args):
    """Get detailed information about a specific asset."""
    api_key = resolve_api_key(args.api_key)

    log_json({"action": "asset_info", "asset_id": args.asset_id})

    try:
        resp = api_request("GET", f"/v1/asset/{args.asset_id}", api_key)
    except HeyGenAPIError as e:
        output_json({
            "error": True,
            "status_code": e.status_code,
            "error_code": e.error_code,
            "message": e.message,
            "endpoint": e.endpoint,
        })
        sys.exit(1)

    data = resp.get("data", resp)

    output_json({
        "success": True,
        "asset_id": data.get("asset_id", data.get("id", args.asset_id)),
        "url": data.get("url", ""),
        "type": data.get("type", ""),
        "filename": data.get("filename", data.get("name", "")),
        "created_at": data.get("created_at", data.get("create_time", "")),
    })


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Lipnardo -- HeyGen Asset Manager (upload, list, info)"
    )
    sub = parser.add_subparsers(dest="command", required=True)

    # upload
    p_upload = sub.add_parser("upload", help="Upload a file as a HeyGen asset")
    p_upload.add_argument("--file", required=True,
                          help="Path to file to upload")
    p_upload.add_argument("--type", choices=["image", "video", "audio"],
                          help="Asset type (auto-detected from extension if not specified)")
    p_upload.add_argument("--api-key",
                          help="HeyGen API key (or set HEYGEN_API_KEY env)")

    # list
    p_list = sub.add_parser("list", help="List HeyGen assets")
    p_list.add_argument("--type", choices=["image", "video", "audio"],
                        help="Filter by asset type")
    p_list.add_argument("--limit", type=int, default=50,
                        help="Maximum number of assets to return (default: 50)")
    p_list.add_argument("--api-key",
                        help="HeyGen API key (or set HEYGEN_API_KEY env)")

    # info
    p_info = sub.add_parser("info", help="Get info about a specific asset")
    p_info.add_argument("--asset-id", required=True,
                        help="HeyGen asset ID")
    p_info.add_argument("--api-key",
                        help="HeyGen API key (or set HEYGEN_API_KEY env)")

    args = parser.parse_args()

    cmds = {
        "upload": cmd_upload,
        "list": cmd_list,
        "info": cmd_info,
    }
    cmds[args.command](args)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations
"""Lipnardo -- Batch Video Generation Queue Manager

Purpose:
    Batch-generate HeyGen avatar videos from a CSV or JSON input file with
    concurrency control, retry logic, crash-recoverable manifests, and
    progress reporting.

Usage:
    # Agent mode -- each CSV row has a "prompt" column:
    python3 batch_generate.py --input data.csv --mode agent \
        [--max-concurrent 3] [--max-retries 3] [--output-dir DIR] [--api-key KEY]

    # Template mode -- CSV columns match template variable names:
    python3 batch_generate.py --input data.csv --mode template \
        --template-id TEMPLATE_ID [options]

    # Resume a batch from an existing manifest (crash recovery):
    python3 batch_generate.py --resume manifest.json

    # Check status/progress of a batch:
    python3 batch_generate.py --status manifest.json

Dependencies: Python 3.8+ stdlib only
"""

import argparse
import csv
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from heygen_client import (
    resolve_api_key,
    api_request,
    download_file,
    ensure_output_dir,
    timestamp_filename,
    output_json,
    log_json,
    error_exit,
    sanitize_id,
    HeyGenAPIError,
)


# ---------------------------------------------------------------------------
# Input Loading
# ---------------------------------------------------------------------------

def load_input(input_file: str) -> list[dict]:
    """Load input rows from a CSV or JSON file.

    CSV: first row is headers, each subsequent row becomes a dict.
    JSON: expects a list of dicts.

    Returns:
        List of row dicts.
    """
    path = Path(input_file)
    if not path.exists():
        error_exit(f"Input file not found: {input_file}")

    ext = path.suffix.lower()

    if ext == ".json":
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, list):
            error_exit("JSON input must be a list of objects")
        return data

    # Default: CSV
    rows = []
    with open(path, "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            rows.append(dict(row))
    if not rows:
        error_exit(f"No data rows found in {input_file}")
    return rows


# ---------------------------------------------------------------------------
# Manifest Management
# ---------------------------------------------------------------------------

def create_manifest(input_file: str, mode: str, template_id: str | None,
                    max_concurrent: int, output_dir: str,
                    rows: list[dict]) -> dict:
    """Create a new batch manifest from input rows."""
    manifest = {
        "created": datetime.now(timezone.utc).isoformat(),
        "mode": mode,
        "template_id": template_id,
        "input_file": os.path.abspath(input_file),
        "max_concurrent": max_concurrent,
        "total": len(rows),
        "completed": 0,
        "failed": 0,
        "pending": len(rows),
        "output_dir": output_dir,
        "jobs": [],
    }

    for i, row in enumerate(rows):
        manifest["jobs"].append({
            "index": i,
            "status": "pending",
            "video_id": None,
            "path": None,
            "attempts": 0,
            "error": None,
            "input_row": row,
        })

    return manifest


def save_manifest(manifest: dict, manifest_path: str) -> None:
    """Write manifest to disk atomically (write tmp then rename)."""
    tmp_path = manifest_path + ".tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(manifest, f, indent=2)
    os.replace(tmp_path, manifest_path)


def load_manifest(manifest_path: str) -> dict:
    """Load an existing manifest from disk."""
    path = Path(manifest_path)
    if not path.exists():
        error_exit(f"Manifest file not found: {manifest_path}")
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def update_manifest_counts(manifest: dict) -> None:
    """Recalculate completed/failed/pending counts from job statuses."""
    completed = 0
    failed = 0
    pending = 0
    for job in manifest["jobs"]:
        if job["status"] == "completed":
            completed += 1
        elif job["status"] == "failed":
            failed += 1
        else:
            pending += 1
    manifest["completed"] = completed
    manifest["failed"] = failed
    manifest["pending"] = pending


# ---------------------------------------------------------------------------
# Video Submission
# ---------------------------------------------------------------------------

def submit_video(job: dict, mode: str, template_id: str | None,
                 api_key: str) -> str:
    """Submit a single video generation request based on mode.

    Returns:
        video_id string

    Raises:
        HeyGenAPIError on submission failure
    """
    row = job["input_row"]

    if mode == "agent":
        prompt = row.get("prompt")
        if not prompt:
            raise HeyGenAPIError(
                400, "MISSING_PROMPT",
                f"Row {job['index']} missing required 'prompt' column",
                "/v3/video-agents"
            )

        body = {"prompt": prompt}
        if row.get("avatar_id"):
            body["avatar_id"] = row["avatar_id"]
        if row.get("voice_id"):
            body["voice_id"] = row["voice_id"]
        if row.get("name"):
            body["title"] = row["name"]

        resp = api_request("POST", "/v3/video-agents", api_key, body=body)
        data = resp.get("data", resp)
        video_id = data.get("video_id")

    elif mode == "template":
        if not template_id:
            raise HeyGenAPIError(
                400, "MISSING_TEMPLATE",
                "Template mode requires --template-id",
                "/v2/template/generate"
            )

        # All CSV columns become template variables (array format per HeyGen API)
        variables = []
        for key, value in row.items():
            variables.append({
                "name": key,
                "type": "text",
                "properties": {"content": value},
            })

        body = {
            "template_id": template_id,
            "variables": variables,
        }

        resp = api_request("POST", "/v2/template/generate", api_key, body=body)
        data = resp.get("data", resp)
        video_id = data.get("video_id")

    else:
        raise HeyGenAPIError(400, "INVALID_MODE", f"Unknown mode: {mode}", "")

    if not video_id:
        raise HeyGenAPIError(
            500, "NO_VIDEO_ID",
            f"No video_id in response for job {job['index']}",
            ""
        )

    return video_id


# ---------------------------------------------------------------------------
# Status Check (single non-blocking poll)
# ---------------------------------------------------------------------------

def check_video_status(video_id: str, api_key: str) -> dict:
    """Check the current status of a video (single non-blocking request).

    Returns:
        Dict with at least 'status' key. May include 'video_url', 'error', etc.
    """
    resp = api_request("GET", f"/v2/videos/{video_id}", api_key)
    data = resp.get("data", resp)
    return data


# ---------------------------------------------------------------------------
# Core Batch Loop
# ---------------------------------------------------------------------------

def run_batch(manifest: dict, manifest_path: str, api_key: str) -> None:
    """Execute the batch generation loop with concurrency control.

    Submits up to max_concurrent videos simultaneously, polls all active
    jobs, and as each completes submits the next from the queue. Failed
    jobs retry with exponential backoff up to max_retries.
    """
    max_concurrent = manifest["max_concurrent"]
    max_retries = manifest.get("max_retries", 3)
    output_dir = ensure_output_dir(manifest["output_dir"])
    jobs = manifest["jobs"]
    poll_interval = 15  # seconds

    active = {}  # video_id -> job index

    # Build queue: pending jobs + failed jobs that can be retried
    queue = []
    for job in jobs:
        if job["status"] == "pending":
            queue.append(job)
        elif job["status"] == "failed" and job["attempts"] < max_retries:
            queue.append(job)
        elif job["status"] == "processing" and job.get("video_id"):
            # Resumed manifest -- re-track active jobs
            active[job["video_id"]] = job["index"]

    log_json({
        "action": "batch_start",
        "total": manifest["total"],
        "queued": len(queue),
        "active": len(active),
        "max_concurrent": max_concurrent,
    })

    # Submit initial batch (fill up to max_concurrent minus already-active)
    while len(active) < max_concurrent and queue:
        job = queue.pop(0)
        try:
            video_id = submit_video(job, manifest["mode"],
                                    manifest["template_id"], api_key)
            active[video_id] = job["index"]
            job["status"] = "processing"
            job["video_id"] = video_id
            job["attempts"] += 1
            job["error"] = None
            log_json({"action": "submitted", "index": job["index"],
                       "video_id": video_id, "attempt": job["attempts"]})
        except HeyGenAPIError as e:
            job["attempts"] += 1
            job["error"] = e.message
            if job["attempts"] < max_retries:
                job["status"] = "pending"
                # Exponential backoff before re-queuing
                backoff = 2 ** job["attempts"]
                log_json({"action": "submit_failed", "index": job["index"],
                           "error": e.message, "retry_in": backoff})
                time.sleep(backoff)
                queue.append(job)
            else:
                job["status"] = "failed"
                log_json({"action": "submit_failed_permanent",
                           "index": job["index"], "error": e.message})

        update_manifest_counts(manifest)
        save_manifest(manifest, manifest_path)

    # Poll loop
    while active:
        for video_id in list(active.keys()):
            job_index = active[video_id]
            job = jobs[job_index]

            try:
                status_data = check_video_status(video_id, api_key)
                status = status_data.get("status", "unknown")

                if status == "completed":
                    video_url = status_data.get("video_url", "")
                    log_json({"action": "completed", "index": job_index,
                               "video_id": video_id})

                    # Download the video
                    if video_url:
                        filename = f"{sanitize_id(video_id, 'video_id')}.mp4"
                        output_path = str(output_dir / filename)
                        try:
                            saved_path = download_file(video_url, output_path)
                            job["path"] = saved_path
                        except Exception as dl_err:
                            log_json({"action": "download_failed",
                                       "index": job_index, "error": str(dl_err)})
                            job["path"] = None

                    job["status"] = "completed"
                    job["error"] = None
                    del active[video_id]

                    # Submit next from queue
                    if queue and len(active) < max_concurrent:
                        next_job = queue.pop(0)
                        try:
                            next_vid = submit_video(
                                next_job, manifest["mode"],
                                manifest["template_id"], api_key
                            )
                            active[next_vid] = next_job["index"]
                            next_job["status"] = "processing"
                            next_job["video_id"] = next_vid
                            next_job["attempts"] += 1
                            next_job["error"] = None
                            log_json({"action": "submitted",
                                       "index": next_job["index"],
                                       "video_id": next_vid,
                                       "attempt": next_job["attempts"]})
                        except HeyGenAPIError as e:
                            next_job["attempts"] += 1
                            next_job["error"] = e.message
                            if next_job["attempts"] < max_retries:
                                next_job["status"] = "pending"
                                queue.append(next_job)
                            else:
                                next_job["status"] = "failed"

                elif status == "failed":
                    error_msg = status_data.get(
                        "error", status_data.get("message", "Video generation failed")
                    )
                    job["error"] = str(error_msg)
                    job["status"] = "failed"
                    log_json({"action": "failed", "index": job_index,
                               "video_id": video_id, "error": job["error"]})

                    # Retry if attempts remain
                    if job["attempts"] < max_retries:
                        job["status"] = "pending"
                        queue.append(job)
                        log_json({"action": "requeued", "index": job_index,
                                   "attempts": job["attempts"],
                                   "max_retries": max_retries})

                    del active[video_id]

                    # Submit next from queue
                    if queue and len(active) < max_concurrent:
                        next_job = queue.pop(0)
                        try:
                            next_vid = submit_video(
                                next_job, manifest["mode"],
                                manifest["template_id"], api_key
                            )
                            active[next_vid] = next_job["index"]
                            next_job["status"] = "processing"
                            next_job["video_id"] = next_vid
                            next_job["attempts"] += 1
                            next_job["error"] = None
                            log_json({"action": "submitted",
                                       "index": next_job["index"],
                                       "video_id": next_vid,
                                       "attempt": next_job["attempts"]})
                        except HeyGenAPIError as e:
                            next_job["attempts"] += 1
                            next_job["error"] = e.message
                            if next_job["attempts"] < max_retries:
                                next_job["status"] = "pending"
                                queue.append(next_job)
                            else:
                                next_job["status"] = "failed"

                # else: still processing, do nothing

            except HeyGenAPIError as e:
                # Tolerate transient errors during polling (5xx)
                if e.status_code >= 500:
                    log_json({"action": "poll_error_transient",
                               "index": job_index, "video_id": video_id,
                               "error": e.message})
                else:
                    # Non-transient error -- mark failed
                    job["error"] = e.message
                    job["status"] = "failed"
                    del active[video_id]
                    log_json({"action": "poll_error_permanent",
                               "index": job_index, "error": e.message})

        update_manifest_counts(manifest)
        save_manifest(manifest, manifest_path)

        if active:
            log_json({
                "action": "poll_cycle",
                "active": len(active),
                "completed": manifest["completed"],
                "failed": manifest["failed"],
                "pending": manifest["pending"],
            })
            time.sleep(poll_interval)

    # Final save
    update_manifest_counts(manifest)
    save_manifest(manifest, manifest_path)


# ---------------------------------------------------------------------------
# Status Report
# ---------------------------------------------------------------------------

def show_status(manifest: dict) -> None:
    """Print a summary of batch progress."""
    total = manifest["total"]
    completed = manifest["completed"]
    failed = manifest["failed"]
    pending = manifest["pending"]

    # Count actively processing jobs
    active_count = sum(1 for j in manifest["jobs"] if j["status"] == "processing")

    # Estimate cost ($1 per video is a rough HeyGen estimate)
    estimated_cost = completed * 1.0

    output_json({
        "success": True,
        "summary": (
            f"Batch Progress: {completed}/{total} completed, "
            f"{failed} failed, {pending} pending"
        ),
        "active": f"{active_count} videos processing",
        "estimated_cost": f"${estimated_cost:.2f}",
        "total": total,
        "completed": completed,
        "failed": failed,
        "pending": pending,
        "processing": active_count,
        "mode": manifest["mode"],
        "created": manifest["created"],
        "manifest_file": manifest.get("input_file", "unknown"),
    })


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Batch-generate HeyGen avatar videos with concurrency control"
    )

    # Mutually exclusive top-level modes
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--input",
                       help="Input CSV or JSON file with video parameters")
    group.add_argument("--resume",
                       help="Resume from an existing manifest JSON file")
    group.add_argument("--status",
                       help="Show progress of a batch manifest")

    parser.add_argument("--mode", choices=["agent", "template"],
                        help="Generation mode (required with --input)")
    parser.add_argument("--template-id",
                        help="Template ID (required for template mode)")
    parser.add_argument("--max-concurrent", type=int, default=3,
                        help="Maximum simultaneous video generations (default: 3, range 1-10)")
    parser.add_argument("--max-retries", type=int, default=3,
                        help="Maximum retry attempts per failed video (default: 3, range 0-10)")
    parser.add_argument("--max-batch", type=int, default=500,
                        help="Maximum rows to process per batch (default: 500, safety limit)")
    parser.add_argument("--output-dir",
                        help="Output directory (default: ~/Documents/lipnardo_videos/)")
    parser.add_argument("--api-key",
                        help="HeyGen API key (or set HEYGEN_API_KEY)")

    args = parser.parse_args()

    # Validate numeric bounds (prevent silent no-ops and runaway retries)
    if args.max_concurrent < 1 or args.max_concurrent > 10:
        parser.error("--max-concurrent must be between 1 and 10")
    if args.max_retries < 0 or args.max_retries > 10:
        parser.error("--max-retries must be between 0 and 10")
    if args.max_batch < 1:
        parser.error("--max-batch must be >= 1")

    # -- Status mode --
    if args.status:
        manifest = load_manifest(args.status)
        update_manifest_counts(manifest)
        show_status(manifest)
        return

    # -- Resume mode --
    if args.resume:
        api_key = resolve_api_key(args.api_key)
        manifest_path = os.path.abspath(args.resume)
        manifest = load_manifest(manifest_path)

        # Allow overriding max_concurrent on resume
        if args.max_concurrent != 3:
            manifest["max_concurrent"] = args.max_concurrent
        if args.max_retries != 3:
            manifest["max_retries"] = args.max_retries

        log_json({"action": "resume", "manifest": manifest_path,
                   "completed": manifest.get("completed", 0),
                   "total": manifest.get("total", 0)})

        try:
            run_batch(manifest, manifest_path, api_key)
        except KeyboardInterrupt:
            update_manifest_counts(manifest)
            save_manifest(manifest, manifest_path)
            log_json({"action": "interrupted", "manifest": manifest_path})
            error_exit("Batch interrupted. Resume with: "
                       f"python3 batch_generate.py --resume {manifest_path}")

        show_status(manifest)
        return

    # -- New batch mode --
    if not args.mode:
        parser.error("--mode is required when using --input")
    if args.mode == "template" and not args.template_id:
        parser.error("--template-id is required for template mode")

    api_key = resolve_api_key(args.api_key)
    rows = load_input(args.input)

    # Safety gate: enforce batch size limit
    if len(rows) > args.max_batch:
        error_exit(
            f"Input has {len(rows)} rows, exceeding --max-batch limit of {args.max_batch}. "
            f"To override, run with: --max-batch {len(rows)}"
        )

    # Cost warning: estimate worst-case cost so the user sees it before charging starts
    avg_duration_seconds = 60
    rate_per_second = 0.0667  # use highest rate (digital_twin/precision) to be conservative
    est_cost_usd = round(len(rows) * avg_duration_seconds * rate_per_second, 2)
    log_json({
        "warning": (
            f"About to start a batch of {len(rows)} videos. "
            f"Worst-case estimated cost: ${est_cost_usd} "
            f"(assuming {avg_duration_seconds}s avg at ${rate_per_second}/sec). "
            f"Run 'credit_check.py balance' to verify your remaining quota."
        )
    })

    output_dir = str(ensure_output_dir(args.output_dir))

    manifest = create_manifest(
        input_file=args.input,
        mode=args.mode,
        template_id=args.template_id,
        max_concurrent=args.max_concurrent,
        output_dir=output_dir,
        rows=rows,
    )
    manifest["max_retries"] = args.max_retries

    # Generate manifest filename with timestamp
    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    manifest_filename = f"batch_manifest_{ts}.json"
    manifest_path = os.path.join(output_dir, manifest_filename)

    save_manifest(manifest, manifest_path)
    log_json({"action": "manifest_created", "path": manifest_path,
               "total_jobs": len(rows)})

    try:
        run_batch(manifest, manifest_path, api_key)
    except KeyboardInterrupt:
        update_manifest_counts(manifest)
        save_manifest(manifest, manifest_path)
        log_json({"action": "interrupted", "manifest": manifest_path})
        error_exit("Batch interrupted. Resume with: "
                   f"python3 batch_generate.py --resume {manifest_path}")

    # Final report
    show_status(manifest)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Lipnardo -- HeyGen Credit & Cost Tracker

Track HeyGen video generation costs, check remaining balance, estimate
batch costs, and view usage summaries.

Usage:
    credit_check.py balance [--api-key KEY]
    credit_check.py estimate --feature FEATURE --duration SECONDS
    credit_check.py estimate-batch --input FILE --feature FEATURE --avg-duration SECONDS
    credit_check.py log --feature FEATURE --duration SECONDS --video-id ID [--prompt SUMMARY]
    credit_check.py summary
"""

import argparse
import csv
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from heygen_client import resolve_api_key, api_request, output_json, error_exit

LEDGER_PATH = Path.home() / ".heygen" / "costs.json"

# HeyGen pricing per second of output (USD)
PRICING_PER_SECOND = {
    "avatar3": 0.0167,
    "video_agent": 0.0333,
    "photo_avatar": 0.05,
    "digital_twin": 0.0667,
    "translate_fast": 0.0333,
    "translate_quality": 0.0667,
    "tts": 0.000667,
}


# ---------------------------------------------------------------------------
# Ledger helpers
# ---------------------------------------------------------------------------

def _load_ledger():
    """Load the cost ledger from disk."""
    if not LEDGER_PATH.exists():
        return {"total_cost": 0.0, "total_videos": 0, "entries": [], "daily": {}}
    with open(LEDGER_PATH, "r") as f:
        return json.load(f)


def _save_ledger(ledger):
    """Save the cost ledger atomically with restrictive permissions.

    Writes to a temp file (mode 0600), then atomically renames over the
    final path. The parent directory is created with mode 0700. This
    prevents corruption from interrupted writes and protects the cost
    history from being read by other local users.
    """
    LEDGER_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    tmp_path = str(LEDGER_PATH) + ".tmp"
    fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(ledger, f, indent=2)
    os.replace(tmp_path, str(LEDGER_PATH))


def _lookup_cost(feature, duration):
    """Calculate cost for a feature and duration."""
    if duration < 0:
        error_exit(f"Duration must be non-negative, got: {duration}")
    rate = PRICING_PER_SECOND.get(feature)
    if rate is None:
        error_exit(
            f"Unknown feature '{feature}'. Valid features: {', '.join(sorted(PRICING_PER_SECOND))}"
        )
    return round(rate * duration, 4)


def _count_rows(file_path):
    """Count data rows in a CSV or JSON file."""
    p = Path(file_path)
    if not p.exists():
        error_exit(f"Input file not found: {file_path}")

    suffix = p.suffix.lower()

    if suffix == ".csv":
        with open(p, "r", newline="") as f:
            reader = csv.reader(f)
            # Skip header row
            try:
                next(reader)
            except StopIteration:
                return 0
            return sum(1 for _ in reader)

    if suffix == ".json":
        with open(p, "r") as f:
            data = json.load(f)
        if isinstance(data, list):
            return len(data)
        error_exit("JSON input must be an array of objects")

    error_exit(f"Unsupported file type '{suffix}'. Use .csv or .json")


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------

def cmd_balance(args):
    """Check remaining HeyGen credit balance."""
    api_key = resolve_api_key(args.api_key)
    try:
        resp = api_request("GET", "/v2/user/remaining_quota", api_key)
    except Exception as e:
        error_exit(f"Failed to fetch balance: {e}")

    data = resp.get("data", resp)
    output_json({
        "command": "balance",
        "remaining_quota": data,
    })


def cmd_estimate(args):
    """Estimate cost for a single video."""
    cost = _lookup_cost(args.feature, args.duration)
    rate = PRICING_PER_SECOND[args.feature]
    output_json({
        "command": "estimate",
        "feature": args.feature,
        "duration_seconds": args.duration,
        "rate_per_second": rate,
        "estimated_cost_usd": cost,
    })


def cmd_estimate_batch(args):
    """Estimate cost for a batch of videos from a CSV/JSON file."""
    row_count = _count_rows(args.input)
    cost_per_video = _lookup_cost(args.feature, args.avg_duration)
    total_cost = round(cost_per_video * row_count, 4)
    rate = PRICING_PER_SECOND[args.feature]
    output_json({
        "command": "estimate-batch",
        "input_file": args.input,
        "row_count": row_count,
        "feature": args.feature,
        "avg_duration_seconds": args.avg_duration,
        "rate_per_second": rate,
        "cost_per_video_usd": cost_per_video,
        "total_estimated_cost_usd": total_cost,
    })


def cmd_log(args):
    """Log a completed video generation to the ledger."""
    cost = _lookup_cost(args.feature, args.duration)
    ledger = _load_ledger()

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")

    entry = {
        "ts": now,
        "feature": args.feature,
        "duration": args.duration,
        "cost": cost,
        "video_id": args.video_id,
        "prompt": (args.prompt or "")[:100],
    }

    ledger["entries"].append(entry)
    ledger["total_cost"] = round(ledger["total_cost"] + cost, 4)
    ledger["total_videos"] += 1

    if today not in ledger["daily"]:
        ledger["daily"][today] = {"cost": 0.0, "count": 0}
    ledger["daily"][today]["count"] += 1
    ledger["daily"][today]["cost"] = round(ledger["daily"][today]["cost"] + cost, 4)

    _save_ledger(ledger)
    output_json({
        "command": "log",
        "logged": True,
        "entry": entry,
        "total_cost": ledger["total_cost"],
        "total_videos": ledger["total_videos"],
    })


def cmd_summary(args):
    """Show usage summary from the ledger."""
    ledger = _load_ledger()

    # Per-feature breakdown
    feature_breakdown = {}
    for entry in ledger.get("entries", []):
        feat = entry.get("feature", "unknown")
        if feat not in feature_breakdown:
            feature_breakdown[feat] = {"cost": 0.0, "count": 0}
        feature_breakdown[feat]["count"] += 1
        feature_breakdown[feat]["cost"] = round(
            feature_breakdown[feat]["cost"] + entry.get("cost", 0), 4
        )

    # Daily breakdown (last 14 days, sorted descending)
    daily = ledger.get("daily", {})
    sorted_days = sorted(daily.keys(), reverse=True)[:14]
    daily_breakdown = {day: daily[day] for day in sorted_days}

    output_json({
        "command": "summary",
        "total_cost_usd": ledger.get("total_cost", 0.0),
        "total_videos": ledger.get("total_videos", 0),
        "daily_breakdown": daily_breakdown,
        "feature_breakdown": feature_breakdown,
    })


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description="Lipnardo -- HeyGen Credit & Cost Tracker")
    sub = parser.add_subparsers(dest="command", required=True)

    # balance
    p_balance = sub.add_parser("balance", help="Check remaining HeyGen credit balance")
    p_balance.add_argument("--api-key", help="HeyGen API key (or set HEYGEN_API_KEY env)")

    # estimate
    p_estimate = sub.add_parser("estimate", help="Estimate cost for a single video")
    p_estimate.add_argument("--feature", required=True,
                            choices=sorted(PRICING_PER_SECOND.keys()),
                            help="HeyGen feature type")
    p_estimate.add_argument("--duration", required=True, type=float,
                            help="Duration in seconds")

    # estimate-batch
    p_batch = sub.add_parser("estimate-batch", help="Estimate cost for a batch of videos")
    p_batch.add_argument("--input", required=True, help="Path to CSV or JSON input file")
    p_batch.add_argument("--feature", required=True,
                         choices=sorted(PRICING_PER_SECOND.keys()),
                         help="HeyGen feature type")
    p_batch.add_argument("--avg-duration", required=True, type=float,
                         help="Average video duration in seconds")

    # log
    p_log = sub.add_parser("log", help="Log a completed video generation")
    p_log.add_argument("--feature", required=True,
                       choices=sorted(PRICING_PER_SECOND.keys()),
                       help="HeyGen feature type")
    p_log.add_argument("--duration", required=True, type=float,
                       help="Duration in seconds")
    p_log.add_argument("--video-id", required=True, help="HeyGen video ID")
    p_log.add_argument("--prompt", default="", help="Brief prompt/description (truncated to 100 chars)")

    # summary
    sub.add_parser("summary", help="Show usage summary from the ledger")

    args = parser.parse_args()
    cmds = {
        "balance": cmd_balance,
        "estimate": cmd_estimate,
        "estimate-batch": cmd_estimate_batch,
        "log": cmd_log,
        "summary": cmd_summary,
    }
    cmds[args.command](args)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations
"""Lipnardo -- Download HeyGen Video

Purpose:
    Download generated HeyGen videos before their 7-day URL expiry.
    Supports downloading by video ID (fetches URL from API) or by direct URL.

Usage:
    # Download by video ID (fetches video_url from API):
    python3 download_video.py --video-id VIDEO_ID [--output-dir DIR] [--api-key KEY]

    # Download by direct URL:
    python3 download_video.py --url URL --output /path/to/file.mp4

Dependencies: Python 3.8+ stdlib only
"""

import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from heygen_client import (
    resolve_api_key,
    api_request,
    download_file,
    ensure_output_dir,
    timestamp_filename,
    output_json,
    log_json,
    error_exit,
    sanitize_id,
    HeyGenAPIError,
)


def download_by_video_id(video_id: str, api_key: str | None, output_dir: str | None) -> None:
    """Fetch video metadata from the API and download the completed video."""
    api_key = resolve_api_key(api_key)

    log_json({"action": "fetch_video_status", "video_id": video_id})

    try:
        resp = api_request("GET", f"/v2/videos/{video_id}", api_key)
    except HeyGenAPIError as e:
        error_exit(e.message, status_code=e.status_code, video_id=video_id)

    data = resp.get("data", resp)
    status = data.get("status", "unknown")

    if status != "completed":
        error_exit(
            f"Video is not ready for download. Current status: {status}",
            video_id=video_id,
            status=status,
        )

    video_url = data.get("video_url", "")
    if not video_url:
        error_exit(
            "Video is completed but no video_url was returned by the API.",
            video_id=video_id,
        )

    out_dir = ensure_output_dir(output_dir)
    output_path = str(out_dir / f"{sanitize_id(video_id, 'video_id')}.mp4")

    log_json({"action": "downloading", "video_id": video_id, "output": output_path})

    try:
        saved_path = download_file(video_url, output_path)
    except Exception as e:
        msg = str(e)
        if "404" in msg or "410" in msg:
            error_exit(
                f"Download failed (URL may have expired): {msg}. "
                f"Try re-fetching the URL with: --video-id {video_id}",
                video_id=video_id,
                url=video_url,
            )
        error_exit(
            f"Download failed: {msg}",
            video_id=video_id,
            url=video_url,
        )

    size_bytes = os.path.getsize(saved_path)
    output_json({
        "success": True,
        "path": saved_path,
        "size_bytes": size_bytes,
        "video_id": video_id,
    })


def download_by_url(url: str, output_path: str) -> None:
    """Download a video directly from a URL to a specified path."""
    log_json({"action": "downloading", "url": url, "output": output_path})

    try:
        saved_path = download_file(url, output_path)
    except Exception as e:
        msg = str(e)
        if "404" in msg or "410" in msg:
            error_exit(
                f"Download failed (URL may have expired): {msg}. "
                "If this is a HeyGen video, try re-fetching via --video-id instead.",
                url=url,
            )
        error_exit(
            f"Download failed: {msg}",
            url=url,
        )

    size_bytes = os.path.getsize(saved_path)
    output_json({
        "success": True,
        "path": saved_path,
        "size_bytes": size_bytes,
    })


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Download HeyGen videos before their 7-day URL expiry.",
    )
    parser.add_argument(
        "--video-id",
        help="HeyGen video ID to fetch and download",
    )
    parser.add_argument(
        "--url",
        help="Direct download URL",
    )
    parser.add_argument(
        "--output",
        help="Exact output file path (required with --url)",
    )
    parser.add_argument(
        "--output-dir",
        help="Output directory (used with --video-id, default: ~/Documents/lipnardo_videos/)",
    )
    parser.add_argument(
        "--api-key",
        help="HeyGen API key override (or set HEYGEN_API_KEY env)",
    )

    args = parser.parse_args()

    # Validation: must provide exactly one of --video-id or --url
    if args.video_id and args.url:
        error_exit("Provide either --video-id or --url, not both.")
    if not args.video_id and not args.url:
        error_exit("Provide either --video-id or --url.")

    # Validation: --url requires --output
    if args.url and not args.output:
        error_exit("--output is required when using --url.")

    if args.video_id:
        download_by_video_id(args.video_id, args.api_key, args.output_dir)
    else:
        download_by_url(args.url, args.output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations
"""Lipnardo -- Video Generation (Agent + Studio modes)

Generate HeyGen avatar videos using either the AI-driven Video Agent (v3)
or the precise multi-scene Studio API (v2).

Usage:
    # Video Agent mode (AI-driven, prompt-to-video):
    generate_video.py --mode agent --prompt "A woman explaining quarterly results"

    # Video Agent with specific avatar/voice:
    generate_video.py --mode agent --prompt "Welcome to our company" --avatar-id Kristin_public_3_20240108 --voice-id en-US-JennyNeural

    # Studio mode (multi-scene, precise control):
    generate_video.py --mode studio --config scenes.json

    # Skip download, just return video_id:
    generate_video.py --mode agent --prompt "Test video" --no-download

Options:
    --mode {agent,studio}   Generation mode (required)
    --prompt TEXT            Script/prompt text (required for agent mode)
    --config FILE            JSON scene config (required for studio mode)
    --avatar-id ID          Avatar ID (optional, AI selects in agent mode)
    --voice-id ID           Voice ID (optional)
    --aspect-ratio RATIO    16:9 or 9:16 (default: 16:9)
    --resolution RES        720p or 1080p (default: 1080p)
    --test                  Generate watermarked test video (free)
    --no-download           Skip download, return video_id only
    --callback-id ID        Webhook callback correlation ID
    --api-key KEY           HeyGen API key override
    --output-dir DIR        Output directory
"""

import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from heygen_client import (
    resolve_api_key,
    api_request,
    poll_video_status,
    download_file,
    ensure_output_dir,
    timestamp_filename,
    output_json,
    log_json,
    error_exit,
    sanitize_id,
    HeyGenAPIError,
)


# ---------------------------------------------------------------------------
# Video Agent (v3)
# ---------------------------------------------------------------------------

def submit_agent_video(api_key: str, prompt: str,
                       avatar_id: str | None = None,
                       voice_id: str | None = None,
                       aspect_ratio: str = "16:9",
                       test: bool = False,
                       callback_id: str | None = None) -> str:
    """Submit a video generation request via the Video Agent API (v3).

    Returns:
        video_id string
    """
    body = {"prompt": prompt}

    if avatar_id:
        body["avatar_id"] = avatar_id
    if voice_id:
        body["voice_id"] = voice_id
    if aspect_ratio:
        body["aspect_ratio"] = aspect_ratio
    if test:
        body["test"] = True
    if callback_id:
        body["callback_id"] = callback_id

    log_json({"action": "submit", "mode": "agent", "prompt": prompt[:80]})

    resp = api_request("POST", "/v3/video-agents", api_key, body=body)
    data = resp.get("data", resp)
    video_id = data.get("video_id")

    if not video_id:
        error_exit(f"No video_id in response: {json.dumps(resp)}")

    log_json({"action": "submitted", "video_id": video_id})
    return video_id


# ---------------------------------------------------------------------------
# Studio Video (v2)
# ---------------------------------------------------------------------------

def submit_studio_video(api_key: str, config: dict,
                        dimension: dict | None = None,
                        aspect_ratio: str = "16:9",
                        test: bool = False,
                        callback_id: str | None = None) -> str:
    """Submit a video generation request via the Studio API (v2).

    Config should contain a 'scenes' or 'video_inputs' array with scene
    definitions including character, voice, and background settings.

    Returns:
        video_id string
    """
    # Accept either 'scenes' or 'video_inputs' key
    video_inputs = config.get("video_inputs", config.get("scenes", []))
    if not video_inputs:
        error_exit("Studio config must contain 'scenes' or 'video_inputs' array")

    body = {
        "video_inputs": video_inputs,
        "aspect_ratio": aspect_ratio,
    }

    # Set dimension based on aspect ratio if not specified
    if dimension:
        body["dimension"] = dimension
    elif aspect_ratio == "9:16":
        body["dimension"] = {"width": 1080, "height": 1920}
    else:
        body["dimension"] = {"width": 1920, "height": 1080}

    if test:
        body["test"] = True
    if callback_id:
        body["callback_id"] = callback_id

    # Copy through any additional top-level config keys
    for key in ("title", "caption", "use_avatar_iv_model"):
        if key in config:
            body[key] = config[key]

    log_json({"action": "submit", "mode": "studio", "scenes": len(video_inputs)})

    resp = api_request("POST", "/v2/video/generate", api_key, body=body)
    data = resp.get("data", resp)
    video_id = data.get("video_id")

    if not video_id:
        error_exit(f"No video_id in response: {json.dumps(resp)}")

    log_json({"action": "submitted", "video_id": video_id})
    return video_id


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Generate HeyGen avatar videos (Agent or Studio mode)"
    )
    parser.add_argument("--mode", required=True, choices=["agent", "studio"],
                        help="Generation mode: agent (AI-driven) or studio (multi-scene)")
    parser.add_argument("--prompt", help="Script/prompt text (required for agent mode)")
    parser.add_argument("--config", help="JSON scene config file (required for studio mode)")
    parser.add_argument("--avatar-id", help="Avatar ID (optional)")
    parser.add_argument("--voice-id", help="Voice ID (optional)")
    parser.add_argument("--aspect-ratio", default="16:9", choices=["16:9", "9:16"],
                        help="Aspect ratio (default: 16:9)")
    parser.add_argument("--resolution", default="1080p", choices=["720p", "1080p"],
                        help="Resolution (default: 1080p)")
    parser.add_argument("--test", action="store_true",
                        help="Generate watermarked test video (free)")
    parser.add_argument("--no-download", action="store_true",
                        help="Skip download, return video_id only")
    parser.add_argument("--callback-id", help="Webhook callback correlation ID")
    parser.add_argument("--api-key", help="HeyGen API key (or set HEYGEN_API_KEY)")
    parser.add_argument("--output-dir",
                        help="Output directory (default: ~/Documents/lipnardo_videos/)")

    args = parser.parse_args()

    # Validate mode-specific requirements
    if args.mode == "agent" and not args.prompt:
        parser.error("--prompt is required for agent mode")
    if args.mode == "studio" and not args.config:
        parser.error("--config is required for studio mode")

    # Resolve API key
    api_key = resolve_api_key(args.api_key)

    # Submit video
    start_time = time.time()
    video_id = None

    try:
        if args.mode == "agent":
            video_id = submit_agent_video(
                api_key=api_key,
                prompt=args.prompt,
                avatar_id=args.avatar_id,
                voice_id=args.voice_id,
                aspect_ratio=args.aspect_ratio,
                test=args.test,
                callback_id=args.callback_id,
            )
        else:
            # Load studio config
            config_path = Path(args.config)
            if not config_path.exists():
                error_exit(f"Config file not found: {args.config}")
            with open(config_path, "r") as f:
                config = json.load(f)

            video_id = submit_studio_video(
                api_key=api_key,
                config=config,
                aspect_ratio=args.aspect_ratio,
                test=args.test,
                callback_id=args.callback_id,
            )

        # If callback_id is set, user wants webhook-based async — don't poll
        if args.callback_id:
            output_json({
                "success": True,
                "video_id": video_id,
                "mode": args.mode,
                "async": True,
                "message": f"Video submitted. Will notify via webhook callback_id={args.callback_id}",
            })
            return

        # If no-download, just return the video_id
        if args.no_download:
            output_json({
                "success": True,
                "video_id": video_id,
                "mode": args.mode,
                "message": "Video submitted. Use --video-id with download_video.py to download later.",
            })
            return

        # Poll for completion
        log_json({"action": "polling", "video_id": video_id})
        result = poll_video_status(video_id, api_key)

        elapsed = round(time.time() - start_time, 1)
        video_url = result.get("video_url", "")
        duration = result.get("duration", 0)

        if not video_url:
            error_exit(f"Video completed but no video_url in response: {json.dumps(result)}")

        # Download
        output_dir = ensure_output_dir(args.output_dir)
        filename = f"{sanitize_id(video_id, 'video_id')}.mp4"
        output_path = str(output_dir / filename)

        log_json({"action": "downloading", "video_id": video_id, "url": video_url[:80]})
        saved_path = download_file(video_url, output_path)

        file_size = os.path.getsize(saved_path)

        output_json({
            "success": True,
            "video_id": video_id,
            "video_url": video_url,
            "path": saved_path,
            "duration_seconds": duration,
            "elapsed_seconds": elapsed,
            "mode": args.mode,
            "size_bytes": file_size,
            "test": args.test,
        })

    except HeyGenAPIError as e:
        output_json({
            "error": True,
            "video_id": video_id,
            "status_code": e.status_code,
            "error_code": e.error_code,
            "message": e.message,
            "endpoint": e.endpoint,
            "elapsed_seconds": round(time.time() - start_time, 1),
        })
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations
"""Lipnardo -- Shared HeyGen API Client

Importable module providing authentication, HTTP requests, polling,
and download utilities for all Lipnardo scripts.

NOT a standalone CLI. Import via:
    import sys, os
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from heygen_client import resolve_api_key, api_request, poll_video_status, ...

Dependencies: Python 3.8+ stdlib only (urllib.request, json, time, etc.)
Optional: pip install requests (auto-detected, used if available)
"""

import json
import os
import re
import ssl
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

API_BASE = "https://api.heygen.com"
CONFIG_PATH = Path.home() / ".heygen" / "config.json"
DEFAULT_TIMEOUT = 30

# Try to use requests if available, fall back to urllib
try:
    import requests as _requests_lib
    _HAS_REQUESTS = True
except ImportError:
    _HAS_REQUESTS = False


class HeyGenAPIError(Exception):
    """Wraps HeyGen API errors with structured information."""

    def __init__(self, status_code: int, error_code: str = "", message: str = "",
                 endpoint: str = ""):
        self.status_code = status_code
        self.error_code = error_code
        self.message = message
        self.endpoint = endpoint
        super().__init__(f"HeyGen API error {status_code} on {endpoint}: {message}")

    def to_dict(self) -> dict:
        return {
            "error": True,
            "status_code": self.status_code,
            "error_code": self.error_code,
            "message": self.message,
            "endpoint": self.endpoint,
        }


# ---------------------------------------------------------------------------
# Output helpers
# ---------------------------------------------------------------------------

def output_json(data: dict) -> None:
    """Print structured JSON to stdout as the script's final result."""
    print(json.dumps(data, indent=2))


def log_json(data: dict) -> None:
    """Print structured JSON to stderr for progress/debug logging."""
    print(json.dumps(data), file=sys.stderr)


def error_exit(message: str, code: int = 1, **extra) -> None:
    """Print a structured error to stdout and exit."""
    result = {"error": True, "message": message}
    result.update(extra)
    output_json(result)
    sys.exit(code)


# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------

def resolve_api_key(cli_key: str | None = None) -> str:
    """Resolve HeyGen API key using tiered priority.

    Priority:
        1. Explicit cli_key argument (from --api-key flag)
        2. HEYGEN_API_KEY environment variable
        3. ~/.heygen/config.json file
        4. Exit with error and setup instructions

    Returns:
        API key string

    Raises:
        SystemExit if no key found
    """
    # Priority 1: CLI argument
    if cli_key:
        return cli_key

    # Priority 2: Environment variable
    env_key = os.environ.get("HEYGEN_API_KEY")
    if env_key:
        return env_key

    # Priority 3: Config file
    if CONFIG_PATH.exists():
        # Warn if config file is readable by group/others (API key exposure risk)
        import stat
        try:
            mode = CONFIG_PATH.stat().st_mode
            if mode & (stat.S_IRGRP | stat.S_IROTH):
                log_json({
                    "warning": f"Config file {CONFIG_PATH} is readable by other users. "
                               f"Run: chmod 600 {CONFIG_PATH}"
                })
        except OSError:
            pass
        try:
            with open(CONFIG_PATH, "r") as f:
                config = json.load(f)
            file_key = config.get("api_key", "").strip()
            if file_key:
                return file_key
        except (json.JSONDecodeError, OSError) as e:
            log_json({"warning": f"Failed to read {CONFIG_PATH}: {e}"})

    # Priority 4: Error with instructions
    error_exit(
        "No HeyGen API key found. Configure one of:\n"
        "  1. Export HEYGEN_API_KEY=your-key-here\n"
        "  2. Pass --api-key your-key-here\n"
        f"  3. Create {CONFIG_PATH} with: {{\"api_key\": \"your-key-here\"}}\n"
        "\nGet your key from: https://app.heygen.com/settings/api"
    )


# ---------------------------------------------------------------------------
# HTTP Client
# ---------------------------------------------------------------------------

def _urllib_request(method: str, url: str, headers: dict,
                    body: bytes | None = None, timeout: int = DEFAULT_TIMEOUT) -> tuple:
    """Make an HTTP request using urllib (stdlib).

    Returns:
        (status_code, response_dict)
    """
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    # Create SSL context that handles most certificate scenarios
    ctx = ssl.create_default_context()
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
            raw = resp.read().decode("utf-8")
            try:
                data = json.loads(raw) if raw else {}
            except json.JSONDecodeError:
                data = {"raw_response": raw}
            return resp.status, data
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            data = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            data = {"raw_response": raw}
        return e.code, data
    except urllib.error.URLError as e:
        raise HeyGenAPIError(0, "CONNECTION_ERROR", str(e.reason), url)
    except TimeoutError:
        raise HeyGenAPIError(0, "TIMEOUT", f"Request timed out after {timeout}s", url)


def _requests_request(method: str, url: str, headers: dict,
                      body: bytes | None = None, timeout: int = DEFAULT_TIMEOUT) -> tuple:
    """Make an HTTP request using the requests library (optional).

    Returns:
        (status_code, response_dict)
    """
    kwargs = {"headers": headers, "timeout": timeout}
    if body:
        kwargs["data"] = body
    resp = _requests_lib.request(method, url, **kwargs)
    try:
        data = resp.json() if resp.text else {}
    except ValueError:
        data = {"raw_response": resp.text}
    return resp.status_code, data


def api_request(method: str, path: str, api_key: str,
                body: dict | None = None, timeout: int = DEFAULT_TIMEOUT,
                max_retries: int = 3) -> dict:
    """Make an authenticated request to the HeyGen API.

    Args:
        method: HTTP method (GET, POST, DELETE)
        path: API path (e.g., /v2/videos/abc123)
        api_key: HeyGen API key
        body: Request body dict (JSON-encoded automatically)
        timeout: Request timeout in seconds
        max_retries: Max retries for 429/5xx errors

    Returns:
        Parsed response dict

    Raises:
        HeyGenAPIError on non-retryable errors or exhausted retries
    """
    url = f"{API_BASE}{path}" if path.startswith("/") else f"{API_BASE}/{path}"
    headers = {
        "X-Api-Key": api_key,
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    encoded_body = json.dumps(body).encode("utf-8") if body else None

    # Select HTTP backend
    do_request = _requests_request if _HAS_REQUESTS else _urllib_request

    last_error = None
    for attempt in range(max_retries + 1):
        try:
            status, data = do_request(method, url, headers, encoded_body, timeout)
        except HeyGenAPIError:
            raise
        except Exception as e:
            raise HeyGenAPIError(0, "REQUEST_FAILED", str(e), path)

        # Success
        if 200 <= status < 300:
            return data

        # Rate limit (429) — retry with exponential backoff
        if status == 429:
            last_error = HeyGenAPIError(status, "RATE_LIMITED",
                                        data.get("message", "Rate limited"), path)
            if attempt < max_retries:
                delay = (2 ** attempt) * 2  # 2s, 4s, 8s
                log_json({"retry": True, "status": 429, "attempt": attempt + 1,
                           "delay_seconds": delay, "endpoint": path})
                time.sleep(delay)
                continue

        # Server errors (5xx) — retry with fixed delay
        if 500 <= status < 600:
            last_error = HeyGenAPIError(status, "SERVER_ERROR",
                                        data.get("message", f"Server error {status}"), path)
            if attempt < max_retries:
                delay = 5
                log_json({"retry": True, "status": status, "attempt": attempt + 1,
                           "delay_seconds": delay, "endpoint": path})
                time.sleep(delay)
                continue

        # Client errors (4xx except 429) — no retry
        error_msg = data.get("message") or data.get("error", {}).get("message", f"HTTP {status}")
        error_code = data.get("code") or data.get("error", {}).get("code", "")
        raise HeyGenAPIError(status, str(error_code), error_msg, path)

    # Exhausted retries
    if last_error:
        raise last_error
    raise HeyGenAPIError(0, "UNKNOWN", "Request failed after retries", path)


# ---------------------------------------------------------------------------
# Multipart upload (for assets)
# ---------------------------------------------------------------------------

def api_upload(path: str, api_key: str, file_path: str,
               fields: dict | None = None, timeout: int = 120) -> dict:
    """Upload a file to the HeyGen API using multipart/form-data.

    Args:
        path: API path (e.g., /v1/asset)
        api_key: HeyGen API key
        file_path: Local path to file to upload
        fields: Additional form fields
        timeout: Request timeout in seconds

    Returns:
        Parsed response dict
    """
    import mimetypes
    import uuid

    url = f"{API_BASE}{path}" if path.startswith("/") else f"{API_BASE}/{path}"
    boundary = uuid.uuid4().hex
    filename = os.path.basename(file_path)
    mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"

    # Build multipart body
    parts = []

    # Add form fields
    if fields:
        for key, value in fields.items():
            parts.append(
                f"--{boundary}\r\n"
                f'Content-Disposition: form-data; name="{key}"\r\n\r\n'
                f"{value}\r\n"
            )

    # Add file
    with open(file_path, "rb") as f:
        file_data = f.read()

    # Escape quotes/backslashes in filename to prevent header injection
    safe_filename = filename.replace("\\", "_").replace('"', "_").replace("\r", "_").replace("\n", "_")
    file_header = (
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="file"; filename="{safe_filename}"\r\n'
        f"Content-Type: {mime_type}\r\n\r\n"
    )
    closing = f"\r\n--{boundary}--\r\n"

    body = b""
    for part in parts:
        body += part.encode("utf-8")
    body += file_header.encode("utf-8") + file_data + closing.encode("utf-8")

    headers = {
        "X-Api-Key": api_key,
        "Content-Type": f"multipart/form-data; boundary={boundary}",
        "Accept": "application/json",
    }

    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    ctx = ssl.create_default_context()
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
            raw = resp.read().decode("utf-8")
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            data = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            data = {"raw_response": raw}
        error_msg = data.get("message", f"Upload failed with HTTP {e.code}")
        raise HeyGenAPIError(e.code, "UPLOAD_FAILED", error_msg, path)


# ---------------------------------------------------------------------------
# Video Status Polling
# ---------------------------------------------------------------------------

def poll_video_status(video_id: str, api_key: str,
                      initial_interval: int = 10,
                      max_interval: int = 60,
                      max_wait: int = 1800) -> dict:
    """Poll for video completion with exponential backoff.

    Polls GET /v2/videos/{video_id} until status is completed or failed.

    Backoff sequence: 10s, 20s, 40s, 60s, 60s, ...
    Maximum wait: 1800s (30 minutes, matching HeyGen's max video length)

    Args:
        video_id: HeyGen video ID
        api_key: HeyGen API key
        initial_interval: Starting poll interval in seconds
        max_interval: Maximum poll interval in seconds
        max_wait: Maximum total wait time in seconds

    Returns:
        Final status dict from the API

    Raises:
        HeyGenAPIError on timeout or API errors
    """
    interval = initial_interval
    elapsed = 0
    attempt = 0

    while elapsed < max_wait:
        attempt += 1
        try:
            resp = api_request("GET", f"/v2/videos/{video_id}", api_key)
        except HeyGenAPIError as e:
            # Tolerate transient errors during polling
            if e.status_code >= 500:
                log_json({"polling": True, "video_id": video_id, "attempt": attempt,
                           "warning": f"Transient error: {e.message}"})
                time.sleep(interval)
                elapsed += interval
                continue
            raise

        # Extract status — API nests data differently depending on version
        data = resp.get("data", resp)
        status = data.get("status", "unknown")

        log_json({
            "polling": True,
            "video_id": video_id,
            "attempt": attempt,
            "elapsed_seconds": elapsed,
            "video_status": status,
        })

        if status == "completed":
            return data

        if status == "failed":
            error_msg = data.get("error", data.get("message", "Video generation failed"))
            raise HeyGenAPIError(
                200, "GENERATION_FAILED", str(error_msg),
                f"/v2/videos/{video_id}"
            )

        # Wait with exponential backoff
        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 2, max_interval)

    raise HeyGenAPIError(
        0, "POLLING_TIMEOUT",
        f"Video {video_id} did not complete within {max_wait}s",
        f"/v2/videos/{video_id}"
    )


# ---------------------------------------------------------------------------
# Security helpers
# ---------------------------------------------------------------------------

def _validate_url(url: str) -> None:
    """Reject non-HTTP(S) URLs and private/loopback hosts.

    Prevents SSRF via file://, ftp://, gopher:// schemes and downloads
    targeted at localhost/private network addresses.
    """
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        error_exit(f"Only http/https URLs allowed, got: {parsed.scheme}://")
    host = (parsed.hostname or "").lower()
    if host in ("localhost", "127.0.0.1", "::1", "0.0.0.0", ""):
        error_exit("Downloads from localhost are not allowed")


def sanitize_id(value: str, name: str = "id") -> str:
    """Ensure an ID contains only safe characters for filenames and URL paths.

    Allows alphanumerics, underscores, dots, and hyphens. Blocks path traversal
    (`/`, `\\`), shell metacharacters, and other dangerous input. Used to
    sanitize IDs returned from the HeyGen API before incorporating them into
    local file paths or downstream URLs.
    """
    if not value or not re.match(r'^[a-zA-Z0-9_.-]+$', value):
        error_exit(f"Invalid {name}: contains unsafe characters")
    return value


# ---------------------------------------------------------------------------
# File Download
# ---------------------------------------------------------------------------

def download_file(url: str, output_path: str, timeout: int = 300) -> str:
    """Download a file from a URL to a local path.

    Args:
        url: URL to download (must be http/https; localhost is blocked)
        output_path: Local file path to save to
        timeout: Download timeout in seconds

    Returns:
        Absolute path to the downloaded file
    """
    _validate_url(url)
    output = Path(output_path)
    output.parent.mkdir(parents=True, exist_ok=True)

    ctx = ssl.create_default_context()

    if _HAS_REQUESTS:
        resp = _requests_lib.get(url, timeout=timeout, stream=True)
        resp.raise_for_status()
        with open(output, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)
    else:
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
            with open(output, "wb") as f:
                while True:
                    chunk = resp.read(8192)
                    if not chunk:
                        break
                    f.write(chunk)

    return str(output.resolve())


# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------

def ensure_output_dir(output_dir: str | None = None) -> Path:
    """Ensure the output directory exists and return its Path.

    Default: ~/Documents/lipnardo_videos/
    """
    if output_dir:
        p = Path(output_dir)
    else:
        p = Path.home() / "Documents" / "lipnardo_videos"
    p.mkdir(parents=True, exist_ok=True)
    return p


def timestamp_filename(prefix: str = "video", ext: str = ".mp4") -> str:
    """Generate a timestamped filename like video_20260415_103022.mp4"""
    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    return f"{prefix}_{ts}{ext}"


def add_common_args(parser: "argparse.ArgumentParser") -> None:
    """Add common CLI arguments shared by all scripts."""
    parser.add_argument("--api-key", help="HeyGen API key (or set HEYGEN_API_KEY env)")
    parser.add_argument("--output-dir", help="Output directory (default: ~/Documents/lipnardo_videos/)")