Spaces:
Running
Running
File size: 3,885 Bytes
9513370 ecf02cb 9513370 070fe77 9513370 7eff50c 9513370 8b0e484 9513370 8b0e484 9513370 8b0e484 9513370 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
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, ChannelParticipantsAdmins
from telethon.tl.custom import InlineKeyboardButton
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")
# Define the list of blacklisted words
blacklist = ["counterfeit", "meth", "shard", "heroin", "ghb"]
@client.on(events.NewMessage(pattern='/start'))
async def start(event):
await event.respond('I am alive!')
@client.on(events.NewMessage(pattern='/help'))
async def help(event):
await event.respond('I can blacklist certain words. Just type them in a chat I am in!')
@client.on(events.NewMessage(incoming=True))
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}"
)
# Get the chat admin permissions
try:
chat_admins = await client.get_participants(event.chat_id, filter=ChannelParticipantsAdmins)
is_admin = any(admin.id == user.id for admin in chat_admins)
except Exception as e:
logger.error(f"Failed to check user admin status: {e}")
is_admin = False
if not is_admin:
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.")
@client.on(events.CallbackQuery())
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() |