Update app.py

app.py

@@ -53,6 +53,13 @@ temp_clients = {}
 
 upload_mode = "telegram"
 ffmpeg_available = True
+
+# ================= AUTO SHARE VARIABLES =================
+auto_share_running = False
+auto_share_task = None
+TARGET_SHARE_GROUP = -1004333630822
+# =========================================================
+
 try:
     subprocess.run(["ffmpeg", "-version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 except FileNotFoundError:
@@ -345,6 +352,130 @@ def api_verify_code():
         return jsonify(result)
     except Exception as e: return jsonify({"status": "error", "msg": str(e)})
 
+
+async def save_progress(source_id, dest_id, msg_id):
+    try:
+        res = await db_query(lambda: supabase.table('clone_progress').select('id').eq('source_id', source_id).eq('dest_id', dest_id).execute())
+        if res.data:
+            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}")
+
+
+# ================= AUTO SHARE LOGIC (5 videos / 5 mins) =================
+async def auto_share_loop(client, target_chat_id, delay=300):
+    global auto_share_running
+    
+    while auto_share_running:
+        try:
+            # Check maximum message ID to avoid infinite blank search
+            latest_msg_id = 0
+            async for m in client.get_chat_history(STORAGE_CHANNEL_ID, limit=1):
+                latest_msg_id = m.id
+                break
+                
+            progress_res = await db_query(lambda: supabase.table('clone_progress').select('last_copied_id').eq('source_id', STORAGE_CHANNEL_ID).eq('dest_id', target_chat_id).execute())
+            last_checked_id = 0
+            if progress_res.data:
+                last_checked_id = progress_res.data[0]['last_copied_id']
+            
+            if latest_msg_id > 0 and last_checked_id >= latest_msg_id:
+                await asyncio.sleep(60) 
+                continue
+
+            batch_to_send = []
+            while len(batch_to_send) < 5 and auto_share_running:
+                if latest_msg_id > 0 and last_checked_id >= latest_msg_id:
+                    break
+                    
+                chunk_ids = list(range(last_checked_id + 1, min(last_checked_id + 51, latest_msg_id + 1)))
+                if not chunk_ids:
+                    break
+                    
+                try:
+                    msgs = await client.get_messages(STORAGE_CHANNEL_ID, chunk_ids)
+                    for idx, msg in enumerate(msgs):
+                        msg_expected_id = chunk_ids[idx]
+                        if msg is None or getattr(msg, "empty", False):
+                            last_checked_id = max(last_checked_id, msg_expected_id)
+                            continue
+                            
+                        if msg.video or (msg.document and msg.document.mime_type and "video" in msg.document.mime_type) or msg.photo:
+                            batch_to_send.append(msg)
+                            last_checked_id = max(last_checked_id, msg.id)
+                            if len(batch_to_send) == 5:
+                                break
+                        else:
+                            last_checked_id = max(last_checked_id, msg.id)
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                except Exception as e:
+                    print(f"Fetch error in auto share: {e}")
+                    await asyncio.sleep(5)
+                    
+            if not batch_to_send:
+                await save_progress(STORAGE_CHANNEL_ID, target_chat_id, last_checked_id)
+                await asyncio.sleep(10)
+                continue
+                
+            sent_message_ids = []
+            for msg in batch_to_send:
+                if not auto_share_running: break
+                try:
+                    sent = await msg.copy(target_chat_id)
+                    sent_message_ids.append(sent.id)
+                    await asyncio.sleep(1.5) 
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                except Exception as e:
+                    print(f"Error copying msg {msg.id}: {e}")
+                    
+            await save_progress(STORAGE_CHANNEL_ID, target_chat_id, last_checked_id)
+            
+            for _ in range(delay):
+                if not auto_share_running: break
+                await asyncio.sleep(1)
+                
+            if sent_message_ids and auto_share_running:
+                try:
+                    await client.delete_messages(target_chat_id, sent_message_ids)
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                    await client.delete_messages(target_chat_id, sent_message_ids)
+                except Exception as e:
+                    print(f"Error deleting msgs: {e}")
+                    
+        except Exception as e:
+            print(f"Auto-share loop error: {e}")
+            await asyncio.sleep(5)
+
+@bot.on_message(filters.command("startshare") & filters.private & filters.user(ADMIN_IDS))
+async def start_sharing_cmd(client, message):
+    global auto_share_running, auto_share_task
+    args = message.command
+    
+    if len(args) > 1 and args[1].lower() == "reset":
+        await save_progress(STORAGE_CHANNEL_ID, TARGET_SHARE_GROUP, 0)
+        await message.reply("🔄 <b>Progress reset!</b> The bot will now start sharing from the 1st video.")
+        
+    if auto_share_running:
+        return await message.reply("⚠️ <b>Auto-share is already running!</b>\nUse `/stopshare` to stop it first.")
+    
+    auto_share_running = True
+    auto_share_task = asyncio.create_task(auto_share_loop(client, TARGET_SHARE_GROUP, 300))
+    await message.reply(f"✅ <b>Auto-sharing started!</b>\n\n📌 <b>Target Group:</b> <code>{TARGET_SHARE_GROUP}</code>\n📦 <b>Batch Size:</b> 5 videos\n⏱ <b>Interval:</b> 5 minutes\n\n<i>Bot will send 5 videos, wait 5 minutes, delete them, and send the next 5!</i>", parse_mode=enums.ParseMode.HTML)
+
+@bot.on_message(filters.command("stopshare") & filters.private & filters.user(ADMIN_IDS))
+async def stop_sharing_cmd(client, message):
+    global auto_share_running
+    if not auto_share_running:
+        return await message.reply("⚠️ Auto-share is not currently running.")
+    
+    auto_share_running = False
+    await message.reply("🛑 <b>Auto-sharing stopped successfully!</b>", parse_mode=enums.ParseMode.HTML)
+
+
 # ================= TELEGRAM BOT COMMANDS =================
 @bot.on_message(filters.command("start"))
 async def start(client, message):
@@ -396,15 +527,6 @@ async def start(client, message):
         await message.reply(welcome_text, parse_mode=enums.ParseMode.HTML, reply_markup=markup)
     except Exception as e: print(f"Start error: {e}")
 
-async def save_progress(source_id, dest_id, msg_id):
-    try:
-        res = await db_query(lambda: supabase.table('clone_progress').select('id').eq('source_id', source_id).eq('dest_id', dest_id).execute())
-        if res.data:
-            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}")
-
 @bot.on_message(filters.new_chat_members)
 async def bot_added_to_group(client, message):
     me = client.me
@@ -821,7 +943,6 @@ async def handle_media_upload(client, message):
             await proc.communicate()
             if not os.path.exists(thumb_path): thumb_path = None
 
-        # 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)
@@ -842,7 +963,6 @@ async def handle_media_upload(client, message):
                     thumb=thumb_path
                 )
             
-            # সেভ করে রাখা হচ্ছে যাতে নতুন গ্রুপে এড হলে এই ভিডিওটা দিতে পারে
             await save_progress(message.chat.id, 0, sent_to_admin.id)
             
         except Exception as e:
@@ -856,7 +976,6 @@ async def handle_media_upload(client, message):
         try: await status_msg.delete()
         except: pass
         
-        # 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:
@@ -866,7 +985,6 @@ async def handle_media_upload(client, message):
         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:
@@ -888,7 +1006,6 @@ async def handle_media_upload(client, message):
                         break
             await asyncio.sleep(1.5)
             
-        # Success notification safely
         retries = 3
         while retries > 0:
             try:
@@ -1057,7 +1174,7 @@ async def auto_clean_channel_loop():
             print(f"Auto clean error: {e}")
         await asyncio.sleep(4 * 3600) 
 
-@bot.on_message(filters.private & filters.user(ADMIN_IDS) & ~filters.command(["start", "stats", "users", "broadcast", "png", "addvideo", "blur", "clean", "sendto", "clone", "upload"]))
+@bot.on_message(filters.private & filters.user(ADMIN_IDS) & ~filters.command(["start", "stats", "users", "broadcast", "png", "addvideo", "blur", "clean", "sendto", "clone", "upload", "startshare", "stopshare"]))
 async def catch_admin_steps(client, message):
     state = admin_states.get(message.chat.id, {})
     if state.get("step") == 1:
@@ -1078,9 +1195,6 @@ async def catch_admin_steps(client, message):
 # ==========================================
 # REDIS QUEUE WATERMARK MULTIPROCESSING
 # ==========================================
-import json
-import redis.asyncio as redis
-
 REDIS_URL = os.environ.get("REDIS_URL_1")
 
 processing_owners = set()
@@ -1094,13 +1208,10 @@ async def process_single_video(task, redis_client):
     wm_text = task['watermark_text']
     task_id = f"{owner_id}_{message_id}"
     
-    # ==============================
-    # FIX: আগে ভেরিয়েবল ডিফাইন করা হলো যাতে finally ব্লকে এরর না আসে
     raw_video_path = None
     watermarked_path = None
     thumb_path = f"thumb_{task_id}.jpg"
     clone_client = None
-    # ==============================
     
     print(f"▶️ [WM] Task Started for User: {owner_id}, Msg: {message_id}")
     
@@ -1122,7 +1233,6 @@ async def process_single_video(task, redis_client):
         
         watermarked_path = f"wm_{task_id}.mp4"
         
-        # Hugging Face এ দ্রুত প্রসেস করার জন্য ultrafast, threads 4 এবং crf 26 ব্যবহার করা হলো
         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)'",
@@ -1146,14 +1256,12 @@ async def process_single_video(task, redis_client):
         
         await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>ফাইনাল ভিডিও স্টোরেজে আপলোড হচ্ছে...</b>", parse_mode=enums.ParseMode.HTML)
         
-        # Background এ অরিজিনাল ভিডিও স্টোরেজে পাঠানো হচ্ছে, যাতে ইউজারকে ওয়েট করতে না হয়
         async def backup_raw_video():
             try:
                 await bot.send_video(chat_id=STORAGE_CHANNEL_ID, video=raw_video_path, caption=f"Original Backup for Clone Owner {owner_id}")
             except Exception: pass
         asyncio.create_task(backup_raw_video())
         
-        # ওয়াটারমার্ক করা ভিডিওটি আপলোড হচ্ছে (সঠিক Stream Link বানানোর জন্য)
         wm_sent = await bot.send_video(
             chat_id=STORAGE_CHANNEL_ID,
             video=watermarked_path,
@@ -1161,7 +1269,6 @@ async def process_single_video(task, redis_client):
             thumb=thumb_path
         )
         
-        # FIX: এখন লিংকগুলো অরিজিনালের বদলে ওয়াটারমার্ক ভিডিও পয়েন্ট করবে
         wm_storage_id = wm_sent.id
         stream_link = f"{BACKEND_URL}/stream/{wm_storage_id}"
         download_link = f"{BACKEND_URL}/download/{wm_storage_id}"
@@ -1191,7 +1298,6 @@ async def process_single_video(task, redis_client):
             print(f"❌ [WM] Failed to send error msg to user: {ex}")
         
     finally:
-        # ফাইল এবং লক ক্লিনআপ
         if raw_video_path and os.path.exists(raw_video_path): os.remove(raw_video_path)
         if watermarked_path and os.path.exists(watermarked_path): os.remove(watermarked_path)
         if thumb_path and os.path.exists(thumb_path): os.remove(thumb_path)
@@ -1218,7 +1324,6 @@ async def watermark_processor_loop():
         print(f"❌ Redis Connection Failed in Hugging Face: {e}")
         return
     
-    # Clear stuck locks
     try:
         async for key in redis_client.scan_iter("wm_processing:*"):
             await redis_client.delete(key)
@@ -1264,5 +1369,4 @@ async def main():
 
 if __name__ == "__main__":
     threading.Thread(target=run_flask, daemon=True).start()
-    main_loop.run_until_complete(main())
-#--- END OF FILE main.py
+    main_loop.run_until_complete(main())