Update app.py

app.py

@@ -1090,17 +1090,23 @@ async def process_single_video(task, redis_client):
     wm_text = task['watermark_text']
     task_id = f"{owner_id}_{message_id}"
     
+    print(f"▶️ [WM] Task Started for User: {owner_id}, Msg: {message_id}")
+    
     clone_client = Client(f"clone_wm_{task_id}", bot_token=clone_token, api_id=API_ID, api_hash=API_HASH, in_memory=True)
-    await clone_client.start()
     
     try:
+        await clone_client.start()
+        print(f"▶️ [WM] Clone Client Started Successfully")
+        
         msg = await clone_client.get_messages(owner_id, message_id)
         if not msg or not (msg.video or msg.document):
-            raise Exception("Video not found or deleted.")
+            raise Exception("Video not found or deleted by user.")
             
         status_msg = await clone_client.send_message(owner_id, "⏳ <b>Video processing started...</b>", parse_mode=enums.ParseMode.HTML)
+        print(f"▶️ [WM] Status Message Sent to User")
         
         raw_video_path = await clone_client.download_media(msg)
+        print(f"▶️ [WM] Original Video Downloaded: {raw_video_path}")
         
         await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>Uploading original file to secure storage...</b>", parse_mode=enums.ParseMode.HTML)
         
@@ -1112,6 +1118,7 @@ async def process_single_video(task, redis_client):
         raw_storage_id = raw_sent.id
         stream_link = f"{BACKEND_URL}/stream/{raw_storage_id}"
         download_link = f"{BACKEND_URL}/download/{raw_storage_id}"
+        print(f"▶️ [WM] Original Backup Uploaded to Storage")
         
         await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>Applying Custom Watermark... (This may take a few minutes)</b>", parse_mode=enums.ParseMode.HTML)
         
@@ -1124,12 +1131,15 @@ async def process_single_video(task, redis_client):
             "-c:a", "copy", "-movflags", "+faststart", watermarked_path
         ]
         
+        print(f"▶️ [WM] Starting FFmpeg processing...")
         process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
-        await process.communicate()
+        stdout, stderr = await process.communicate()
         
-        if not os.path.exists(watermarked_path) or os.path.getsize(watermarked_path) == 0:
-            raise Exception("FFmpeg processing failed.")
+        if process.returncode != 0 or not os.path.exists(watermarked_path) or os.path.getsize(watermarked_path) == 0:
+            print(f"❌ [WM] FFmpeg Error: {stderr.decode()}")
+            raise Exception("FFmpeg processing failed! The video format might be unsupported.")
             
+        print(f"▶️ [WM] FFmpeg Processing Complete!")
         await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>Uploading final video...</b>", parse_mode=enums.ParseMode.HTML)
         
         await bot.send_video(
@@ -1137,6 +1147,7 @@ async def process_single_video(task, redis_client):
             video=watermarked_path,
             caption=f"Watermarked Backup for Clone Owner {owner_id}"
         )
+        print(f"▶️ [WM] Watermarked Backup Uploaded to Storage")
         
         final_caption = (
             f"✅ <b>Watermark Successfully Added!</b>\n\n"
@@ -1150,6 +1161,7 @@ async def process_single_video(task, redis_client):
             caption=final_caption,
             parse_mode=enums.ParseMode.HTML
         )
+        print(f"▶️ [WM] Final Video Sent to User Successfully!")
         
         await clone_client.delete_messages(owner_id, status_msg.id)
         
@@ -1157,14 +1169,17 @@ async def process_single_video(task, redis_client):
         if os.path.exists(watermarked_path): os.remove(watermarked_path)
         
     except Exception as e:
+        print(f"❌ [WM] Process Error: {e}")
         try:
             await clone_client.send_message(owner_id, f"❌ <b>Error processing video:</b> {e}", parse_mode=enums.ParseMode.HTML)
-        except: pass
+        except Exception as ex: 
+            print(f"❌ [WM] Failed to send error msg to user: {ex}")
         
     finally:
         await clone_client.stop()
         await redis_client.delete(f"wm_processing:{owner_id}")
         processing_owners.discard(owner_id) 
+        print(f"▶️ [WM] Task Cleaned Up and Client Stopped")
 
 async def watermark_processor_loop():
     global processing_owners
@@ -1181,6 +1196,7 @@ async def watermark_processor_loop():
         print(f"❌ Redis Connection Failed in Hugging Face: {e}")
         return
     
+    # সার্ভার রিস্টার্ট হলে আটকে থাকা লকগুলো ক্লিয়ার করবে
     try:
         async for key in redis_client.scan_iter("wm_processing:*"):
             await redis_client.delete(key)
@@ -1192,19 +1208,23 @@ async def watermark_processor_loop():
                 await asyncio.sleep(2)
                 continue 
             
-            result = await redis_client.brpop("watermark_task_queue", timeout=5)
+            # BRPOP এর বদলে RPOP ব্যবহার করা হলো (Upstash Connection Drop এড়াতে)
+            result = await redis_client.rpop("watermark_task_queue")
             
             if result:
-                _, task_json = result
-                task = json.loads(task_json)
+                print(f"📥 [WM] Received New Video Task from Queue!")
+                task = json.loads(result)
                 owner_id = task['owner_id']
                 
                 processing_owners.add(owner_id)
                 asyncio.create_task(process_single_video(task, redis_client))
+            else:
+                await asyncio.sleep(2) # ভিডিও না থাকলে ২ সেকেন্ড পর আবার চেক করবে
                     
         except Exception as e:
+            print(f"❌ [WM] Loop Error: {e}")
             await asyncio.sleep(2)
-
+            
 def run_flask(): app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), threaded=True) 
 
 async def main():