# 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 .
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('Unsupported')
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('Invalid user type')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
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'{self.name} banned!'
+ f'\n{"Cause: " + text_c.split(maxsplit=1)[1] + "" 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('Unsupported')
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('Invalid user type')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
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'{self.name} unbanned!'
+ f'\n{"Cause: " + text_c.split(maxsplit=1)[1] + "" 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('Unsupported')
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('Reply on user msg')
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('user_id or username')
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('Invalid user type')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
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'{self.name} kicked!'
+ f'\n{"Cause: " + text_c.split(maxsplit=1)[1] + "" 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('Kicking deleted accounts...')
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'Successfully kicked {self.kicked_count} deleted account(s)',
)
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('Unsupported')
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'{name} already in tmute')
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'{name} in tmute'
+ f'\n{"Cause: " + self.cause.split(maxsplit=1)[1] + "" 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'{name} in tmute'
+ f'\n{"Cause: " + self.cause.split(maxsplit=2)[2] + "" if len(self.cause.split()) > 2 else ""}',
)
else:
await self.message.edit(
f'{name} already in tmute',
)
else:
await self.message.edit('user_id or username')
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('Invalid user type')
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('Unsupported')
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'{name} not in tmute')
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'{name} tunmuted'
+ f'\n{"Cause: " + self.cause.split(maxsplit=1)[1] + "" 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'{name} not in tmute',
)
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'{name} tunmuted'
+ f'\n{"Cause: " + self.cause.split(maxsplit=2)[2] + "" if len(self.cause.split()) > 2 else ""}',
)
else:
await self.message.edit('user_id or username')
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('Invalid user type')
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'All users {self.message.chat.title} who are now in tmute\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}. {name}\n'
except PeerIdInvalid:
pass
if count == 0:
await self.message.edit('No users in tmute')
else:
text += f'\nTotal users in tmute {count}'
await self.message.edit(text)
else:
await self.message.edit('Unsupported')
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('Unsupported')
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'{user_for_unmute.first_name} unmuted'
+ f'\n{"Cause: " + self.cause.split(" ", maxsplit=1)[1] + "" if len(self.cause.split()) > 1 else ""}'
)
except UserAdminInvalid:
await self.message.edit('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('Reply on user msg')
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'{user_to_unmute.first_name} unmuted!'
+ f'\n{"Cause: " + self.cause.split(" ", maxsplit=2)[2] + "" if len(self.cause.split()) > 2 else ""}'
)
except UserAdminInvalid:
await self.message.edit('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('User is not found')
else:
await self.message.edit('user_id or username')
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('Invalid user type')
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('Unsupported')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('Reply on user msg')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('User is not found')
else:
await self.message.edit('user_id or username')
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('Invalid user type')
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'{user.first_name} 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 "")}'
+ f'\n{"Cause: " + self.cause.split(" ", maxsplit=2)[2] + "" 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('Unsupported')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('Reply on user msg')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('User is not found')
else:
await self.message.edit('user_id or username not provided!')
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('Invalid user type')
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'{user.first_name} demoted!'
+ f'\n{"Cause: " + self.cause.split(" ", maxsplit=2)[2] + "" 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('Unsupported')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('Reply on user msg')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('User is not found')
else:
await self.message.edit('user_id or username not provided!')
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('Invalid user type')
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'{user.first_name} promoted!'
+ f'\n{"Title: " + self.cause.split(" ", maxsplit=1)[1] + "" 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('Not supported in non-group chats')
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'Usage: {self.prefix}antich [enable|disable]')
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('Blocking channels in this chat enabled.')
else:
await self.message.edit('Blocking channels in this chat disabled.')
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('Blocking channels in this chat enabled.')
async def disable_anti_channels(self):
db.set('core.ats', f'antich{self.chat_id}', False)
await self.message.edit('Blocking channels in this chat disabled.')
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('Unsupported')
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('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
except Exception as e:
await self.message.edit(format_exc(e))
else:
await self.message.edit('Reply on user msg')
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('User is not found')
except PeerIdInvalid:
await self.message.edit('User is not found')
except UsernameInvalid:
await self.message.edit('User is not found')
except IndexError:
await self.message.edit('User is not found')
else:
await self.message.edit('user_id or username')
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('Invalid user type')
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'History from {name} was deleted!'
+ f'\n{"Cause: " + self.cause.split(" ", maxsplit=1)[1] + "" if len(self.cause.split()) > 1 else ""}'
)
except UserAdminInvalid:
await self.message.edit('No rights')
except ChatAdminRequired:
await self.message.edit('No rights')
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'Anti-raid mode enabled!\nDisable with: {self.prefix}antiraid off')
async def disable_antiraid(self):
db.set('core.ats', f'antiraid{self.chat_id}', False)
await self.message.edit('Anti-raid mode disabled')
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'Anti-raid mode enabled!\nDisable with: {self.prefix}antiraid off'
)
else:
await self.message.edit('Anti-raid mode disabled')
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('Loading...')
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(
'Sorry, but this note is unavailable.\n\n'
f'You can delete this note with '
f'{self.prefix}clear {note_name}'
)
else:
await self.message.edit('There is no such note')
else:
await self.message.edit(
f'Example: {self.prefix}note note_name',
)
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)