Feat: warn and mute command

src/spotted/config/db/post_db_init.sql

@@ -40,6 +40,13 @@ CREATE TABLE IF NOT EXISTS banned_users
   PRIMARY KEY (user_id)
 );
 -----
+CREATE TABLE IF NOT EXISTS warned_users
+(
+  user_id BIGINT NOT NULL,
+  warn_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (user_id, warn_date)
+);
+-----
 CREATE TABLE IF NOT EXISTS spot_report
 (
   user_id BIGINT NOT NULL,

src/spotted/data/user.py

@@ -1,10 +1,11 @@
 """Users management"""
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timedelta
 from random import choice
 
 from telegram import Bot
 
+from .config import Config
 from .data_reader import read_md
 from .db_manager import DbManager
 from .pending_post import PendingPost
@@ -41,6 +42,15 @@ def is_credited(self) -> bool:
         """If the user is in the credited list"""
         return DbManager.count_from(table_name="credited_users", where="user_id = %s", where_args=(self.user_id,)) == 1
 
+    def get_n_warns(self) -> int:
+        """_summary_
+
+        Returns:
+            int: _description_
+        """
+        count = DbManager.count_from(table_name="warned_users", where="user_id = %s", where_args=(self.user_id,))
+        return count if count else 0
+
     @classmethod
     def banned_users(cls) -> "list[User]":
         """Returns a list of all the banned users"""
@@ -83,6 +93,7 @@ def ban(self):
 
         if not self.is_banned:
             DbManager.insert_into(table_name="banned_users", columns=("user_id",), values=(self.user_id,))
+        DbManager.delete_from(table_name="warned_users", where="user_id = %s", where_args=(self.user_id,))
 
     def sban(self) -> bool:
         """Removes the user from the banned list
@@ -95,6 +106,40 @@ def sban(self) -> bool:
             return True
         return False
 
+    def mute(self, bot: Bot, days: int = 0):
+        """_summary_
+
+        Args:
+            bot (Bot): the telegram bot
+            days (int, optional): The number of days the user should be muted for.
+                                    Defaults to 0.
+        """
+        bot.restrict_chat_member(
+            chat_id=Config.post_get("channel_id"),
+            user_id=self.user_id,
+            can_send_messages=False,
+            can_send_media_messages=False,
+            can_send_other_messages=False,
+            can_add_web_page_previews=False,
+        )
+        current_date_time = datetime.now()
+        expiration_date_time = (current_date_time + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
+        DbManager.insert_into(
+            table_name="muted_users",
+            columns=("user_id", "expiration_date"),
+            values=(self.user_id, expiration_date_time),
+        )
+
+    def warn(self):
+        """Increase the number of warns of a user
+        If this is number would reach 3 the user is banned
+
+        Args:
+            bot (Bot): the telegram bot
+
+        """
+        DbManager.insert_into(table_name="warned_users", columns=("user_id",), values=(self.user_id,))
+
     def become_anonym(self) -> bool:
         """Removes the user from the credited list, if he was present
 

src/spotted/handlers/__init__.py

@@ -38,6 +38,7 @@
 from .settings import settings_callback, settings_cmd
 from .spot import spot_conv_handler
 from .start import start_cmd
+from .warn import warn_cmd
 
 
 async def add_commands(app: Application):
@@ -84,7 +85,8 @@ def add_handlers(app: Application):
     if Config.settings_get("debug", "local_log"):  # add MessageHandler only if log_message is enabled
         app.add_handler(MessageHandler(filters.ALL, log_message), 1)
 
-    admin_filter = filters.Chat(chat_id=Config.post_get("admin_group_id"))
+    # admin_message_filter = filters.MessageFilter
+    admin_group_filter = filters.Chat(chat_id=Config.post_get("admin_group_id"))
     community_filter = filters.Chat(chat_id=Config.post_get("community_group_id"))
 
     # Error handler
@@ -97,23 +99,24 @@ def add_handlers(app: Application):
 
     # Command handlers
     app.add_handler(CommandHandler("start", start_cmd, filters=filters.ChatType.PRIVATE))
-    app.add_handler(CommandHandler("help", help_cmd, filters=filters.ChatType.PRIVATE | admin_filter))
+    app.add_handler(CommandHandler("help", help_cmd, filters=filters.ChatType.PRIVATE | admin_group_filter))
     app.add_handler(CommandHandler("rules", rules_cmd, filters=filters.ChatType.PRIVATE))
     app.add_handler(CommandHandler("settings", settings_cmd, filters=filters.ChatType.PRIVATE))
     # it must be after the conversation handler's 'cancel'
     app.add_handler(CommandHandler("cancel", cancel_cmd, filters=filters.ChatType.PRIVATE))
 
     # Command handlers: Admin commands
-    app.add_handler(CommandHandler("sban", sban_cmd, filters=admin_filter))
-    app.add_handler(CommandHandler("clean_pending", clean_pending_cmd, filters=admin_filter))
-    app.add_handler(CommandHandler("db_backup", db_backup_cmd, filters=admin_filter))
-    app.add_handler(CommandHandler("purge", purge_cmd, filters=admin_filter))
-    app.add_handler(CommandHandler("reload", reload_cmd, filters=admin_filter))
+    app.add_handler(CommandHandler("sban", sban_cmd, filters=admin_group_filter))
+    app.add_handler(CommandHandler("clean_pending", clean_pending_cmd, filters=admin_group_filter))
+    app.add_handler(CommandHandler("db_backup", db_backup_cmd, filters=admin_group_filter))
+    app.add_handler(CommandHandler("purge", purge_cmd, filters=admin_group_filter))
+    app.add_handler(CommandHandler("reload", reload_cmd, filters=admin_group_filter))
+    app.add_handler(CommandHandler("warn", warn_cmd, filters=community_filter))
 
     # MessageHandler
-    app.add_handler(MessageHandler(filters.REPLY & admin_filter & filters.Regex(r"^/ban$"), ban_cmd))
-    app.add_handler(MessageHandler(filters.REPLY & admin_filter & filters.Regex(r"^/reply"), reply_cmd))
-    app.add_handler(MessageHandler(filters.REPLY & admin_filter & filters.Regex(r"^/autoreply"), autoreply_cmd))
+    app.add_handler(MessageHandler(filters.REPLY & admin_group_filter & filters.Regex(r"^/ban$"), ban_cmd))
+    app.add_handler(MessageHandler(filters.REPLY & admin_group_filter & filters.Regex(r"^/reply"), reply_cmd))
+    app.add_handler(MessageHandler(filters.REPLY & admin_group_filter & filters.Regex(r"^/autoreply"), autoreply_cmd))
 
     # Callback handlers
     app.add_handler(CallbackQueryHandler(settings_callback, pattern=r"^settings\.*"))

src/spotted/handlers/warn.py

@@ -0,0 +1,35 @@
+"""/warn command"""
+from telegram import Update
+from telegram.ext import CallbackContext
+
+from spotted.data import Config, User
+from spotted.utils import EventInfo
+
+
+async def warn_cmd(update: Update, context: CallbackContext):
+    """Handles the /warn command.
+    Warn a user by replying to one of his message in the comment group with /warn
+
+    Args:
+        update: update event
+        context: context passed by the handler
+    """
+    info = EventInfo.from_message(update, context)
+    g_message = update.message.reply_to_message
+    if g_message is None:
+        await info.bot.send_message(
+            chat_id=info.chat_id,
+            text="Per warnare qualcuno, rispondi al suo commento con /warn",
+        )
+        return
+    user = User(g_message.from_user.id)
+    n_warns = user.get_n_warns()
+    text = f"L'utente {g_message.from_user.name} "
+    if n_warns < 2:
+        user.warn()
+        text += f"ha ricevuto {n_warns + 1} warn(s)"
+    else:
+        text += "è stato bannato."
+        await info.bot.ban_chat_member(Config.post_get("community_group_id"), user.user_id)
+        user.ban()
+    await info.bot.send_message(chat_id=info.chat_id, text=text)