added antichannelpin

just a duplicate of the last commit with minor changes Co-Authored-By: Sayan Biswas <dankguy6146@gmail.com>
vinuth2 · Dec 22, 2021 · 521e478 · 521e478
1 parent dbb16cc
commit 521e478
Show file tree

Hide file tree

Showing 3 changed files with 106 additions and 10 deletions.
diff --git a/tg_bot/langs/en.yaml b/tg_bot/langs/en.yaml
@@ -369,7 +369,8 @@ misc_help: |
       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
+      • /antilinkedchan <on/off>: Makes Kigyo automatically delete linked channel posts from chatroom
+      • /antichannelpin <on/off>: Makes Kigyo automatically unpin linked channel posts from chatroom
 
 muting_help: |
       *Admins only:*

diff --git a/tg_bot/modules/antilinkedchannel.py b/tg_bot/modules/antilinkedchannel.py
@@ -3,13 +3,15 @@
 from telegram import Update, TelegramError
 from telegram.ext import CallbackContext
 from telegram.ext.filters import Filters
+from tg_bot.modules.helper_funcs.chat_status import bot_admin, bot_can_delete
 
 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)
+@bot_can_delete
 @user_admin(AdminPerms.CAN_RESTRICT_MEMBERS)
 def set_antilinkedchannel(update: Update, context: CallbackContext):
     message = update.effective_message
@@ -18,25 +20,68 @@ def set_antilinkedchannel(update: Update, context: CallbackContext):
     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)))
+            if sql.status_pin(chat.id):
+                sql.disable_pin(chat.id)
+                sql.enable_pin(chat.id)
+                message.reply_html("Enabled Linked channel deletion and Disabled anti channel pin in {}".format(html.escape(chat.title)))
+            else:
+                sql.enable_linked(chat.id)
+                message.reply_html("Enabled anti linked channel in {}".format(html.escape(chat.title)))
         elif s in ["off", "no"]:
-            sql.disable(chat.id)
+            sql.disable_linked(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)))
+        "Linked channel deletion is currently {} in {}".format(sql.status_linked(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):
+    if not sql.status_linked(chat.id):
         return
     try:
         message.delete()
     except TelegramError:
         return
+
+@kigcmd(command="antichannelpin", group=114)
+@bot_admin
+@user_admin(AdminPerms.CAN_RESTRICT_MEMBERS)
+def set_antipinchannel(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"]:
+            if sql.status_linked(chat.id):
+                sql.disable_linked(chat.id)
+                sql.enable_pin(chat.id)
+                message.reply_html("Disabled Linked channel deletion and Enabled anti channel pin in {}".format(html.escape(chat.title)))
+            else:
+                sql.enable_pin(chat.id)
+                message.reply_html("Enabled anti channel pin in {}".format(html.escape(chat.title)))
+        elif s in ["off", "no"]:
+            sql.disable_pin(chat.id)
+            message.reply_html("Disabled anti channel pin in {}".format(html.escape(chat.title)))
+        else:
+            message.reply_text("Unrecognized arguments {}".format(s))
+        return
+    message.reply_html(
+        "Linked channel message unpin is currently {} in {}".format(sql.status_pin(chat.id), html.escape(chat.title)))
+
+
+@kigmsg(Filters.is_automatic_forward | Filters.status_update.pinned_message, group=113)
+def eliminate_linked_channel_msg(update: Update, _: CallbackContext):
+    message = update.effective_message
+    chat = update.effective_chat
+    if not sql.status_pin(chat.id):
+        return
+    try:
+        message.unpin()
+    except TelegramError:
+        return
diff --git a/tg_bot/modules/sql/antilinkedchannel_sql.py b/tg_bot/modules/sql/antilinkedchannel_sql.py
@@ -18,14 +18,30 @@ def __init__(self, chat_id: int, disabled: bool):
         self.setting = disabled
 
     def __repr__(self):
-        return "<Antiflood setting {} ({})>".format(self.chat_id, self.setting)
+        return "<Antilinked setting {} ({})>".format(self.chat_id, self.setting)
+
+class AntiPinChannelSettings(BASE):
+    __tablename__ = "anti_pin_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 "<Antipin setting {} ({})>".format(self.chat_id, self.setting)
 
 
 AntiLinkedChannelSettings.__table__.create(checkfirst=True)
 ANTI_LINKED_CHANNEL_SETTING_LOCK = threading.RLock()
 
+AntiPinChannelSettings.__table__.create(checkfirst=True)
+ANTI_PIN_CHANNEL_SETTING_LOCK = threading.RLock()
 
-def enable(chat_id: int):
+
+def enable_linked(chat_id: int):
     with ANTI_LINKED_CHANNEL_SETTING_LOCK:
         chat = SESSION.query(AntiLinkedChannelSettings).get(str(chat_id))
         if not chat:
@@ -35,8 +51,18 @@ def enable(chat_id: int):
         SESSION.add(chat)
         SESSION.commit()
 
+def enable_pin(chat_id: int):
+    with ANTI_PIN_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiPinChannelSettings).get(str(chat_id))
+        if not chat:
+            chat = AntiPinChannelSettings(chat_id, True)
+
+        chat.setting = True
+        SESSION.add(chat)
+        SESSION.commit()
+
 
-def disable(chat_id: int):
+def disable_linked(chat_id: int):
     with ANTI_LINKED_CHANNEL_SETTING_LOCK:
         chat = SESSION.query(AntiLinkedChannelSettings).get(str(chat_id))
         if not chat:
@@ -46,14 +72,31 @@ def disable(chat_id: int):
         SESSION.add(chat)
         SESSION.commit()
 
+def disable_pin(chat_id: int):
+    with ANTI_PIN_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiPinChannelSettings).get(str(chat_id))
+        if not chat:
+            chat = AntiPinChannelSettings(chat_id, False)
+
+        chat.setting = False
+        SESSION.add(chat)
+        SESSION.commit()
+
 
-def status(chat_id: int) -> bool:
+def status_linked(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 status_pin(chat_id: int) -> bool:
+    with ANTI_PIN_CHANNEL_SETTING_LOCK:
+        d = SESSION.query(AntiPinChannelSettings).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:
@@ -63,3 +106,10 @@ def migrate_chat(old_chat_id, new_chat_id):
             SESSION.add(chat)
 
         SESSION.commit()
+    with ANTI_PIN_CHANNEL_SETTING_LOCK:
+        chat = SESSION.query(AntiPinChannelSettings).get(str(old_chat_id))
+        if chat:
+            chat.chat_id = new_chat_id
+            SESSION.add(chat)
+
+        SESSION.commit()