File size: 4,899 Bytes
618430a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
# 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
|