New: linked channel message auto-deletion

Signed-off-by: Sayan Biswas <sayan@pokurt.me>
fiction7799 · Dec 22, 2021 · dc942aa · dc942aa
1 parent 4773013
commit dc942aa
Show file tree

Hide file tree

Showing 3 changed files with 114 additions and 1 deletion.
diff --git a/tg_bot/langs/en.yaml b/tg_bot/langs/en.yaml
@@ -357,13 +357,19 @@ misc_help: |
         example syntax: /cash 1 USD INR
       • /getid : get IDs of chat, user and chat message.
       • /spbinfo : get info about a user from @Intellivoid's SpamProtection API
-      ───────────────────────────────
+      ~~~~~~
       *Last.FM*
       Share what you're what listening to with the help of this module!
       *Available commands:*
       • /setuser <username>: sets your last.fm username.
       • /clearuser: removes your last.fm username from the bot's database.
       • /lastfm: returns what you're scrobbling on last.fm.
+      ~~~~~~
+      *Anti channel*
+      Tired of telegram's stupidity? well here you go
+      *Available commands:*
+      • /antichannel <on/off>: Bans and deletes anyone who tries to talk as channel and forces em to talk as themselves
+      • /antilinkedchan <on/off: Makes Kigyo automatically delete linked channel posts from chatroom
 
 muting_help: |
       *Admins only:*

diff --git a/tg_bot/modules/antilinkedchannel.py b/tg_bot/modules/antilinkedchannel.py
@@ -0,0 +1,42 @@
+import html
+
+from telegram import Update, TelegramError
+from telegram.ext import CallbackContext
+from telegram.ext.filters import Filters
+
+from tg_bot.modules.helper_funcs.decorators import kigcmd, kigmsg
+from ..modules.helper_funcs.anonymous import user_admin, AdminPerms
+import tg_bot.modules.sql.antilinkedchannel_sql as sql
+
+
+@kigcmd(command="antilinkedchan", group=112)
+@user_admin(AdminPerms.CAN_RESTRICT_MEMBERS)
+def set_antilinkedchannel(update: Update, context: CallbackContext):
+    message = update.effective_message
+    chat = update.effective_chat
+    args = context.args
+    if len(args) > 0:
+        s = args[0].lower()
+        if s in ["yes", "on"]:
+            sql.enable(chat.id)
+            message.reply_html("Enabled anti linked channel in {}".format(html.escape(chat.title)))
+        elif s in ["off", "no"]:
+            sql.disable(chat.id)
+            message.reply_html("Disabled anti linked channel in {}".format(html.escape(chat.title)))
+        else:
+            message.reply_text("Unrecognized arguments {}".format(s))
+        return
+    message.reply_html(
+        "Linked channel deletion is currently {} in {}".format(sql.status(chat.id), html.escape(chat.title)))
+
+
+@kigmsg(Filters.is_automatic_forward, group=111)
+def eliminate_linked_channel_msg(update: Update, _: CallbackContext):
+    message = update.effective_message
+    chat = update.effective_chat
+    if not sql.status(chat.id):
+        return
+    try:
+        message.delete()
+    except TelegramError:
+        return
diff --git a/tg_bot/modules/sql/antilinkedchannel_sql.py b/tg_bot/modules/sql/antilinkedchannel_sql.py
@@ -0,0 +1,65 @@
+import threading
+
+from sqlalchemy import Boolean
+from sqlalchemy.sql.sqltypes import String
+from sqlalchemy import Column
+
+from tg_bot.modules.sql import BASE, SESSION
+
+
+class AntiLinkedChannelSettings(BASE):
+    __tablename__ = "anti_linked_channel_settings"
+
+    chat_id = Column(String(14), primary_key=True)
+    setting = Column(Boolean, default=False, nullable=False)
+
+    def __init__(self, chat_id: int, disabled: bool):
+        self.chat_id = str(chat_id)
+        self.setting = disabled
+
+    def __repr__(self):
+        return "<Antiflood setting {} ({})>".format(self.chat_id, self.setting)
+
+
+AntiLinkedChannelSettings.__table__.create(checkfirst=True)
+ANTI_LINKED_CHANNEL_SETTING_LOCK = threading.RLock()
+
+
+def enable(chat_id: int):
+    with ANTI_LINKED_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiLinkedChannelSettings).get(str(chat_id))
+        if not chat:
+            chat = AntiLinkedChannelSettings(chat_id, True)
+
+        chat.setting = True
+        SESSION.add(chat)
+        SESSION.commit()
+
+
+def disable(chat_id: int):
+    with ANTI_LINKED_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiLinkedChannelSettings).get(str(chat_id))
+        if not chat:
+            chat = AntiLinkedChannelSettings(chat_id, False)
+
+        chat.setting = False
+        SESSION.add(chat)
+        SESSION.commit()
+
+
+def status(chat_id: int) -> bool:
+    with ANTI_LINKED_CHANNEL_SETTING_LOCK:
+        d = SESSION.query(AntiLinkedChannelSettings).get(str(chat_id))
+        if not d:
+            return False
+        return d.setting
+
+
+def migrate_chat(old_chat_id, new_chat_id):
+    with ANTI_LINKED_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiLinkedChannelSettings).get(str(old_chat_id))
+        if chat:
+            chat.chat_id = new_chat_id
+            SESSION.add(chat)
+
+        SESSION.commit()