Мне нужно открывать одновременно несколько селениум браузеров и чтобы они работали параллельно друг от друга, значит мне нужны именно Потоки (threading), а selenium синхронный из коробки, значит мне нужно перейти на потоки.

import asyncio
import random
import re
import threading
import time
from pathlib import Path
from typing import List, Dict

import aiosocks
from opentele.api import UseCurrentSession, API
from opentele.td import TDesktop
from opentele.tl import TelegramClient as OTClient
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from telethon.errors import SessionPasswordNeededError, UserAlreadyParticipantError
from telethon.network.connection.tcpabridged import ConnectionTcpAbridged
from telethon.tl.functions.channels import JoinChannelRequest
from telethon.tl.functions.messages import ImportChatInviteRequest
from telethon.tl.types import InputChannel
from aiohttp_socks import ProxyType
import logging
from logging.handlers import RotatingFileHandler

MILLION_CHANNEL = "https://t.me/+8AqnUALHTpZkNmJi"

# Настройка логгера
def setup_logger():
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    file_handler = RotatingFileHandler(
        'telegram_session.log',
        maxBytes=10 * 1024 * 1024,
        backupCount=3
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s'
    ))

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(
        '%(levelname)s - %(message)s'
    ))

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

logger = setup_logger()

# Настройки задержек
DELAYS = {
    'short': (0.5, 1.5),
    'medium': (1.5, 3),
    'long': (3, 6)
}

def human_delay(delay_type):
    """Возвращает случайную задержку в заданном диапазоне"""
    min_d, max_d = DELAYS[delay_type]
    delay = random.uniform(min_d, max_d)
    time.sleep(delay)
    return delay

async def get_phone_from_session(client):
    """Получает номер телефона из существующей сессии"""
    me = await client.get_me()
    phone = me.phone
    await client.get_dialogs(limit=3)
    await asyncio.sleep(human_delay('short'))
    return phone

def extract_code_from_text(text, code_type='web'):
    """Извлекает код подтверждения из текста сообщения"""
    if not text:
        return None

    if code_type == 'web':
        patterns = [
            r'`([A-Za-z0-9_\-]{8,12})`',
            r'(?:Вот он:|код:|code:)\s*`?([A-Za-z0-9_\-]{8,12})`?',
            r'\b([A-Za-z0-9_\-]{8,12})\b(?=.*my\.telegram\.org)'
        ]
    else:
        patterns = [
            r'\b(\d{5})\b',
            r'\b(\d{6})\b',
            r'код:\s*(\d{5,6})',
            r'code:\s*(\d{5,6})'
        ]

    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            return match.group(1)
    return None


async def get_code_from_telegram(client, phone_number, code_type='web', timeout=60):
    """Получает код подтверждения из Telegram через существующую сессию"""
    start_time = time.time()
    last_message_id = 0

    while time.time() - start_time < timeout:
        messages = []
        async for message in client.iter_messages(777000, limit=5):
            if message.id > last_message_id:
                messages.append(message)
                last_message_id = message.id

        for message in reversed(messages):
            if message.text:
                logger.info(f"Анализируем сообщение:\n{message.text}\n")
                code = extract_code_from_text(message.text, code_type)
                if code:
                    logger.info(f"Найден {code_type} код: {code}")
                    return code

        await asyncio.sleep(human_delay('short'))

    logger.error("Не удалось найти код в сообщениях.")
    return None


async def create_telegram_app(client, phone_number):
    """Создает новое приложение на my.telegram.org"""
    options = webdriver.ChromeOptions()
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")
    options.add_argument("--window-size=1920,1080")
    options.add_argument("--disable-notifications")
    options.add_argument("--disable-blink-features=AutomationControlled")
    options.add_argument("--incognito")

    driver = webdriver.Chrome(options=options)
    try:
        driver.get("https://my.telegram.org/apps")

        phone_input = WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.ID, "my_login_phone"))
        )
        phone_input.clear()
        phone_input.send_keys(phone_number)

        next_button = WebDriverWait(driver, 15).until(
            EC.element_to_be_clickable((By.XPATH, "//button[contains(@class, 'btn-primary')]"))
        )
        driver.execute_script("arguments[0].click();", next_button)
        human_delay('long')

        try:
            alert_error = WebDriverWait(driver, 5).until(
                EC.visibility_of_element_located(
                    (By.XPATH, "//div[@id='my_login_alert']//div[contains(@class, 'alert-danger')]"))
            )
            if "too many tries" in alert_error.text:
                logger.error("❌ Ошибка: Слишком много попыток.")
                return None, None
        except:
            pass

        code = await get_code_from_telegram(client, phone_number, code_type='web')
        if not code:
            raise Exception("Не удалось получить код подтверждения")

        code_input = WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.ID, "my_password"))
        )
        code_input.clear()
        code_input.send_keys(code)

        sign_in_button = WebDriverWait(driver, 15).until(
            EC.element_to_be_clickable((By.XPATH, "//button[contains(text(), 'Sign In')]"))
        )
        driver.execute_script("arguments[0].click();", sign_in_button)

        try:
            api_id_element = WebDriverWait(driver, 10).until(
                EC.presence_of_element_located(
                    (By.XPATH, "//label[contains(text(), 'App api_id:')]/following-sibling::div/span"))
            )
            api_id = api_id_element.text.strip()

            api_hash_element = WebDriverWait(driver, 15).until(
                EC.presence_of_element_located(
                    (By.XPATH, "//label[contains(text(), 'App api_hash:')]/following-sibling::div/span"))
            )
            api_hash = api_hash_element.text.strip()

            logger.info(f"Успешно получены API данные: api_id={api_id}, api_hash={api_hash}")
            return int(api_id), str(api_hash)
        except:
            logger.info("Приложение еще не создано, перехожу к созданию.")

        WebDriverWait(driver, 30).until(
            EC.presence_of_element_located((By.NAME, "app_title"))
        )

        title_field = driver.find_element(By.NAME, "app_title")
        title_field.send_keys("My App")

        shortname_field = driver.find_element(By.NAME, "app_shortname")
        shortname_field.send_keys("myapp")

        desktop_radio = WebDriverWait(driver, 15).until(
            EC.element_to_be_clickable((By.XPATH, "//input[@name='app_platform' and @value='desktop']"))
        )
        desktop_radio.click()

        desc_field = driver.find_element(By.NAME, "app_desc")
        desc_field.send_keys("My Telegram Application")

        max_attempts = 66
        for attempt in range(max_attempts):
            try:
                create_button = WebDriverWait(driver, 15).until(
                    EC.element_to_be_clickable((By.ID, "app_save_btn"))
                )
                human_delay('long')
                driver.execute_script("arguments[0].click();", create_button)
                human_delay('long')

                try:
                    WebDriverWait(driver, 3).until(EC.alert_is_present())
                    alert = driver.switch_to.alert
                    if "Error" in alert.text:
                        logger.warning(f"Ошибка при попытке {attempt + 1}/{max_attempts}")
                        alert.accept()
                        human_delay('medium')
                        continue
                    alert.accept()
                except:
                    break
            except Exception as e:
                logger.warning(f"Ошибка при создании приложения (попытка {attempt + 1}): {e}")
                human_delay('short')

        WebDriverWait(driver, 30).until(
            EC.presence_of_element_located((By.XPATH, "//label[contains(text(), 'App api_id:')]"))
        )

        api_id_element = WebDriverWait(driver, 15).until(
            EC.presence_of_element_located(
                (By.XPATH, "//label[contains(text(), 'App api_id:')]/following-sibling::div/span"))
        )
        api_id = api_id_element.text.strip()

        api_hash_element = WebDriverWait(driver, 15).until(
            EC.presence_of_element_located(
                (By.XPATH, "//label[contains(text(), 'App api_hash:')]/following-sibling::div/span"))
        )
        api_hash = api_hash_element.text.strip()

        logger.info(f"Успешно получены API данные: api_id={api_id}, api_hash={api_hash}")
        return int(api_id), str(api_hash)

    except Exception as e:
        logger.error(f"Критическая ошибка: {e}")
        driver.save_screenshot("error.png")
        raise
    finally:
        driver.quit()
        human_delay('short')

async def get_channel_entity(client, invite_link):
    try:
        if any(invite_link.startswith(prefix) for prefix in ["https://t.me/+", "http://t.me/+", "t.me/+"]):
            invite_hash = invite_link.split("+")[-1]
            try:
                result = await client(ImportChatInviteRequest(invite_hash))
                return InputChannel(result.chats[0].id, result.chats[0].access_hash)
            except UserAlreadyParticipantError:
                entity = await client.get_entity(invite_link)
                return entity
        else:
            entity = await client.get_entity(invite_link)
            return entity
    except Exception as e:
        logger.error(f"Ошибка получения entity: {e}")
        raise

async def subscribe_to_channel(client, channel_link):
    try:
        entity = await get_channel_entity(client, channel_link)
        try:
            await client(JoinChannelRequest(channel=entity))
            logger.info(f"Успешно подписались на канал: {channel_link}")
        except UserAlreadyParticipantError:
            logger.info(f"Уже подписан на канал: {channel_link}")
    except Exception as e:
        logger.error(f"Ошибка подписки: {e}")

async def create_new_session(phone_number: str, api_id: int, api_hash: str, proxy_config: Dict, session_number: int):
    max_attempts = 3
    last_error = None
    session_name = f"new_session_{phone_number}"

    # Удаляем старый session-файл если он есть
    session_file = Path(f"{session_name}.session")
    if session_file.exists():
        try:
            session_file.unlink()
        except PermissionError:
            logger.warning(f"Не удалось удалить файл сессии {session_name}.session, возможно он занят другим процессом")
            return None

    for attempt in range(max_attempts):
        client = None
        try:
            logger.info(f"[{session_number}] Попытка {attempt + 1}/{max_attempts} создания сессии для {phone_number}")

            client = OTClient(
                session=session_name,
                api_id=int(api_id),
                api_hash=str(api_hash),
                api=API.TelegramDesktop.Generate(),
                connection=ConnectionTcpAbridged,
                proxy=(
                    ProxyType.SOCKS5,
                    proxy_config["host"],
                    proxy_config["port"],
                    True,  # rdns
                    proxy_config["username"],
                    proxy_config["password"]
                ),
                use_ipv6=True
            )

            logger.debug(f"[{session_number}] Инициализация клиента завершена, подключаемся...")
            await client.connect()
            logger.info(f"[{session_number}] Успешное подключение к Telegram")

            logger.debug(f"[{session_number}] Отправка запроса кода...")
            sent_code = await client.send_code_request(phone_number)
            logger.info(f"[{session_number}] Код подтверждения отправлен")

            code = await get_code_from_telegram(client, phone_number, code_type='login')
            if not code:
                raise Exception("Не удалось получить код подтверждения")
            logger.debug(f"[{session_number}] Получен код: {code}")

            try:
                await client.sign_in(
                    phone=phone_number,
                    code=code,
                    phone_code_hash=sent_code.phone_code_hash
                )
                logger.info(f"[{session_number}] Успешная авторизация по коду")
            except SessionPasswordNeededError:
                logger.warning(f"[{session_number}] Требуется пароль двухфакторной аутентификации")
                password = input(f"Введите пароль 2FA для {phone_number}: ")
                await client.sign_in(password=password)
                logger.info(f"[{session_number}] Успешная авторизация с паролем 2FA")

            await subscribe_to_channel(client, MILLION_CHANNEL)
            logger.info(f"[{session_number}] Сессия {phone_number} успешно создана!")
            return client

        except Exception as e:
            last_error = e
            logger.error(f"[{session_number}] Ошибка при попытке {attempt + 1}: {str(e)}", exc_info=True)
            if client and client.is_connected():
                await client.disconnect()
            human_delay('long')

    logger.critical(f"[{session_number}] Не удалось создать сессию после {max_attempts} попыток")
    raise Exception(f"[{session_number}] Последняя ошибка: {last_error}")

def find_all_tdata_paths(base_path="all_tdatas") -> List[str]:
    tdata_paths = []
    base_path = Path(base_path)

    if not base_path.exists():
        raise FileNotFoundError(f"Директория {base_path} не найдена")

    for entry in base_path.iterdir():
        if entry.is_dir():
            tdata_path = entry / "tdata"
            if tdata_path.exists():
                tdata_paths.append(str(tdata_path))

    if not tdata_paths:
        raise FileNotFoundError(f"Не найдено ни одной папки tdata в {base_path}")

    return tdata_paths

def load_proxies(filename="proxies.txt") -> List[Dict]:
    proxies = []
    try:
        with open(filename, 'r') as f:
            for line in f:
                line = line.strip()
                if line:
                    parts = line.split(':')
                    if len(parts) == 4:
                        proxy = {
                            "host": parts[0],
                            "port": int(parts[1]),
                            "username": parts[2],
                            "password": parts[3],
                            "check_timeout": 3,
                            "rdns": True
                        }
                        proxies.append(proxy)

        if not proxies:
            raise ValueError("Файл проксей пуст или имеет неверный формат")

        logger.info(f"Загружено {len(proxies)} проксей")
        return proxies
    except Exception as e:
        logger.error(f"Ошибка загрузки проксей: {e}")
        raise


async def process_session(tdata_path: str, proxy_config: Dict, session_number: int):
    """Обрабатывает одну сессию tdata в отдельном потоке"""
    logger.info(f"\n{'=' * 50}")
    logger.info(f"Обработка сессии #{session_number} ({tdata_path})")
    logger.info(f"{'=' * 50}\n")

    try:
        logger.info(f"[{session_number}] 1. Загружаем tdata...")
        tdata = TDesktop(tdata_path)
        if not tdata.isLoaded():
            logger.error(f"[{session_number}] ⚠️ tdata не загружено или повреждено")
            return

        session_name = f"session_{session_number}"

        # Конвертируем tdata в telethon сессию
        converter_client = await tdata.ToTelethon(
            session=session_name,
            flag=UseCurrentSession,
            api=API.TelegramDesktop.Generate()
        )
        await converter_client.disconnect()
        logger.info(f"[{session_number}] Сессия сохранена")

        # Подключаемся к существующей сессии
        existing_client = OTClient(
            session=session_name,
            api_id=2040,
            api_hash="b18441a1ff607e10a989891a5462e627",
            api=API.TelegramDesktop.Generate(),
            connection=ConnectionTcpAbridged,
        )

        await existing_client.connect()
        me = await existing_client.get_me()
        logger.info(me)

        if not await existing_client.is_user_authorized():
            logger.error(f"[{session_number}] ⚠️ Сессия не авторизована")
            return

        logger.info(f"[{session_number}] 2. Получаю номер телефона...")
        phone_number = await get_phone_from_session(existing_client)
        logger.info(f"[{session_number}] Номер: {phone_number}")

        logger.info(f"[{session_number}] 3. Создаю приложение...")
        api_id, api_hash = await create_telegram_app(existing_client, phone_number)
        if not api_id or not api_hash:
            logger.error(f"[{session_number}] Не удалось получить API данные")
            return

        logger.info(f"[{session_number}] 4. Создаю новую сессию...")
        await create_new_session(str(phone_number), int(api_id), str(api_hash), proxy_config, session_number)

        logger.info(f"[{session_number}] ✅ Сессия успешно обработана!")

    except Exception as e:
        logger.error(f"[{session_number}] ⚠️ Критическая ошибка: {str(e)[:100]}", exc_info=True)
    finally:
        if 'existing_client' in locals() and existing_client.is_connected():
            await existing_client.disconnect()

async def main():
    try:
        # Загружаем прокси и tdata сессии
        proxies = load_proxies()
        tdata_paths = find_all_tdata_paths()
        logger.info(f"Найдено {len(tdata_paths)} сессий для обработки")

        # Создаем задачи для каждой сессии с уникальным прокси
        tasks = []
        for i, tdata_path in enumerate(tdata_paths, 1):
            proxy = proxies[i % len(proxies)]  # Распределяем прокси по кругу
            task = asyncio.create_task(process_session(tdata_path, proxy, i))
            tasks.append(task)
            await asyncio.sleep(1)  # Небольшая задержка между запуском задач

        # Ждем завершения всех задач
        await asyncio.gather(*tasks)
        logger.info("Все задачи выполнены!")

    except Exception as e:
        logger.error(f"Ошибка в main: {e}", exc_info=True)
    finally:
        # Закрываем все соединения при завершении
        for task in asyncio.all_tasks():
            if not task.done():
                task.cancel()

if __name__ == "__main__":
    asyncio.run(main())