import os import time import random import string import asyncio from datetime import datetime, timedelta from collections import namedtuple import aiohttp from dotenv import load_dotenv from telethon import TelegramClient, events from telethon.utils import get_peer_id, get_display_name from telethon.tl.functions.channels import EditBannedRequest from telethon.tl.types import ChatBannedRights from captcha.image import ImageCaptcha def get_user_name(user): full_name = get_display_name(user) if not full_name: full_name = user.username if not full_name: full_name = "user" return full_name def get_link_to_user(user): full_name = get_user_name(user) if user.username: return f"[{full_name}](@{user.username})" return f"[{full_name}](tg://user?id={user.id})" load_dotenv() API_ID = int(os.getenv("API_ID")) API_HASH = os.getenv("API_HASH") BOT_TOKEN = os.getenv("BOT_TOKEN") ALLOWED_GROUPS = set( map(lambda cid: int("-100" + cid), os.getenv("ALLOWED_GROUPS").split(",")) ) CAPTCHA_WAIT_TIME = int(os.getenv("CAPTCHA_WAIT_TIME", 80)) MAX_CAPTCHA_ATTEMPTS = int(os.getenv("MAX_CAPTCHA_ATTEMPTS", 8)) USE_LOLS_API = os.getenv("USE_LOLS_API", "yes") == "yes" bot = TelegramClient("captcha_bot", API_ID, API_HASH).start(bot_token=BOT_TOKEN) pending = {} async def lols_check(user_id): try: async with aiohttp.ClientSession() as session: async with session.get(f"https://api.lols.bot/account?id={user_id}") as response: data = await response.json() return data["banned"] or data["scammer"] except Exception: return False class PendingUser: def __init__(self, peer_id, user_id): self.peer_id = peer_id self.user_id = user_id self.text = None self.message_id = None self.captcha = None self.captcha_im = None self.join_ts = time.time() self.retries = 0 @property def expired(self): return time.time() - self.join_ts >= CAPTCHA_WAIT_TIME async def start(self, event): if USE_LOLS_API: if not await lols_check(self.user_id): await self.kick(forever=True) return await self.update_captcha() self.text = f"Welcome, {get_link_to_user(event.user)}! Please write a message with the **four latin letters (case insensitive)** that appear in this image to verify that you are a human. If you don't solve the captcha in **{CAPTCHA_WAIT_TIME}** seconds, you will be automatically kicked out of the group." message = await bot.send_message( event.chat_id, self.text, file=self.captcha_im, ) self.message_id = message async def update_captcha(self, only_text=False): if not only_text: self.captcha = "".join(random.sample(string.ascii_uppercase, 4)) self.captcha_im = ImageCaptcha().generate(self.captcha) self.captcha_im.name = "captcha.png" if self.message_id is not None: try: attempts = MAX_CAPTCHA_ATTEMPTS - self.retries message = await bot.edit_message( self.peer_id, self.message_id, text=self.text + f"\n\nYou have **{attempts}** attempt{'' if attempts == 1 else 's'} left.", file=None if only_text else self.captcha_im, ) self.message_id = message.id except Exception: pass async def check_captcha(self, captcha): captcha = captcha.strip().upper() if captcha == self.captcha: del pending[self.user_id] try: await bot.delete_messages(self.peer_id, self.message_id) except Exception: pass return True self.retries += 1 if self.retries > MAX_CAPTCHA_ATTEMPTS: await self.kick() else: await self.update_captcha( only_text=not ( len(captcha) == 4 and all(map(lambda c: c in string.ascii_uppercase, captcha)) ) ) return False async def remove(self): del pending[self.user_id] if self.message_id is not None: try: await bot.delete_messages(self.peer_id, self.message_id) except Exception: pass async def kick(self, forever=False): if self.user_id not in pending: return await self.remove() try: await bot( EditBannedRequest( self.peer_id, self.user_id, ChatBannedRights( until_date=None if forever else datetime.now() + timedelta(minutes=5), view_messages=True, ), ) ) except Exception: pass @bot.on(events.ChatAction) async def handler(event): if event.chat_id not in ALLOWED_GROUPS: return if (event.user_left or event.user_kicked) and event.user_id in pending: await pending[event.user_id].remove() return if not event.user_joined: return pending[event.user_id] = PendingUser(peer_id=event.chat_id, user_id=event.user_id) await pending[event.user_id].start(event) @bot.on(events.NewMessage) async def handler(event): peer_id = get_peer_id(event.peer_id) if peer_id not in ALLOWED_GROUPS: return pending_user = pending.get(event.sender.id) if not pending_user: return try: await event.delete() except Exception: pass text = getattr(event, "text", "") await pending_user.check_captcha(text) async def check_pending(): while True: for user_id in dict(pending): pending_user = pending[user_id] if pending_user.expired: await pending_user.kick() await asyncio.sleep(1) if __name__ == "__main__": bot.loop.create_task(check_pending()) bot.start() bot.run_until_disconnected()