from telethon.tl.types import JsonObject, JsonObjectValue, JsonString, JsonNumber, Message from telethon.tl.types import MessageEntityTextUrl from telethon import TelegramClient, errors from urllib.parse import urlparse, parse_qs import asyncio, random, json, os import random, string from typing import Union from utils.captcha import (capslover_create_task, wait_for_captcha, telegram_check_captcha) from utils.logger import logger class Telegram: def __init__(self, phone_number: str, proxy: dict = None) -> None: self.telegram_spam_message = 'Hey, I think my account is limited mistakenly, please check it.' self.telegram_message = 'Hey, how are you today bro ?' self.telegram_username = '@d24680j' self.phone_number = phone_number self.push_token = self._get_push_token() self.installer = 'com.android.vending' self.package_id = 'org.telegram.messenger' self.tz_offset = -3600 # Load JSON data FIRST (BUG FIX) self.json_data = self._load_json(f'valid_sessions/{self.phone_number}.json') # Load proxy list self.proxy_list = self._load_proxies("utils/proxies.txt") self.proxy_index = random.randint(0, len(self.proxy_list) - 1) if self.proxy_list: proxy = self.proxy_list[self.proxy_index] else: proxy = None self.client = TelegramClient( session=f'valid_sessions/{self.phone_number}', api_hash='eb06d4abfb49dc3eeb1aeb98ae0f581e', api_id=6, device_model=self.json_data.get('device_model') or self.json_data.get('device'), system_version=self.json_data.get('system_version') or self.json_data.get('sdk'), app_version=self.json_data.get('app_version'), lang_code=self.json_data.get('lang_code') or 'en', system_lang_code=self.json_data.get('system_lang_code') or 'en-US', proxy=proxy ) # ---------------- JSON LOADER ---------------- def _load_json(self, path: str) -> Union[dict, bool]: try: with open(path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as error: logger.error(f'[-] Telegram._load_json -> {error}') return False # ---------------- PROXY SYSTEM ---------------- def _load_proxies(self, path: str): proxies = [] if not os.path.exists(path): logger.error("Proxy file not found.") return proxies with open(path, "r") as f: for line in f: line = line.strip() if not line: continue host, port, user, password = line.split(":") proxies.append({ "proxy_type": "socks5", "addr": host, "port": int(port), "username": user, "password": password }) logger.info(f"Loaded {len(proxies)} proxies.") return proxies def _switch_proxy(self): if not self.proxy_list: logger.error("No proxies available to switch.") return False # انتخاب پروکسی کاملاً رندومی self.proxy_index = random.randint(0, len(self.proxy_list) - 1) new_proxy = self.proxy_list[self.proxy_index] logger.warning(f"Random proxy selected → {new_proxy['addr']}:{new_proxy['port']}") self.client._proxy = new_proxy return True # ---------------- MISC UTILS ---------------- def _get_push_token(self) -> str: try: with open('utils/push_tokens.txt', 'r', encoding='utf-8') as f: tokens = [l.strip() for l in f if l.strip() and not l.startswith('#')] return random.choice(tokens) if tokens else '' except FileNotFoundError: logger.warning('[-] utils/push_tokens.txt not found.') return '' def _random_text(self) -> str: allowed = string.ascii_lowercase + string.digits text_parts = [] for _ in range(3): part_length = random.randint(5, 10) part = ''.join(random.choice(allowed) for _ in range(part_length)) text_parts.append(part) return ' '.join(text_parts) # ---------------- CONNECTION SYSTEM ---------------- async def _patch_client(self) -> None: try: self.client._init_request.lang_pack = 'android' self.client._init_request.params = JsonObject([ JsonObjectValue('device_token', JsonString(self.push_token)), JsonObjectValue('data', JsonString('49C1522548EBACD46CE322B6FD47F6092BB745D0F88082145CAF35E14DCC38E1')), JsonObjectValue('installer', JsonString(self.installer)), JsonObjectValue('package_id', JsonString(self.package_id)), JsonObjectValue('tz_offset', JsonNumber(self.tz_offset)), JsonObjectValue('perf_cat', JsonNumber(2)), ]) except Exception as error: logger.error(f'[-] Telegram._patch_client -> {error}') async def _connect(self, retries: int = 5, delay: int = 2) -> bool: failure_count = 0 for attempt in range(1, retries + 1): try: logger.info(f"Attempt {attempt}/{retries} connecting...") if self.client.is_connected(): return True await self.client.connect() if self.client.is_connected(): logger.success("Connected successfully.") return True except Exception as e: logger.error(f"Connect error: {type(e).__name__}: {e}") failure_count += 1 # Auto-switch proxy after 3 failures if failure_count >= 3: logger.warning("3 consecutive failures → switching proxy...") if not self._switch_proxy(): return False failure_count = 0 await asyncio.sleep(delay) logger.error("All attempts failed → connection aborted.") return False async def _disconnect(self) -> bool: try: if self.client.is_connected(): await self.client.disconnect() return True except Exception as error: logger.error(f'[-] Telegram._disconnect -> {error}') return False # ---------------- SPAMBOT CHECK ---------------- async def check_spam(self, send_request: bool = False) -> bool: try: await self._patch_client() await self._connect() try: entity = await self.client.get_entity(self.telegram_username) try: await self.client.send_message(entity=entity, message=self.telegram_message) except errors.rpcerrorlist.UserBlockedError: logger.warning(f'[-] You blocked {self.telegram_username} → Unblocking...') await self.client(UnblockRequest(self.telegram_username)) logger.success(f'[+] {self.telegram_username} unblocked successfully.') # دوباره پیام await self.client.send_message(entity=entity, message=self.telegram_message) except Exception as e: if "blocked" in str(e).lower(): logger.warning(f'[-] Block detected → Unblocking {self.telegram_username}...') await self.client(UnblockRequest(self.telegram_username)) logger.success(f'[+] {self.telegram_username} unblocked successfully.') await self.client.send_message(entity=entity, message=self.telegram_message) else: raise e # بقیه ارورها return True except errors.rpcerrorlist.PeerFloodError: if not send_request: return False # === SPAMBOT FLOW بدون تغییر === async with self.client.conversation('@SpamBot', timeout=100) as conv: await conv.send_message('/start') response = (await conv.get_response()).message if 'Good news, no limits' in response or 'مژده' in response: return True if 'Unfortunately,' in response: await conv.send_message('This is a mistake') response = (await conv.get_response()).message if 'If you think the limitations' in response: await conv.send_message('Yes') response = (await conv.get_response()).message if 'Great! Please confirm' in response: await conv.send_message('No! Never did that!') response = await conv.get_response() if 'Please verify you are a human' in response.message: url, scope, actor = None, None, None for entity in getattr(response, 'entities', []) or []: if isinstance(entity, MessageEntityTextUrl): url = entity.url result = parse_qs(urlparse(url).query) scope = result.get('scope', [None])[0] actor = result.get('actor', [None])[0] break if not url: return False create_task = await capslover_create_task(url) if not create_task: return False task_id = create_task.get('taskId') if not task_id: return False captcha_token = await wait_for_captcha(task_id, timeout=100) if not captcha_token: return False await telegram_check_captcha(captcha_token, actor, scope) await asyncio.sleep(random.randint(1, 3)) await conv.send_message('Done') response = (await conv.get_response()).message if 'Great! I’m very sorry' in response: random_text = self._random_text() await conv.send_message(random_text) response = (await conv.get_response()).message if 'Thank you! Your complaint has been successfully submitted.' in response: return True except Exception as error: logger.error(f'[-] Telegram.check_spam -> {error}') if os.path.exists(f'valid_sessions/{self.phone_number}.session'): os.rename(f'valid_sessions/{self.phone_number}.session', f'invalid_sessions/{self.phone_number}.session') if os.path.exists(f'valid_sessions/{self.phone_number}.json'): os.rename(f'valid_sessions/{self.phone_number}.json', f'invalid_sessions/{self.phone_number}.json') return False finally: await self._disconnect()