Spaces:
Running
Running
import asyncio | |
import logging | |
import os | |
import re | |
from datetime import datetime, timedelta | |
from telethon import TelegramClient, events | |
from telethon.tl.custom import Button | |
from telethon.tl.types import ChatAdminRights, ChatBannedRights, ChannelParticipantsAdmins | |
import re | |
from telethon import events | |
from telethon.tl.functions.channels import EditBannedRequest | |
from telethon.tl.types import ChatBannedRights | |
from telethon.tl.types import MessageEntityMentionName | |
API_ID = os.getenv("API_ID") | |
API_HASH = os.getenv("API_HASH") | |
BOT_TOKEN = os.getenv("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") | |
blacklist = ["counterfeit", "meth", "shard", "heroin", "ghb"] | |
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}") | |
# Check if the user is an admin | |
chat_admins = await client.get_participants(event.chat_id, filter=ChannelParticipantsAdmins) | |
is_admin = any(admin.id == user.id for admin in chat_admins) | |
if not is_admin: | |
try: | |
await event.delete() | |
logger.info("User is not an admin. Deleted the message.") | |
# Mute the user by revoking their ability to send messages | |
await event.client.edit_permissions( | |
event.chat_id, | |
user.id, | |
send_messages=False | |
) | |
except Exception as e: | |
logger.error(f"Failed to mute the user: {e}") | |
buttons = [ | |
[Button.inline("Unmute", b"unmute")], | |
[Button.inline("Ban", b"ban")], | |
] | |
await event.respond( | |
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 unmute_user(event): | |
user_id = event.data_match.group(1) | |
try: | |
await event.client.edit_permissions( | |
event.chat_id, | |
entity=int(user_id), | |
send_messages=True | |
) | |
await event.respond("User has been unmuted.") | |
except Exception as e: | |
await event.respond(f"Failed to unmute user: {e}") | |
async def ban_user(event): | |
user_id = event.data_match.group(1) | |
try: | |
await event.client.edit_permissions( | |
event.chat_id, | |
entity=int(user_id), | |
send_messages=False | |
) | |
await event.respond("User has been banned.") | |
except Exception as e: | |
await event.respond(f"Failed to ban user: {e}") | |
# Register the check_blacklist function to listen for new messages | |
client.add_event_handler(check_blacklist, events.NewMessage(incoming=True)) | |
client.run_until_disconnected() |