Upload 3 files

Akeno/utils/database.py

@@ -0,0 +1,583 @@
+import datetime
+import time
+
+from motor import motor_asyncio
+from motor.core import AgnosticClient
+
+from config import MONGO_URL
+from Akeno.utils.logger import LOGS
+
+
+class Database:
+    def __init__(self, uri: str) -> None:
+        self.client: AgnosticClient = motor_asyncio.AsyncIOMotorClient(uri)
+        self.db = self.client["Akeno"]
+
+        self.afk = self.db["afk"]
+        self.antiflood = self.db["antiflood"]
+        self.autopost = self.db["autopost"]
+        self.blacklist = self.db["blacklist"]
+        self.echo = self.db["echo"]
+        self.env = self.db["env"]
+        self.filter = self.db["filter"]
+        self.forcesub = self.db["forcesub"]
+        self.gachabots = self.db["gachabots"]
+        self.gban = self.db["gban"]
+        self.gmute = self.db["gmute"]
+        self.greetings = self.db["greetings"]
+        self.mute = self.db["mute"]
+        self.pmpermit = self.db["pmpermit"]
+        self.session = self.db["session"]
+        self.snips = self.db["snips"]
+        self.stan_users = self.db["stan_users"]
+
+    async def connect(self):
+        try:
+            await self.client.admin.command("ping")
+            LOGS.info(
+                f"{Symbols.bullet * 3} Database Connection Established! {Symbols.bullet * 3}"
+            )
+        except Exception as e:
+            LOGS.info(f"{Symbols.cross_mark} DatabaseErr: {e} ")
+            quit(1)
+
+    def get_datetime(self) -> str:
+        return datetime.datetime.now().strftime("%d/%m/%Y - %H:%M")
+
+    async def set_env(self, name: str, value: str) -> None:
+        await self.env.update_one(
+            {"name": name}, {"$set": {"value": value}}, upsert=True
+        )
+
+    async def get_env(self, name: str) -> str | None:
+        if await self.is_env(name):
+            data = await self.env.find_one({"name": name})
+            return data["value"]
+        return None
+
+    async def rm_env(self, name: str) -> None:
+        await self.env.delete_one({"name": name})
+
+    async def is_env(self, name: str) -> bool:
+        if await self.env.find_one({"name": name}):
+            return True
+        return False
+
+    async def get_all_env(self) -> list:
+        return [i async for i in self.env.find({})]
+
+    async def is_stan(self, client: int, user_id: int) -> bool:
+        if await self.stan_users.find_one({"client": client, "user_id": user_id}):
+            return True
+        return False
+
+    async def add_stan(self, client: int, user_id: int) -> bool:
+        if await self.is_stan(client, user_id):
+            return False
+        await self.stan_users.insert_one(
+            {"client": client, "user_id": user_id, "date": self.get_datetime()}
+        )
+        return True
+
+    async def rm_stan(self, client: int, user_id: int) -> bool:
+        if not await self.is_stan(client, user_id):
+            return False
+        await self.stan_users.delete_one({"client": client, "user_id": user_id})
+        return True
+
+    async def get_stans(self, client: int) -> list:
+        return [i async for i in self.stan_users.find({"client": client})]
+
+    async def get_all_stans(self) -> list:
+        return [i async for i in self.stan_users.find({})]
+
+    async def is_session(self, user_id: int) -> bool:
+        if await self.session.find_one({"user_id": user_id}):
+            return True
+        return False
+
+    async def update_session(self, user_id: int, session: str) -> None:
+        await self.session.update_one(
+            {"user_id": user_id},
+            {"$set": {"session": session, "date": self.get_datetime()}},
+            upsert=True,
+        )
+
+    async def rm_session(self, user_id: int) -> None:
+        await self.session.delete_one({"user_id": user_id})
+
+    async def get_session(self, user_id: int):
+        if not await self.is_session(user_id):
+            return False
+        data = await self.session.find_one({"user_id": user_id})
+        return data
+
+    async def get_all_sessions(self) -> list:
+        return [i async for i in self.session.find({})]
+
+    async def is_gbanned(self, user_id: int) -> bool:
+        if await self.gban.find_one({"user_id": user_id}):
+            return True
+        return False
+
+    async def add_gban(self, user_id: int, reason: str) -> bool:
+        if await self.is_gbanned(user_id):
+            return False
+        await self.gban.insert_one(
+            {"user_id": user_id, "reason": reason, "date": self.get_datetime()}
+        )
+        return True
+
+    async def rm_gban(self, user_id: int):
+        if not await self.is_gbanned(user_id):
+            return None
+        reason = (await self.gban.find_one({"user_id": user_id}))["reason"]
+        await self.gban.delete_one({"user_id": user_id})
+        return reason
+
+    async def get_gban(self) -> list:
+        return [i async for i in self.gban.find({})]
+
+    async def get_gban_user(self, user_id: int) -> dict | None:
+        if not await self.is_gbanned(user_id):
+            return None
+        return await self.gban.find_one({"user_id": user_id})
+
+    async def is_gmuted(self, user_id: int) -> bool:
+        if await self.gmute.find_one({"user_id": user_id}):
+            return True
+        return False
+
+    async def add_gmute(self, user_id: int, reason: str) -> bool:
+        if await self.is_gmuted(user_id):
+            return False
+        await self.gmute.insert_one(
+            {"user_id": user_id, "reason": reason, "date": self.get_datetime()}
+        )
+        return True
+
+    async def rm_gmute(self, user_id: int):
+        if not await self.is_gmuted(user_id):
+            return None
+        reason = (await self.gmute.find_one({"user_id": user_id}))["reason"]
+        await self.gmute.delete_one({"user_id": user_id})
+        return reason
+
+    async def get_gmute(self) -> list:
+        return [i async for i in self.gmute.find({})]
+
+    async def add_mute(self, client: int, user_id: int, chat_id: int, reason: str):
+        await self.mute.update_one(
+            {"client": client, "user_id": user_id, "chat_id": chat_id},
+            {"$set": {"reason": reason, "date": self.get_datetime()}},
+            upsert=True,
+        )
+
+    async def rm_mute(self, client: int, user_id: int, chat_id: int) -> str:
+        reason = (await self.get_mute(client, user_id, chat_id))["reason"]
+        await self.mute.delete_one({"client": client, "user_id": user_id, "chat_id": chat_id})
+        return reason
+
+    async def is_muted(self, client: int, user_id: int, chat_id: int) -> bool:
+        if await self.get_mute(client, user_id, chat_id):
+            return True
+        return False
+
+    async def get_mute(self, client: int, user_id: int, chat_id: int):
+        data = await self.mute.find_one({"client": client, "user_id": user_id, "chat_id": chat_id})
+        return data
+
+    async def set_afk(
+        self, user_id: int, reason: str, media: int, media_type: str
+    ) -> None:
+        await self.afk.update_one(
+            {"user_id": user_id},
+            {
+                "$set": {
+                    "reason": reason,
+                    "time": time.time(),
+                    "media": media,
+                    "media_type": media_type,
+                }
+            },
+            upsert=True,
+        )
+
+    async def get_afk(self, user_id: int):
+        data = await self.afk.find_one({"user_id": user_id})
+        return data
+
+    async def is_afk(self, user_id: int) -> bool:
+        if await self.afk.find_one({"user_id": user_id}):
+            return True
+        return False
+
+    async def rm_afk(self, user_id: int) -> None:
+        await self.afk.delete_one({"user_id": user_id})
+
+    async def set_flood(self, client_chat: tuple[int, int], settings: dict):
+        await self.antiflood.update_one(
+            {"client": client_chat[0], "chat": client_chat[1]},
+            {"$set": settings},
+            upsert=True,
+        )
+
+    async def get_flood(self, client_chat: tuple[int, int]):
+        data = await self.antiflood.find_one(
+            {"client": client_chat[0], "chat": client_chat[1]}
+        )
+        return data or {}
+
+    async def is_flood(self, client_chat: tuple[int, int]) -> bool:
+        data = await self.get_flood(client_chat)
+
+        if not data:
+            return False
+
+        if data["limit"] == 0:
+            return False
+
+        return True
+
+    async def get_all_floods(self) -> list:
+        return [i async for i in self.antiflood.find({})]
+
+    async def set_autopost(self, client: int, from_channel: int, to_channel: int):
+        await self.autopost.update_one(
+            {"client": client},
+            {
+                "$push": {
+                    "autopost": {
+                        "from_channel": from_channel,
+                        "to_channel": to_channel,
+                        "date": self.get_datetime(),
+                    }
+                }
+            },
+            upsert=True,
+        )
+
+    async def get_autopost(self, client: int, from_channel: int):
+        data = await self.autopost.find_one(
+            {
+                "client": client,
+                "autopost": {"$elemMatch": {"from_channel": from_channel}},
+            }
+        )
+        return data
+
+    async def is_autopost(
+        self, client: int, from_channel: int, to_channel: int = None
+    ) -> bool:
+        if to_channel:
+            data = await self.autopost.find_one(
+                {
+                    "client": client,
+                    "autopost": {
+                        "$elemMatch": {
+                            "from_channel": from_channel,
+                            "to_channel": to_channel,
+                        }
+                    },
+                }
+            )
+        else:
+            data = await self.autopost.find_one(
+                {
+                    "client": client,
+                    "autopost": {"$elemMatch": {"from_channel": from_channel}},
+                }
+            )
+        return True if data else False
+
+    async def rm_autopost(self, client: int, from_channel: int, to_channel: int):
+        await self.autopost.update_one(
+            {"client": client},
+            {
+                "$pull": {
+                    "autopost": {
+                        "from_channel": from_channel,
+                        "to_channel": to_channel,
+                    }
+                }
+            },
+        )
+
+    async def get_all_autoposts(self, client: int) -> list:
+        return [i async for i in self.autopost.find({"client": client})]
+
+    async def add_blacklist(self, client: int, chat: int, blacklist: str):
+        await self.blacklist.update_one(
+            {"client": client, "chat": chat},
+            {"$push": {"blacklist": blacklist}},
+            upsert=True,
+        )
+
+    async def rm_blacklist(self, client: int, chat: int, blacklist: str):
+        await self.blacklist.update_one(
+            {"client": client, "chat": chat},
+            {"$pull": {"blacklist": blacklist}},
+        )
+
+    async def is_blacklist(self, client: int, chat: int, blacklist: str) -> bool:
+        blacklists = await self.get_all_blacklists(client, chat)
+        if blacklist in blacklists:
+            return True
+        return False
+
+    async def get_all_blacklists(self, client: int, chat: int) -> list:
+        data = await self.blacklist.find_one({"client": client, "chat": chat})
+
+        if not data:
+            return []
+
+        return data["blacklist"]
+
+    async def get_blacklist_clients(self) -> list:
+        return [i async for i in self.blacklist.find({})]
+
+    async def set_echo(self, client: int, chat: int, user: int):
+        await self.echo.update_one(
+            {"client": client, "chat": chat},
+            {"$push": {"echo": user}},
+            upsert=True,
+        )
+
+    async def rm_echo(self, client: int, chat: int, user: int):
+        await self.echo.update_one(
+            {"client": client, "chat": chat},
+            {"$pull": {"echo": user}},
+        )
+
+    async def is_echo(self, client: int, chat: int, user: int) -> bool:
+        data = await self.get_all_echo(client, chat)
+        if user in data:
+            return True
+        return False
+
+    async def get_all_echo(self, client: int, chat: int) -> list:
+        data = await self.echo.find_one({"client": client, "chat": chat})
+
+        if not data:
+            return []
+
+        return data["echo"]
+
+    async def set_filter(self, client: int, chat: int, keyword: str, msgid: int):
+        await self.filter.update_one(
+            {"client": client, "chat": chat},
+            {"$push": {"filter": {"keyword": keyword, "msgid": msgid}}},
+            upsert=True,
+        )
+
+    async def rm_filter(self, client: int, chat: int, keyword: str):
+        await self.filter.update_one(
+            {"client": client, "chat": chat},
+            {"$pull": {"filter": {"keyword": keyword}}},
+        )
+
+    async def rm_all_filters(self, client: int, chat: int):
+        await self.filter.delete_one({"client": client, "chat": chat})
+
+    async def is_filter(self, client: int, chat: int, keyword: str) -> bool:
+        data = await self.get_filter(client, chat, keyword)
+        return True if data else False
+
+    async def get_filter(self, client: int, chat: int, keyword: str):
+        data = await self.filter.find_one(
+            {
+                "client": client,
+                "chat": chat,
+                "filter": {"$elemMatch": {"keyword": keyword}},
+            }
+        )
+        return data
+
+    async def get_all_filters(self, client: int, chat: int) -> list:
+        data = await self.filter.find_one({"client": client, "chat": chat})
+
+        if not data:
+            return []
+
+        return data["filter"]
+
+    async def set_snip(self, client: int, chat: int, keyword: str, msgid: int):
+        await self.snips.update_one(
+            {"client": client, "chat": chat},
+            {"$push": {"snips": {"keyword": keyword, "msgid": msgid}}},
+            upsert=True,
+        )
+
+    async def rm_snip(self, client: int, chat: int, keyword: str):
+        await self.snips.update_one(
+            {"client": client, "chat": chat},
+            {"$pull": {"snips": {"keyword": keyword}}},
+        )
+
+    async def rm_all_snips(self, client: int, chat: int):
+        await self.snips.delete_one({"client": client, "chat": chat})
+
+    async def is_snip(self, client: int, chat: int, keyword: str) -> bool:
+        data = await self.get_snip(client, chat, keyword)
+        return True if data else False
+
+    async def get_snip(self, client: int, chat: int, keyword: str):
+        data = await self.snips.find_one(
+            {
+                "client": client,
+                "chat": chat,
+                "snips": {"$elemMatch": {"keyword": keyword}},
+            }
+        )
+        return data
+
+    async def get_all_snips(self, client: int, chat: int) -> list:
+        data = await self.snips.find_one({"client": client, "chat": chat})
+
+        if not data:
+            return []
+
+        return data["snips"]
+
+    async def add_pmpermit(self, client: int, user: int):
+        await self.pmpermit.update_one(
+            {"client": client, "user": user},
+            {"$set": {"date": self.get_datetime()}},
+            upsert=True,
+        )
+
+    async def rm_pmpermit(self, client: int, user: int):
+        await self.pmpermit.delete_one({"client": client, "user": user})
+
+    async def is_pmpermit(self, client: int, user: int) -> bool:
+        data = await self.get_pmpermit(client, user)
+        return True if data else False
+
+    async def get_pmpermit(self, client: int, user: int):
+        data = await self.pmpermit.find_one({"client": client, "user": user})
+        return data
+
+    async def get_all_pmpermits(self, client: int) -> list:
+        return [i async for i in self.pmpermit.find({"client": client})]
+
+    async def set_welcome(self, client: int, chat: int, message: int):
+        await self.greetings.update_one(
+            {"client": client, "chat": chat, "welcome": True},
+            {"$set": {"message": message}},
+            upsert=True,
+        )
+
+    async def rm_welcome(self, client: int, chat: int):
+        await self.greetings.delete_one(
+            {"client": client, "chat": chat, "welcome": True}
+        )
+
+    async def is_welcome(self, client: int, chat: int) -> bool:
+        data = await self.get_welcome(client, chat)
+        return True if data else False
+
+    async def get_welcome(self, client: int, chat: int):
+        data = await self.greetings.find_one(
+            {"client": client, "chat": chat, "welcome": True}
+        )
+        return data
+
+    async def set_goodbye(self, client: int, chat: int, message: int):
+        await self.greetings.update_one(
+            {"client": client, "chat": chat, "welcome": False},
+            {"$set": {"message": message}},
+            upsert=True,
+        )
+
+    async def rm_goodbye(self, client: int, chat: int):
+        await self.greetings.delete_one(
+            {"client": client, "chat": chat, "welcome": False}
+        )
+
+    async def is_goodbye(self, client: int, chat: int) -> bool:
+        data = await self.get_goodbye(client, chat)
+        return True if data else False
+
+    async def get_goodbye(self, client: int, chat: int):
+        data = await self.greetings.find_one(
+            {"client": client, "chat": chat, "welcome": False}
+        )
+        return data
+
+    async def get_all_greetings(self, client: int) -> list:
+        return [i async for i in self.greetings.find({"client": client})]
+
+    async def add_forcesub(self, chat: int, must_join: int):
+        await self.forcesub.update_one(
+            {"chat": chat},
+            {"$push": {"must_join": must_join}},
+            upsert=True,
+        )
+
+    async def rm_forcesub(self, chat: int, must_join: int) -> int:
+        await self.forcesub.update_one(
+            {"chat": chat},
+            {"$pull": {"must_join": must_join}},
+        )
+        data = await self.forcesub.find_one({"chat": chat})
+        return len(data["must_join"])
+
+    async def rm_all_forcesub(self, in_chat: int):
+        await self.forcesub.delete_one({"chat": in_chat})
+
+    async def is_forcesub(self, chat: int, must_join: int) -> bool:
+        data = await self.get_forcesub(chat)
+        if must_join in data["must_join"]:
+            return True
+        return False
+
+    async def get_forcesub(self, in_chat: int):
+        data = await self.forcesub.find_one({"chat": in_chat})
+        return data
+
+    async def get_all_forcesubs(self) -> list:
+        return [i async for i in self.forcesub.find({})]
+
+    async def add_gachabot(
+        self, client: int, bot: tuple[int, str], catch_command: str, chat_id: int
+    ):
+        await self.gachabots.update_one(
+            {"client": client, "bot": bot[0]},
+            {
+                "$set": {
+                    "username": bot[1],
+                    "catch_command": catch_command,
+                    "chat_id": chat_id,
+                    "date": self.get_datetime(),
+                }
+            },
+            upsert=True,
+        )
+
+    async def rm_gachabot(self, client: int, bot: int, chat_id: int = None):
+        if chat_id:
+            await self.gachabots.delete_one(
+                {"client": client, "bot": bot, "chat_id": chat_id}
+            )
+        else:
+            await self.gachabots.delete_one({"client": client, "bot": bot})
+
+    async def is_gachabot(self, client: int, bot: int, chat_id: int) -> bool:
+        data = await self.get_gachabot(client, bot, chat_id)
+        return True if data else False
+
+    async def get_gachabot(self, client: int, bot: int, chat_id: int):
+        data = await self.gachabots.find_one(
+            {"client": client, "bot": bot, "chat_id": chat_id}
+        )
+
+        return data
+
+    async def get_all_gachabots(self, client: int) -> list:
+        return [i async for i in self.gachabots.find({"client": client})]
+
+    async def get_all_gachabots_id(self) -> list:
+        data = await self.gachabots.distinct("bot")
+        return data
+
+db = Database(MONGO_URL)

Akeno/utils/formatter.py

@@ -0,0 +1,94 @@
+import math
+import re
+
+
+def format_text(text: str) -> str:
+    emoji_pattern = re.compile(
+        "["
+        "\U0001F600-\U0001F64F"  # emoticons
+        "\U0001F300-\U0001F5FF"  # symbols & pictographs
+        "\U0001F680-\U0001F6FF"  # transport & map symbols
+        "\U0001F700-\U0001F77F"  # alchemical symbols
+        "\U0001F780-\U0001F7FF"  # Geometric Shapes Extended
+        "\U0001F800-\U0001F8FF"  # Supplemental Arrows-C
+        "\U0001F900-\U0001F9FF"  # Supplemental Symbols and Pictographs
+        "\U0001FA00-\U0001FA6F"  # Chess Symbols
+        "\U0001FA70-\U0001FAFF"  # Symbols and Pictographs Extended-A
+        "\U00002702-\U000027B0"  # Dingbats
+        "\U000024C2-\U0001F251"  # enclosed characters
+        "]+",
+        flags=re.UNICODE,
+    )
+
+    return re.sub(emoji_pattern, "", text)
+
+
+def superscript(text: str) -> str:
+    superscript_digits = str.maketrans("0123456789", "⁰¹²³⁴⁵⁶⁷⁸⁹")
+    return text.translate(superscript_digits)
+
+
+def subscript(text: str) -> str:
+    subscript_digits = str.maketrans("0123456789", "₀₁₂₃₄₅₆₇₈₉")
+    return text.translate(subscript_digits)
+
+
+def readable_time(seconds: int) -> str:
+    count = 0
+    out_time = ""
+    time_list = []
+    time_suffix_list = ["secs", "mins", "hrs", "days"]
+
+    while count < 4:
+        count += 1
+        remainder, result = divmod(seconds, 60) if count < 3 else divmod(seconds, 24)
+        if seconds == 0 and remainder == 0:
+            break
+        time_list.append(int(result))
+        seconds = int(remainder)
+
+    for x in range(len(time_list)):
+        time_list[x] = str(time_list[x]) + time_suffix_list[x]
+
+    if len(time_list) == 4:
+        out_time += time_list.pop() + ", "
+
+    time_list.reverse()
+    out_time += " ".join(time_list)
+
+    return out_time or "0 secs"
+
+
+def humanbytes(size: int):
+    if not size:
+        return ""
+    power = 2**10
+    number = 0
+    dict_power_n = {0: " ", 1: "Ki", 2: "Mi", 3: "Gi", 4: "Ti"}
+    while size > power:
+        size /= power
+        number += 1
+    return str(round(size, 2)) + " " + dict_power_n[number] + "B"
+
+
+def add_to_dict(data: dict, keys: list, value: str | int | bool = None) -> None:
+    current_level = data
+    for key in keys[:-1]:
+        current_level = current_level.setdefault(key, {})
+    current_level[keys[-1]] = value
+
+
+def get_from_dict(data: dict, key: list):
+    current_level = data
+    for k in key:
+        current_level = current_level[k]
+    return current_level
+
+
+def limit_per_page(limit: int) -> int:
+    return math.ceil(limit / 10)
+
+
+def secs_to_mins(secs: int) -> str:
+    mins, secs = divmod(secs, 60)
+    return f"{mins}:{secs}"

Akeno/utils/handler.py

@@ -1,4 +1,9 @@
 from pyrogram import Client, filters
+from pyrogram import Client, filters
+from pyrogram.enums import ChatType
+from pyrogram.handlers import MessageHandler
+from pyrogram.types import Message
 
+group_only = [ChatType.GROUP, ChatType.SUPERGROUP]
 Akeno = Client.on_message
 Akeno_chat_member_updated = Client.on_chat_member_updated()