diff --git a/bot_logic.py b/bot_logic.py index 7211145..18f8e26 100644 --- a/bot_logic.py +++ b/bot_logic.py @@ -1,7 +1,7 @@ from utils.orga2Utils import noitip, asm from handlers.basic import start, help_command, estasvivo, colaborar from handlers.info import campusvivo, flan, flanviejo, aulas, cuandovence, listarlabos -from handlers.admin import checodepers, checodeppers, sugerirNoticia, get_logs, joder, movergrupo +from handlers.admin import checodepers, checodeppers, sugerirNoticia, get_logs, joder, movergrupo, powerban, unpowerban from handlers.groups import listar, listaroptativa, listareci, listarotro, cubawiki, agregargrupo, agregaroptativa, agregarotros, agregareci, sugerirgrupo, sugeriroptativa, sugerireci, sugerirotro, actualizar_grupos, listararchivado, archivar COMMANDS = { @@ -137,4 +137,12 @@ 'sugerirotro': { 'handler': sugerirotro, }, + 'powerban': { + 'handler': powerban, + 'description': 'Banea al usuario de todos los grupos en donde el bot es admin.' + }, + 'unpowerban': { + 'handler': unpowerban, + 'description': 'Desbanea al usuario de todos los grupos en donde el bot es admin.' + } } diff --git a/handlers/admin.py b/handlers/admin.py index bbc2ae8..d03ae50 100644 --- a/handlers/admin.py +++ b/handlers/admin.py @@ -4,8 +4,9 @@ from telegram.ext import ContextTypes from telegram.constants import ParseMode from tg_ids import CODEPERS_CHATID, ROZEN_CHATID, DGARRO_CHATID, DC_GROUP_CHATID -from models import Noticia +from models import Noticia, BannedUser from handlers.db import get_session +from utils.db import process_unban_state logger = logging.getLogger("DCUBABOT") admin_ids = [ROZEN_CHATID, DGARRO_CHATID] @@ -172,3 +173,111 @@ async def get_logs(update: Update, context: ContextTypes.DEFAULT_TYPE): ) except Exception as e: await update.effective_message.reply_text(f"Error al leer logs (¿falta permiso roles/logging.viewer en la Service Account?): {e}") + +async def powerban(update: Update, context: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + if user_id not in admin_ids and str(user_id) not in admin_ids: + logger.warning(f"Unauthorized user {user_id} tried to access /powerban") + return + + msg = update.message.reply_to_message + + if not msg and len(context.args) < 2: + await update.effective_message.reply_text("Uso: \"/powerban \" o respondiendo al mensaje del usuario a banear") + return + + user_to_ban = None + if msg: + user_to_ban = msg.from_user + else: + try: + user_to_ban = (await context.bot.getChatMember(context._chat_id, context.args[0])).user + except: + await update.effective_message.reply_text("user_id invalido") + return + context.args = context.args[1:] + + # No confundir user_id del admin del user_id del usuario baneado + logger.info(f"Admin '{user_id}' execute /powerban to '{user_to_ban}' ") + + reason = "" + if context.args: + reason = " ".join(context.args) + + username = None + with get_session() as session: + banned_user = session.query(BannedUser).filter_by(user_id=user_to_ban.id).first() + + if banned_user and not banned_user.confirmed: + # TODO: Agregar que expire despues de unos minutos o un día para que no quede softlockeado sin poder banear al usuario. + await update.effective_chat.send_message("Esperando confirmación de powerban") + return + + if banned_user and banned_user.confirmed: + await update.effective_chat.send_message(f"Ese usuario ya se encuentra baneado de los grupos por \"{banned_user.reason}\"") + return + + username = user_to_ban.id + if user_to_ban.username: + username = "@" + user_to_ban.username + + bot_message = await update.effective_message.reply_text(f"@{update.effective_user.username} Andá a confirmar el baneo de \"{username}\"") + + banned_user = BannedUser( + user_id=user_to_ban.id, confirmed=False, username=username, banned_by_id=user_id, + reason=reason, bot_chat_id=bot_message.chat_id, bot_msg_id=bot_message.id + ) + session.add(banned_user) + + keyboard = [ + [ + InlineKeyboardButton("Confirmar", callback_data=f"Powerban|{user_to_ban.id}|Confirm", api_kwargs={"style": "success"}), + InlineKeyboardButton("Cancelar", callback_data=f"Powerban|{user_to_ban.id}|Cancel", api_kwargs={"style": "danger"}) + ] + ] + reply_markup = InlineKeyboardMarkup(keyboard) + await context.bot.send_message(chat_id=user_id, text=f"Banear usuario {user_to_ban.id} {username}?", + reply_markup=reply_markup) + +async def unpowerban(update: Update, context: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + if user_id not in admin_ids and str(user_id) not in admin_ids: + logger.warning(f"Unauthorized user {user_id} tried to access /unpowerban") + return + + msg = update.message.reply_to_message + + if not msg and len(context.args) < 1: + await update.effective_message.reply_text("Uso: \"/unpowerban \"") + return + + user_to_unban = None + try: + user_to_unban = (await context.bot.getChatMember(context._chat_id, context.args[0])).user + except: + await update.effective_message.reply_text("user_id invalido") + return + context.args = context.args[1:] + + # No confundir user_id del admin del user_id del usuario baneado + logger.info(f"Admin '{user_id}' execute /unpowerban to '{user_to_unban}' ") + + username = None + with get_session() as session: + banned_user = session.query(BannedUser).filter_by(user_id=user_to_unban.id).first() + + if not banned_user: + await update.effective_chat.send_message(f"El usuario no tiene powerban") + return + + if banned_user and not banned_user.processed_lock: + await update.effective_chat.send_message("El usuario aún se encuentra en proceso de powerban") + return + + await process_unban_state(banned_user.user_id, session, context) + session.delete(banned_user) + username = "" + if banned_user.username: + username = f"({banned_user.username})" + unban_msg = f"Usuario '{banned_user.user_id}' {username} desbaneado de todos los grupos" + await update.effective_chat.send_message(unban_msg) diff --git a/handlers/callbacks.py b/handlers/callbacks.py index e0b920f..ede51a1 100644 --- a/handlers/callbacks.py +++ b/handlers/callbacks.py @@ -1,9 +1,13 @@ from telegram import Update from telegram.ext import ContextTypes from telegram.constants import ParseMode -from models import Listable, Noticia +from models import Listable, Noticia, BannedUser from handlers.db import get_session from tg_ids import NOTICIAS_CHATID +from utils.db import process_ban_state +import logging + +logger = logging.getLogger("DCUBABOT") async def button(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query @@ -15,9 +19,33 @@ async def button(update: Update, context: ContextTypes.DEFAULT_TYPE): buttonType = data_parts[0] id_val = data_parts[1] if len(data_parts) > 1 else None action = data_parts[2] if len(data_parts) > 2 else None - + with get_session() as session: - if buttonType == "Listable": + user_to_ban = session.query(BannedUser).filter_by(user_id=id_val).first() + # Powerban Callback + if buttonType == "Powerban" and action == "Confirm": + user_to_ban.confirmed = True + username = "" + if user_to_ban.username: + username = f"({user_to_ban.username})" + logger.info(f"User '{user_to_ban.user_id}' {username} banned by '{user_to_ban.user_id}', reason: \"{user_to_ban.reason}\"") + ban_msg = f"Usuario '{user_to_ban.user_id}' {username} baneado de todos los grupos por razón \"{user_to_ban.reason}\" 🚫" + await update.effective_message.edit_text(ban_msg) + await context.bot.editMessageText(ban_msg, user_to_ban.bot_chat_id, user_to_ban.bot_msg_id) + await process_ban_state(session, context) + + if buttonType == "Powerban" and action == "Cancel": + user_to_ban = session.query(BannedUser).filter_by(user_id=id_val).first() + session.delete(user_to_ban) + logger.info(f"User ban '{user_to_ban.user_id}' cancelled by '{user_to_ban.banned_by_id}'") + username = "" + if user_to_ban.username: + username = f"({user_to_ban.username})" + cancel_msg = f"Baneo de '{user_to_ban.user_id}' {username} cancelado" + await update.effective_message.edit_text(cancel_msg) + await context.bot.editMessageText(cancel_msg, user_to_ban.bot_chat_id, user_to_ban.bot_msg_id) + + elif buttonType == "Listable": group = session.query(Listable).filter_by(id=int(id_val)).first() if group: if action == "1": @@ -101,3 +129,5 @@ async def button(update: Update, context: ContextTypes.DEFAULT_TYPE): await query.edit_message_text(text=message.text + action_text) else: await query.edit_message_text(text=message.text + "\n[Botón huérfano: La noticia ya no existe en la base de datos]") + + diff --git a/models.py b/models.py index 2fb26d6..324eaa6 100644 --- a/models.py +++ b/models.py @@ -145,3 +145,16 @@ class File(Base): id = Column(Integer, primary_key=True) path = Column(String, nullable=False, unique=True) file_id = Column(String, nullable=False) + +class BannedUser(Base): + __tablename__ = 'banned_users' + user_id = Column(BigInteger, primary_key=True) + username = Column(String, nullable=True) + banned_by_id = Column(BigInteger, nullable=False) + reason = Column(String, nullable=True) + date = Column(Date, nullable=False, default=datetime.date.today) + confirmed = Column(Boolean, default=False) + processed_lock = Column(Boolean, default=False) # True = Ya fue procesado; False = Falta procesar. + # Para editar el mensaje del bot. + bot_chat_id = Column(String, nullable=True) + bot_msg_id = Column(String, nullable=True) diff --git a/utils/db.py b/utils/db.py new file mode 100644 index 0000000..0764b3d --- /dev/null +++ b/utils/db.py @@ -0,0 +1,49 @@ +from sqlalchemy.orm import Session +from models import BannedUser, Listable +from telegram.ext import ContextTypes +import logging + +logger = logging.getLogger("DCUBABOT") + +async def process_ban_state(session: Session, context: ContextTypes.DEFAULT_TYPE) -> None: + groups: list[Listable] = ( + session + .query(Listable) + .filter(Listable.validated == True) + .all() + ) + + users_to_process: list[BannedUser] = ( + session + .query(BannedUser) + .filter(BannedUser.processed_lock == False) + .all() + ) + + for group in groups: + for user in users_to_process: + try: + await context.bot.ban_chat_member(group.chat_id, user.user_id) + # logger.info(f"Processed ban user with id:\'{user.user_id}\' from group with id:\'{group.chat_id}\'") + except Exception as e: + # logger.error(f"Failed to process ban user with id:\'{user.user_id}\' from group with id:\'{group.chat_id}\', reason: {e}\n") + pass + + for user in users_to_process: + user.processed_lock = True + +async def process_unban_state(user_id: int, session: Session, context: ContextTypes.DEFAULT_TYPE) -> None: + groups: list[Listable] = ( + session + .query(Listable) + .filter(Listable.validated == True) + .all() + ) + + for group in groups: + try: + await context.bot.unban_chat_member(group.chat_id, user_id) + # logger.info(f"Processed unban of user with id:\'{user.user_id}\' from group with id:\'{group.chat_id}\'") + except Exception as e: + # logger.error(f"Failed to process unban user with id:\'{user.user_id}\' from group with id:\'{group.chat_id}\', reason: {e}\n") + pass \ No newline at end of file