bot

Author: iterweb

README.md

@@ -0,0 +1,25 @@
+# Async Telegram Bot Moderator
+This is a simple realization of an asynchronous telegram bot moderator.
+## Bot Functions
+- deleting duplicate messages
+- deleting abusive messages
+- deleting messages with external urls
+- muting users
+
+## Requirements
+```pip install -r requirements.txt```
+
+## Usage
+- create a bot in ```@BotFather``` then run the command in BotFather's menu ```/setprivacy``` and set up current status in: **Disable**
+- add the bot to your group and set administrator rights for it
+- add received bot's token in ```config.py``` and fill other fields
+
+## Mute Users
+First you need to get the user ID
+- run ```!info <full name>```<br>
+Example: ```!info John Doe```<br>
+You will receive a message from the bot with user data, something like this ```[('1234567', 'John Doe', '2021-08-25 23:02:37')]```
+- for mute user, run ```!mute <id>```<br>
+Example: ```!mute 1234567```
+## Start
+```python bot.py```

bot.py

@@ -0,0 +1,54 @@
+import logging
+import filters
+from aiogram import Bot, Dispatcher, executor, types
+from config import TOKEN, ACCESS_DENIED_MESSAGE, BAN_MESSAGE, MUTE_TIME
+from db_writer import DataWriter
+
+
+db_write = DataWriter()
+
+logging.basicConfig(level=logging.INFO)
+
+bot = Bot(token=TOKEN)
+dp = Dispatcher(bot)
+
+
+@dp.message_handler(commands=['mute', 'info'], commands_prefix='!')
+async def admin_command(message: types.Message):
+    user_status = ['creator', 'administrator']
+    member = await bot.get_chat_member(message.chat.id, message.from_user.id)
+
+    if member['status'] in user_status:
+        if message.text.startswith('!mute'):
+            await message.delete()
+            user_id = message.text.replace('!mute', '').strip()
+            await bot.restrict_chat_member(message.chat.id, user_id, until_date=int(message.date.timestamp() + MUTE_TIME * 3600))
+            await message.answer(f'{db_write.get_user_name(user_id)} - {BAN_MESSAGE}!')
+        elif message.text.startswith('!info'):
+            full_name = message.text.replace('!info', '').strip()
+            await bot.send_message(message.from_user.id, db_write.get_user_info(full_name))
+    else:
+        await message.reply(ACCESS_DENIED_MESSAGE)
+
+
+@dp.message_handler()
+async def filter_message(message: types.Message):
+    user_id = message.from_user.id
+    user_name = message.from_user.username
+    full_name = message.from_user.full_name
+    msg_date = message.date
+    msg = message.text.lower()
+    timestamp = message.date.timestamp()
+
+    check_msg = await filters.filter_text(msg, user_id, int(timestamp))
+    if check_msg:
+        await message.delete()
+
+    if str(message.from_user.id) not in db_write.get_all_users_id():
+        db_write.save_user_data(user_id, user_name, full_name, msg_date, msg, timestamp)
+    else:
+        db_write.update_user_data(user_id, user_name, full_name, msg_date, msg, timestamp)
+
+
+if __name__ == '__main__':
+    executor.start_polling(dp, skip_updates=True)

config.py

@@ -0,0 +1,19 @@
+# bot token
+TOKEN = ''
+
+# abusive words
+BAD_WORDS = ['bad', 'word', 'very bad']
+
+# allowed domains
+# if you want to allow all urls, this array must be empty
+GOOD_DOMAINS = ['domain.com', 'example.net']
+
+ACCESS_DENIED_MESSAGE = 'Access denied'
+
+BAN_MESSAGE = 'You were blocked'
+
+# maximum time to user mute in hours
+MUTE_TIME = 24
+
+# maximum message sending frequency in seconds
+MESSAGE_SEND_FREQUENCY = 3

db_writer.py

@@ -0,0 +1,64 @@
+import sqlite3
+
+
+class DataWriter:
+    def __init__(self):
+        self.conn = sqlite3.connect('data.sqlite3')
+        self.cur = self.conn.cursor()
+        self.cur.execute(
+            """
+            CREATE TABLE IF NOT EXISTS users(
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            tg_id TEXT NOT NULL,
+            user_name TEXT,
+            full_name TEXT,
+            date_time TEXT,
+            message TEXT,
+            active_time INTEGER
+            );
+            """
+        )
+        self.conn.commit()
+
+    def get_user_msg(self, user_id):
+        self.cur.execute(f"SELECT message, active_time FROM users WHERE tg_id = '{str(user_id)}';")
+        user_info = self.cur.fetchall()
+        return user_info
+
+    def get_user_info(self, name):
+        self.cur.execute(f"SELECT tg_id, full_name, date_time FROM users WHERE full_name = '{name}';")
+        db_user = self.cur.fetchall()
+        print(db_user)
+        return db_user
+
+    def get_user_name(self, user_id):
+        self.cur.execute(f"SELECT full_name FROM users WHERE tg_id = '{user_id}';")
+        name_user = self.cur.fetchall()
+        return name_user[0][0]
+
+    def get_all_users_id(self):
+        user_data = []
+        self.cur.execute("SELECT tg_id FROM users;")
+        db_users = self.cur.fetchall()
+
+        for i in db_users:
+            user_data.append(''.join(i))
+
+        return user_data
+
+    def save_user_data(self, tg_id, user_name, full_name, date_time, msg, timestamp):
+        self.cur.execute(f"""INSERT INTO users(tg_id, user_name, full_name, date_time, message, active_time)
+                           VALUES('{tg_id}', '{user_name}', '{full_name}', '{date_time}', '{msg}', '{timestamp}');""")
+        self.conn.commit()
+
+    def update_user_data(self, user_tg_id, username, fullname, datetime, msg, timestamp):
+        self.cur.execute(f"""UPDATE users 
+                                SET date_time = '{datetime}',
+                                    user_name = '{username}',
+                                    full_name = '{fullname}',
+                                    message = '{msg}',
+                                    active_time = '{timestamp}'
+                                WHERE tg_id = '{user_tg_id}'
+                        """)
+        self.conn.commit()
+

filters.py

@@ -0,0 +1,36 @@
+import re
+
+from config import GOOD_DOMAINS, BAD_WORDS, MESSAGE_SEND_FREQUENCY
+from db_writer import DataWriter
+
+
+db_info = DataWriter()
+
+
+regex = r'(?:[a-zA-Z0-9](?:[a-zA-Z0-9\-]{,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,6}'
+
+
+async def filter_text(message: str, user_tg_id: int, timestamp: float):
+    # check spam
+    msg_and_time = db_info.get_user_msg(user_tg_id)
+    msg = msg_and_time[0][0]
+    time = msg_and_time[0][1]
+    sec = timestamp - time
+    if sec <= MESSAGE_SEND_FREQUENCY or message.lower() == msg:
+        return True
+
+    # urls filter
+    domains = re.findall(regex, message.lower())
+    if GOOD_DOMAINS:
+        if domains:
+            for domain in GOOD_DOMAINS:
+                if domain in domains:
+                    domains.remove(domain)
+            if domains != []:
+                return True
+
+    # bad words filter
+    for word in BAD_WORDS:
+        if word.lower() in message.lower():
+            return True
+

requirements.txt

@@ -0,0 +1,12 @@
+aiogram==2.14.3
+aiohttp==3.7.4.post0
+async-timeout==3.0.1
+attrs==21.2.0
+Babel==2.9.1
+certifi==2021.5.30
+chardet==4.0.0
+idna==3.2
+multidict==5.1.0
+pytz==2021.1
+typing-extensions==3.10.0.0
+yarl==1.6.3