Update server.py

server.py

@@ -195,7 +195,7 @@ async def load_progress_file() -> Dict:
         return progress
     except Exception as e:
         print(f"⊘ No existing progress file or load failed: {str(e)[:100]}")
-        return {"compressed": [], "failed": [], "last_updated": datetime.now().isoformat()}
+        return {"compressed": [], "failed": [], "processing": [], "last_updated": datetime.now().isoformat()}
 
 
 async def save_progress_file(progress: Dict):
@@ -238,8 +238,46 @@ async def scan_and_compress_videos():
         compressed_files = {item["path"] for item in progress.get("compressed", [])}
         failed_files = progress.get("failed", [])
         failed_paths = {item["path"] for item in failed_files}
+        processing_files = progress.get("processing", [])
+        processing_paths = {item["path"] for item in processing_files}
         
-        print(f"Current state: {len(compressed_files)} compressed, {len(failed_files)} failed")
+        # Clean up stale processing files (>4 hours old = likely crashed server)
+        print("\n" + "="*80)
+        print("CHECKING FOR STALE PROCESSING FILES")
+        print("="*80)
+        
+        current_time = datetime.now()
+        stale_threshold = 4 * 3600  # 4 hours in seconds
+        stale_files = []
+        
+        for processing_item in processing_files:
+            started_at = datetime.fromisoformat(processing_item["started_at"])
+            elapsed_seconds = (current_time - started_at).total_seconds()
+            
+            if elapsed_seconds > stale_threshold:
+                print(f"  ⚠ {processing_item['file_name']} (stale, {elapsed_seconds/3600:.1f}h old)")
+                stale_files.append(processing_item)
+        
+        # Move stale files back to failed for retry
+        if stale_files:
+            print(f"\nRecovering {len(stale_files)} stale processing files...")
+            for stale_item in stale_files:
+                progress["processing"] = [f for f in progress["processing"] if f["path"] != stale_item["path"]]
+                progress["failed"].append({
+                    "path": stale_item["path"],
+                    "file_name": stale_item["file_name"],
+                    "error": "Server crash detected (stale processing)",
+                    "retry_count": 0,
+                    "timestamp": datetime.now().isoformat()
+                })
+            await save_progress_file(progress)
+            # Reload processing paths after cleanup
+            processing_files = progress.get("processing", [])
+            processing_paths = {item["path"] for item in processing_files}
+        else:
+            print("✓ No stale processing files")
+        
+        print(f"Current state: {len(compressed_files)} compressed, {len(failed_files)} failed, {len(processing_paths)} processing")
         
         print("\n" + "="*80)
         print("RETRYING PREVIOUSLY FAILED FILES")
@@ -264,6 +302,16 @@ async def scan_and_compress_videos():
                 "file_name": failed_item["file_name"]
             }
             
+            # Mark as processing
+            progress["failed"] = [f for f in progress["failed"] if f["path"] != video_info["path"]]
+            progress["processing"].append({
+                "path": video_info["path"],
+                "file_name": video_info["file_name"],
+                "status": "retrying",
+                "started_at": datetime.now().isoformat()
+            })
+            await save_progress_file(progress)
+            
             try:
                 # Download
                 print(f"\nRetrying: {video_info['path']}")
@@ -307,8 +355,8 @@ async def scan_and_compress_videos():
                 compression_state["total_uploaded"] += 1
                 compression_state["total_space_saved_mb"] += compression_stats["saved_mb"]
                 
-                # Remove from failed list and add to compressed
-                progress["failed"] = [f for f in progress["failed"] if f["path"] != video_info["path"]]
+                # Remove from processing and add to compressed
+                progress["processing"] = [f for f in progress["processing"] if f["path"] != video_info["path"]]
                 progress["compressed"].append({
                     "path": video_info["path"],
                     "file_name": video_info["file_name"],
@@ -329,13 +377,18 @@ async def scan_and_compress_videos():
                 
             except Exception as e:
                 print(f"✗ Retry failed: {e}")
-                # Increment retry count
-                for i, f in enumerate(progress["failed"]):
-                    if f["path"] == video_info["path"]:
-                        progress["failed"][i]["retry_count"] = f.get("retry_count", 0) + 1
-                        progress["failed"][i]["last_error"] = str(e)
-                        progress["failed"][i]["last_attempt"] = datetime.now().isoformat()
-                        break
+                # Remove from processing and re-add to failed with incremented retry count
+                progress["processing"] = [f for f in progress["processing"] if f["path"] != video_info["path"]]
+                retry_count = failed_item.get("retry_count", 0) + 1
+                progress["failed"].append({
+                    "path": video_info["path"],
+                    "file_name": video_info["file_name"],
+                    "error": str(e),
+                    "retry_count": retry_count,
+                    "last_error": str(e),
+                    "last_attempt": datetime.now().isoformat(),
+                    "timestamp": failed_item.get("timestamp", datetime.now().isoformat())
+                })
             
             # Save progress after each retry attempt
             await save_progress_file(progress)
@@ -378,6 +431,11 @@ async def scan_and_compress_videos():
                     print(f"  ⊘ {f.split('/')[-1]} (queued for retry next scan)")
                     continue
                 
+                # Skip if currently being processed by another server
+                if f in processing_paths:
+                    print(f"  ⟳ {f.split('/')[-1]} (currently processing)")
+                    continue
+                
                 try:
                     # Parse file path: ready_videos/moviename/segment-XX.mp4
                     parts = f.split("/")
@@ -405,6 +463,15 @@ async def scan_and_compress_videos():
         for video_info in video_files:
             compression_state["current_video"] = video_info["file_name"]
             
+            # Mark as processing
+            progress["processing"].append({
+                "path": video_info["path"],
+                "file_name": video_info["file_name"],
+                "status": "compressing",
+                "started_at": datetime.now().isoformat()
+            })
+            await save_progress_file(progress)
+            
             try:
                 # Download
                 print(f"\nDownloading: {video_info['path']}")
@@ -463,7 +530,8 @@ async def scan_and_compress_videos():
                 compression_state["total_uploaded"] += 1
                 compression_state["total_space_saved_mb"] += compression_stats["saved_mb"]
                 
-                # Track in progress file
+                # Remove from processing and add to compressed
+                progress["processing"] = [f for f in progress["processing"] if f["path"] != video_info["path"]]
                 progress["compressed"].append({
                     "path": video_info["path"],
                     "file_name": video_info["file_name"],
@@ -485,10 +553,13 @@ async def scan_and_compress_videos():
             except Exception as e:
                 print(f"✗ Error: {e}")
                 compression_state["last_error"] = str(e)
+                # Remove from processing and add to failed
+                progress["processing"] = [f for f in progress["processing"] if f["path"] != video_info["path"]]
                 progress["failed"].append({
                     "path": video_info["path"],
                     "file_name": video_info["file_name"],
                     "error": str(e),
+                    "retry_count": 0,
                     "timestamp": datetime.now().isoformat()
                 })
                 await save_progress_file(progress)
@@ -573,6 +644,21 @@ async def get_status():
     })
 
 
+@app.get("/progress")
+async def get_progress():
+    """Get full progress file state with counts and details."""
+    progress = await load_progress_file()
+    return JSONResponse({
+        "compressed_count": len(progress.get("compressed", [])),
+        "failed_count": len(progress.get("failed", [])),
+        "processing_count": len(progress.get("processing", [])),
+        "compressed": progress.get("compressed", []),
+        "failed": progress.get("failed", []),
+        "processing": progress.get("processing", []),
+        "last_updated": progress.get("last_updated")
+    })
+
+
 @app.post("/scan")
 async def trigger_scan():
     """Manually trigger a scan and compression run."""