Add initial implementation of Powerban model and callback mechanism

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.'
+    }
 }

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 <user_id> <razón>\" 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 <user_id>\"")
+        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)

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]")
+
+

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)

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