Upload 4 files

pyUltroid/_misc/_assistant.py

@@ -0,0 +1,150 @@
+# Ultroid - UserBot
+# Copyright (C) 2021-2025 TeamUltroid
+#
+# This file is a part of < https://github.com/TeamUltroid/Ultroid/ >
+# PLease read the GNU Affero General Public License in
+# <https://github.com/TeamUltroid/pyUltroid/blob/main/LICENSE>.
+
+import inspect
+import re
+from traceback import format_exc
+
+from telethon import Button
+from telethon.errors import QueryIdInvalidError
+from telethon.events import CallbackQuery, InlineQuery, NewMessage
+from telethon.tl.types import InputWebDocument
+
+from .. import LOGS, asst, udB, ultroid_bot
+from ..fns.admins import admin_check
+from . import append_or_update, owner_and_sudos
+
+OWNER = ultroid_bot.full_name
+
+MSG = f"""
+**Ultroid - UserBot**
+➖➖➖➖➖➖➖➖➖➖
+**Owner**: [{OWNER}](tg://user?id={ultroid_bot.uid})
+**Support**: @TeamUltroid
+➖➖➖➖➖➖➖➖➖➖
+"""
+
+IN_BTTS = [
+    [
+        Button.url(
+            "Repository",
+            url="https://github.com/TeamUltroid/Ultroid",
+        ),
+        Button.url("Support", url="https://t.me/UltroidSupportChat"),
+    ]
+]
+
+
+# decorator for assistant
+
+
+def asst_cmd(pattern=None, load=None, owner=False, **kwargs):
+    """Decorator for assistant's command"""
+    name = inspect.stack()[1].filename.split("/")[-1].replace(".py", "")
+    kwargs["forwards"] = False
+
+    def ult(func):
+        if pattern:
+            kwargs["pattern"] = re.compile(f"^/{pattern}")
+
+        async def handler(event):
+            if owner and event.sender_id not in owner_and_sudos():
+                return
+            try:
+                await func(event)
+            except Exception as er:
+                LOGS.exception(er)
+
+        asst.add_event_handler(handler, NewMessage(**kwargs))
+        if load is not None:
+            append_or_update(load, func, name, kwargs)
+
+    return ult
+
+
+def callback(data=None, from_users=[], admins=False, owner=False, **kwargs):
+    """Assistant's callback decorator"""
+    if "me" in from_users:
+        from_users.remove("me")
+        from_users.append(ultroid_bot.uid)
+
+    def ultr(func):
+        async def wrapper(event):
+            if admins and not await admin_check(event):
+                return
+            if from_users and event.sender_id not in from_users:
+                return await event.answer("Not for You!", alert=True)
+            if owner and event.sender_id not in owner_and_sudos():
+                return await event.answer(f"This is {OWNER}'s bot!!")
+            try:
+                await func(event)
+            except Exception as er:
+                LOGS.exception(er)
+
+        asst.add_event_handler(wrapper, CallbackQuery(data=data, **kwargs))
+
+    return ultr
+
+
+def in_pattern(pattern=None, owner=False, **kwargs):
+    """Assistant's inline decorator."""
+
+    def don(func):
+        async def wrapper(event):
+            if owner and event.sender_id not in owner_and_sudos():
+                res = [
+                    await event.builder.article(
+                        title="Ultroid Userbot",
+                        url="https://t.me/TeamUltroid",
+                        description="(c) TeamUltroid",
+                        text=MSG,
+                        thumb=InputWebDocument(
+                            "https://graph.org/file/dde85d441fa051a0d7d1d.jpg",
+                            0,
+                            "image/jpeg",
+                            [],
+                        ),
+                        buttons=IN_BTTS,
+                    )
+                ]
+                return await event.answer(
+                    res,
+                    switch_pm=f"🤖: Assistant of {OWNER}",
+                    switch_pm_param="start",
+                )
+            try:
+                await func(event)
+            except QueryIdInvalidError:
+                pass
+            except Exception as er:
+                err = format_exc()
+
+                def error_text():
+                    return f"**#ERROR #INLINE**\n\nQuery: `{asst.me.username} {pattern}`\n\n**Traceback:**\n`{format_exc()}`"
+
+                LOGS.exception(er)
+                try:
+                    await event.answer(
+                        [
+                            await event.builder.article(
+                                title="Unhandled Exception has Occured!",
+                                text=error_text(),
+                                buttons=Button.url(
+                                    "Report", "https://t.me/UltroidSupportChat"
+                                ),
+                            )
+                        ]
+                    )
+                except QueryIdInvalidError:
+                    LOGS.exception(err)
+                except Exception as er:
+                    LOGS.exception(er)
+                    await asst.send_message(udB.get_key("LOG_CHANNEL"), error_text())
+
+        asst.add_event_handler(wrapper, InlineQuery(pattern=pattern, **kwargs))
+
+    return don

pyUltroid/_misc/_decorators.py

@@ -0,0 +1,340 @@
+# Ultroid - UserBot
+# Copyright (C) 2021-2025 TeamUltroid
+#
+# This file is a part of < https://github.com/TeamUltroid/Ultroid/ >
+# PLease read the GNU Affero General Public License in
+# <https://github.com/TeamUltroid/pyUltroid/blob/main/LICENSE>.
+
+import asyncio
+import inspect
+import re
+import sys
+from io import BytesIO
+from pathlib import Path
+from time import gmtime, strftime
+from traceback import format_exc
+
+from telethon import Button
+from telethon import __version__ as telever
+from telethon import events
+from telethon.errors.common import AlreadyInConversationError
+from telethon.errors.rpcerrorlist import (
+    AuthKeyDuplicatedError,
+    BotInlineDisabledError,
+    BotMethodInvalidError,
+    ChatSendInlineForbiddenError,
+    ChatSendMediaForbiddenError,
+    ChatSendStickersForbiddenError,
+    FloodWaitError,
+    MessageDeleteForbiddenError,
+    MessageIdInvalidError,
+    MessageNotModifiedError,
+    UserIsBotError,
+)
+from telethon.events import MessageEdited, NewMessage
+from telethon.utils import get_display_name
+
+from pyUltroid.exceptions import DependencyMissingError
+from strings import get_string
+
+from .. import *
+from .. import _ignore_eval
+from ..dB import DEVLIST
+from ..dB._core import LIST, LOADED
+from ..fns.admins import admin_check
+from ..fns.helper import bash
+from ..fns.helper import time_formatter as tf
+from ..version import __version__ as pyver
+from ..version import ultroid_version as ult_ver
+from . import SUDO_M, owner_and_sudos
+from ._wrappers import eod
+
+MANAGER = udB.get_key("MANAGER")
+TAKE_EDITS = udB.get_key("TAKE_EDITS")
+black_list_chats = udB.get_key("BLACKLIST_CHATS")
+allow_sudo = SUDO_M.should_allow_sudo
+
+
+def compile_pattern(data, hndlr):
+    if data.startswith("^"):
+        data = data[1:]
+    if data.startswith("."):
+        data = data[1:]
+    if hndlr in [" ", "NO_HNDLR"]:
+        # No Hndlr Feature
+        return re.compile("^" + data)
+    return re.compile("\\" + hndlr + data)
+
+
+def ultroid_cmd(
+    pattern=None, manager=False, ultroid_bot=ultroid_bot, asst=asst, **kwargs
+):
+    owner_only = kwargs.get("owner_only", False)
+    groups_only = kwargs.get("groups_only", False)
+    admins_only = kwargs.get("admins_only", False)
+    fullsudo = kwargs.get("fullsudo", False)
+    only_devs = kwargs.get("only_devs", False)
+    func = kwargs.get("func", lambda e: not e.via_bot_id)
+
+    def decor(dec):
+        async def wrapp(ult):
+            if udB.get_key("COMMAND_LOGGER"):
+                user_id = ult.sender_id
+                chat_id = ult.chat_id
+                command_name = pattern if pattern else ult.text.split()[0].lstrip(HNDLR)
+                chat_name = get_display_name(ult.chat)
+                LOGS.info(f"Command '{command_name}' executed by user ID {user_id} in chat {chat_id} ({chat_name})")
+                log_channel = udB.get_key("LOG_CHANNEL")
+                if log_channel:
+                    try:
+                        await asst.send_message(
+                            log_channel,
+                            f"Command '{command_name}' executed by user ID {user_id} in chat {chat_id} ({chat_name})"
+                        )
+                    except Exception as e:
+                        LOGS.warning(f"Failed to send command log to log channel {log_channel}: {e}")
+            if not ult.out:
+                if owner_only:
+                    return
+                if ult.sender_id not in owner_and_sudos():
+                    return
+                if ult.sender_id in _ignore_eval:
+                    return await eod(
+                        ult,
+                        get_string("py_d1"),
+                    )
+                if fullsudo and ult.sender_id not in SUDO_M.fullsudos:
+                    return await eod(ult, get_string("py_d2"), time=15)
+            chat = ult.chat
+            if hasattr(chat, "title"):
+                if (
+                    "#noub" in chat.title.lower()
+                    and not (chat.admin_rights or chat.creator)
+                    and not (ult.sender_id in DEVLIST)
+                ):
+                    return
+            if ult.is_private and (groups_only or admins_only):
+                return await eod(ult, get_string("py_d3"))
+            elif admins_only and not (chat.admin_rights or chat.creator):
+                return await eod(ult, get_string("py_d5"))
+            if only_devs and not udB.get_key("I_DEV"):
+                return await eod(
+                    ult,
+                    get_string("py_d4").format(HNDLR),
+                    time=10,
+                )
+            try:
+                await dec(ult)
+            except FloodWaitError as fwerr:
+                await asst.send_message(
+                    udB.get_key("LOG_CHANNEL"),
+                    f"`FloodWaitError:\n{str(fwerr)}\n\nSleeping for {tf((fwerr.seconds + 10)*1000)}`",
+                )
+                await ultroid_bot.disconnect()
+                await asyncio.sleep(fwerr.seconds + 10)
+                await ultroid_bot.connect()
+                await asst.send_message(
+                    udB.get_key("LOG_CHANNEL"),
+                    "`Bot is working again`",
+                )
+                return
+            except ChatSendInlineForbiddenError:
+                return await eod(ult, "`Inline Locked In This Chat.`")
+            except (ChatSendMediaForbiddenError, ChatSendStickersForbiddenError):
+                return await eod(ult, get_string("py_d8"))
+            except (BotMethodInvalidError, UserIsBotError):
+                return await eod(ult, get_string("py_d6"))
+            except AlreadyInConversationError:
+                return await eod(
+                    ult,
+                    get_string("py_d7"),
+                )
+            except (BotInlineDisabledError, DependencyMissingError) as er:
+                return await eod(ult, f"`{er}`")
+            except (
+                MessageIdInvalidError,
+                MessageNotModifiedError,
+                MessageDeleteForbiddenError,
+            ) as er:
+                LOGS.exception(er)
+            except AuthKeyDuplicatedError as er:
+                LOGS.exception(er)
+                await asst.send_message(
+                    udB.get_key("LOG_CHANNEL"),
+                    "Session String expired, create new session from 👇",
+                    buttons=[
+                        Button.url("Bot", "t.me/SessionGeneratorBot?start="),
+                        Button.url(
+                            "Repl",
+                            "https://replit.com/@TheUltroid/UltroidStringSession",
+                        ),
+                    ],
+                )
+                sys.exit()
+            except events.StopPropagation:
+                raise events.StopPropagation
+            except KeyboardInterrupt:
+                pass
+            except Exception as e:
+                LOGS.exception(e)
+                date = strftime("%Y-%m-%d %H:%M:%S", gmtime())
+                naam = get_display_name(chat)
+                ftext = "**Ultroid Client Error:** `Forward this to` @UltroidSupportChat\n\n"
+                ftext += "**Py-Ultroid Version:** `" + str(pyver)
+                ftext += "`\n**Ultroid Version:** `" + str(ult_ver)
+                ftext += "`\n**Telethon Version:** `" + str(telever)
+                ftext += f"`\n**Hosted At:** `{HOSTED_ON}`\n\n"
+                ftext += "--------START ULTROID CRASH LOG--------"
+                ftext += "\n**Date:** `" + date
+                ftext += "`\n**Group:** `" + str(ult.chat_id) + "` " + str(naam)
+                ftext += "\n**Sender ID:** `" + str(ult.sender_id)
+                ftext += "`\n**Replied:** `" + str(ult.is_reply)
+                ftext += "`\n\n**Event Trigger:**`\n"
+                ftext += str(ult.text)
+                ftext += "`\n\n**Traceback info:**`\n"
+                ftext += str(format_exc())
+                ftext += "`\n\n**Error text:**`\n"
+                ftext += str(sys.exc_info()[1])
+                ftext += "`\n\n--------END ULTROID CRASH LOG--------"
+                ftext += "\n\n\n**Last 5 commits:**`\n"
+
+                stdout, stderr = await bash('git log --pretty=format:"%an: %s" -5')
+                result = stdout + (stderr or "")
+
+                ftext += f"{result}`"
+
+                if len(ftext) > 4096:
+                    with BytesIO(ftext.encode()) as file:
+                        file.name = "logs.txt"
+                        error_log = await asst.send_file(
+                            udB.get_key("LOG_CHANNEL"),
+                            file,
+                            caption="**Ultroid Client Error:** `Forward this to` @UltroidSupportChat\n\n",
+                        )
+                else:
+                    error_log = await asst.send_message(
+                        udB.get_key("LOG_CHANNEL"),
+                        ftext,
+                    )
+                if ult.out:
+                    await ult.edit(
+                        f"<b><a href={error_log.message_link}>[An error occurred]</a></b>",
+                        link_preview=False,
+                        parse_mode="html",
+                    )
+
+        cmd = None
+        blacklist_chats = False
+        chats = None
+        if black_list_chats:
+            blacklist_chats = True
+            chats = list(black_list_chats)
+        _add_new = allow_sudo and HNDLR != SUDO_HNDLR
+        if _add_new:
+            if pattern:
+                cmd = compile_pattern(pattern, SUDO_HNDLR)
+            ultroid_bot.add_event_handler(
+                wrapp,
+                NewMessage(
+                    pattern=cmd,
+                    incoming=True,
+                    forwards=False,
+                    func=func,
+                    chats=chats,
+                    blacklist_chats=blacklist_chats,
+                ),
+            )
+        if pattern:
+            cmd = compile_pattern(pattern, HNDLR)
+        ultroid_bot.add_event_handler(
+            wrapp,
+            NewMessage(
+                outgoing=True if _add_new else None,
+                pattern=cmd,
+                forwards=False,
+                func=func,
+                chats=chats,
+                blacklist_chats=blacklist_chats,
+            ),
+        )
+        if TAKE_EDITS:
+
+            def func_(x):
+                return not x.via_bot_id and not (x.is_channel and x.chat.broadcast)
+
+            ultroid_bot.add_event_handler(
+                wrapp,
+                MessageEdited(
+                    pattern=cmd,
+                    forwards=False,
+                    func=func_,
+                    chats=chats,
+                    blacklist_chats=blacklist_chats,
+                ),
+            )
+        if manager and MANAGER:
+            allow_all = kwargs.get("allow_all", False)
+            allow_pm = kwargs.get("allow_pm", False)
+            require = kwargs.get("require", None)
+
+            async def manager_cmd(ult):
+                if not allow_all and not (await admin_check(ult, require=require)):
+                    return
+                if not allow_pm and ult.is_private:
+                    return
+                try:
+                    await dec(ult)
+                except Exception as er:
+                    if chat := udB.get_key("MANAGER_LOG"):
+                        text = f"**#MANAGER_LOG\n\nChat:** `{get_display_name(ult.chat)}` `{ult.chat_id}`"
+                        text += f"\n**Replied :** `{ult.is_reply}`\n**Command :** {ult.text}\n\n**Error :** `{er}`"
+                        try:
+                            return await asst.send_message(
+                                chat, text, link_preview=False
+                            )
+                        except Exception as er:
+                            LOGS.exception(er)
+                    LOGS.info(f"• MANAGER [{ult.chat_id}]:")
+                    LOGS.exception(er)
+
+            if pattern:
+                cmd = compile_pattern(pattern, "/")
+            asst.add_event_handler(
+                manager_cmd,
+                NewMessage(
+                    pattern=cmd,
+                    forwards=False,
+                    incoming=True,
+                    func=func,
+                    chats=chats,
+                    blacklist_chats=blacklist_chats,
+                ),
+            )
+        if DUAL_MODE and not (manager and DUAL_HNDLR == "/"):
+            if pattern:
+                cmd = compile_pattern(pattern, DUAL_HNDLR)
+            asst.add_event_handler(
+                wrapp,
+                NewMessage(
+                    pattern=cmd,
+                    incoming=True,
+                    forwards=False,
+                    func=func,
+                    chats=chats,
+                    blacklist_chats=blacklist_chats,
+                ),
+            )
+        file = Path(inspect.stack()[1].filename)
+        if "addons/" in str(file):
+            if LOADED.get(file.stem):
+                LOADED[file.stem].append(wrapp)
+            else:
+                LOADED.update({file.stem: [wrapp]})
+        if pattern:
+            if LIST.get(file.stem):
+                LIST[file.stem].append(pattern)
+            else:
+                LIST.update({file.stem: [pattern]})
+        return wrapp
+
+    return decor

pyUltroid/_misc/_supporter.py

@@ -0,0 +1,126 @@
+# Ultroid - UserBot
+# Copyright (C) 2021-2025 TeamUltroid
+#
+# This file is a part of < https://github.com/TeamUltroid/Ultroid/ >
+# PLease read the GNU Affero General Public License in
+# <https://github.com/TeamUltroid/pyUltroid/blob/main/LICENSE>.
+#
+#   To Install Other USERBOTs plugin Support
+#
+#   ULTROID Don't Need This Stuffs
+#
+
+import inspect
+import os
+from pathlib import Path
+
+from telethon import events, types
+
+from pyUltroid._misc._decorators import compile_pattern, ultroid_cmd
+from pyUltroid._misc._wrappers import eod, eor
+
+from .. import *
+from ..dB._core import LIST
+from . import CMD_HELP, SUDO_M  # ignore: pylint
+
+ALIVE_NAME = ultroid_bot.me.first_name
+BOTLOG_CHATID = BOTLOG = udB.get_key("LOG_CHANNEL")
+
+
+bot = borg = catub = friday = ultroid_bot
+catub.cat_cmd = ultroid_cmd
+
+black_list_chats = udB.get_key("BLACKLIST_CHATS")
+
+
+def admin_cmd(pattern=None, command=None, **args):
+    args["func"] = lambda e: not e.via_bot_id
+    args["chats"] = black_list_chats
+    args["blacklist_chats"] = True
+    args["outgoing"] = True
+    args["forwards"] = False
+    if pattern:
+        args["pattern"] = compile_pattern(pattern, HNDLR)
+        file = Path(inspect.stack()[1].filename)
+        if LIST.get(file.stem):
+            LIST[file.stem].append(pattern)
+        else:
+            LIST.update({file.stem: [pattern]})
+    return events.NewMessage(**args)
+
+
+friday_on_cmd = admin_cmd
+command = ultroid_cmd
+register = ultroid_cmd
+
+
+def sudo_cmd(allow_sudo=True, pattern=None, command=None, **args):
+    args["func"] = lambda e: not e.via_bot_id
+    args["chats"] = black_list_chats
+    args["blacklist_chats"] = True
+    args["forwards"] = False
+    if pattern:
+        args["pattern"] = compile_pattern(pattern, SUDO_HNDLR)
+    if allow_sudo:
+        args["from_users"] = SUDO_M.get_sudos
+        args["incoming"] = True
+    return events.NewMessage(**args)
+
+
+edit_or_reply = eor
+edit_delete = eod
+
+
+ENV = bool(os.environ.get("ENV", False))
+
+
+class Config((object)):
+    if ENV:
+        from .. import asst, udB
+
+        LOGGER = True
+        LOCATION = os.environ.get("LOCATION", None)
+        OPEN_WEATHER_MAP_APPID = os.environ.get("OPEN_WEATHER_MAP_APPID", None)
+        SUDO_COMMAND_HAND_LER = SUDO_HNDLR
+        TMP_DOWNLOAD_DIRECTORY = os.environ.get(
+            "TMP_DOWNLOAD_DIRECTORY", "resources/downloads/"
+        )
+        TEMP_DOWNLOAD_DIRECTORY = TMP_DOWNLOAD_DIRECTORY
+        TEMP_DIR = TMP_DOWNLOAD_DIRECTORY
+        TELEGRAPH_SHORT_NAME = os.environ.get("TELEGRAPH_SHORT_NAME", "Ultroid")
+        OCR_SPACE_API_KEY = os.environ.get("OCR_SPACE_API_KEY", None)
+        TG_BOT_USER_NAME_BF_HER = asst.me.username
+        UB_BLACK_LIST_CHAT = [
+            int(blacklist) for blacklist in udB.get_key("BLACKLIST_CHATS")
+        ]
+        MAX_ANTI_FLOOD_MESSAGES = 10
+        ANTI_FLOOD_WARN_MODE = types.ChatBannedRights(
+            until_date=None, view_messages=None, send_messages=True
+        )
+        REM_BG_API_KEY = os.environ.get("REM_BG_API_KEY", None)
+        GITHUB_ACCESS_TOKEN = os.environ.get("GITHUB_ACCESS_TOKEN", None)
+        GIT_REPO_NAME = os.environ.get("GIT_REPO_NAME", None)
+        PRIVATE_GROUP_BOT_API_ID = BOTLOG
+        PM_LOGGR_BOT_API_ID = BOTLOG
+        DB_URI = os.environ.get("DATABASE_URL", None)
+        HANDLR = HNDLR
+        SUDO_USERS = SUDO_M.get_sudos()
+        CHANNEL_ID = int(os.environ.get("CHANNEL_ID", -100))
+        BLACKLIST_CHAT = UB_BLACK_LIST_CHAT
+        MONGO_URI = os.environ.get("MONGO_URI", None)
+        ALIVE_PHOTTO = os.environ.get("ALIVE_PHOTTO", None)
+        ALIVE_PIC = os.environ.get("ALIVE_PIC", None)
+        ALIVE_MSG = os.environ.get("ALIVE_MSG", None)
+        DEFAULT_BIO = os.environ.get("DEFAULT_BIO", None)
+        BIO_MSG = os.environ.get("BIO_MSG", None)
+        LYDIA_API = os.environ.get("LYDIA_API", None)
+        PLUGIN_CHANNEL = int(os.environ.get("PLUGIN_CHANNEL", -69))
+        PM_DATA = os.environ.get("PM_DATA", "ENABLE")
+        DEEP_AI = os.environ.get("DEEP_AI", None)
+        TAG_LOG = os.environ.get("TAG_LOG", None)
+
+    else:
+        DB_URI = None
+
+
+CMD_HNDLR = HNDLR

pyUltroid/_misc/_wrappers.py

@@ -0,0 +1,65 @@
+# Ultroid - UserBot
+# Copyright (C) 2021-2025 TeamUltroid
+#
+# This file is a part of < https://github.com/TeamUltroid/Ultroid/ >
+# PLease read the GNU Affero General Public License in
+# <https://github.com/TeamUltroid/pyUltroid/blob/main/LICENSE>.
+
+from asyncio import sleep
+
+from telethon.errors import MessageDeleteForbiddenError, MessageNotModifiedError
+from telethon.tl.custom import Message
+from telethon.tl.types import MessageService
+
+# edit or reply
+
+
+async def eor(event, text=None, time=None, link_preview=False, edit_time=None, **args):
+    reply_to = event.reply_to_msg_id or event
+    if event.out and not isinstance(event, MessageService):
+        if edit_time:
+            await sleep(edit_time)
+        if args.get("file") and not event.media:
+            await event.delete()
+            ok = await event.client.send_message(
+                event.chat_id,
+                text,
+                link_preview=link_preview,
+                reply_to=reply_to,
+                **args
+            )
+        else:
+            try:
+                ok = await event.edit(text, link_preview=link_preview, **args)
+            except MessageNotModifiedError:
+                ok = event
+    else:
+        ok = await event.client.send_message(
+            event.chat_id, text, link_preview=link_preview, reply_to=reply_to, **args
+        )
+
+    if time:
+        await sleep(time)
+        return await ok.delete()
+    return ok
+
+
+async def eod(event, text=None, **kwargs):
+    kwargs["time"] = kwargs.get("time", 8)
+    return await eor(event, text, **kwargs)
+
+
+async def _try_delete(event):
+    try:
+        return await event.delete()
+    except (MessageDeleteForbiddenError):
+        pass
+    except BaseException as er:
+        from . import LOGS
+
+        LOGS.error("Error while Deleting Message..")
+        LOGS.exception(er)
+
+
+setattr(Message, "eor", eor)
+setattr(Message, "try_delete", _try_delete)