Предисловие: Привет, написал небольшого бота для полу-автоматического создания и полного хранения аккаунтов. Не знаю кому он может пригодится(очень узкая ЦА), но кому он будет нужен и тем кто крутится в этой сфере торговли, думаю будет удобно его использовать Telegram Account Registation Bot Бот для автоматической регистрации и управления аккаунтами Telegram с поддержкой ******, 2FA и временной почты. Основные возможности Регистрация аккаунтов Автоматическая регистрация с использованием временной почты Поддержка SMS-верификации Опциональная настройка 2FA Генерация случайных имен и фамилий Автоматическое обновление профиля Настройки Управление ****** (добавление, удаление, просмотр) Настройка 2FA (включение/выключение, длина пароля, сложность) Настройка уведомлений (**** регистрации, ошибки) Автоматические действия (проверка валидности, присоединение к каналам) Настройка приветственных сообщений Управление задержками между действиями Управление аккаунтами Просмотр списка всех аккаунтов Детальная информация об аккаунте Система тегов для организации аккаунтов Экспорт сессий Проверка валидности аккаунтов Процесс регистрации Подготовка Бот генерирует случайные имя и фамилию для аккаунта Создается временная почта через mail.tm Выбирается случайный ****** из списка (если настроены) Этапы регистрации Ввод номера телефона Пользователь вводит номер в международном формате Бот проверяет формат номера Проверяется, не зарегистрирован ли уже этот номер SMS-верификация Бот отправляет запрос на получение кода Пользователь вводит полученный код Проверяется валидность кода 2FA (опционально) Если аккаунт требует 2FA: Генерируется сложный пароль (если включена автоматическая 2FA) Пользователь вводит пароль Проверяется валидность пароля Завершение регистрации Обновляется профиль (имя, фамилия) Сохраняются данные аккаунта Проверяется валидность аккаунта Выполняются автоматические действия: Присоединение к каналам Отправка приветственного сообщения Сохранение данных Создается .session файл Сохраняется информация в базу данных: Номер телефона Имя и фамилия Данные временной почты Настройки 2FA Использованный ****** Дата регистрации Теги Система тегов Назначение тегов Теги используются для организации и категоризации аккаунтов. Они помогают: Группировать аккаунты по назначению Быстро находить нужные аккаунты Отслеживать использование аккаунтов Управлять большим количеством аккаунтов Предустановленные теги TG - аккаунты для Telegram VK - аккаунты для ВКонтакте WA - аккаунты для WhatsApp YA - аккаунты для Яндекс AV - аккаунты для Avito Работа с тегами Добавление тега Выберите аккаунт из списка Нажмите "Добавить тег" Введите название тега Тег будет добавлен к аккаунту Удаление тега Выберите аккаунт Нажмите "Удалить тег" Выберите тег для удаления Тег будет удален из аккаунта Просмотр по тегам В списке аккаунтов теги отображаются рядом с номером Можно фильтровать аккаунты по тегам Группировка аккаунтов по тегам в списке Преимущества использования тегов Удобная организация аккаунтов Быстрый поиск нужных аккаунтов Групповые операции с аккаунтами Отслеживание использования аккаунтов Упрощение управления большим количеством аккаунтов Установка Создайте папку в любом удобном для вас месте и скопируйте код 2-ух файлов из конца темы Установите зависимости: pip install -r requirements.txt Code pip install -r requirements.txt aiogram==3.20.0 telethon==1.32.1 python-dotenv==1.0.0 requests==2.31.0 faker==22.6.0 pymailtm==0.1.0 Code aiogram==3.20.0 telethon==1.32.1 python-dotenv==1.0.0 requests==2.31.0 faker==22.6.0 pymailtm==0.1.0 Создайте файл .env и добавьте необходимые переменные: BOT_TOKEN=your_bot_token API_ID=your_api_id API_HASH=your_api_hash Code BOT_TOKEN=your_bot_token API_ID=your_api_id API_HASH=your_api_hash Конфигурация Настройка ****** Поддерживаются ****** в формате ip:port:user:pass Можно добавить несколько ****** ****** выбираются случайным образом при регистрации Настройка 2FA Включение/выключение автоматической 2FA Настройка длины пароля (8-32 символа) Настройка сложности пароля Настройка уведомлений **** регистрации **** ошибок Сохранение ***** Автоматические действия Проверка валидности после регистрации Автоматическое присоединение к каналам Отправка приветственных сообщений Использование Команды бота /start - Главное меню /settings - Настройки бота Регистрация аккаунта Нажмите " Регистрация аккаунта" Введите номер телефона в международном формате Введите код подтверждения из SMS При необходимости введите пароль 2FA Управление аккаунтами Нажмите " Список аккаунтов" Выберите аккаунт для просмотра Используйте доступные действия: Просмотр деталей Управление тегами Экспорт сессии Безопасность Все данные хранятся локально Поддержка ****** для анонимности Безопасное хранение сессий Временная почта для 2FA Отказ от ответственности Этот бот предназначен только для образовательных целей. Используйте его на свой страх и риск. Автор не несет ответственности за любые последствия использования бота. В случае находки бага/недоработки Пишите в эту тему, буду отвечать по мере возможности и помогать вам разобраться с тем или иным багом или недоработкой Дальнейщие планы Хочу так же добавить автоматическую регистрацию аккаунтов которые имеются в тегах, а именно: VKontakte, WhatsApp, Avito, Yandex Если все-таки дойдут до этого руки, кину апдейт и сюда bot.py https://pastebin.com/6UJtE5YU database.py [/CENTER] import json import os from dataclasses import dataclass, asdict from datetime import datetime from typing import Dict, List, Set, Optional import random import asyncio from telethon import TelegramClient from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError @dataclass class NotificationSettings: registration_logs: bool = True error_logs: bool = True save_logs: bool = True @dataclass class AutoActions: check_validity: bool = True join_channels: bool = False send_welcome: bool = False channels: List[str] = None welcome_message: str = "" def __post_init__(self): if self.channels is None: self.channels = [] @dataclass class Settings: proxies: List[str] = None auto_2fa: bool = False password_complexity: bool = True min_password_length: int = 8 max_password_length: int = 32 notifications: NotificationSettings = None auto_actions: AutoActions = None min_delay: int = 1 max_delay: int = 5 admin_ids: List[int] = None def __post_init__(self): if self.proxies is None: self.proxies = [] if self.notifications is None: self.notifications = NotificationSettings() if self.auto_actions is None: self.auto_actions = AutoActions() if self.admin_ids is None: self.admin_ids = [] @dataclass class AccountData: phone: str first_name: str last_name: str email: str email_password: str registration_date: datetime has_2fa: bool = False proxy: Optional[str] = None tags: Set[str] = None def __post_init__(self): if self.tags is None: self.tags = set() def to_dict(self): data = asdict(self) data['registration_date'] = self.registration_date.isoformat() data['tags'] = list(self.tags) return data @classmethod def from_dict(cls, data: dict): data['registration_date'] = datetime.fromisoformat(data['registration_date']) data['tags'] = set(data['tags']) return cls(**data) class Database: def __init__(self, data_file: str = 'data/database.json'): self.data_file = data_file self.settings = Settings() self.accounts: Dict[str, AccountData] = {} self.load_data() def load_data(self): if not os.path.exists(self.data_file): os.makedirs(os.path.dirname(self.data_file), exist_ok=True) self.save_data() return try: with open(self.data_file, 'r', encoding='utf-8') as f: data = json.load(f) self.settings = Settings(**data.get('settings', {})) self.accounts = { phone: AccountData.from_dict(account_data) for phone, account_data in data.get('accounts', {}).items() } except Exception as e: print(f"Error loading database: {e}") self.settings = Settings() self.accounts = {} def save_data(self): os.makedirs(os.path.dirname(self.data_file), exist_ok=True) data = { 'settings': asdict(self.settings), 'accounts': { phone: account.to_dict() for phone, account in self.accounts.items() } } with open(self.data_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=4) def add_account(self, account: AccountData): self.accounts[account.phone] = account self.save_data() def get_account(self, phone: str) -> Optional[AccountData]: return self.accounts.get(phone) def get_all_accounts(self) -> Dict[str, AccountData]: return self.accounts def update_account_tags(self, phone: str, tags: Set[str]): if phone in self.accounts: self.accounts[phone].tags = tags self.save_data() def get_random_proxy(self) -> Optional[str]: if not self.settings.proxies: return None return random.choice(self.settings.proxies) async def send_registration_summary(self, client: TelegramClient, account: AccountData): if not self.settings.notifications.registration_logs: return summary = ( f" Регистрация успешно завершена!\n\n" f" Телефон: {account.phone}\n" f" Имя: {account.first_name}\n" f" Фамилия: {account.last_name}\n" f" Email: {account.email}\n" f" Пароль от email: {account.email_password}\n" f" 2FA: {'Включена' if account.has_2fa else 'Выключена'}\n" f" Дата регистрации: {account.registration_date.strftime('%Y-%m-%d %H:%M:%S')}\n" f" Теги: {', '.join(account.tags) if account.tags else 'Нет'}" ) for admin_id in self.settings.admin_ids: try: await client.send_message(admin_id, summary) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error sending registration summary to {admin_id}: {e}") async def send_welcome_message(self, client: TelegramClient, account: AccountData): if not self.settings.auto_actions.send_welcome or not self.settings.auto_actions.welcome_message: return try: await client.send_message('me', self.settings.auto_actions.welcome_message) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error sending welcome message to {account.phone}: {e}") async def join_channels(self, client: TelegramClient, account: AccountData): if not self.settings.auto_actions.join_channels or not self.settings.auto_actions.channels: return for channel in self.settings.auto_actions.channels: try: await client.join_channel(channel) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error joining channel {channel} for {account.phone}: {e}") async def check_account(self, client: TelegramClient, account: AccountData) -> bool: if not self.settings.auto_actions.check_validity: return True try: me = await client.get_me() return me is not None except Exception as e: print(f"Error checking account {account.phone}: {e}") return False def settings_check(self) -> bool: if not self.settings.admin_ids: print("Error: No admin IDs configured") return False return True def settings_notifications(self) -> bool: if not self.settings.notifications: print("Error: Notification settings not configured") return False return True db = Database() [CENTER] Python [/CENTER] import json import os from dataclasses import dataclass, asdict from datetime import datetime from typing import Dict, List, Set, Optional import random import asyncio from telethon import TelegramClient from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError @dataclass class NotificationSettings: registration_logs: bool = True error_logs: bool = True save_logs: bool = True @dataclass class AutoActions: check_validity: bool = True join_channels: bool = False send_welcome: bool = False channels: List[str] = None welcome_message: str = "" def __post_init__(self): if self.channels is None: self.channels = [] @dataclass class Settings: proxies: List[str] = None auto_2fa: bool = False password_complexity: bool = True min_password_length: int = 8 max_password_length: int = 32 notifications: NotificationSettings = None auto_actions: AutoActions = None min_delay: int = 1 max_delay: int = 5 admin_ids: List[int] = None def __post_init__(self): if self.proxies is None: self.proxies = [] if self.notifications is None: self.notifications = NotificationSettings() if self.auto_actions is None: self.auto_actions = AutoActions() if self.admin_ids is None: self.admin_ids = [] @dataclass class AccountData: phone: str first_name: str last_name: str email: str email_password: str registration_date: datetime has_2fa: bool = False proxy: Optional[str] = None tags: Set[str] = None def __post_init__(self): if self.tags is None: self.tags = set() def to_dict(self): data = asdict(self) data['registration_date'] = self.registration_date.isoformat() data['tags'] = list(self.tags) return data @classmethod def from_dict(cls, data: dict): data['registration_date'] = datetime.fromisoformat(data['registration_date']) data['tags'] = set(data['tags']) return cls(**data) class Database: def __init__(self, data_file: str = 'data/database.json'): self.data_file = data_file self.settings = Settings() self.accounts: Dict[str, AccountData] = {} self.load_data() def load_data(self): if not os.path.exists(self.data_file): os.makedirs(os.path.dirname(self.data_file), exist_ok=True) self.save_data() return try: with open(self.data_file, 'r', encoding='utf-8') as f: data = json.load(f) self.settings = Settings(**data.get('settings', {})) self.accounts = { phone: AccountData.from_dict(account_data) for phone, account_data in data.get('accounts', {}).items() } except Exception as e: print(f"Error loading database: {e}") self.settings = Settings() self.accounts = {} def save_data(self): os.makedirs(os.path.dirname(self.data_file), exist_ok=True) data = { 'settings': asdict(self.settings), 'accounts': { phone: account.to_dict() for phone, account in self.accounts.items() } } with open(self.data_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=4) def add_account(self, account: AccountData): self.accounts[account.phone] = account self.save_data() def get_account(self, phone: str) -> Optional[AccountData]: return self.accounts.get(phone) def get_all_accounts(self) -> Dict[str, AccountData]: return self.accounts def update_account_tags(self, phone: str, tags: Set[str]): if phone in self.accounts: self.accounts[phone].tags = tags self.save_data() def get_random_proxy(self) -> Optional[str]: if not self.settings.proxies: return None return random.choice(self.settings.proxies) async def send_registration_summary(self, client: TelegramClient, account: AccountData): if not self.settings.notifications.registration_logs: return summary = ( f" Регистрация успешно завершена!\n\n" f" Телефон: {account.phone}\n" f" Имя: {account.first_name}\n" f" Фамилия: {account.last_name}\n" f" Email: {account.email}\n" f" Пароль от email: {account.email_password}\n" f" 2FA: {'Включена' if account.has_2fa else 'Выключена'}\n" f" Дата регистрации: {account.registration_date.strftime('%Y-%m-%d %H:%M:%S')}\n" f" Теги: {', '.join(account.tags) if account.tags else 'Нет'}" ) for admin_id in self.settings.admin_ids: try: await client.send_message(admin_id, summary) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error sending registration summary to {admin_id}: {e}") async def send_welcome_message(self, client: TelegramClient, account: AccountData): if not self.settings.auto_actions.send_welcome or not self.settings.auto_actions.welcome_message: return try: await client.send_message('me', self.settings.auto_actions.welcome_message) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error sending welcome message to {account.phone}: {e}") async def join_channels(self, client: TelegramClient, account: AccountData): if not self.settings.auto_actions.join_channels or not self.settings.auto_actions.channels: return for channel in self.settings.auto_actions.channels: try: await client.join_channel(channel) await asyncio.sleep(random.uniform(self.settings.min_delay, self.settings.max_delay)) except Exception as e: print(f"Error joining channel {channel} for {account.phone}: {e}") async def check_account(self, client: TelegramClient, account: AccountData) -> bool: if not self.settings.auto_actions.check_validity: return True try: me = await client.get_me() return me is not None except Exception as e: print(f"Error checking account {account.phone}: {e}") return False def settings_check(self) -> bool: if not self.settings.admin_ids: print("Error: No admin IDs configured") return False return True def settings_notifications(self) -> bool: if not self.settings.notifications: print("Error: Notification settings not configured") return False return True db = Database() [CENTER] P.s. Код прогонял через ИИ для добавления форматирования, эмодзи и оптимизации, структруированния кода, не думаю что это прям тяжелый грех
CelestialDev, и что? если ему потоки не надо рвать и безумно насиловать процессор то почему бы и нет? основное преимущество его использование это то что он позволяет переключаться между задачами. пока одна из них ждет результата. НО. При большом количестве задач он само собой проигрывает и теряем производительность. в некоторых моментах этим можно пренебречь. при этом получив просто и легко код который не блокирует процессы соседние.