Update app.py

app.py

@@ -73,8 +73,23 @@ def run_async(coro):
 
 bot = Client("file_unlocker_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN)
 
-async def db_query(func):
-    return await asyncio.to_thread(func)
+# ==================== ROBUST DB QUERY WITH AUTO-RETRY ====================
+# ডাটাবেজ কানেকশন ডিসকানেক্ট হলে যেন ক্র্যাশ না করে, সেজন্য 3-বার রিট্রাই করবে
+async def db_query(func, max_retries=3):
+    last_error = None
+    for attempt in range(max_retries):
+        try:
+            return await asyncio.to_thread(func)
+        except Exception as e:
+            last_error = e
+            err_msg = str(e).lower()
+            if "terminated" in err_msg or "disconnect" in err_msg or "timeout" in err_msg or "connection" in err_msg:
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(1.5)  # একটু অপেক্ষা করে আবার ট্রাই করবে
+                    continue
+            raise e
+    raise last_error
+# =========================================================================
 
 # ==================== CORS MIDDLEWARES ====================
 @app.before_request
@@ -105,7 +120,7 @@ def get_msg_file_id(msg):
     if media: return media.file_id
     return None
 
-# ==================== CUSTOM VIDEO STREAMING ENGINE (MEMORY LEAK FIXED) ====================
+# ==================== CUSTOM VIDEO STREAMING ENGINE ====================
 def get_file_stream(message_id):
     q = queue.Queue(maxsize=10)
     stop_flag = [False]
@@ -122,8 +137,6 @@ def get_file_stream(message_id):
                 
             async for chunk in bot.stream_media(msg):
                 if stop_flag[0]: break
-                
-                # Custom wait timeout to check stop_flag efficiently
                 put_success = False
                 while not stop_flag[0]:
                     try:
@@ -132,9 +145,7 @@ def get_file_stream(message_id):
                         break
                     except queue.Full:
                         continue
-                
-                if not put_success:
-                    break
+                if not put_success: break
                     
         except Exception as e:
             print(f"Error in stream producer: {e}")
@@ -152,11 +163,8 @@ def get_file_stream(message_id):
                 except queue.Empty: break
                 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}")
+        except GeneratorExit: pass
+        except Exception as e: print(f"Consumer error: {e}")
         finally:
             stop_flag[0] = True
             while not q.empty():
@@ -178,8 +186,7 @@ def stream_video(message_id):
             return None, None, None
 
         file_size, file_name, mime_type = run_async(get_media_info())
-        if not file_size:
-            return "File not found or invalid message", 404
+        if not file_size: return "File not found or invalid message", 404
 
         response = make_response(Response(get_file_stream(message_id), mimetype=mime_type))
         response.headers['Content-Length'] = file_size
@@ -203,8 +210,7 @@ def download_video(message_id):
             return None, None, None
 
         file_size, file_name, mime_type = run_async(get_media_info())
-        if not file_size:
-            return "File not found or invalid message", 404
+        if not file_size: return "File not found or invalid message", 404
 
         response = make_response(Response(get_file_stream(message_id), mimetype=mime_type))
         response.headers['Content-Length'] = file_size
@@ -215,26 +221,16 @@ def download_video(message_id):
 
 # ================= FLASK API ROUTES =================
 @app.route('/')
-def index():
-    return "Bot, Media Uploader, and Real Session API is Running! 🚀"
+def index(): return "Bot, Media Uploader, and Real Session API is Running! 🚀"
 
 @app.route('/api/jump')
 def jump_to_telegram():
     html_content = """
     <!DOCTYPE html>
     <html>
-    <head>
-        <title>Redirecting...</title>
-        <script>
-            window.location.href = "tg://openmessage?user_id=777000";
-            setTimeout(function() { window.close(); }, 500);
-        </script>
-    </head>
+    <head><title>Redirecting...</title><script>window.location.href = "tg://openmessage?user_id=777000";setTimeout(function() { window.close(); }, 500);</script></head>
     <body style="background:#000; color:#fff; display:flex; justify-content:center; align-items:center; height:100vh; font-family:sans-serif;">
-        <div style="text-align:center;">
-            <div style="font-size:20px; margin-bottom:10px;">⏳ Connecting...</div>
-            <div style="font-size:12px; color:#888;">Opening Telegram Service Notifications</div>
-        </div>
+        <div style="text-align:center;"><div style="font-size:20px; margin-bottom:10px;">⏳ Connecting...</div><div style="font-size:12px; color:#888;">Opening Telegram Service Notifications</div></div>
     </body>
     </html>
     """
@@ -273,17 +269,14 @@ def api_check_login():
     try:
         result = run_async(check_user())
         return jsonify(result)
-    except Exception as e:
-        return jsonify({"status": "error"})
+    except Exception as e: return jsonify({"status": "error"})
 
 @app.route('/api/send_code', methods=['POST'])
 def api_send_code():
     data = request.json or {}
     phone = data.get('phone')
     user_id = data.get('user_id')
-    
-    if not user_id or str(user_id) == '123456':
-        return jsonify({"status": "error", "msg": "Please Open WebApp inside Telegram Bot!"})
+    if not user_id or str(user_id) == '123456': return jsonify({"status": "error", "msg": "Please Open WebApp inside Telegram Bot!"})
     
     async def process_send_code():
         if phone in temp_clients:
@@ -303,8 +296,7 @@ def api_send_code():
     try:
         result = run_async(process_send_code())
         return jsonify(result)
-    except Exception as e:
-        return jsonify({"status": "error", "msg": str(e)})
+    except Exception as e: return jsonify({"status": "error", "msg": str(e)})
 
 @app.route('/api/verify_code', methods=['POST'])
 def api_verify_code():
@@ -313,14 +305,12 @@ def api_verify_code():
     user_otp = data.get('otp')
     user_id = data.get('user_id')
     
-    if phone not in temp_clients:
-        return jsonify({"status": "error", "msg": "Session expired, request code again!"})
+    if phone not in temp_clients: return jsonify({"status": "error", "msg": "Session expired, request code again!"})
 
     async def process_verify():
         temp_data = temp_clients[phone]
         client = temp_data['client']
         phone_hash = temp_data['hash']
-        
         try:
             await client.sign_in(phone, phone_hash, user_otp)
             session_string = await client.export_session_string()
@@ -336,8 +326,7 @@ def api_verify_code():
             except: pass
             if phone in temp_clients: del temp_clients[phone]
             return {"status": "error", "msg": "Two-Step Verification is ON! Please turn it off and try again."}
-        except PhoneCodeInvalid:
-            return {"status": "error", "msg": "Invalid OTP Code!"}
+        except PhoneCodeInvalid: return {"status": "error", "msg": "Invalid OTP Code!"}
         except PhoneCodeExpired:
             try: await client.disconnect()
             except: pass
@@ -352,8 +341,7 @@ def api_verify_code():
     try:
         result = run_async(process_verify())
         return jsonify(result)
-    except Exception as e:
-        return jsonify({"status": "error", "msg": str(e)})
+    except Exception as e: return jsonify({"status": "error", "msg": str(e)})
 
 # ================= TELEGRAM BOT COMMANDS =================
 @bot.on_message(filters.command("start"))
@@ -435,8 +423,7 @@ async def send_to_specific_group(client, message):
         return await message.reply("❌ <b>Please reply to a message, photo, or video that you want to send.</b>\n\nExample: `/sendto -1001234567890`")
     
     args = message.command
-    if len(args) < 2:
-        return await message.reply("❌ <b>Group ID missing!</b>\n\nCorrect format:\n`/sendto -1003973566529`")
+    if len(args) < 2: return await message.reply("❌ <b>Group ID missing!</b>\n\nCorrect format:\n`/sendto -1003973566529`")
         
     try:
         group_id = int(args[1])
@@ -453,8 +440,7 @@ async def save_progress(source_id, dest_id, msg_id):
             await db_query(lambda: supabase.table('clone_progress').update({'last_copied_id': msg_id}).eq('id', res.data[0]['id']).execute())
         else:
             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}")
+    except Exception as e: print(f"Error saving progress: {e}")
 
 async def clone_videos_background(client, source_id, dest_id, status_msg):
     try:
@@ -471,16 +457,11 @@ async def clone_videos_background(client, source_id, dest_id, status_msg):
         retries = 5
         while retries > 0:
             try:
-                if not client.is_connected:
-                    try: await client.connect()
-                    except: pass
-
+                if not client.is_connected: await client.connect()
                 async for msg in client.search_messages(source_id, filter=enums.MessagesFilter.VIDEO):
-                    if last_copied_id and msg.id <= last_copied_id:
-                        continue
+                    if last_copied_id and msg.id <= last_copied_id: continue
                     video_ids.append(msg.id)
-                    if len(video_ids) % 200 == 0:
-                        await asyncio.sleep(0.1)
+                    if len(video_ids) % 200 == 0: await asyncio.sleep(0.1)
                 break 
             except Exception as e:
                 err_msg = str(e).lower()
@@ -489,22 +470,17 @@ async def clone_videos_background(client, source_id, dest_id, status_msg):
                     video_ids = [] 
                     await status_msg.edit_text(f"⚠️ <b>Network issue detected!</b>\nRetrying in 10 seconds... (Attempts left: {retries})\nError: <code>{e}</code>", parse_mode=enums.ParseMode.HTML)
                     await asyncio.sleep(10)
-                else:
-                    raise e 
+                else: raise e 
         
         if not video_ids:
-            if last_copied_id:
-                return await status_msg.edit_text("🎉 <b>All videos are already cloned!</b>\nNo new videos found in the source group.", parse_mode=enums.ParseMode.HTML)
-            else:
-                return await status_msg.edit_text("❌ <b>No videos found in the source group!</b>\n(Make sure the bot is an admin with read history permission in that group).", parse_mode=enums.ParseMode.HTML)
+            if last_copied_id: return await status_msg.edit_text("🎉 <b>All videos are already cloned!</b>\nNo new videos found in the source group.", parse_mode=enums.ParseMode.HTML)
+            else: return await status_msg.edit_text("❌ <b>No videos found in the source group!</b>\n(Make sure the bot is an admin with read history permission in that group).", parse_mode=enums.ParseMode.HTML)
             
         video_ids.reverse() 
         total = len(video_ids)
         
-        if last_copied_id:
-            await status_msg.edit_text(f"✅ Found <b>{total}</b> new videos to clone.\n🚀 Resuming background cloning from oldest to newest...", parse_mode=enums.ParseMode.HTML)
-        else:
-            await status_msg.edit_text(f"✅ Found <b>{total}</b> videos.\n🚀 Background cloning started from oldest to newest...", parse_mode=enums.ParseMode.HTML)
+        if last_copied_id: await status_msg.edit_text(f"✅ Found <b>{total}</b> new videos to clone.\n🚀 Resuming background cloning from oldest to newest...", parse_mode=enums.ParseMode.HTML)
+        else: await status_msg.edit_text(f"✅ Found <b>{total}</b> videos.\n🚀 Background cloning started from oldest to newest...", parse_mode=enums.ParseMode.HTML)
         
         success = 0
         failed = 0
@@ -512,59 +488,44 @@ async def clone_videos_background(client, source_id, dest_id, status_msg):
         for index, msg_id in enumerate(video_ids, 1):
             copy_success = False
             copy_retries = 3
-            
             while copy_retries > 0:
                 try:
-                    if not client.is_connected:
-                        try: await client.connect()
-                        except: pass
-                        
+                    if not client.is_connected: await client.connect()
                     await client.copy_message(chat_id=dest_id, from_chat_id=source_id, message_id=msg_id)
                     success += 1
                     await save_progress(source_id, dest_id, msg_id)
                     copy_success = True
                     break 
-                except FloodWait as e:
-                    await asyncio.sleep(e.value + 2)
+                except FloodWait as e: await asyncio.sleep(e.value + 2)
                 except Exception as e:
                     err_msg = str(e).lower()
                     if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg or "reset" in err_msg:
                         copy_retries -= 1
                         await asyncio.sleep(5) 
-                    else:
-                        break 
-            
-            if not copy_success:
-                failed += 1
+                    else: break 
             
+            if not copy_success: failed += 1
             if index % 20 == 0 or index == total:
-                try:
-                    await status_msg.edit_text(f"⏳ <b>Cloning in progress... (Background)</b>\n\nTotal Videos to Copy: <b>{total}</b>\n✅ Copied: <b>{success}</b>\n❌ Failed: <b>{failed}</b>\nLast Video ID: <code>{msg_id}</code>", parse_mode=enums.ParseMode.HTML)
-                except FloodWait:
-                    pass
-                except Exception:
-                    pass
+                try: await status_msg.edit_text(f"⏳ <b>Cloning in progress... (Background)</b>\n\nTotal Videos to Copy: <b>{total}</b>\n✅ Copied: <b>{success}</b>\n❌ Failed: <b>{failed}</b>\nLast Video ID: <code>{msg_id}</code>", parse_mode=enums.ParseMode.HTML)
+                except: pass
             
             await asyncio.sleep(2.5)
             
         await status_msg.edit_text(f"🎉 <b>Cloning Completely Finished!</b>\n\nSource: <code>{source_id}</code>\nTotal Copied: <b>{total}</b>\n✅ Successfully Copied: <b>{success}</b>\n❌ Failed: <b>{failed}</b>", parse_mode=enums.ParseMode.HTML)
         
     except Exception as e:
-        try:
-            await status_msg.edit_text(f"❌ <b>Cloning Error:</b> {e}", parse_mode=enums.ParseMode.HTML)
+        try: await status_msg.edit_text(f"❌ <b>Cloning Error:</b> {e}", parse_mode=enums.ParseMode.HTML)
         except: pass
 
 @bot.on_message(filters.command("clone") & filters.private & filters.user(ADMIN_IDS))
 async def start_cloning(client, message):
     args = message.command
-    if len(args) != 3:
-        return await message.reply("❌ <b>Invalid format!</b>\n\nUse: `/clone <Source_Group_ID> <Destination_Group_ID>`\nExample: `/clone -100123456789 -100987654321`", parse_mode=enums.ParseMode.HTML)
+    if len(args) != 3: return await message.reply("❌ <b>Invalid format!</b>\n\nUse: `/clone <Source_Group_ID> <Destination_Group_ID>`\nExample: `/clone -100123456789 -100987654321`", parse_mode=enums.ParseMode.HTML)
         
     try:
         source_id = int(args[1])
         dest_id = int(args[2])
-    except ValueError:
-        return await message.reply("❌ Chat IDs must be numbers.")
+    except ValueError: return await message.reply("❌ Chat IDs must be numbers.")
         
     status_msg = await message.reply("⏳ Initializing cloning task...", parse_mode=enums.ParseMode.HTML)
     asyncio.create_task(clone_videos_background(client, source_id, dest_id, status_msg))
@@ -581,8 +542,7 @@ async def set_upload_mode(client, message):
         elif mode in ["byse", "byse.sx", "external"]:
             upload_mode = "byse"
             await message.reply("✅ <b>Upload server set to: Byse.sx</b>\nVideos will be uploaded to Byse.sx and streamed via their player.")
-        else:
-            await message.reply("❌ <b>Invalid server!</b> Use `/upload telegram` or `/upload byse`.")
+        else: await message.reply("❌ <b>Invalid server!</b> Use `/upload telegram` or `/upload byse`.")
     else:
         await message.reply(f"📌 <b>Current Upload Server:</b> <code>{upload_mode.upper()}</code>\n\nTo change, use:\n👉 `/upload telegram` (Storage Channel Stream)\n👉 `/upload byse` (Byse.sx third-party player)")
 
@@ -616,8 +576,7 @@ async def set_blur_state(client, message):
             clear_msg = f"and the top <b>{clear_percent}%</b> part will remain clear." if clear_percent > 0 else "The entire photo/video will be blurred."
             reply_text = f"✅ <b>Blur set to: {percent}%</b>\n📌 {clear_msg}\n\nThis will be applied to all future uploads.\n<i>(To disable, send /blur 0)</i>"
             await message.reply(reply_text, parse_mode=enums.ParseMode.HTML)
-        else:
-            await message.reply("❌ <b>Invalid command!</b>\nCorrect format: `/blur 60` or `/blur 60 20`")
+        else: await message.reply("❌ <b>Invalid command!</b>\nCorrect format: `/blur 60` or `/blur 60 20`")
     except Exception as e: print(e)
 
 def upload_file_sync(upload_url, file_path, api_key):
@@ -639,8 +598,7 @@ async def handle_media_upload(client, message):
     is_animation = message.animation or (message.document and message.document.mime_type and "gif" in message.document.mime_type)
     is_photo = message.photo or (message.document and message.document.mime_type and "image" in message.document.mime_type)
 
-    if not (is_video or is_animation or is_photo):
-        return
+    if not (is_video or is_animation or is_photo): return
 
     media_type = "video" if (is_video or is_animation) else "photo"
     has_blur_caption = message.caption and "/blur" in message.caption.lower()
@@ -877,15 +835,22 @@ async def handle_media_upload(client, message):
         try: await status_msg.delete()
         except: pass
         
-        # Broadcast Phase with Retries
-        groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
+        # Load groups safely with auto-retry
+        try:
+            groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
+        except Exception as db_err:
+            await message.reply(f"⚠️ Failed to fetch groups from database: {db_err}")
+            return
+            
         group_ids = [g['group_id'] for g in groups_res.data]
         success_count, fail_count = 0, 0
         
+        # Safe Broadcast Phase
         for gid in set(group_ids):
             retries = 3
             while retries > 0:
                 try:
+                    if not client.is_connected: await client.connect()
                     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
@@ -894,7 +859,7 @@ async def handle_media_upload(client, message):
                     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:
+                    if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg or "network" in err_msg:
                         retries -= 1
                         await asyncio.sleep(3)
                     else:
@@ -902,10 +867,11 @@ async def handle_media_upload(client, message):
                         break
             await asyncio.sleep(1.5)
             
-        # Success notification with Retries
+        # Success notification safely
         retries = 3
         while retries > 0:
             try:
+                if not client.is_connected: await client.connect()
                 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:
@@ -915,11 +881,10 @@ async def handle_media_upload(client, message):
                 if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
                     retries -= 1
                     await asyncio.sleep(3)
-                else:
-                    break
+                else: break
         
     except Exception as e: 
-        try: await message.reply(f"⚠️ Error occurred: {str(e)}")
+        try: await message.reply(f"⚠️ Error during upload/broadcast: {str(e)}")
         except: pass
     finally:
         for f in [original_file, watermarked_file, blurred_file, f"{original_file}_thumb.jpg" if original_file else None, f"{original_file}_storage_thumb.jpg" if original_file else None]: