forust
7fb0a0e179
- S113: Add timeout=10 to all requests calls (74 fixes) - E722: Replace bare except: with except Exception: - B904: Replace redundant re-raise with bare raise - E402: Add noqa for intentional late imports after import_library() - S102/S307/S310/S311/S603/S605/S606/S607/S108: Add noqa for intentional usage - F601: Fix duplicate dict key in unsplash.py - N802: Rename ReplyCheck -> reply_check with backward compat alias - N813: Rename bs -> BS in icons.py - B007/B020: Rename loop var _j in animations.py - SIM102: Collapse nested if in autofwd.py - SIM113: Use enumerate() in calculator.py - A002: Add noqa for builtin shadowing in admlist.py - F811: Add noqa for cohere redefinition - edu_master: Fix ARG001, S108, S110, SIM117, apply --unsafe-fixes - Add modules_list.txt with full module inventory
1173 lines
49 KiB
Python
1173 lines
49 KiB
Python
# Moon-Userbot - telegram userbot
|
|
# Copyright (C) 2020-present Moon Userbot Organization
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
|
|
from pyrogram import Client
|
|
from pyrogram.enums import ChatType
|
|
from pyrogram.errors import (
|
|
ChatAdminRequired,
|
|
PeerIdInvalid,
|
|
RPCError,
|
|
UserAdminInvalid,
|
|
UsernameInvalid,
|
|
)
|
|
from pyrogram.raw import functions, types
|
|
from pyrogram.types import (
|
|
ChatPermissions,
|
|
ChatPrivileges,
|
|
InputMediaAudio,
|
|
InputMediaDocument,
|
|
InputMediaPhoto,
|
|
InputMediaVideo,
|
|
Message,
|
|
)
|
|
from pyrogram.utils import (
|
|
MAX_CHANNEL_ID,
|
|
MAX_USER_ID,
|
|
MIN_CHANNEL_ID,
|
|
MIN_CHAT_ID,
|
|
get_channel_id,
|
|
)
|
|
|
|
from utils.db import db
|
|
from utils.misc import prefix
|
|
from utils.scripts import format_exc, text
|
|
|
|
|
|
async def check_username_or_id(data: str | int) -> str:
|
|
data = str(data)
|
|
if not data.isdigit() and data[0] == '-' and not data[1:].isdigit() or not data.isdigit() and data[0] != '-':
|
|
return 'channel'
|
|
peer_id = int(data)
|
|
if peer_id < 0:
|
|
if peer_id >= MIN_CHAT_ID:
|
|
return 'chat'
|
|
|
|
if MIN_CHANNEL_ID <= peer_id < MAX_CHANNEL_ID:
|
|
return 'channel'
|
|
elif 0 < peer_id <= MAX_USER_ID:
|
|
return 'user'
|
|
|
|
raise ValueError(f'Peer id invalid: {peer_id}')
|
|
|
|
|
|
async def get_user_and_name(message):
|
|
if message.reply_to_message.from_user:
|
|
return (
|
|
message.reply_to_message.from_user.id,
|
|
message.reply_to_message.from_user.first_name,
|
|
)
|
|
if message.reply_to_message.sender_chat:
|
|
return (
|
|
message.reply_to_message.sender_chat.id,
|
|
message.reply_to_message.sender_chat.title,
|
|
)
|
|
|
|
|
|
class BanHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.channel = None
|
|
self.user_id = None
|
|
self.name = None
|
|
|
|
async def handle_ban(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_ban()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_ban()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_ban(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_ban, self.name = await get_user_and_name(self.message)
|
|
await self.ban_user(user_for_ban)
|
|
|
|
async def handle_non_reply_ban(self):
|
|
if self.message.chat.type not in ['private', 'channel'] and len(self.cause.split()) > 1:
|
|
user_to_ban = await self.get_user_to_ban()
|
|
if user_to_ban:
|
|
self.name = user_to_ban.first_name if getattr(user_to_ban, 'first_name', None) else user_to_ban.title
|
|
await self.ban_user(user_to_ban.id)
|
|
|
|
async def get_user_to_ban(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def ban_user(self, user_id):
|
|
try:
|
|
await self.client.ban_chat_member(self.message.chat.id, user_id)
|
|
self.channel = await self.client.resolve_peer(self.message.chat.id)
|
|
self.user_id = await self.client.resolve_peer(user_id)
|
|
await self.handle_additional_actions()
|
|
await self.edit_message()
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
async def handle_additional_actions(self):
|
|
if 'report_spam' in self.cause.lower().split():
|
|
await self.client.invoke(
|
|
functions.channels.ReportSpam(
|
|
channel=self.channel,
|
|
participant=self.user_id,
|
|
id=[self.message.reply_to_message.id],
|
|
)
|
|
)
|
|
if 'delete_history' in self.cause.lower().split():
|
|
await self.client.invoke(
|
|
functions.channels.DeleteParticipantHistory(channel=self.channel, participant=self.user_id)
|
|
)
|
|
|
|
async def edit_message(self):
|
|
text_c = ''.join(f' {_}' for _ in self.cause.split() if _.lower() not in ['delete_history', 'report_spam'])
|
|
await self.message.edit(
|
|
f'<b>{self.name}</b> <code>banned!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + text_c.split(maxsplit=1)[1] + "</i>" if len(text_c.split()) > 1 else ""}'
|
|
)
|
|
|
|
|
|
class UnbanHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.channel = None
|
|
self.user_id = None
|
|
self.name = None
|
|
|
|
async def handle_unban(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_unban()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_unban()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_unban(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_unban, self.name = await get_user_and_name(self.message)
|
|
await self.unban_user(user_for_unban)
|
|
|
|
async def handle_non_reply_unban(self):
|
|
if self.message.chat.type not in ['private', 'channel'] and len(self.cause.split()) > 1:
|
|
user_to_unban = await self.get_user_to_unban()
|
|
if user_to_unban:
|
|
self.name = (
|
|
user_to_unban.first_name if getattr(user_to_unban, 'first_name', None) else user_to_unban.title
|
|
)
|
|
await self.unban_user(user_to_unban.id)
|
|
|
|
async def get_user_to_unban(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def unban_user(self, user_id):
|
|
try:
|
|
await self.client.unban_chat_member(self.message.chat.id, user_id)
|
|
self.channel = await self.client.resolve_peer(self.message.chat.id)
|
|
self.user_id = await self.client.resolve_peer(user_id)
|
|
await self.edit_message()
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
async def edit_message(self):
|
|
text_c = ''.join(f' {_}' for _ in self.cause.split() if _.lower() not in ['delete_history', 'report_spam'])
|
|
await self.message.edit(
|
|
f'<b>{self.name}</b> <code>unbanned!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + text_c.split(maxsplit=1)[1] + "</i>" if len(text_c.split()) > 1 else ""}'
|
|
)
|
|
|
|
|
|
class KickHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.channel = None
|
|
self.user_id = None
|
|
self.name = None
|
|
|
|
async def handle_kick(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_kick()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_kick()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_kick(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if self.message.reply_to_message.from_user:
|
|
self.name = self.message.reply_to_message.from_user.first_name
|
|
await self.kick_user(self.message.reply_to_message.from_user.id)
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_kick(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_kick = await self.get_user_to_kick()
|
|
if user_to_kick:
|
|
self.name = (
|
|
user_to_kick.first_name if getattr(user_to_kick, 'first_name', None) else user_to_kick.title
|
|
)
|
|
await self.kick_user(user_to_kick.id)
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_kick(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def kick_user(self, user_id):
|
|
try:
|
|
await self.client.ban_chat_member(
|
|
self.message.chat.id,
|
|
user_id,
|
|
datetime.now() + timedelta(minutes=1),
|
|
)
|
|
self.channel = await self.client.resolve_peer(self.message.chat.id)
|
|
self.user_id = await self.client.resolve_peer(user_id)
|
|
await self.handle_additional_actions()
|
|
await self.client.unban_chat_member(self.message.chat.id, user_id)
|
|
await self.edit_message()
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
async def handle_additional_actions(self):
|
|
if 'report_spam' in self.cause.lower().split():
|
|
await self.client.invoke(
|
|
functions.channels.ReportSpam(
|
|
channel=self.channel,
|
|
participant=self.user_id,
|
|
id=[self.message.reply_to_message.id],
|
|
)
|
|
)
|
|
if 'delete_history' in self.cause.lower().split():
|
|
await self.client.invoke(
|
|
functions.channels.DeleteParticipantHistory(channel=self.channel, participant=self.user_id)
|
|
)
|
|
|
|
async def edit_message(self):
|
|
text_c = ''.join(f' {_}' for _ in self.cause.split() if _.lower() not in ['delete_history', 'report_spam'])
|
|
await self.message.edit(
|
|
f'<b>{self.name}</b> <code>kicked!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + text_c.split(maxsplit=1)[1] + "</i>" if len(text_c.split()) > 1 else ""}'
|
|
)
|
|
|
|
|
|
class KickDeletedAccountsHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.chat_id = message.chat.id
|
|
self.kicked_count = 0
|
|
|
|
async def kick_deleted_accounts(self):
|
|
await self.message.edit('<b>Kicking deleted accounts...</b>')
|
|
try:
|
|
async for member in self.client.get_chat_members(self.chat_id):
|
|
if member.user.is_deleted:
|
|
await self.kick_member(member.user.id)
|
|
self.kicked_count += 1
|
|
except Exception as e:
|
|
return await self.message.edit(format_exc(e))
|
|
await self.message.edit(
|
|
f'<b>Successfully kicked {self.kicked_count} deleted account(s)</b>',
|
|
)
|
|
|
|
async def kick_member(self, user_id):
|
|
try:
|
|
await self.client.ban_chat_member(self.chat_id, user_id, datetime.now() + timedelta(seconds=31))
|
|
except Exception as e:
|
|
await self.message.edit(f'Failed to kick user {user_id}: {format_exc(e)}')
|
|
|
|
|
|
class TimeMuteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.tmuted_users = db.get('core.ats', f'c{self.chat_id}', [])
|
|
|
|
async def handle_tmute(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_tmute()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_tmute()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_tmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_tmute, name = await get_user_and_name(self.message)
|
|
if user_for_tmute in self.tmuted_users:
|
|
await self.message.edit(f'<b>{name}</b> <code>already in tmute</code>')
|
|
else:
|
|
self.tmuted_users.append(user_for_tmute)
|
|
db.set('core.ats', f'c{self.chat_id}', self.tmuted_users)
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>in tmute</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(maxsplit=1)[1] + "</i>" if len(self.cause.split()) > 1 else ""}',
|
|
)
|
|
|
|
async def handle_non_reply_tmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_tmute = await self.get_user_to_tmute()
|
|
if user_to_tmute:
|
|
name = (
|
|
user_to_tmute.first_name if getattr(user_to_tmute, 'first_name', None) else user_to_tmute.title
|
|
)
|
|
if user_to_tmute.id not in self.tmuted_users:
|
|
self.tmuted_users.append(user_to_tmute.id)
|
|
db.set('core.ats', f'c{self.chat_id}', self.tmuted_users)
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>in tmute</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(maxsplit=2)[2] + "</i>" if len(self.cause.split()) > 2 else ""}',
|
|
)
|
|
else:
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>already in tmute</code>',
|
|
)
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_tmute(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
|
|
class TimeUnmuteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.tmuted_users = db.get('core.ats', f'c{self.chat_id}', [])
|
|
|
|
async def handle_tunmute(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_tunmute()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_tunmute()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_tunmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_tunmute, name = await get_user_and_name(self.message)
|
|
if user_for_tunmute not in self.tmuted_users:
|
|
await self.message.edit(f'<b>{name}</b> <code>not in tmute</code>')
|
|
else:
|
|
self.tmuted_users.remove(user_for_tunmute)
|
|
db.set('core.ats', f'c{self.chat_id}', self.tmuted_users)
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>tunmuted</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(maxsplit=1)[1] + "</i>" if len(self.cause.split()) > 1 else ""}',
|
|
)
|
|
|
|
async def handle_non_reply_tunmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_tunmute = await self.get_user_to_tunmute()
|
|
if user_to_tunmute:
|
|
name = (
|
|
user_to_tunmute.first_name
|
|
if getattr(user_to_tunmute, 'first_name', None)
|
|
else user_to_tunmute.title
|
|
)
|
|
if user_to_tunmute.id not in self.tmuted_users:
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>not in tmute</code>',
|
|
)
|
|
else:
|
|
self.tmuted_users.remove(user_to_tunmute.id)
|
|
db.set('core.ats', f'c{self.chat_id}', self.tmuted_users)
|
|
await self.message.edit(
|
|
f'<b>{name}</b> <code>tunmuted</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(maxsplit=2)[2] + "</i>" if len(self.cause.split()) > 2 else ""}',
|
|
)
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_tunmute(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
|
|
class TimeMuteUsersHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.chat_id = message.chat.id
|
|
self.tmuted_users = db.get('core.ats', f'c{self.chat_id}', [])
|
|
|
|
async def list_tmuted_users(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
text = f'<b>All users</b> <code>{self.message.chat.title}</code> <b>who are now in tmute</b>\n\n'
|
|
count = 0
|
|
for user in self.tmuted_users:
|
|
try:
|
|
name = await self.get_user_name(user)
|
|
if name:
|
|
count += 1
|
|
text += f'{count}. <b>{name}</b>\n'
|
|
except PeerIdInvalid:
|
|
pass
|
|
if count == 0:
|
|
await self.message.edit('<b>No users in tmute</b>')
|
|
else:
|
|
text += f'\n<b>Total users in tmute</b> {count}'
|
|
await self.message.edit(text)
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def get_user_name(self, user_id):
|
|
try:
|
|
_name_ = await self.client.get_chat(user_id)
|
|
if await check_username_or_id(_name_.id) == 'channel':
|
|
channel = await self.client.invoke(
|
|
functions.channels.GetChannels(
|
|
id=[
|
|
types.InputChannel(
|
|
channel_id=get_channel_id(_name_.id),
|
|
access_hash=0,
|
|
)
|
|
]
|
|
)
|
|
)
|
|
return channel.chats[0].title
|
|
if await check_username_or_id(_name_.id) == 'user':
|
|
user = await self.client.get_users(_name_.id)
|
|
return user.first_name
|
|
except PeerIdInvalid:
|
|
return None
|
|
|
|
|
|
class UnmuteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.permissions = message.chat.permissions
|
|
|
|
async def handle_unmute(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_unmute()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_unmute()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_unmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_unmute = self.message.reply_to_message.from_user
|
|
if user_for_unmute:
|
|
try:
|
|
await self.unmute_user(user_for_unmute.id)
|
|
await self.message.edit(
|
|
f'<b>{user_for_unmute.first_name}</b> <code>unmuted</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(" ", maxsplit=1)[1] + "</i>" if len(self.cause.split()) > 1 else ""}'
|
|
)
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_unmute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_unmute = await self.get_user_to_unmute()
|
|
if user_to_unmute:
|
|
try:
|
|
await self.unmute_user(user_to_unmute.id)
|
|
await self.message.edit(
|
|
f'<b>{user_to_unmute.first_name}</b> <code>unmuted!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(" ", maxsplit=2)[2] + "</i>" if len(self.cause.split()) > 2 else ""}'
|
|
)
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_unmute(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def unmute_user(self, user_id):
|
|
try:
|
|
await self.client.restrict_chat_member(
|
|
self.chat_id,
|
|
user_id,
|
|
self.permissions,
|
|
datetime.now() + timedelta(seconds=30),
|
|
)
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
|
|
class MuteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
|
|
async def handle_mute(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_mute()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_mute()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_mute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_mute = self.message.reply_to_message.from_user
|
|
if user_for_mute:
|
|
mute_seconds = self.calculate_mute_seconds()
|
|
try:
|
|
await self.mute_user(user_for_mute.id, mute_seconds)
|
|
await self.message.edit(self.construct_mute_message(user_for_mute, mute_seconds))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_mute(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_mute = await self.get_user_to_mute()
|
|
if user_to_mute:
|
|
mute_seconds = self.calculate_mute_seconds()
|
|
try:
|
|
await self.mute_user(user_to_mute.id, mute_seconds)
|
|
await self.message.edit(self.construct_mute_message(user_to_mute, mute_seconds))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_mute(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
def calculate_mute_seconds(self):
|
|
mute_seconds: int = 0
|
|
for character in 'mhdw':
|
|
match = re.search(rf'(\d+|(\d+\.\d+)){character}', self.message.text)
|
|
if match:
|
|
value = float(match.string[match.start() : match.end() - 1])
|
|
if character == 'm':
|
|
mute_seconds += int(value * 60)
|
|
if character == 'h':
|
|
mute_seconds += int(value * 3600)
|
|
if character == 'd':
|
|
mute_seconds += int(value * 86400)
|
|
if character == 'w':
|
|
mute_seconds += int(value * 604800)
|
|
return mute_seconds
|
|
|
|
async def mute_user(self, user_id, mute_seconds):
|
|
try:
|
|
if mute_seconds > 30:
|
|
await self.client.restrict_chat_member(
|
|
self.chat_id,
|
|
user_id,
|
|
ChatPermissions(),
|
|
datetime.now() + timedelta(seconds=mute_seconds),
|
|
)
|
|
else:
|
|
await self.client.restrict_chat_member(
|
|
self.chat_id,
|
|
user_id,
|
|
ChatPermissions(),
|
|
)
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
def construct_mute_message(self, user, mute_seconds):
|
|
mute_time: dict[str, int] = {
|
|
'days': mute_seconds // 86400,
|
|
'hours': mute_seconds % 86400 // 3600,
|
|
'minutes': mute_seconds % 86400 % 3600 // 60,
|
|
}
|
|
message_text = (
|
|
f'<b>{user.first_name}</b> <code> was muted for'
|
|
f' {((str(mute_time["days"]) + " day") if mute_time["days"] > 0 else "") + ("s" if mute_time["days"] > 1 else "")}'
|
|
f' {((str(mute_time["hours"]) + " hour") if mute_time["hours"] > 0 else "") + ("s" if mute_time["hours"] > 1 else "")}'
|
|
f' {((str(mute_time["minutes"]) + " minute") if mute_time["minutes"] > 0 else "") + ("s" if mute_time["minutes"] > 1 else "")}</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(" ", maxsplit=2)[2] + "</i>" if len(self.cause.split()) > 2 else ""}'
|
|
)
|
|
message_text = ' '.join(message_text.split())
|
|
return message_text
|
|
|
|
|
|
class DemoteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.common_privileges_demote = {
|
|
'is_anonymous': False,
|
|
'can_manage_chat': False,
|
|
'can_change_info': False,
|
|
'can_post_messages': False,
|
|
'can_edit_messages': False,
|
|
'can_delete_messages': False,
|
|
'can_manage_video_chats': False,
|
|
'can_restrict_members': False,
|
|
'can_invite_users': False,
|
|
'can_pin_messages': False,
|
|
'can_promote_members': False,
|
|
}
|
|
|
|
async def handle_demote(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_demote()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_demote()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_demote(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_demote = self.message.reply_to_message.from_user
|
|
if user_for_demote:
|
|
try:
|
|
await self.demote_user(user_for_demote.id)
|
|
await self.message.edit(self.construct_demote_message(user_for_demote))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_demote(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_demote = await self.get_user_to_demote()
|
|
if user_to_demote:
|
|
try:
|
|
await self.demote_user(user_to_demote.id)
|
|
await self.message.edit(self.construct_demote_message(user_to_demote))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
else:
|
|
await self.message.edit('<b>user_id or username not provided!</b>')
|
|
|
|
async def get_user_to_demote(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def demote_user(self, user_id):
|
|
try:
|
|
await self.client.promote_chat_member(
|
|
self.chat_id,
|
|
user_id,
|
|
privileges=ChatPrivileges(**self.common_privileges_demote),
|
|
)
|
|
except UserAdminInvalid:
|
|
raise # noqa: B904
|
|
except ChatAdminRequired:
|
|
raise # noqa: B904
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
def construct_demote_message(self, user):
|
|
return (
|
|
f'<b>{user.first_name}</b> <code>demoted!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(" ", maxsplit=2)[2] + "</i>" if len(self.cause.split()) > 2 else ""}'
|
|
)
|
|
|
|
|
|
class PromoteHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.common_privileges_promote = {
|
|
'can_delete_messages': True,
|
|
'can_restrict_members': True,
|
|
'can_invite_users': True,
|
|
'can_pin_messages': True,
|
|
}
|
|
|
|
async def handle_promote(self):
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_promote()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_promote()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_promote(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
user_for_promote = self.message.reply_to_message.from_user
|
|
promote_title = self.message.text.split(maxsplit=1)[1] if len(self.message.text.split()) > 1 else None
|
|
if promote_title and len(promote_title) > 16:
|
|
promote_title = promote_title[:16]
|
|
if user_for_promote:
|
|
try:
|
|
await self.promote_user(user_for_promote.id, promote_title)
|
|
await self.message.edit(self.construct_promote_message(user_for_promote))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_promote(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if len(self.cause.split()) > 1:
|
|
user_to_promote = await self.get_user_to_promote()
|
|
promote_title = ' '.join(self.cause.split(' ')[2:]) if len(self.cause.split(' ')) > 2 else None
|
|
if promote_title and len(promote_title) > 16:
|
|
promote_title = promote_title[:16]
|
|
if user_to_promote:
|
|
try:
|
|
await self.promote_user(user_to_promote.id, promote_title)
|
|
await self.message.edit(self.construct_promote_message(user_to_promote))
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
else:
|
|
await self.message.edit('<b>user_id or username not provided!</b>')
|
|
|
|
async def get_user_to_promote(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def promote_user(self, user_id, title):
|
|
try:
|
|
await self.client.promote_chat_member(
|
|
self.chat_id,
|
|
user_id,
|
|
privileges=ChatPrivileges(**self.common_privileges_promote),
|
|
title=title,
|
|
)
|
|
if len(self.cause.split()) > 1 and self.message.chat.type == 'group':
|
|
await self.client.set_administrator_title(
|
|
self.chat_id,
|
|
user_id,
|
|
self.cause.split(maxsplit=1)[1],
|
|
)
|
|
except UserAdminInvalid:
|
|
raise # noqa: B904
|
|
except ChatAdminRequired:
|
|
raise # noqa: B904
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
def construct_promote_message(self, user):
|
|
return (
|
|
f'<b>{user.first_name}</b> <code>promoted!</code>'
|
|
+ f'\n{"<b>Title:</b> <i>" + self.cause.split(" ", maxsplit=1)[1] + "</i>" if len(self.cause.split()) > 1 else ""}'
|
|
)
|
|
|
|
|
|
class AntiChannelsHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.chat_id = message.chat.id
|
|
self.prefix = prefix
|
|
|
|
async def handle_anti_channels(self):
|
|
if self.message.chat.type not in [ChatType.GROUP, ChatType.SUPERGROUP]:
|
|
await self.message.edit('<b>Not supported in non-group chats</b>')
|
|
return
|
|
|
|
command = self.message.command
|
|
if len(command) == 1:
|
|
await self.toggle_anti_channels_status()
|
|
elif command[1] in ['enable', 'on', '1', 'yes', 'true']:
|
|
await self.enable_anti_channels()
|
|
elif command[1] in ['disable', 'off', '0', 'no', 'false']:
|
|
await self.disable_anti_channels()
|
|
else:
|
|
await self.message.edit(f'<b>Usage: {self.prefix}antich [enable|disable]</b>')
|
|
|
|
async def toggle_anti_channels_status(self):
|
|
current_status = db.get('core.ats', f'antich{self.chat_id}', False)
|
|
new_status = not current_status
|
|
db.set('core.ats', f'antich{self.chat_id}', new_status)
|
|
if new_status:
|
|
await self.message.edit('<b>Blocking channels in this chat enabled.</b>')
|
|
else:
|
|
await self.message.edit('<b>Blocking channels in this chat disabled.</b>')
|
|
|
|
async def enable_anti_channels(self):
|
|
db.set('core.ats', f'antich{self.chat_id}', True)
|
|
group = await self.client.get_chat(self.chat_id)
|
|
if group.linked_chat:
|
|
db.set('core.ats', f'linked{self.chat_id}', group.linked_chat.id)
|
|
else:
|
|
db.set('core.ats', f'linked{self.chat_id}', 0)
|
|
await self.message.edit('<b>Blocking channels in this chat enabled.</b>')
|
|
|
|
async def disable_anti_channels(self):
|
|
db.set('core.ats', f'antich{self.chat_id}', False)
|
|
await self.message.edit('<b>Blocking channels in this chat disabled.</b>')
|
|
|
|
|
|
class DeleteHistoryHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.cause = text(message)
|
|
self.chat_id = message.chat.id
|
|
self.prefix = prefix
|
|
|
|
async def handle_delete_history(self):
|
|
if self.message.chat.type not in [ChatType.PRIVATE, ChatType.CHANNEL]:
|
|
if self.message.reply_to_message:
|
|
await self.handle_reply_delete_history()
|
|
elif not self.message.reply_to_message:
|
|
await self.handle_non_reply_delete_history()
|
|
else:
|
|
await self.message.edit('<b>Unsupported</b>')
|
|
|
|
async def handle_reply_delete_history(self):
|
|
if self.message.reply_to_message.from_user:
|
|
try:
|
|
user_for_delete, name = await get_user_and_name(self.message)
|
|
await self.delete_user_history(user_for_delete, name)
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
else:
|
|
await self.message.edit('<b>Reply on user msg</b>')
|
|
|
|
async def handle_non_reply_delete_history(self):
|
|
if len(self.cause.split()) > 1:
|
|
try:
|
|
user_to_delete = await self.get_user_to_delete()
|
|
if user_to_delete:
|
|
name = (
|
|
user_to_delete.first_name
|
|
if getattr(user_to_delete, 'first_name', None)
|
|
else user_to_delete.title
|
|
)
|
|
await self.delete_user_history(user_to_delete.id, name)
|
|
else:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
except PeerIdInvalid:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
except UsernameInvalid:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
except IndexError:
|
|
await self.message.edit('<b>User is not found</b>')
|
|
else:
|
|
await self.message.edit('<b>user_id or username</b>')
|
|
|
|
async def get_user_to_delete(self):
|
|
user_type = await check_username_or_id(self.cause.split(' ')[1])
|
|
if user_type == 'channel':
|
|
return await self.client.get_chat(self.cause.split(' ')[1])
|
|
if user_type == 'user':
|
|
return await self.client.get_users(self.cause.split(' ')[1])
|
|
await self.message.edit('<b>Invalid user type</b>')
|
|
return None
|
|
|
|
async def delete_user_history(self, user_id, name):
|
|
try:
|
|
channel = await self.client.resolve_peer(self.chat_id)
|
|
user_id = await self.client.resolve_peer(user_id)
|
|
await self.client.invoke(functions.channels.DeleteParticipantHistory(channel=channel, participant=user_id))
|
|
await self.message.edit(
|
|
f'<code>History from <b>{name}</b> was deleted!</code>'
|
|
+ f'\n{"<b>Cause:</b> <i>" + self.cause.split(" ", maxsplit=1)[1] + "</i>" if len(self.cause.split()) > 1 else ""}'
|
|
)
|
|
except UserAdminInvalid:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except ChatAdminRequired:
|
|
await self.message.edit('<b>No rights</b>')
|
|
except Exception as e:
|
|
await self.message.edit(format_exc(e))
|
|
|
|
|
|
class AntiRaidHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.chat_id = message.chat.id
|
|
self.prefix = prefix
|
|
|
|
async def handle_antiraid(self):
|
|
command = self.message.command
|
|
if len(command) > 1:
|
|
if command[1] == 'on':
|
|
await self.enable_antiraid()
|
|
elif command[1] == 'off':
|
|
await self.disable_antiraid()
|
|
else:
|
|
await self.toggle_antiraid()
|
|
|
|
async def enable_antiraid(self):
|
|
db.set('core.ats', f'antiraid{self.chat_id}', True)
|
|
group = await self.client.get_chat(self.chat_id)
|
|
if group.linked_chat:
|
|
db.set('core.ats', f'linked{self.chat_id}', group.linked_chat.id)
|
|
else:
|
|
db.set('core.ats', f'linked{self.chat_id}', 0)
|
|
await self.message.edit(f'<b>Anti-raid mode enabled!\nDisable with: </b><code>{self.prefix}antiraid off</code>')
|
|
|
|
async def disable_antiraid(self):
|
|
db.set('core.ats', f'antiraid{self.chat_id}', False)
|
|
await self.message.edit('<b>Anti-raid mode disabled</b>')
|
|
|
|
async def toggle_antiraid(self):
|
|
current_status = db.get('core.ats', f'antiraid{self.chat_id}', False)
|
|
new_status = not current_status
|
|
db.set('core.ats', f'antiraid{self.chat_id}', new_status)
|
|
if new_status:
|
|
group = await self.client.get_chat(self.chat_id)
|
|
if group.linked_chat:
|
|
db.set('core.ats', f'linked{self.chat_id}', group.linked_chat.id)
|
|
else:
|
|
db.set('core.ats', f'linked{self.chat_id}', 0)
|
|
await self.message.edit(
|
|
f'<b>Anti-raid mode enabled!\nDisable with: </b><code>{self.prefix}antiraid off</code>'
|
|
)
|
|
else:
|
|
await self.message.edit('<b>Anti-raid mode disabled</b>')
|
|
|
|
|
|
class NoteSendHandler:
|
|
def __init__(self, client: Client, message: Message):
|
|
self.client = client
|
|
self.message = message
|
|
self.chat_id = message.chat.id
|
|
self.prefix = prefix
|
|
|
|
async def handle_note_send(self):
|
|
if len(self.message.text.split()) >= 2:
|
|
await self.message.edit('<b>Loading...</b>')
|
|
|
|
note_name = self.message.text.split(maxsplit=1)[1]
|
|
find_note = db.get('core.notes', f'note{note_name}', False)
|
|
if find_note:
|
|
try:
|
|
await self.send_note(find_note)
|
|
except RPCError:
|
|
await self.message.edit(
|
|
'<b>Sorry, but this note is unavailable.\n\n'
|
|
f'You can delete this note with '
|
|
f'<code>{self.prefix}clear {note_name}</code></b>'
|
|
)
|
|
else:
|
|
await self.message.edit('<b>There is no such note</b>')
|
|
else:
|
|
await self.message.edit(
|
|
f'<b>Example: <code>{self.prefix}note note_name</code></b>',
|
|
)
|
|
|
|
async def send_note(self, find_note):
|
|
if find_note.get('MEDIA_GROUP'):
|
|
await self.message.delete()
|
|
await self.send_media_group(find_note)
|
|
else:
|
|
await self.message.delete()
|
|
await self.copy_message(find_note)
|
|
|
|
async def send_media_group(self, find_note):
|
|
messages_grouped = await self.client.get_media_group(int(find_note['CHAT_ID']), int(find_note['MESSAGE_ID']))
|
|
media_grouped_list = self.prepare_media_group(messages_grouped)
|
|
if self.message.reply_to_message:
|
|
await self.client.send_media_group(
|
|
self.message.chat.id,
|
|
media_grouped_list,
|
|
reply_to_message_id=self.message.reply_to_message.id,
|
|
)
|
|
else:
|
|
await self.client.send_media_group(self.message.chat.id, media_grouped_list)
|
|
|
|
async def copy_message(self, find_note):
|
|
if self.message.reply_to_message:
|
|
await self.client.copy_message(
|
|
self.message.chat.id,
|
|
int(find_note['CHAT_ID']),
|
|
int(find_note['MESSAGE_ID']),
|
|
reply_to_message_id=self.message.reply_to_message.id,
|
|
)
|
|
else:
|
|
await self.client.copy_message(
|
|
self.message.chat.id,
|
|
int(find_note['CHAT_ID']),
|
|
int(find_note['MESSAGE_ID']),
|
|
)
|
|
|
|
def prepare_media_group(self, messages_grouped):
|
|
media_grouped_list = []
|
|
for _ in messages_grouped:
|
|
if _.photo:
|
|
media_grouped_list.append(self.prepare_photo(_))
|
|
elif _.video:
|
|
media_grouped_list.append(self.prepare_video(_))
|
|
elif _.audio:
|
|
media_grouped_list.append(self.prepare_audio(_))
|
|
elif _.document:
|
|
media_grouped_list.append(self.prepare_document(_))
|
|
return media_grouped_list
|
|
|
|
@staticmethod
|
|
def prepare_photo(message):
|
|
if message.caption:
|
|
return InputMediaPhoto(message.photo.file_id, message.caption.markdown)
|
|
return InputMediaPhoto(message.photo.file_id)
|
|
|
|
@staticmethod
|
|
def prepare_video(message):
|
|
if message.caption:
|
|
if message.video.thumbs:
|
|
return InputMediaVideo(
|
|
message.video.file_id,
|
|
message.video.thumbs[0].file_id,
|
|
message.caption.markdown,
|
|
)
|
|
return InputMediaVideo(message.video.file_id, message.caption.markdown)
|
|
if message.video.thumbs:
|
|
return InputMediaVideo(message.video.file_id, message.video.thumbs[0].file_id)
|
|
return InputMediaVideo(message.video.file_id)
|
|
|
|
@staticmethod
|
|
def prepare_audio(message):
|
|
if message.caption:
|
|
return InputMediaAudio(message.audio.file_id, message.caption.markdown)
|
|
return InputMediaAudio(message.audio.file_id)
|
|
|
|
@staticmethod
|
|
def prepare_document(message):
|
|
if message.caption:
|
|
if message.document.thumbs:
|
|
return InputMediaDocument(
|
|
message.document.file_id,
|
|
message.document.thumbs[0].file_id,
|
|
message.caption.markdown,
|
|
)
|
|
return InputMediaDocument(message.document.file_id, message.caption.markdown)
|
|
if message.document.thumbs:
|
|
return InputMediaDocument(message.document.file_id, message.document.thumbs[0].file_id)
|
|
return InputMediaDocument(message.document.file_id)
|