Processing Videos

# main.py:423-433
def detect_scenes(video_path):
    video_manager = VideoManager([video_path])
    scene_manager = SceneManager()
    scene_manager.add_detector(ContentDetector())  # Default threshold
    video_manager.set_downscale_factor()
    video_manager.start()
    scene_manager.detect_scenes(frame_source=video_manager)
    scene_list = scene_manager.get_scene_list()
    fps = video_manager.get_framerate()
    video_manager.release()
    return scene_list, fps

# main.py:375-421
def analyze_scenes_strategy(video_path, scenes):
    # Samples 3 frames per scene (start, middle, end)
    # Counts faces using MediaPipe
    # Returns 'TRACK' or 'GENERAL' strategy for each scene
    
    if avg_faces > 1.2 or avg_faces < 0.5:
        strategies.append('GENERAL')  # Multiple people or no faces
    else:
        strategies.append('TRACK')     # Single subject tracking

# app.py:28-39
MAX_CONCURRENT_JOBS = int(os.environ.get("MAX_CONCURRENT_JOBS", "5"))
MAX_FILE_SIZE_MB = 2048  # 2GB limit
JOB_RETENTION_SECONDS = 3600  # 1 hour retention

job_queue = asyncio.Queue()
jobs: Dict[str, Dict] = {}
concurrency_semaphore = asyncio.Semaphore(MAX_CONCURRENT_JOBS)

# app.py:113-144
async def process_queue():
    while True:
        job_id = await job_queue.get()
        await concurrency_semaphore.acquire()
        asyncio.create_task(run_job_wrapper(job_id))

async def run_job_wrapper(job_id):
    try:
        job = jobs.get(job_id)
        if job:
            await run_job(job_id, job)
    finally:
        concurrency_semaphore.release()
        job_queue.task_done()

# Job states
jobs[job_id] = {
    'status': 'queued',        # queued -> processing -> completed/failed
    'logs': [],                # Real-time processing logs
    'cmd': cmd,                # Shell command being executed
    'env': env,                # Environment with API keys
    'output_dir': job_output_dir,
    'result': {                # Populated when completed
        'clips': [...],
        'cost_analysis': {...}
    }
}

# main.py:31-68 - Gemini Prompt Template
GEMINI_PROMPT_TEMPLATE = """
You are a senior short-form video editor. Read the ENTIRE transcript and 
word-level timestamps to choose the 3–15 MOST VIRAL moments for TikTok/IG 
Reels/YouTube Shorts. Each clip must be between 15 and 60 seconds long.

⚠️ FFMPEG TIME CONTRACT — STRICT REQUIREMENTS:
- Return timestamps in ABSOLUTE SECONDS from the start of the video
- Only NUMBERS with decimal point, up to 3 decimals (examples: 0, 1.250, 17.350)
- Ensure 0 ≤ start < end ≤ VIDEO_DURATION_SECONDS
- Each clip between 15 and 60 s (inclusive)
- Prefer starting 0.2–0.4 s BEFORE the hook and ending 0.2–0.4 s AFTER the payoff
- Use silence moments for natural cuts; never cut mid-word
"""

// main.py:56-67
{
  "shorts": [
    {
      "start": 12.340,
      "end": 37.900,
      "video_description_for_tiktok": "<description>",
      "video_description_for_instagram": "<description>",
      "video_title_for_youtube_short": "<title 100 chars max>",
      "viral_hook_text": "<SHORT punchy text (max 10 words)>"
    }
  ]
}

# main.py:836-869
# Gemini 2.5 Flash Pricing (Dec 2025)
# Input: $0.10 per 1M tokens
# Output: $0.40 per 1M tokens

if usage:
    prompt_tokens = usage.prompt_token_count
    output_tokens = usage.candidates_token_count
    
    input_cost = (prompt_tokens / 1_000_000) * 0.10
    output_cost = (output_tokens / 1_000_000) * 0.40
    total_cost = input_cost + output_cost

# main.py:898-899
python main.py -i input.mp4 -o output.mp4 --skip-analysis

# main.py:29
ASPECT_RATIO = 9 / 16  # Output aspect ratio (vertical)

# main.py:169-178 - Speaker tracking
stabilization_frames=15    # Frames needed to confirm new speaker
cooldown_frames=30         # Minimum frames before switching

# main.py:104 - Camera safe zone
self.safe_zone_radius = self.crop_width * 0.25  # 25% safe zone

# app.py:176-188
def enqueue_output(out, job_id):
    for line in iter(out.readline, b''):
        decoded_line = line.decode('utf-8').strip()
        if decoded_line:
            print(f"📝 [Job Output] {decoded_line}")
            if job_id in jobs:
                jobs[job_id]['logs'].append(decoded_line)