Update app.py

app.py

@@ -8,6 +8,8 @@ import re
 import urllib3
 import subprocess
 import logging
+import json
+import redis.asyncio as redis
 from flask import Flask, jsonify, make_response, request, Response
 from supabase import create_client
 from pyrogram import Client, filters, enums, idle, utils  
@@ -63,10 +65,6 @@ except RuntimeError:
     main_loop = asyncio.new_event_loop()
     asyncio.set_event_loop(main_loop)
 
-def run_async(coro):
-    future = asyncio.run_coroutine_threadsafe(coro, main_loop)
-    return future.result()
-
 bot = Client("file_unlocker_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN)
 
 # ==================== ROBUST DB QUERY WITH AUTO-RETRY ====================
@@ -312,7 +310,6 @@ def api_verify_code():
             try: await client.disconnect()
             except: pass
             
-            # 📌 UPDATE: Added 'phone' field to be saved in DB
             await db_query(lambda: supabase.table('user_sessions').insert({
                 "user_id": user_id, 
                 "phone": phone, 
@@ -404,7 +401,6 @@ async def save_progress(source_id, dest_id, msg_id):
             await db_query(lambda: supabase.table('clone_progress').insert({'source_id': source_id, 'dest_id': dest_id, 'last_copied_id': msg_id}).execute())
     except Exception as e: print(f"Error saving progress: {e}")
 
-# ================= AUTO BROADCAST TO NEW GROUPS =================
 @bot.on_message(filters.new_chat_members)
 async def bot_added_to_group(client, message):
     me = client.me
@@ -442,7 +438,6 @@ async def bot_added_to_group(client, message):
                     await client.copy_message(message.chat.id, src_chat, msg_id, caption=caption_text, reply_markup=group_markup)
             except Exception as e:
                 print(f"Error handling new group logic: {e}")
-# =======================================================================
 
 @bot.on_message(filters.command("sendto") & filters.private & filters.user(ADMIN_IDS))
 async def send_to_specific_group(client, message):
@@ -1077,18 +1072,23 @@ async def catch_admin_steps(client, message):
         await process_broadcast(client, message)
 
 # ==========================================
-# WATERMARK MULTIPROCESSING SYSTEM
+# REDIS QUEUE WATERMARK MULTIPROCESSING
 # ==========================================
+import json
+import redis.asyncio as redis
+
+REDIS_URL = os.environ.get("REDIS_URL_1")
+
 processing_owners = set()
 MAX_CONCURRENT_VIDEOS = 5 
 
-async def process_single_video(task):
+async def process_single_video(task, redis_client):
     global processing_owners
-    task_id = task['id']
     clone_token = task['clone_token']
     owner_id = task['owner_id']
     message_id = task['message_id']
     wm_text = task['watermark_text']
+    task_id = f"{owner_id}_{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()
@@ -1096,7 +1096,7 @@ async def process_single_video(task):
     try:
         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.")
+            raise Exception("Video not found or deleted.")
             
         status_msg = await clone_client.send_message(owner_id, "⏳ <b>Video processing started...</b>", parse_mode=enums.ParseMode.HTML)
         
@@ -1156,50 +1156,54 @@ async def process_single_video(task):
         if os.path.exists(raw_video_path): os.remove(raw_video_path)
         if os.path.exists(watermarked_path): os.remove(watermarked_path)
         
-        await db_query(lambda: supabase.table('watermark_queue').update({'status': 'completed'}).eq('id', task_id).execute())
-        
     except Exception as e:
         try:
             await clone_client.send_message(owner_id, f"❌ <b>Error processing video:</b> {e}", parse_mode=enums.ParseMode.HTML)
         except: pass
-        await db_query(lambda: supabase.table('watermark_queue').update({'status': 'failed'}).eq('id', task_id).execute())
         
     finally:
         await clone_client.stop()
+        await redis_client.delete(f"wm_processing:{owner_id}")
         processing_owners.discard(owner_id) 
 
 async def watermark_processor_loop():
     global processing_owners
+    
+    if not REDIS_URL:
+        print("⚠️ REDIS_URL missing in Hugging Space! Watermark feature disabled.")
+        return
+        
     try:
-        await db_query(lambda: supabase.table('watermark_queue').update({'status': 'pending'}).eq('status', 'processing').execute())
+        redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
+        await redis_client.ping()
+        print("💧 Super-Fast Redis Watermark Processor Started!")
+    except Exception as e:
+        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)
     except: pass
 
-    await asyncio.sleep(5)
-    print("💧 Multi-Threaded Watermark Processor Started!")
-    
     while True:
-        await asyncio.sleep(3) 
         try:
             if len(processing_owners) >= MAX_CONCURRENT_VIDEOS:
+                await asyncio.sleep(2)
                 continue 
             
-            res = await db_query(lambda: supabase.table('watermark_queue').select('*').eq('status', 'pending').order('id', desc=False).limit(10).execute())
+            result = await redis_client.brpop("watermark_task_queue", timeout=5)
             
-            if res and res.data:
-                for task in res.data:
-                    if len(processing_owners) >= MAX_CONCURRENT_VIDEOS:
-                        break
-                    
-                    owner_id = task['owner_id']
-                    if owner_id in processing_owners:
-                        continue 
-                    
-                    processing_owners.add(owner_id)
-                    await db_query(lambda: supabase.table('watermark_queue').update({'status': 'processing'}).eq('id', task['id']).execute())
-                    asyncio.create_task(process_single_video(task))
+            if result:
+                _, task_json = result
+                task = json.loads(task_json)
+                owner_id = task['owner_id']
+                
+                processing_owners.add(owner_id)
+                asyncio.create_task(process_single_video(task, redis_client))
                     
-        except Exception:
-            pass
+        except Exception as e:
+            await asyncio.sleep(2)
 
 def run_flask(): app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), threaded=True) 
 
@@ -1208,7 +1212,7 @@ async def main():
         await bot.start()
         print("🤖 Pyrogram Bot & Real Session API is running!")
         asyncio.create_task(auto_clean_channel_loop())
-        asyncio.create_task(watermark_processor_loop())
+        asyncio.create_task(watermark_processor_loop()) 
         await idle()
     except Exception as e:
         print(f"❌ Failed to start Bot: {e}")