Create Unique Video Variations at Scale

Total combinations = crop_values x brightness_values x noise_values x pitch_values x crf_values x hue_values

import random
import math

def euclidean_distance(p1, p2):
    """Calculate normalized distance between parameter sets."""
    dims = ['crop', 'bright', 'noise', 'pitch', 'crf', 'hue']
    ranges = {'crop': 8, 'bright': 0.04, 'noise': 4, 'pitch': 0.012, 'crf': 4, 'hue': 6}
    total = 0
    for d in dims:
        normalized_diff = (p1[d] - p2[d]) / ranges[d]
        total += normalized_diff ** 2
    return math.sqrt(total)

def generate_diverse_params(count, min_distance=0.3, max_attempts=1000):
    """Generate parameters that are maximally spread across the space."""
    params = []

    for _ in range(count):
        best_candidate = None
        best_min_dist = -1

        for _ in range(max_attempts):
            candidate = {
                'crop': random.randint(2, 10),
                'bright': round(random.uniform(-0.02, 0.02), 3),
                'noise': random.randint(3, 7),
                'pitch': round(random.uniform(0.994, 1.006), 4),
                'crf': random.randint(21, 25),
                'hue': random.randint(-3, 3),
            }

            if not params:
                best_candidate = candidate
                break

            min_dist = min(euclidean_distance(candidate, p) for p in params)

            if min_dist > best_min_dist:
                best_min_dist = min_dist
                best_candidate = candidate

            if min_dist >= min_distance:
                break

        params.append(best_candidate)

    return params

import requests
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Optional

API_KEY = "ffsk_your_key"
BASE_URL = "https://renderio.dev/api/v1"
HEADERS = {"Content-Type": "application/json", "X-API-KEY": API_KEY}

@dataclass
class Job:
    index: int
    command_id: str
    params: dict
    status: str = "submitted"
    output_url: Optional[str] = None
    error: Optional[str] = None

def build_command(p):
    vf_parts = [
        f'crop=iw-{p["crop"]}:ih-{p["crop"]}:{p["crop"]//2}:{p["crop"]//2}',
        f'eq=brightness={p["bright"]:.3f}',
        f'noise=alls={p["noise"]}:allf=t',
        f'hue=h={p["hue"]}',
    ]
    vf = ",".join(vf_parts)
    af = f'asetrate=44100*{p["pitch"]:.4f},aresample=44100'
    return f'-i {{in_video}} -vf "{vf}" -af "{af}" -c:v libx264 -crf {p["crf"]} -map_metadata -1 {{out_video}}'

def submit_batch(source_url, params_list, start_index=0):
    """Submit a batch of jobs."""
    jobs = []
    for i, params in enumerate(params_list):
        idx = start_index + i
        cmd = build_command(params)
        try:
            res = requests.post(f"{BASE_URL}/run-ffmpeg-command", headers=HEADERS, json={
                "ffmpeg_command": cmd,
                "input_files": {"in_video": source_url},
                "output_files": {"out_video": f"var_{idx:04d}.mp4"}
            })
            res.raise_for_status()
            jobs.append(Job(index=idx, command_id=res.json()["command_id"], params=params))
        except Exception as e:
            jobs.append(Job(index=idx, command_id="", params=params, status="submit_failed", error=str(e)))
    return jobs

def poll_jobs(jobs, timeout=600):
    """Poll all jobs until complete or timeout."""
    pending = [j for j in jobs if j.status == "submitted"]
    start = time.time()

    while pending and (time.time() - start) < timeout:
        for job in pending:
            try:
                res = requests.get(f"{BASE_URL}/commands/{job.command_id}", headers=HEADERS).json()
                if res["status"] == "SUCCESS":
                    job.status = "SUCCESS"
                    job.output_url = res["output_files"]["out_video"]["storage_url"]
                elif res["status"] == "FAILED":
                    job.status = "FAILED"
                    job.error = res.get("error", "Unknown")
            except Exception as e:
                pass  # Retry on next poll

        pending = [j for j in jobs if j.status == "submitted"]
        if pending:
            time.sleep(3)

    # Mark remaining as timed out
    for job in pending:
        job.status = "timeout"

    return jobs

def generate_variations(source_url, count, batch_size=20):
    """Generate N unique variations with batched processing."""
    params = generate_diverse_params(count)
    all_jobs = []

    for batch_start in range(0, count, batch_size):
        batch_params = params[batch_start:batch_start + batch_size]
        print(f"Submitting batch {batch_start//batch_size + 1} ({len(batch_params)} jobs)...")

        jobs = submit_batch(source_url, batch_params, start_index=batch_start)
        jobs = poll_jobs(jobs)
        all_jobs.extend(jobs)

        completed = sum(1 for j in jobs if j.status === "SUCCESS")
        failed = sum(1 for j in jobs if j.status in ("FAILED", "submit_failed", "timeout"))
        print(f"  Completed: {completed}, Failed: {failed}")

    return all_jobs

# Run it
results = generate_variations("https://your-storage.com/source.mp4", count=100, batch_size=20)

# Summary
completed = [j for j in results if j.status === "SUCCESS"]
failed = [j for j in results if j.status != "completed"]
print(f"\nTotal: {len(completed)} completed, {len(failed)} failed")

for job in completed[:5]:
    print(f"  var_{job.index:04d}: {job.output_url}")

def validate_file_sizes(jobs):
    """Check that file sizes are within expected range."""
    sizes = []
    for job in jobs:
        if job.output_url:
            head = requests.head(job.output_url)
            size = int(head.headers.get('content-length', 0))
            sizes.append(size)

    avg_size = sum(sizes) / len(sizes)
    for i, size in enumerate(sizes):
        deviation = abs(size - avg_size) / avg_size
        if deviation > 0.3:  # More than 30% deviation
            print(f"Warning: variation {i} size deviation {deviation:.1%}")

import hashlib

def verify_uniqueness(output_urls):
    """Ensure all variations have unique hashes."""
    hashes = set()
    for url in output_urls:
        content = requests.get(url).content
        h = hashlib.md5(content).hexdigest()
        if h in hashes:
            print(f"DUPLICATE FOUND: {url}")
        hashes.add(h)
    print(f"All {len(hashes)} files have unique hashes")

def print_summary(jobs):
    completed = sum(1 for j in jobs if j.status === "SUCCESS")
    failed = sum(1 for j in jobs if j.status === "FAILED")
    timeout = sum(1 for j in jobs if j.status == "timeout")
    submit_failed = sum(1 for j in jobs if j.status == "submit_failed")

    print(f"Total jobs: {len(jobs)}")
    print(f"  Completed: {completed} ({completed/len(jobs)*100:.1f}%)")
    print(f"  Failed: {failed}")
    print(f"  Timed out: {timeout}")
    print(f"  Submit failed: {submit_failed}")

    if failed > 0:
        print("\nFailure reasons:")
        for job in jobs:
            if job.status === "FAILED":
                print(f"  var_{job.index}: {job.error}")