Update server.py

server.py

@@ -1,4 +1,4 @@
- #!/usr/bin/env python3
+
 import os
 import json
 import asyncio
@@ -6,6 +6,7 @@ import tempfile
 import subprocess
 import shutil
 import time
+import threading
 from pathlib import Path
 from datetime import datetime
 from dotenv import load_dotenv
@@ -193,32 +194,102 @@ async def run_processing_loop():
         
         add_log("Starting repository scan...")
         files = list_repo_files(repo_id=HF_DATASET_REPO, repo_type="dataset", token=HF_TOKEN)
-        movies = sorted(list(set(f.split("/")[1] for f in files if f.startswith(HOOKS_FOLDER + "/") and f.endswith(".json"))))
         
-        add_log(f"Found {len(movies)} movies to process")
-        for movie in movies:
+        # 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
+                processed_segments = processed_videos[movie_name]
+                unprocessed_hooks = [h for h in hooks if not any(f"segment-{json.loads(open(h).read()).get('segment_number', 1):02d}.mp4" in s for s in processed_segments)]
+                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} ---")
-            video_path = hf_hub_download(repo_id=HF_DATASET_REPO, filename=f"{movie}.mkv", repo_type="dataset", token=HF_TOKEN)
-            movie_hooks = sorted([f for f in files if f.startswith(f"{HOOKS_FOLDER}/{movie}/") and f.endswith(".json")])
-            add_log(f"Found {len(movie_hooks)} segments for {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()
-            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"
-                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")
-            shutil.rmtree(temp_dir)
+            
+            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}")
@@ -232,8 +303,18 @@ async def run_processing_loop():
 
 @app.on_event("startup")
 async def startup_event():
-    # Only kick off the main loop, which now handles the 5s delay and model loading
-    asyncio.create_task(run_processing_loop())
+    """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")