Update app.py

app.py

@@ -7,6 +7,7 @@ import asyncio
 import re
 import urllib3
 import subprocess
+import logging
 from flask import Flask, jsonify, make_response, request, Response
 from supabase import create_client
 from pyrogram import Client, filters, enums, idle, utils  
@@ -15,6 +16,10 @@ from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton, WebAppInf
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
+# Flask এর অপ্রয়োজনীয় socket.send() এরর লগস বন্ধ করার জন্য
+log = logging.getLogger('werkzeug')
+log.setLevel(logging.ERROR)
+
 # ==================== PYROGRAM NEW ID RANGE FIX (MONKEY PATCH) ====================
 def get_peer_type_new(peer_id: int) -> str:
     peer_id_str = str(peer_id)
@@ -100,9 +105,10 @@ def get_msg_file_id(msg):
     if media: return media.file_id
     return None
 
-# ==================== CUSTOM VIDEO STREAMING ENGINE ====================
+# ==================== CUSTOM VIDEO STREAMING ENGINE (MEMORY LEAK FIXED) ====================
 def get_file_stream(message_id):
     q = queue.Queue(maxsize=10)
+    stop_flag = [False]
 
     async def producer():
         try:
@@ -110,14 +116,32 @@ def get_file_stream(message_id):
             msg = await bot.get_messages(STORAGE_CHANNEL_ID, message_id)
             media = get_media_obj(msg)
             if not media:
-                await asyncio.to_thread(q.put, None)
+                try: await asyncio.to_thread(q.put, None, True, 1.0)
+                except queue.Full: pass
                 return
+                
             async for chunk in bot.stream_media(msg):
-                await asyncio.to_thread(q.put, chunk)
+                if stop_flag[0]: break
+                
+                # Custom wait timeout to check stop_flag efficiently
+                put_success = False
+                while not stop_flag[0]:
+                    try:
+                        await asyncio.to_thread(q.put, chunk, True, 2.0)
+                        put_success = True
+                        break
+                    except queue.Full:
+                        continue
+                
+                if not put_success:
+                    break
+                    
         except Exception as e:
             print(f"Error in stream producer: {e}")
         finally:
-            await asyncio.to_thread(q.put, None)
+            if not stop_flag[0]:
+                try: await asyncio.to_thread(q.put, None, True, 1.0)
+                except queue.Full: pass
 
     asyncio.run_coroutine_threadsafe(producer(), main_loop)
 
@@ -129,6 +153,12 @@ def get_file_stream(message_id):
                 if chunk is None: break
                 yield chunk
         except GeneratorExit:
+            # Client closed video player or browser disconnected
+            pass
+        except Exception as e:
+            print(f"Consumer error: {e}")
+        finally:
+            stop_flag[0] = True
             while not q.empty():
                 try: q.get_nowait()
                 except: break
@@ -816,51 +846,77 @@ async def handle_media_upload(client, message):
             await proc.communicate()
             if not os.path.exists(thumb_path): thumb_path = None
 
-        if media_type == "photo":
-            sent_to_admin = await client.send_photo(message.chat.id, telegram_file, caption=admin_cap, parse_mode=enums.ParseMode.HTML)
-        else:
-            media = get_media_obj(message)
-            vid_duration = media.duration if media and hasattr(media, 'duration') and media.duration else 0
-            vid_width = media.width if media and hasattr(media, 'width') and media.width else 0
-            vid_height = media.height if media and hasattr(media, 'height') and media.height else 0
-            
-            sent_to_admin = await client.send_video(
-                message.chat.id, 
-                telegram_file, 
-                caption=admin_cap, 
-                parse_mode=enums.ParseMode.HTML, 
-                duration=vid_duration, 
-                width=vid_width, 
-                height=vid_height, 
-                thumb=thumb_path
-            )
+        # Auto-retry logic for sending final result to admin
+        try:
+            if media_type == "photo":
+                sent_to_admin = await client.send_photo(message.chat.id, telegram_file, caption=admin_cap, parse_mode=enums.ParseMode.HTML)
+            else:
+                media = get_media_obj(message)
+                vid_duration = media.duration if media and hasattr(media, 'duration') and media.duration else 0
+                vid_width = media.width if media and hasattr(media, 'width') and media.width else 0
+                vid_height = media.height if media and hasattr(media, 'height') and media.height else 0
+                
+                sent_to_admin = await client.send_video(
+                    message.chat.id, 
+                    telegram_file, 
+                    caption=admin_cap, 
+                    parse_mode=enums.ParseMode.HTML, 
+                    duration=vid_duration, 
+                    width=vid_width, 
+                    height=vid_height, 
+                    thumb=thumb_path
+                )
+        except Exception as e:
+            await message.reply(f"❌ Failed to send final file to you: {e}")
+            return
             
         tg_file_id = get_msg_file_id(sent_to_admin)
         if not tg_file_id:
             tg_file_id = get_msg_file_id(message)
             
-        await status_msg.delete()
+        try: await status_msg.delete()
+        except: pass
         
+        # Broadcast Phase with Retries
         groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
         group_ids = [g['group_id'] for g in groups_res.data]
         success_count, fail_count = 0, 0
         
         for gid in set(group_ids):
-            try:
-                if media_type == "photo": await client.send_photo(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
-                else: await client.send_video(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
-                success_count += 1
-                await asyncio.sleep(1.5)
-            except FloodWait as e:
-                await asyncio.sleep(e.value + 1)
+            retries = 3
+            while retries > 0:
                 try:
                     if media_type == "photo": await client.send_photo(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
                     else: await client.send_video(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
                     success_count += 1
-                except Exception: fail_count += 1
-            except Exception: fail_count += 1 
-                
-        await message.reply(f"📢 <b>Broadcast Complete!</b>\n\n✅ Success: {success_count} groups\n❌ Failed: {fail_count} groups", parse_mode=enums.ParseMode.HTML)
+                    break
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                except Exception as e:
+                    err_msg = str(e).lower()
+                    if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
+                        retries -= 1
+                        await asyncio.sleep(3)
+                    else:
+                        fail_count += 1
+                        break
+            await asyncio.sleep(1.5)
+            
+        # Success notification with Retries
+        retries = 3
+        while retries > 0:
+            try:
+                await message.reply(f"📢 <b>Broadcast Complete!</b>\n\n✅ Success: {success_count} groups\n❌ Failed: {fail_count} groups", parse_mode=enums.ParseMode.HTML)
+                break
+            except FloodWait as e:
+                await asyncio.sleep(e.value + 1)
+            except Exception as e:
+                err_msg = str(e).lower()
+                if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
+                    retries -= 1
+                    await asyncio.sleep(3)
+                else:
+                    break
         
     except Exception as e: 
         try: await message.reply(f"⚠️ Error occurred: {str(e)}")