Feat: warn and mute command

src/spotted/config/db/post_db_del.sql

@@ -14,3 +14,7 @@ DROP TABLE IF EXISTS spot_report
 DROP TABLE IF EXISTS user_report
 -----
 DROP TABLE IF EXISTS user_follow
+-----
+DROP TABLE IF EXISTS warned_users
+-----
+DROP TABLE IF EXISTS muted_users

src/spotted/config/db/post_db_init.sql

@@ -40,6 +40,19 @@ 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 muted_users
+(
+  user_id BIGINT NOT NULL,
+  PRIMARY KEY user_id
+);
+-----
 CREATE TABLE IF NOT EXISTS spot_report
 (
   user_id BIGINT NOT NULL,

src/spotted/config/yaml/settings.yaml

@@ -18,5 +18,7 @@ post:
   delete_anonymous_comments: true
   reject_after_autoreply: true
   autoreplies_per_page: 6
+  warn_after_days: 7
+  max_n_warns : 3
 token: ""
-bot_tag: "@bot_tag"
+bot_tag: "@bot_tag"

src/spotted/config/yaml/settings.yaml.types

@@ -17,5 +17,7 @@ post:
   delete_anonymous_comments: bool
   reject_after_autoreply: bool
   autoreplies_per_page: int
+  warn_after_days: int
+  max_n_warns: int
 token: str
 bot_tag: str

src/spotted/data/user.py

@@ -5,6 +5,7 @@
 
 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,11 @@ 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:
+        """Returns the count of consecutive warns of the user"""
+        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 +89,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 +102,53 @@ def sban(self) -> bool:
             return True
         return False
 
+    def mute(self, bot: Bot):
+        """Mute a user restricting its actions inside the community group
+
+        Args:
+            bot: the telegram bot
+            days(optional): The number of days the user should be muted for.
+        """
+        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,
+        )
+        DbManager.insert_into(
+            table_name="muted_users",
+            columns=("user_id"),
+            values=(self.user_id),
+        )
+
+    def unmute(self, bot: Bot):
+        """Unmute a user taking back all restrictions
+
+        Args:
+            bot : the telegram bot
+        """
+        bot.restrict_chat_member(
+            chat_id=Config.post_get("channel_id"),
+            user_id=self.user_id,
+            can_send_messages=True,
+            can_send_media_messages=True,
+            can_send_other_messages=True,
+            can_add_web_page_previews=True,
+        )
+        DbManager.delete_from(table_name="muted_users", where="user_id = %s", where_args=(self.user_id,))
+
+    def warn(self):
+        """Increase the number of warns of a user
+        If this is number would reach 3 the user is banned
+
+        Args:
+            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

@@ -1,5 +1,5 @@
 """Modules that handle the events the bot recognizes and reacts to"""
-from datetime import time
+from datetime import time, timedelta
 from warnings import filterwarnings
 
 from pytz import utc
@@ -22,12 +22,14 @@
 from .ban import ban_cmd
 from .cancel import cancel_cmd
 from .clean_pending import clean_pending_cmd
+from .custom_filters import IsAdminFilter
 from .db_backup import db_backup_cmd
 from .follow_comment import follow_spot_comment
 from .follow_spot import follow_spot_callback
 from .forwarded_post import forwarded_post_msg
 from .help import help_cmd
-from .job_handlers import clean_pending_job, db_backup_job
+from .job_handlers import clean_pending_job, clean_warned_users, db_backup_job
+from .mute import mute_cmd
 from .purge import purge_cmd
 from .reload import reload_cmd
 from .reply import reply_cmd
@@ -38,6 +40,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,9 +87,9 @@ 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_group_filter = filters.Chat(chat_id=Config.post_get("admin_group_id"))
     community_filter = filters.Chat(chat_id=Config.post_get("community_group_id"))
-
+    is_admin_filter = IsAdminFilter()
     # Error handler
     app.add_error_handler(error_handler)
 
@@ -97,23 +100,27 @@ 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 & is_admin_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))
+    app.add_handler(
+        MessageHandler(filters.REPLY & community_filter & is_admin_filter & filters.Regex(r"^/mute"), mute_cmd)
+    )
 
     # Callback handlers
     app.add_handler(CallbackQueryHandler(settings_callback, pattern=r"^settings\.*"))
@@ -144,3 +151,4 @@ def add_jobs(app: Application):
     """
     app.job_queue.run_daily(clean_pending_job, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc
     app.job_queue.run_daily(db_backup_job, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc
+    app.job_queue.run_daily(clean_warned_users, time=time(hour=5, tzinfo=utc))  # run each day at 05:00 utc

src/spotted/handlers/custom_filters.py

@@ -0,0 +1,18 @@
+"""Defines custom filters to use for commands"""
+from telegram import Chat, Message
+from telegram.ext.filters import MessageFilter
+
+
+class IsAdminFilter(MessageFilter):
+    """Check if the message from the update was sent by
+        one of the administrators of the group
+    Args:
+        MessageFilter: the superclass for the filter
+    """
+
+    def filter(self, message: Message):
+        chat = message.chat
+        sender_id = message.from_user.id
+        if chat.type in [Chat.SUPERGROUP, Chat.GROUP]:
+            return sender_id in [admin.id for admin in chat.get_administrators(chat.id)]
+        return False

src/spotted/handlers/job_handlers.py

@@ -4,7 +4,7 @@
 from telegram.error import BadRequest, Forbidden
 from telegram.ext import CallbackContext
 
-from spotted.data import Config, PendingPost
+from spotted.data import Config, DbManager, PendingPost
 from spotted.debug import logger
 from spotted.utils import EventInfo
 
@@ -62,3 +62,27 @@ async def db_backup_job(context: CallbackContext):
             )
         except Exception as ex:  # pylint: disable=broad-except
             await context.bot.send_message(chat_id=admin_group_id, text=f"✖️ Impossibile effettuare il backup\n\n{ex}")
+
+
+async def clean_warned_users():
+    """Job called each day at 05:00 utc.
+    Removed users who have been warned for longer than setting duration
+    Args:
+        context: context passed by the jobqueue
+    """
+    warn_expiration = datetime.now() + timedelta(days=Config.post_get("warn_after_days"))
+    DbManager.delete_from(
+        table_name="warned_users",
+        where="warn_time > %s",
+        where_args=(warn_expiration,),
+    )
+
+
+async def unmute_user(context: CallbackContext):
+    """A callback function that unmute the user
+
+    Args:
+        context: context passed by the job queue
+    """
+    user = context.job.context
+    user.unmute(context.bot)

src/spotted/handlers/mute.py

@@ -0,0 +1,39 @@
+"""/mute command"""
+import re
+from datetime import timedelta
+
+from telegram import Update
+from telegram.ext import CallbackContext
+
+from spotted.data import User
+from spotted.utils import EventInfo
+
+from .job_handlers import unmute_user
+
+
+async def mute_cmd(update: Update, context: CallbackContext):
+    """Handles the /mute command.
+    Mute a user by replying to one of his message in the comment group with /mute <n_days>
+
+    Args:
+        update: update event
+        context: context passed by the handler
+    """
+    info = EventInfo.from_message(update, context)
+    g_message = update.message.reply_to_message
+    user = User(g_message.from_user.id)
+    match = re.search(r"^/mute (?P<days>\\d*)$", info.text)
+
+    if g_message is None or match is None:
+        await info.bot.send_message(
+            chat_id=info.chat_id,
+            text="Per mutare qualcuno, rispondi al suo commento con /mute <giorni>",
+        )
+        return
+
+    days = 1 if match == "" else int(match)  # if no argv is provided, default is 1 day
+
+    user.mute(info.bot)
+    text = f"L'utente è stato mutato per {days} giorn{'o' if days == 1 else 'i'}."
+    await info.bot.send_message(chat_id=info.chat_id, text=text)
+    context.job_queue.run_once(unmute_user, timedelta(days=days), context=user)

src/spotted/handlers/warn.py

@@ -0,0 +1,36 @@
+"""/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 = "L'utente "
+    max_warns = Config.post_get("max_n_warns")
+    if n_warns < max_warns:
+        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)