Update To V0.0.5 🔥

RknDeveloper · web-flow · commit 60fae953d179 · 2024-03-21T20:21:15.000+05:45
diff --git a/Rkn_Bots/Caption.py b/Rkn_Bots/Caption.py
@@ -0,0 +1,131 @@
+# (c) @RknDeveloperr
+# Rkn Developer 
+# Don't Remove Credit 😔
+# Telegram Channel @RknDeveloper & @Rkn_Bots
+# Developer @RknDeveloperr
+
+from pyrogram import Client, filters, errors
+from Config import Rkn_Bots
+import asyncio, re, time, sys
+from .database import *
+from pyrogram.errors import FloodWait
+
+@Client.on_message(filters.private & filters.user(Rkn_Bots.ADMIN)  & filters.command(["rknusers"]))
+async def all_db_users_here(client,message):
+    x = await message.reply_text("Please Wait....")
+    total = await total_user()
+    await x.edit(f"Tᴏᴛᴀʟ Usᴇʀ :- `{total}`")
+
+@Client.on_message(filters.private & filters.user(Rkn_Bots.ADMIN) & filters.command(["broadcast"]))
+async def broadcast(bot, message):
+    if (message.reply_to_message):
+        rkn = await message.reply_text("Geting All ids from database..\n Please wait")
+        ids = await getid()
+        tot = await total_user()
+        success = 0
+        failed = 0
+        deactivated = 0
+        blocked = 0
+        await rkn.edit(f"ʙʀᴏᴀᴅᴄᴀsᴛɪɴɢ...")
+        for id in ids:
+            try:
+                time.sleep(1)
+                await message.reply_to_message.copy(id)
+                success += 1
+            except errors.InputUserDeactivated:
+                deactivated +=1
+                await delete({"_id": id})
+            except errors.UserIsBlocked:
+                blocked +=1
+                await delete({"_id": id})
+            except Exception as e:
+                failed += 1
+                await delete({"_id": bot_id})
+                pass
+            try:
+                await rkn.edit(f"<u>ʙʀᴏᴀᴅᴄᴀsᴛ ᴘʀᴏᴄᴇssɪɴɢ</u>\n\n• ᴛᴏᴛᴀʟ ᴜsᴇʀs: {tot}\n• sᴜᴄᴄᴇssғᴜʟ: {success}\n• ʙʟᴏᴄᴋᴇᴅ ᴜsᴇʀs: {blocked}\n• ᴅᴇʟᴇᴛᴇᴅ ᴀᴄᴄᴏᴜɴᴛs: {deactivated}\n• ᴜɴsᴜᴄᴄᴇssғᴜʟ: {failed}")
+            except FloodWait as e:
+                await asyncio.sleep(t.x)
+        await rkn.edit(f"<u>ʙʀᴏᴀᴅᴄᴀsᴛ ᴄᴏᴍᴘʟᴇᴛᴇᴅ</u>\n\n• ᴛᴏᴛᴀʟ ᴜsᴇʀs: {tot}\n• sᴜᴄᴄᴇssғᴜʟ: {success}\n• ʙʟᴏᴄᴋᴇᴅ ᴜsᴇʀs: {blocked}\n• ᴅᴇʟᴇᴛᴇᴅ ᴀᴄᴄᴏᴜɴᴛs: {deactivated}\n• ᴜɴsᴜᴄᴄᴇssғᴜʟ: {failed}")
+        
+#Restart to cancell all process 
+@Client.on_message(filters.private & filters.user(Rkn_Bots.ADMIN) & filters.command("restart"))
+async def restart_bot(b, m):
+    msg = await b.send_message(text="**🔄 𝙿𝚁𝙾𝙲𝙴𝚂𝚂𝙴𝚂 𝚂𝚃𝙾𝙿𝙴𝙳. 𝙱𝙾𝚃 𝙸𝚂 𝚁𝙴𝚂𝚃𝙰𝚁𝚃𝙸𝙽𝙶...**", chat_id=m.chat.id)       
+    await asyncio.sleep(3)
+    await msg.edit("**✅️ 𝙱𝙾𝚃 𝙸𝚂 𝚁𝙴𝚂𝚃𝙰𝚁𝚃𝙴𝙳. 𝙽𝙾𝚆 𝚈𝙾𝚄 𝙲𝙰𝙽 𝚄𝚂𝙴 𝙼𝙴**")
+    os.execl(sys.executable, sys.executable, *sys.argv)
+    
+@Client.on_message(filters.command("start") & filters.private)
+async def start_cmd(bot, message):
+    user_id = int(message.from_user.id)
+    user_det = {"_id": user_id)
+    await users.insert_one(user_det)
+    await message.reply(
+        f"<b>Hey, {message.from_user.mention}\n\nI'm an auto-caption bot. I automatically edit captions for videos, audio files, and documents posted on channels.\n\nuse <code>/set_caption</code> to set caption\nUse<code>/delcaption</code> To delete caption and set caption to default.\n\nNote:All commands works on channels only</b>"
+    )
+
+
+@Client.on_message(filters.command("set_caption") & filters.channel)
+async def setCap(bot, message):
+    if len(message.command) < 2:
+        return await message.reply(
+            "Usage: /set_caption <code>your caption (use {file_name} to show file name</code>)"
+        )
+    chnl_id = message.chat.id
+    caption = (
+        message.text.split(" ", 1)[1] if len(message.text.split(" ", 1)) > 1 else None
+    )
+    chkData = await chnl_ids.find_one({"chnl_id": chnl_id})
+    if chkData:
+        await updateCap(chnl_id, caption)
+        return await message.reply(f"Your New Caption: {caption}")
+    else:
+        await addCap(chnl_id, caption)
+        return await message.reply(f"Your New Caption: {caption}")
+
+
+@Client.on_message(filters.command(["delcaption", "del_caption", "delete_caption"]) & filters.channel)
+async def delCap(_, msg):
+    chnl_id = msg.chat.id
+    try:
+        await chnl_ids.delete_one({"chnl_id": chnl_id})
+        return await msg.reply("<b>Success..From now i will use my default caption</b>")
+    except Exception as e:
+        e_val = await msg.replay(f"ERR I GOT: {e}")
+        await asyncio.sleep(5)
+        await e_val.delete()
+        return
+
+
+@Client.on_message(filters.channel)
+async def auto_edit_caption(bot, message):
+    chnl_id = message.chat.id
+    if message.media:
+        for file_type in ("video", "audio", "document", "voice"):
+            obj = getattr(message, file_type, None)
+            if obj and hasattr(obj, "file_name"):
+                file_name = obj.file_name
+                file_name = (
+                    re.sub(r"@\w+\s*", "", file_name)
+                    .replace("_", " ")
+                    .replace(".", " ")
+                )
+                cap_dets = await chnl_ids.find_one({"chnl_id": chnl_id})
+                try:
+                    if cap_dets:
+                        cap = cap_dets["caption"]
+                        replaced_caption = cap.format(file_name=file_name)
+                        await message.edit(replaced_caption)
+                    else:
+                        replaced_caption = Rkn_Bots.DEF_CAP.format(file_name=file_name)
+                        await message.edit(replaced_caption)
+                except FloodWait as e:
+                    await asyncio.sleep(e.x)
+                    continue
+    return
+
+# Rkn Developer 
+# Don't Remove Credit 😔
+# Telegram Channel @RknDeveloper & @Rkn_Bots
+# Developer @RknDeveloperr
diff --git a/Rkn_Bots/Force_Sub.py b/Rkn_Bots/Force_Sub.py
@@ -0,0 +1,35 @@
+from pyrogram import Client, filters, enums 
+from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup
+from pyrogram.errors import UserNotParticipant
+from config import Rkn_Bots as Config
+from .database import *
+
+async def not_subscribed(_, client, message):
+    user_id = int(message.from_user.id)
+    user_det = {"_id": user_id)
+    await users.insert_one(user_det)
+    if not Config.FORCE_SUB:
+        return False
+    try:             
+        user = await client.get_chat_member(Config.FORCE_SUB, message.from_user.id) 
+        if user.status == enums.ChatMemberStatus.BANNED:
+            return True 
+        else:
+            return False                
+    except UserNotParticipant:
+        pass
+    return True
+
+
+@Client.on_message(filters.private & filters.create(not_subscribed))
+async def forces_sub(client, message):
+    buttons = [[InlineKeyboardButton(text="📢 Join Update Channel 📢", url=f"https://t.me/{Config.FORCE_SUB}") ]]
+    text = "**Sᴏʀʀy Dᴜᴅᴇ Yᴏᴜ'ʀᴇ Nᴏᴛ Jᴏɪɴᴇᴅ My Cʜᴀɴɴᴇʟ 😐. Sᴏ Pʟᴇᴀꜱᴇ Jᴏɪɴ Oᴜʀ Uᴩᴅᴀᴛᴇ Cʜᴀɴɴᴇʟ Tᴏ Cᴄᴏɴᴛɪɴᴜᴇ**"
+    try:
+        user = await client.get_chat_member(Config.FORCE_SUB, message.from_user.id)    
+        if user.status == enums.ChatMemberStatus.BANNED:                                   
+            return await client.send_message(message.from_user.id, text="Sᴏʀʀy Yᴏᴜ'ʀᴇ Bᴀɴɴᴇᴅ Tᴏ Uꜱᴇ Mᴇ")  
+    except UserNotParticipant:                       
+        return await message.reply_text(text=text, reply_markup=InlineKeyboardMarkup(buttons))
+    return await message.reply_text(text=text, reply_markup=InlineKeyboardMarkup(buttons))
+          
diff --git a/Rkn_Bots/database.py b/Rkn_Bots/database.py
@@ -0,0 +1,37 @@
+# (c) @RknDeveloperr
+# Rkn Developer 
+# Don't Remove Credit 😔
+# Telegram Channel @RknDeveloper & @Rkn_Bots
+# Developer @RknDeveloperr
+
+import motor.motor_asyncio
+from config import Rkn_Bots
+
+client = motor.motor_asyncio.AsyncIOMotorClient(Rkn_Bots.DB_URL)
+db = client.captions_with_chnl
+chnl_ids = db.chnl_ids
+users = db.users
+
+# Total User
+async def total_user():
+    user = users.count_documents({})
+    return user
+
+async def getid():
+    all_users = users.find({})
+    return all_users
+
+async def delete(id):
+    users.delete_one(id)
+                     
+async def addCap(chnl_id, caption):
+    dets = {"chnl_id": chnl_id, "caption": caption}
+    await chnl_ids.insert_one(dets)
+
+async def updateCap(chnl_id, caption):
+    await chnl_ids.update_one({"chnl_id": chnl_id}, {"$set": {"caption": caption}})
+
+# Rkn Developer 
+# Don't Remove Credit 😔
+# Telegram Channel @RknDeveloper & @Rkn_Bots
+# Developer @RknDeveloperr
diff --git a/Rkn_Bots/web_support.py b/Rkn_Bots/web_support.py
@@ -0,0 +1,12 @@
+from aiohttp import web
+
+Rkn_AutoCaptionBot = web.RouteTableDef()
+
+@Rkn_AutoCaptionBot.get("/", allow_head=True)
+async def root_route_handler(request):
+    return web.json_response("Rkn_AutoCaptionBot")
+
+async def web_server():
+    web_app = web.Application(client_max_size=30000000)
+    web_app.add_routes(Rkn_AutoCaptionBot)
+    return web_app
