#!/usr/bin/env python3 import os import json import asyncio import tempfile import subprocess import shutil import time import threading from pathlib import Path from datetime import datetime from dotenv import load_dotenv from typing import List, Dict, Optional, Tuple from fastapi import FastAPI from fastapi.responses import JSONResponse import uvicorn try: from huggingface_hub import list_repo_files, hf_hub_download, upload_file import cv2 import numpy as np from PIL import Image, ImageDraw, ImageFont from faster_whisper import WhisperModel except ImportError as e: print(f"Missing dependency: {e}") exit(1) # Load environment variables load_dotenv() HF_TOKEN = os.getenv("HF_TOKEN") HF_DATASET_REPO = "factorstudios/movs" HOOKS_FOLDER = "hooks" READY_VIDEOS_FOLDER = "ready_videos" app = FastAPI(title="Video Processing Service") # Global state processing_state = { "is_running": False, "total_processed": 0, "current_file": None, "error_count": 0, "last_error": None, "processed_files": [], "whisper_ready": False, "log": [] } whisper_model = None def add_log(msg): # Print to console as requested timestamp = datetime.now().strftime('%H:%M:%S') formatted_msg = f"[{timestamp}] {msg}" print(formatted_msg) # Also keep in state for API status checks processing_state["log"].append(formatted_msg) if len(processing_state["log"]) > 100: processing_state["log"].pop(0) def _load_whisper_model(): """Load model in a way that doesn't block the event loop.""" global whisper_model try: add_log("Starting Whisper model load...") whisper_model = WhisperModel("small", device="auto", compute_type="int8") processing_state["whisper_ready"] = True add_log("✓ Whisper model loaded successfully") except Exception as e: add_log(f"✗ Failed to load Whisper model: {e}") def timestamp_to_seconds(timestamp: str) -> float: try: parts = timestamp.split(":") if len(parts) == 3: return int(parts[0]) * 3600 + int(parts[1]) * 60 + float(parts[2]) return 0.0 except: return 0.0 def apply_color_grading(frame): lab = cv2.cvtColor(frame, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8)) l = clahe.apply(l) frame = cv2.cvtColor(cv2.merge([l, a, b]), cv2.COLOR_LAB2BGR) kernel = np.array([[-1,-1,-1], [-1,9,-1], [-1,-1,-1]]) / 1.2 sharpened = cv2.filter2D(frame, -1, kernel) return cv2.addWeighted(frame, 0.4, sharpened, 0.6, 0) def burn_captions(frame, text, font_size=40): h, w = frame.shape[:2] pil_img = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)).convert('RGBA') draw = ImageDraw.Draw(pil_img) try: font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", font_size) except: font = ImageFont.load_default() lines, curr = [], [] for word in text.split(): test = ' '.join(curr + [word]) if draw.textbbox((0, 0), test, font=font)[2] < w - 100: curr.append(word) else: lines.append(' '.join(curr)) curr = [word] if curr: lines.append(' '.join(curr)) y = int(h * 0.8) for line in lines: bbox = draw.textbbox((0, 0), line, font=font) x = (w - (bbox[2] - bbox[0])) // 2 draw.text((x+2, y+2), line, font=font, fill=(0,0,0,180)) draw.text((x, y), line, font=font, fill=(255,255,255,255)) y += font_size + 10 return cv2.cvtColor(np.array(pil_img.convert('RGB')), cv2.COLOR_RGB2BGR) def process_video_sync(video_path, output_path, start_t, end_t): temp_seg = output_path + ".seg.mp4" temp_no_audio = output_path + ".noaudio.mp4" temp_wav = output_path + ".wav" try: start_s = timestamp_to_seconds(start_t) end_s = timestamp_to_seconds(end_t) subprocess.run(["ffmpeg", "-y", "-ss", str(start_s), "-to", str(end_s), "-i", video_path, "-c", "copy", temp_seg], capture_output=True) subprocess.run(["ffmpeg", "-y", "-i", temp_seg, "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", temp_wav], capture_output=True) captions = [] add_log(f"[process_video_sync] Whisper model ready: {processing_state["whisper_ready"]}") add_log(f"[process_video_sync] Whisper model instance: {whisper_model is not None}") if whisper_model and processing_state["whisper_ready"]: segs, _ = whisper_model.transcribe(temp_wav) captions = [(s.start, s.end, s.text.strip()) for s in segs if s.text.strip()] add_log(f"[process_video_sync] Transcribed {len(captions)} captions for {temp_wav}") if not captions: add_log("[process_video_sync] WARNING: No captions transcribed. Check audio or model.") cap = cv2.VideoCapture(temp_seg) fps = cap.get(cv2.CAP_PROP_FPS) or 24 width, height = 1080, 1350 ffmpeg_cmd = [ "ffmpeg", "-y", "-f", "rawvideo", "-vcodec", "rawvideo", "-s", f"{width}x{height}", "-pix_fmt", "bgr24", "-r", str(fps), "-i", "pipe:0", "-vcodec", "libx264", "-preset", "veryfast", "-crf", "22", "-pix_fmt", "yuv420p", temp_no_audio ] proc = subprocess.Popen(ffmpeg_cmd, stdin=subprocess.PIPE, stderr=subprocess.DEVNULL) f_idx = 0 while True: ret, frame = cap.read() if not ret: break h, w = frame.shape[:2] target_ratio = width / height if w/h > target_ratio: nw = int(h * target_ratio) off = (w - nw) // 2 frame = frame[:, off:off+nw] else: nh = int(w / target_ratio) off = (h - nh) // 2 frame = frame[off:off+nh, :] frame = cv2.resize(frame, (width, height)) frame = apply_color_grading(frame) ts = f_idx / fps for s, e, t in captions: if s <= ts <= e: frame = burn_captions(frame, t) break proc.stdin.write(frame.tobytes()) f_idx += 1 proc.stdin.close() proc.wait() cap.release() subprocess.run(["ffmpeg", "-y", "-i", temp_no_audio, "-i", temp_seg, "-map", "0:v:0", "-map", "1:a:0", "-c", "copy", "-shortest", output_path], capture_output=True) return os.path.exists(output_path) except Exception as e: add_log(f"Error in sync process: {e}") return False finally: for f in [temp_seg, temp_no_audio, temp_wav]: if os.path.exists(f): os.remove(f) async def run_processing_loop(): if processing_state["is_running"]: return processing_state["is_running"] = True try: add_log("Waiting 5 seconds for server to settle...") await asyncio.sleep(5) # Start model loading after the 5s delay add_log("Initiating background tasks...") asyncio.create_task(asyncio.to_thread(_load_whisper_model)) while not processing_state["whisper_ready"]: await asyncio.sleep(2) add_log("Starting repository scan...") files = list_repo_files(repo_id=HF_DATASET_REPO, repo_type="dataset", token=HF_TOKEN) # Find all movies with hooks add_log("Scanning hooks directory...") all_hooks_movies = {} for f in files: if f.startswith(HOOKS_FOLDER + "/") and f.endswith(".json"): parts = f.split("/") if len(parts) >= 3: movie_name = parts[1] if movie_name not in all_hooks_movies: all_hooks_movies[movie_name] = [] all_hooks_movies[movie_name].append(f) add_log(f"Found {len(all_hooks_movies)} movies in hooks folder") # Find all movies with ready videos add_log("Scanning ready_videos directory...") processed_videos = {} for f in files: if f.startswith(READY_VIDEOS_FOLDER + "/") and f.endswith(".mp4"): parts = f.split("/") if len(parts) >= 3: movie_name = parts[1] if movie_name not in processed_videos: processed_videos[movie_name] = set() processed_videos[movie_name].add(parts[2]) add_log(f"Found {len(processed_videos)} movies with ready videos") # Find unprocessed movies unprocessed_movies = [] for movie_name, hooks in all_hooks_movies.items(): if movie_name not in processed_videos: # Movie has no ready videos at all unprocessed_movies.append((movie_name, hooks, [])) add_log(f" ⊘ {movie_name} (no ready videos, process all {len(hooks)} segments)") else: # Check which segments are already processed by matching filenames processed_segments = processed_videos[movie_name] unprocessed_hooks = [] for h in hooks: # Extract segment filename and convert to .mp4 version segment_filename = h.split('/')[-1] # e.g., "segment-01.json" segment_mp4 = segment_filename.replace('.json', '.mp4') # e.g., "segment-01.mp4" if segment_mp4 not in processed_segments: unprocessed_hooks.append(h) if unprocessed_hooks: unprocessed_movies.append((movie_name, unprocessed_hooks, list(processed_segments))) add_log(f" ⊘ {movie_name} (already has {len(processed_segments)} videos, {len(unprocessed_hooks)} segments remaining)") else: add_log(f" ✓ {movie_name} (already complete with {len(processed_segments)} videos)") add_log(f"\nTotal unprocessed movies to process: {len(unprocessed_movies)}\n") if not unprocessed_movies: add_log("All movies already processed!") return for movie, movie_hooks, existing_videos in unprocessed_movies: processing_state["current_file"] = movie add_log(f"--- Processing Movie: {movie} ---") try: video_path = hf_hub_download(repo_id=HF_DATASET_REPO, filename=f"{READY_VIDEOS_FOLDER}/{movie}.mkv", repo_type="dataset", token=HF_TOKEN) except: try: # Try alternative path video_path = hf_hub_download(repo_id=HF_DATASET_REPO, filename=f"{movie}.mkv", repo_type="dataset", token=HF_TOKEN) except Exception as e: add_log(f"✗ Could not find video file for {movie}: {e}") processing_state["error_count"] += 1 continue add_log(f"Found {len(movie_hooks)} unprocessed segments for {movie}") temp_dir = tempfile.mkdtemp() try: for hook_file in movie_hooks: await asyncio.sleep(0.1) hook_path = hf_hub_download(repo_id=HF_DATASET_REPO, filename=hook_file, repo_type="dataset", token=HF_TOKEN) with open(hook_path, 'r') as f: data = json.load(f) num, start, end = data.get("segment_number", 1), data.get("start_time", "00:00:00"), data.get("end_time", "00:00:10") out_name = f"segment-{num:02d}.mp4" # Skip if already exists if out_name in existing_videos: add_log(f" ⊘ Segment {num} (already processed)") continue out_path = os.path.join(temp_dir, out_name) add_log(f"Processing Segment {num} ({start} to {end})") success = await asyncio.to_thread(process_video_sync, video_path, out_path, start, end) if success: upload_file(path_or_fileobj=out_path, path_in_repo=f"{READY_VIDEOS_FOLDER}/{movie}/{out_name}", repo_id=HF_DATASET_REPO, repo_type="dataset", token=HF_TOKEN) add_log(f"✓ Segment {num} uploaded successfully") else: add_log(f"✗ Segment {num} failed") processing_state["error_count"] += 1 finally: shutil.rmtree(temp_dir, ignore_errors=True) processing_state["processed_files"].append(movie) processing_state["total_processed"] += 1 add_log(f"Finished movie: {movie}") except Exception as e: add_log(f"CRITICAL ERROR: {e}") processing_state["last_error"] = str(e) finally: processing_state["is_running"] = False add_log("Background worker idle.") @app.on_event("startup") async def startup_event(): """Schedule video processing loop on server startup with background thread.""" add_log("\n" + "="*80) add_log("STARTUP EVENT TRIGGERED - Video Segment Processing Service") add_log("="*80) # Schedule processing in a background thread (more reliable for deployment) def run_loop(): asyncio.run(run_processing_loop()) process_thread = threading.Thread(target=run_loop, daemon=True) process_thread.start() add_log("✓ Background processing thread scheduled") @app.get("/") @app.get("/status") async def status(): return processing_state if __name__ == "__main__": add_log("Starting Video Processing Service on port 7860...") uvicorn.run(app, host="0.0.0.0", port=7860)