Update app.py

app.py

@@ -1076,6 +1076,131 @@ async def catch_admin_steps(client, message):
     elif state.get("step") == "broadcast":
         await process_broadcast(client, message)
 
+# ==========================================
+# WATERMARK MULTIPROCESSING SYSTEM
+# ==========================================
+processing_owners = set()
+MAX_CONCURRENT_VIDEOS = 5 
+
+async def process_single_video(task):
+    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']
+    
+    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:
+        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.")
+            
+        status_msg = await clone_client.send_message(owner_id, "⏳ <b>Video processing started...</b>", parse_mode=enums.ParseMode.HTML)
+        
+        raw_video_path = await clone_client.download_media(msg)
+        
+        await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>Uploading original file to secure storage...</b>", parse_mode=enums.ParseMode.HTML)
+        
+        raw_sent = await bot.send_video(
+            chat_id=STORAGE_CHANNEL_ID,
+            video=raw_video_path,
+            caption=f"Original Backup for Clone Owner {owner_id}"
+        )
+        raw_storage_id = raw_sent.id
+        stream_link = f"{BACKEND_URL}/stream/{raw_storage_id}"
+        download_link = f"{BACKEND_URL}/download/{raw_storage_id}"
+        
+        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)
+        
+        watermarked_path = f"wm_{task_id}.mp4"
+        
+        cmd = [
+            "ffmpeg", "-y", "-i", raw_video_path,
+            "-vf", f"drawtext=text='{wm_text}':x=W-tw-20:y=H-th-20:fontsize=22:fontcolor=white@0.7:shadowcolor=black@0.8:shadowx=2:shadowy=2:enable='gte(t,5)'",
+            "-c:v", "libx264", "-preset", "superfast", "-crf", "23",
+            "-c:a", "copy", "-movflags", "+faststart", watermarked_path
+        ]
+        
+        process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        await process.communicate()
+        
+        if not os.path.exists(watermarked_path) or os.path.getsize(watermarked_path) == 0:
+            raise Exception("FFmpeg processing failed.")
+            
+        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(
+            chat_id=STORAGE_CHANNEL_ID,
+            video=watermarked_path,
+            caption=f"Watermarked Backup for Clone Owner {owner_id}"
+        )
+        
+        final_caption = (
+            f"✅ <b>Watermark Successfully Added!</b>\n\n"
+            f"🎬 <b>Stream/Watch Online Link:</b>\n<code>{stream_link}</code>\n\n"
+            f"📥 <b>Direct Download Link:</b>\n<code>{download_link}</code>"
+        )
+        
+        await clone_client.send_video(
+            chat_id=owner_id,
+            video=watermarked_path,
+            caption=final_caption,
+            parse_mode=enums.ParseMode.HTML
+        )
+        
+        await clone_client.delete_messages(owner_id, status_msg.id)
+        
+        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()
+        processing_owners.discard(owner_id) 
+
+async def watermark_processor_loop():
+    global processing_owners
+    try:
+        await db_query(lambda: supabase.table('watermark_queue').update({'status': 'pending'}).eq('status', 'processing').execute())
+    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:
+                continue 
+            
+            res = await db_query(lambda: supabase.table('watermark_queue').select('*').eq('status', 'pending').order('id', desc=False).limit(10).execute())
+            
+            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))
+                    
+        except Exception:
+            pass
+
 def run_flask(): app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), threaded=True) 
 
 async def main():
@@ -1083,6 +1208,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())
         await idle()
     except Exception as e:
         print(f"❌ Failed to start Bot: {e}")