Spaces:
Build error
Build error
| import asyncio | |
| import logging | |
| import os | |
| import re | |
| from datetime import datetime, timedelta | |
| from telethon import TelegramClient, events | |
| from telethon.tl.types import ChatAdminRights, ChatBannedRights | |
| API_ID = 28810829 | |
| API_HASH = "d3f304bbd0b69b8c30dbec4be5824748" | |
| BOT_TOKEN = os.environ["BOT_TOKEN"] | |
| logging.basicConfig( | |
| level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" | |
| ) | |
| logger = logging.getLogger("TruesAutoBlacklister") | |
| logger.info("Logging has been set up.") | |
| client = TelegramClient("TruesBlacklister", API_ID, API_HASH).start(bot_token=BOT_TOKEN) | |
| logger.info("Initialized Telethon client") | |
| # Define the list of blacklisted words | |
| blacklist = ["counterfeit", "meth", "shard", "heroin", "gbh"] | |
| async def start(event): | |
| await event.respond('I am alive!') | |
| async def help(event): | |
| await event.respond('I can blacklist certain words. Just type them in a chat I am in!') | |
| async def check_blacklist(event): | |
| if any( | |
| re.search(rf"\b{re.escape(word)}\b", event.raw_text, re.IGNORECASE) | |
| for word in blacklist | |
| ): | |
| logger.info("Message contains a blacklisted word.") | |
| user = await event.get_sender() | |
| if user: | |
| logger.info( | |
| f"User id: {user.id}, username: {user.username}" | |
| ) | |
| if not await client.is_admin(event.chat_id, user.id): | |
| try: | |
| await event.delete() | |
| logger.info("User is not an admin. Deleted the message.") | |
| await client.edit_permissions( | |
| entity=event.chat_id, | |
| user_id=user.id, | |
| view_messages=False | |
| ) | |
| except Exception as e: | |
| logger.error("Failed to mute the user. Skipping action.") | |
| buttons = [ | |
| [InlineKeyboardButton("Unmute", callback_data="unmute")], | |
| [InlineKeyboardButton("Ban", callback_data="ban")], | |
| ] | |
| await client.send_message( | |
| entity=event.chat_id, | |
| message=f"{user.username}, Your message has been deleted due to a blacklisted word. Please select an action:", | |
| buttons=buttons | |
| ) | |
| else: | |
| logger.info("User is an admin. Skipping action.") | |
| else: | |
| logger.info("Message does not contain a blacklisted word.") | |
| async def handle_button(event): | |
| user = await event.get_sender() | |
| try: | |
| if event.data.decode("utf-8") == "unmute": | |
| await client.edit_permissions( | |
| entity=event.chat_id, | |
| user_id=user.id, | |
| view_messages=True | |
| ) | |
| await event.respond(f"{user.username} has been unmuted.") | |
| await asyncio.sleep(10) # Pause for 10 seconds | |
| await event.delete() | |
| elif event.data.decode("utf-8") == "ban": | |
| await client.edit_permissions( | |
| entity=event.chat_id, | |
| user_id=user.id, | |
| view_messages=False | |
| ) | |
| await event.respond(f"{user.username} has been banned.") | |
| await asyncio.sleep(10) # Pause for 10 seconds | |
| await event.delete() | |
| except Exception as e: | |
| logger.error(f"An error occurred: {e}") | |
| client.run_until_disconnected() |