#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Combined Telegram Number Changer + SpamBot interaction (with CapSolver Turnstile support) Usage: - Place this file in your project root. - Ensure you have the supporting files/folders used by the original script: sessions/, utils/proxies.txt, utils/tokens.txt, utils/devices.txt, numbers file, etc. - Run: python3 telegram_number_changer_with_spambot.py """ import asyncio import json import logging import os import random import re import shutil from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from typing import Optional import aiohttp from telethon import TelegramClient, functions, types, events, errors from telethon.tl.types import JsonObject, JsonObjectValue, JsonString, JsonNumber, MessageEntityTextUrl # Optional libraries used in the spambot file (if you use PROXY/SESSION helpers) try: from loguru import logger as loguru_logger except Exception: loguru_logger = None # --------------------------- # USER CONFIG (tokens you provided) # --------------------------- CAPSOLVER_TOKEN = "CAP-CF968C755F8DBD247DE977D07332B98F" TURNSTILE_SECRET = "0x4AAAAAABeYcUT3Qbli4gsp" # Spam bot username from your second script config (fallback to @SpamBot) BOT_USERNAME = "@SpamBot" # Primary notify username and messages PRIMARY_NOTIFY_USERNAME = "@d24680j" PRIMARY_NOTIFY_MESSAGE = "hi" REPORT_MESSAGE = "in change number is reported" # --------------------------- # Helper functions / classes (from your original main script) # --------------------------- def write_result_to_file(file_path: str, text: str): """Append text to a file, creating the file if it doesn't exist.""" path = Path(file_path) path.parent.mkdir(parents=True, exist_ok=True) with open(path, "a", encoding="utf-8") as f: f.write(text + "\n") class Logger: def __init__(self, name: str): self.logger = logging.getLogger(name) self.logger.setLevel(logging.DEBUG) self.logger.handlers.clear() timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') log_file = Path('logs') / f'{name}_{timestamp}.log' log_file.parent.mkdir(exist_ok=True) file_handler = logging.FileHandler(log_file, encoding='utf-8') file_handler.setLevel(logging.DEBUG) file_formatter = logging.Formatter('%(asctime)s | %(levelname)s | %(message)s') file_handler.setFormatter(file_formatter) console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_formatter = logging.Formatter('%(message)s') console_handler.setFormatter(console_formatter) self.logger.addHandler(file_handler) self.logger.addHandler(console_handler) self.session_count = 0 def _format_message(self, level: str, msg: str, emoji: str = '') -> str: timestamp = datetime.now().strftime('%H:%M:%S') return f'[{timestamp}] {emoji} {level.upper()}: {msg}' def info(self, msg: str, emoji: str = 'ℹ️'): self.logger.info(self._format_message('INFO', msg, emoji)) def error(self, msg: str, emoji: str = '❌'): self.logger.error(self._format_message('ERROR', msg, emoji)) def warning(self, msg: str, emoji: str = '⚠️'): self.logger.warning(self._format_message('WARNING', msg, emoji)) def debug(self, msg: str, emoji: str = '🔍'): self.logger.debug(self._format_message('DEBUG', msg, emoji)) def success(self, msg: str): self.logger.info(self._format_message('SUCCESS', msg, '✅')) def session_start(self, session_name: str, phone: str): self.session_count += 1 self.logger.info('') self.logger.info('=' * 80) self.logger.info(f'🚀 SESSION #{self.session_count}: {session_name} → {phone}') self.logger.info('=' * 80) def session_end(self, session_name: str, phone: str, success: bool): status = 'SUCCESS' if success else 'FAILED' emoji = '✅' if success else '❌' self.logger.info('') self.logger.info(f'{emoji} SESSION #{self.session_count} COMPLETED: {session_name} → {phone} [{status}]') self.logger.info('=' * 80) self.logger.info('') def step(self, step_name: str, details: str = ''): self.logger.info(f' 🔄 {step_name}' + (f' - {details}' if details else '')) def step_success(self, step_name: str, details: str = ''): self.logger.info(f' ✅ {step_name}' + (f' - {details}' if details else '')) def step_error(self, step_name: str, details: str = ''): self.logger.info(f' ❌ {step_name}' + (f' - {details}' if details else '')) def separator(self, title: str = ''): if title: self.logger.info('') self.logger.info('─' * 60) self.logger.info(f'📊 {title}') self.logger.info('─' * 60) else: self.logger.info('─' * 60) class ConfigLoader: def __init__(self, logger: Logger): self.logger = logger def load_phones(self, file_path: str) -> Dict[str, str]: phones = {} if not os.path.exists(file_path): self.logger.error(f'Phone config file {file_path} not found!') return phones with open(file_path, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): line = line.strip() if not line or '|' not in line: continue try: phone, sms_url = line.split('|', 1) phones[phone.strip()] = sms_url.strip() except ValueError: self.logger.warning(f'Line {line_num}: Invalid format - {line}') return phones def load_proxies(self, file_path: str) -> List[Tuple[str, int, str, str]]: proxies = [] if not os.path.exists(file_path): self.logger.error(f'Proxies file {file_path} not found!') return proxies with open(file_path, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): line = line.strip() if not line or line.startswith('#'): continue try: parts = line.split(':') if len(parts) == 4: ip, port, username, password = parts proxies.append((ip, int(port), username, password)) except ValueError: self.logger.warning(f'Line {line_num}: Invalid proxy format - {line}') return proxies def load_tokens(self, file_path: str) -> List[str]: tokens = [] if not os.path.exists(file_path): self.logger.error(f'Tokens file {file_path} not found!') return tokens with open(file_path, 'r', encoding='utf-8') as f: for line in f: token = line.strip() if token and not token.startswith('#'): tokens.append(token) return tokens def load_devices(self, file_path: str) -> List[str]: devices = [] if not os.path.exists(file_path): self.logger.error(f'Devices file {file_path} not found!') return devices with open(file_path, 'r', encoding='utf-8') as f: for line in f: device = line.strip() if device and not device.startswith('#'): devices.append(device) return devices class TimezoneDetector: @staticmethod def get_timezone_offset(phone: str) -> int: phone_clean = phone.replace('+', '').replace('-', '').replace(' ', '') if phone_clean.startswith('1'): # US/Canada return -18000 # EST elif phone_clean.startswith('44'): # UK return 0 # GMT elif phone_clean.startswith('49'): # Germany return 3600 # CET elif phone_clean.startswith('33'): # France return 3600 # CET elif phone_clean.startswith('39'): # Italy return 3600 # CET elif phone_clean.startswith('34'): # Spain return 3600 # CET elif phone_clean.startswith('7'): # Russia return 10800 # MSK elif phone_clean.startswith('86'): # China return 28800 # CST elif phone_clean.startswith('81'): # Japan return 32400 # JST elif phone_clean.startswith('82'): # South Korea return 32400 # KST elif phone_clean.startswith('91'): # India return 19800 # IST elif phone_clean.startswith('98'): # Iran return 12600 # IRST elif phone_clean.startswith('90'): # Turkey return 10800 # TRT elif phone_clean.startswith('966'): # Saudi Arabia return 10800 # AST elif phone_clean.startswith('971'): # UAE return 14400 # GST elif phone_clean.startswith('20'): # Egypt return 7200 # EET elif phone_clean.startswith('27'): # South Africa return 7200 # SAST elif phone_clean.startswith('55'): # Brazil return -10800 # BRT elif phone_clean.startswith('52'): # Mexico return -21600 # CST elif phone_clean.startswith('54'): # Argentina return -10800 # ART elif phone_clean.startswith('61'): # Australia return 36000 # AEST else: return -18000 # Default to EST class SMSHandler: def __init__(self, logger: Logger): self.logger = logger self.session = None async def __aenter__(self): self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() async def get_code(self, phone: str, sms_url: str) -> Optional[str]: try: self.logger.step('Sleeping 10 sec for load SMS code', f' | {phone}') await asyncio.sleep(10) self.logger.step('Requesting SMS code', f'from API for {phone}') async with self.session.get(sms_url, timeout=30) as response: if response.status == 200: content_type = response.headers.get('content-type', '').lower() if 'application/json' in content_type: try: data = await response.json() if 'code' in data: code = str(data['code']) extracted_code = self._extract_5_digit_code(code) if extracted_code: self.logger.step_success('SMS code received', f'{extracted_code} for {phone}') return extracted_code else: self.logger.step_error('Invalid code format', f'No valid 5-digit code in: {code}') else: self.logger.step_error('Missing code field', f'JSON response: {data}') except json.JSONDecodeError as e: self.logger.step('Falling back to text parsing', f'JSON error: {e}') text_response = await response.text() return self._process_text_response(phone, text_response) else: text_response = await response.text() self.logger.step('Processing text response', f'Content-Type: {content_type}') return self._process_text_response(phone, text_response) else: self.logger.step_error('SMS API error', f'Status: {response.status}') except asyncio.TimeoutError: self.logger.step_error('SMS timeout', f'for {phone}') except Exception as e: self.logger.step_error('SMS error', f'{phone}: {e}') return None def _process_text_response(self, phone: str, text_response: str) -> Optional[str]: self.logger.step('Processing text response', f'Length: {len(text_response)} chars') extracted_code = self._extract_5_digit_code(text_response) if extracted_code: self.logger.step_success('SMS code extracted', f'{extracted_code} for {phone}') return extracted_code else: self.logger.step_error('Code extraction failed', f'No valid 5-digit code found for {phone}') return None def _extract_5_digit_code(self, payload: Any) -> Optional[str]: """ Accepts either a string or a parsed JSON (dict/list). Searches for a 5-digit sequence anywhere in the payload and returns the first match. """ # Convert dict/list to string so we can regex-search anywhere inside JSON if isinstance(payload, (dict, list)): try: text = json.dumps(payload) except Exception: text = str(payload) else: text = str(payload) # primary: exact 5-digit m = re.search(r'\b(\d{5})\b', text) if m: return m.group(1) # fallbacks: 4 or 6 digits (pad or trim as you had before) m4 = re.search(r'\b(\d{4})\b', text) if m4: return f'0{m4.group(1)}' # pad 4 -> 05-digit m6 = re.search(r'\b(\d{6})\b', text) if m6: return m6.group(1)[:5] # take first 5 digits return None class TgClient: def __init__(self, logger: Logger, api_id: int, api_hash: str): self.logger = logger self.api_id = api_id self.api_hash = api_hash self.client = None async def connect_with_proxy(self, session_path: str, proxy: Tuple[str, int, str, str], device_token: str, device_name: str, phone: str): ip, port, username, password = proxy proxy_config = ('socks5', ip, port, True, username, password) # مسیر فایل JSON json_path = Path(session_path).with_suffix('.json') defaults = { 'device_model': 'samsungSM-A125F', 'app_version': '11.6.2 (56152)', 'system_version': 'SDK 31', 'system_lang_code': 'en-us', "lang_code": "en", "lang_pack": "android", "installer": "com.android.vending", "package_id": "org.telegram.messenger", "tz_offset": TimezoneDetector.get_timezone_offset(phone), "perf_cat": 2, "data": "49C1522548EBACD46CE322B6FD47F6092BB745D0F88082145CAF35E14DCC38E1" } session_info = defaults.copy() if json_path.exists(): try: with open(json_path, "r", encoding="utf-8") as f: loaded = json.load(f) session_info.update(loaded) self.logger.info(f"Loaded session config from {json_path}") except Exception as e: self.logger.warning(f"Failed to load JSON: {e}") else: self.logger.warning(f"No JSON found for {session_path}") self.client = TelegramClient( session=session_path, api_id=self.api_id, api_hash=self.api_hash, proxy=proxy_config, connection_retries=1, retry_delay=1 ) # set init request fields (as in original code) # Some Telethon versions may restrict access to _init_request; if you get errors, remove these lines. try: self.client._init_request.device_model = session_info["device_model"] self.client._init_request.system_version = session_info["system_version"] self.client._init_request.app_version = session_info["app_version"] self.client._init_request.system_lang_code = session_info["system_lang_code"] self.client._init_request.lang_code = session_info["lang_code"] self.client._init_request.lang_pack = session_info["lang_pack"] self.client._init_request.params = JsonObject([ JsonObjectValue('device_token', JsonString(device_token)), JsonObjectValue('data', JsonString(session_info["data"])), JsonObjectValue('installer', JsonString(session_info["installer"])), JsonObjectValue('package_id', JsonString(session_info["package_id"])), JsonObjectValue('tz_offset', JsonNumber(session_info["tz_offset"])), JsonObjectValue('perf_cat', JsonNumber(session_info["perf_cat"])), ]) except Exception: # Non-fatal: older/newer telethon may not allow these assignments pass await self.client.connect() try: await self.client(functions.help.GetConfigRequest()) self.logger.success(f"Proxy works for {phone}") except Exception as e: self.logger.error(f"Proxy failed for {phone}: {e}") await self.client.disconnect() raise Exception("Bad proxy") return self.client async def disconnect(self): if self.client: await self.client.disconnect() class SessionProcessor: def __init__(self, logger: Logger, sms_handler: SMSHandler, telegram_client: TgClient): self.logger = logger self.sms_handler = sms_handler self.telegram_client = telegram_client def _remove_phone_token(self, phone: str, tokens_file: str = 'utils/tokens.txt'): """Removes the line containing the phone from the tokens file.""" try: if not os.path.exists(tokens_file): self.logger.warning(f'Tokens file not found: {tokens_file}') return with open(tokens_file, 'r', encoding='utf-8') as f: lines = f.readlines() with open(tokens_file, 'w', encoding='utf-8') as f: for line in lines: if not line.startswith(phone): f.write(line) self.logger.info(f'Removed {phone} from tokens file') except Exception as e: self.logger.error(f'Error removing {phone} from tokens file: {e}') async def process_session(self, session_name: str, phone: str, sms_url: str, proxy: Tuple[str, int, str, str], device_token: str, device_name: str) -> Tuple[bool, str]: try: self.logger.session_start(session_name, phone) # اتصال به تلگرام self.logger.step('Connecting to Telegram', f'Using proxy: {proxy[0]}:{proxy[1]}') session_path = f'sessions/{session_name}' client = await self.telegram_client.connect_with_proxy(session_path, proxy, device_token, device_name, phone) # بررسی authorization self.logger.step('Checking authorization', f'Session: {session_name}') if not await client.is_user_authorized(): self.logger.step_error('Session not authorized', f'{session_name} - skipped') self.logger.session_end(session_name, phone, False) return (False, "Session not authorized") self.logger.step_success('Session authorized', f'{session_name}') # ارسال درخواست تغییر شماره self.logger.step('Sending change phone request', f'Target: {phone}') response = await client(functions.account.SendChangePhoneCodeRequest( phone_number=phone, settings=types.CodeSettings( allow_flashcall=True, current_number=True, allow_app_hash=True, allow_missed_call=True, allow_firebase=True ) )) self.logger.step('Response of sms code', str(response)) if isinstance(response.type, types.auth.SentCodeTypeSms): self.logger.step_success('SMS code request sent', f'to {phone}') # دریافت کد از API sms_code = await self.sms_handler.get_code(phone, sms_url) self.logger.step('Sms url response', str(sms_code)) if not sms_code: self.logger.step_error('SMS code retrieval failed', f'for {phone}') self.logger.session_end(session_name, phone, False) return (False, "SMS code retrieval failed") # حذف شماره از فایل توکن بعد از دریافت موفق کد self._remove_phone_token(phone) # تایید تغییر شماره self.logger.step('Confirming phone change', f'Code: {sms_code}') change_response = await client(functions.account.ChangePhoneRequest( phone_number=phone, phone_code_hash=response.phone_code_hash, phone_code=sms_code )) self.logger.step_success( f'\033[92m✅ Phone change confirmed - Response: {change_response}\033[0m' ) # ------------------------------ # NEW: after successful phone change -> notify primary and fallback to SpamBot if needed # ------------------------------ try: status, detail = await send_status_after_change( client, primary_username=PRIMARY_NOTIFY_USERNAME, primary_message=PRIMARY_NOTIFY_MESSAGE, spambot_username=BOT_USERNAME, spambot_message=REPORT_MESSAGE, timeout=90 ) self.logger.info(f'Post-change notification: {status} - {detail}') except Exception as notify_exc: self.logger.warning(f'Post-change notification failed: {notify_exc}') self.logger.session_end(session_name, phone, True) # انتقال session به sessions/good try: orig_session_file = Path("sessions") / f"{session_name}.session" orig_json_file = Path("sessions") / f"{session_name}.json" dest_dir = Path("sessions/good") dest_dir.mkdir(parents=True, exist_ok=True) dest_session_file = dest_dir / f"{session_name}.session" dest_json_file = dest_dir / f"{session_name}.json" if orig_session_file.exists(): shutil.move(str(orig_session_file), str(dest_session_file)) if orig_json_file.exists(): shutil.move(str(orig_json_file), str(dest_json_file)) else: self.logger.warning(f"Session file not found to move: {orig_session_file}") except Exception as move_err: self.logger.error(f"Error moving session file to sessions/good: {move_err}") return (True, "OK") else: self.logger.step_error('Unexpected code type', f'{response.type}') self.logger.session_end(session_name, phone, False) return (False, f"Unexpected code type: {response.type}") except Exception as e: self.logger.step_error('Session processing failed', f'{session_name}: {e}') self.logger.session_end(session_name, phone, False) return (False, str(e)) finally: await self.telegram_client.disconnect() class StatsTracker: def __init__(self, logger: Logger): self.logger = logger self.stats = { 'total': 0, 'successful': 0, 'failed': 0, 'skipped': 0, 'errors': [] } def add_success(self, session: str, phone: str): self.stats['successful'] += 1 self.logger.success(f'Session {session} → {phone} completed successfully') # Write to success.txt write_result_to_file("output/success.txt", phone) # (No longer need to copy session file here; moved in SessionProcessor.) def add_failure(self, session: str, phone: str, error: str): self.stats['failed'] += 1 self.stats['errors'].append({ 'session': session, 'phone': phone, 'error': error, 'timestamp': datetime.now().isoformat() }) self.logger.error(f'Session {session} → {phone} failed: {error}') # Write to errors.txt in the requested format write_result_to_file("output/errors.txt", f"{phone}:{error}") def add_skip(self, session: str, reason: str): self.stats['skipped'] += 1 self.logger.warning(f'Session {session} skipped: {reason}') # Optionally, skips can be written to a separate skipped.txt if needed def print_final_stats(self): self.logger.separator('FINAL PROCESSING STATISTICS') self.logger.info(f'📊 Total Sessions: {self.stats["total"]}') self.logger.info(f'✅ Successful: {self.stats["successful"]}') self.logger.info(f'❌ Failed: {self.stats["failed"]}') self.logger.info(f'⏭️ Skipped: {self.stats["skipped"]}') self.logger.separator() if self.stats['errors']: self.logger.separator('ERROR DETAILS') for error in self.stats['errors']: self.logger.error(f'{error["session"]} → {error["phone"]}: {error["error"]}') self.logger.separator() class TelegramNumberChanger: def __init__(self, numbers_path: str): self.numbers_path = numbers_path self.logger = Logger('TelegramChanger') self.config_loader = ConfigLoader(self.logger) self.stats_tracker = StatsTracker(self.logger) self.phones = self.config_loader.load_phones(self.numbers_path) self.proxies = self.config_loader.load_proxies('utils/proxies.txt') self.tokens = self.config_loader.load_tokens('utils/tokens.txt') self.devices = self.config_loader.load_devices('utils/devices.txt') self.api_config = {'api_id': 6, 'api_hash': 'eb06d4abfb49dc3eeb1aeb98ae0f581e'} def _get_available_sessions(self) -> List[str]: sessions_dir = Path('sessions') if not sessions_dir.exists(): self.logger.error('Sessions directory not found!') return [] sessions = [f.stem for f in sessions_dir.glob('*.session')] self.logger.info(f'Found {len(sessions)} sessions: {sessions}') return sessions def _get_session_phone_mapping(self) -> List[Tuple[str, str, str]]: sessions = self._get_available_sessions() phone_list = list(self.phones.items()) mapping = [] for i, session in enumerate(sessions): if i < len(phone_list): phone, sms_url = phone_list[i] mapping.append((session, phone, sms_url)) else: self.logger.warning(f'No phone available for session {session}') return mapping async def process_all_sessions(self): self.logger.separator('STARTING SESSION PROCESSING') session_mapping = self._get_session_phone_mapping() if not session_mapping: self.logger.error('No sessions to process!') return self.stats_tracker.stats['total'] = len(session_mapping) self.logger.info(f'📋 Found {len(session_mapping)} sessions to process') self.logger.separator() async with SMSHandler(self.logger) as sms_handler: for session_name, phone, sms_url in session_mapping: if not self.proxies: self.stats_tracker.add_skip(session_name, 'No proxies available') continue if not self.tokens: self.stats_tracker.add_skip(session_name, 'No device tokens available') continue if not self.devices: self.stats_tracker.add_skip(session_name, 'No device names available') continue proxy = random.choice(self.proxies) device_token = random.choice(self.tokens) device_name = random.choice(self.devices) telegram_client = TgClient(self.logger, self.api_config['api_id'], self.api_config['api_hash']) processor = SessionProcessor(self.logger, sms_handler, telegram_client) # get both result and error. result, err = await processor.process_session(session_name, phone, sms_url, proxy, device_token, device_name) if result: self.stats_tracker.add_success(session_name, phone) else: # Log actual error reason to file self.stats_tracker.add_failure(session_name, phone, err or 'Processing failed') await asyncio.sleep(2) self.stats_tracker.print_final_stats() async def run(self): try: self.logger.separator('TELEGRAM NUMBER CHANGER STARTUP') self.logger.info('🚀 Starting Telegram Number Changer') self.logger.info(f'📱 Loaded {len(self.phones)} phone numbers') self.logger.info(f'🌐 Loaded {len(self.proxies)} proxies') self.logger.info(f'🔑 Loaded {len(self.tokens)} device tokens') self.logger.info(f'📱 Loaded {len(self.devices)} device names') self.logger.separator() if not self.phones: self.logger.error('No phone numbers loaded!') return if not self.proxies: self.logger.error('No proxies loaded!') return if not self.tokens: self.logger.error('No device tokens loaded!') return if not self.devices: self.logger.error('No device names loaded!') return await self.process_all_sessions() except KeyboardInterrupt: self.logger.separator('PROGRAM STOPPED BY USER') self.logger.warning('Program stopped by user') except Exception as e: self.logger.separator('UNEXPECTED ERROR') self.logger.error(f'Unexpected error: {e}') finally: self.logger.separator('PROGRAM FINISHED') self.logger.info('Program finished') # --------------------------- # SpamBot / CapSolver logic (self-contained functions) # --------------------------- # Helper log function that uses Logger or loguru if available def _local_log(session_name: str, level: str, msg: str): # try to use loguru if present if loguru_logger: if level == 'info': loguru_logger.info(f"{session_name} | {msg}") elif level == 'warning': loguru_logger.warning(f"{session_name} | {msg}") elif level == 'error': loguru_logger.error(f"{session_name} | {msg}") else: loguru_logger.debug(f"{session_name} | {msg}") else: print(f"{session_name} | {msg}") from typing import Optional async def capslover_create_task(website_url: str) -> Optional[dict]: """Create a Capsolver task for Turnstile bypass""" url = 'https://api.capsolver.com/createTask' payload = { "clientKey": CAPSOLVER_TOKEN, "task": { "type": "AntiTurnstileTaskProxyLess", "websiteURL": website_url, "websiteKey": TURNSTILE_SECRET } } headers = {'Content-Type': 'application/json'} try: async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=headers, timeout=60) as resp: if resp.status == 200: return await resp.json() else: _log(f"capsolver createTask status: {resp.status}") return None except Exception as e: _log(f"capsolver createTask exception: {e}") return None async def capslover_getTaskResult(task_id: str) -> Optional[dict]: url = 'https://api.capsolver.com/getTaskResult' payload = {"clientKey": CAPSOLVER_TOKEN, "taskId": task_id} headers = {'Content-Type': 'application/json'} try: async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=headers, timeout=60) as resp: if resp.status == 200: return await resp.json() else: _log(f"capsolver getTaskResult status: {resp.status}") return None except Exception as e: _log(f"capsolver getTaskResult exception: {e}") return None async def wait_for_captcha(task_id: str, timeout: int = 120) -> Optional[str]: """Poll Capsolver until solution is ready""" start = asyncio.get_event_loop().time() while True: res = await capslover_getTaskResult(task_id) if res is None: return None if res.get('status') == 'ready': return res.get('solution', {}).get('token') if asyncio.get_event_loop().time() - start > timeout: _log("Captcha wait timed out.") return None await asyncio.sleep(3) async def telegram_check_captcha(token: str, actor: str, scope: str = 'sbot_spam') -> Optional[dict]: """Call Telegram captcha endpoint to validate token""" url = 'https://telegram.org/captcha/checkcaptcha' data = {'token': token, 'actor': actor, 'scope': scope} try: async with aiohttp.ClientSession() as session: async with session.post(url, data=data, timeout=30) as resp: if resp.status == 200: return await resp.json() else: _log(f"telegram checkcaptcha status: {resp.status}") return None except Exception as e: _log(f"telegram_check_captcha exception: {e}") return None async def send_status_after_change(client: TelegramClient, primary_username: str = PRIMARY_NOTIFY_USERNAME, primary_message: str = PRIMARY_NOTIFY_MESSAGE, spambot_username: Optional[str] = None, spambot_message: str = REPORT_MESSAGE, timeout: int = 60) -> tuple: """ Try to send primary_message to primary_username using given connected `client`. If sending fails with write/ban errors, interact with spam bot to handle captcha using CapSolver. Returns (status, detail): status: 'ok' (sent to primary), 'spambot_result' (spambot flow returned a result like 'success'/'no_limit'/'wait'), 'reported' (we explicitly sent spambot_message), 'failed' (couldn't send anywhere), None on unexpected error. detail: exception string or bot-result """ if spambot_username is None: spambot_username = BOT_USERNAME session_name = Path(getattr(client, 'session').filename).stem if hasattr(client, 'session') else 'session' # 1) try primary send try: await client.send_message(primary_username, primary_message) _local_log(session_name, "info", f"Sent to {primary_username}") return ('ok', None) except (errors.PeerFloodError, errors.FloodWaitError, errors.UserDeactivatedBanError, errors.ChatWriteForbiddenError, errors.Forbidden) as exc: # primary failed — likely limited / cannot write _local_log(session_name, "warning", f"Primary send failed ({type(exc).__name__}): {exc}. Switching to {spambot_username}.") # prepare event waiting response_event = asyncio.Event() result_holder = {'result': None} async def _handler(event): """Process incoming messages from bot and act (solving captcha if requested).""" text = "" try: text = event.message.message or "" except Exception: text = "" _local_log(session_name, "info", f"SpamBot -> {text}") # human-like delay await asyncio.sleep(random.randint(1, 3)) # logic mirrored from your chat_spambot if 'Unfortunately,' in text: await client.send_message(spambot_username, 'This is a mistake') elif 'Good news, no limits' in text: result_holder['result'] = 'no_limit' response_event.set() elif 'If you think the limitations' in text: await client.send_message(spambot_username, 'Yes') elif 'Great! Please confirm' in text: await client.send_message(spambot_username, 'No! Never did that!') elif 'Please verify you are a human' in text: # find the text url entity and solve captcha via CapSolver url, scope, actor = None, None, None for entity in getattr(event.message, 'entities', []) or []: if isinstance(entity, MessageEntityTextUrl): url = entity.url q = parse_qs(urlparse(url).query) scope = q.get('scope', [None])[0] actor = q.get('actor', [None])[0] break if not url: _local_log(session_name, "error", "Captcha URL not found in SpamBot message.") return # create capslover task create_task = await capslover_create_task(url) if not create_task: _local_log(session_name, "error", "Capsolver createTask failed.") return task_id = create_task.get('taskId') if not task_id: _local_log(session_name, "error", "Capsolver didn't return taskId.") return _local_log(session_name, "info", f"Capsolver task created: {task_id}") # wait for result captcha_token = await wait_for_captcha(task_id) if not captcha_token: _local_log(session_name, "error", "Capsolver returned no token.") return _local_log(session_name, "info", "Captcha token received, checking with Telegram") await telegram_check_captcha(captcha_token, actor, scope) await asyncio.sleep(random.randint(1,3)) await client.send_message(spambot_username, 'Done') elif 'Great! I’m very sorry if your account was limited by mistake.' in text: # send either random message from MESSAGES or a default text msgs = globals().get('MESSAGES', None) IS_RANDOM = globals().get('IS_RANDOM_MESSAGE', False) if IS_RANDOM and isinstance(msgs, (list, tuple)) and msgs: msg = random.choice(msgs) else: msg = msgs if isinstance(msgs, str) else (msgs[0] if msgs else 'I did nothing wrong.') await client.send_message(spambot_username, msg) elif "You've already submitted a complaint recently" in text: result_holder['result'] = 'wait' response_event.set() elif 'Thank you! Your complaint has been successfully submitted.' in text: result_holder['result'] = 'success' response_event.set() # register handler client.add_event_handler(_handler, events.NewMessage(chats=[spambot_username])) # start conversation try: await client.send_message(spambot_username, '/start') _local_log(session_name, "info", f"Sent /start to {spambot_username}") try: await asyncio.wait_for(response_event.wait(), timeout=timeout) except asyncio.TimeoutError: _local_log(session_name, "error", "Timeout waiting for SpamBot response") except Exception as send_exc: # couldn't send to spambot either _local_log(session_name, "error", f"Failed sending to SpamBot: {send_exc}") client.remove_event_handler(_handler, events.NewMessage) return ('failed', str(send_exc)) finally: client.remove_event_handler(_handler, events.NewMessage) # if bot flow produced a result, return it if result_holder['result']: # Ensure explicit report message is sent (best-effort) try: await client.send_message(spambot_username, spambot_message) except Exception: pass return ('spambot_result', result_holder['result']) # fallback: try to send direct report text once try: await client.send_message(spambot_username, spambot_message) return ('reported', None) except Exception as final_e: _local_log(session_name, "error", f"Final spambot send failed: {final_e}") return ('failed', str(final_e)) except Exception as e: return (None, str(e)) # --------------------------- # Entrypoint main() to run the whole pipeline (similar to your original main) # --------------------------- async def main(): numbers_path = "5.txt" changer = TelegramNumberChanger(numbers_path=numbers_path) await changer.run() if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print("\n> Stopped by user")