import asyncio
import random
from pathlib import Path
from typing import Dict, List, Tuple
from opentele.api import UseCurrentSession, API
from opentele.td import TDesktop
from opentele.tl import TelegramClient as OTClient
from telethon.network.connection.tcpabridged import ConnectionTcpAbridged
from aiohttp_socks import ProxyType
import logging
import sys
import sqlite3
import time
from datetime import datetime


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

    class SmileyFormatter(logging.Formatter):
        def format(self, record):
            level = record.levelname
            if level == 'INFO':
                record.msg = f"ℹ️ {record.msg}"
            elif level == 'WARNING':
                record.msg = f"⚠️ {record.msg}"
            elif level == 'ERROR':
                record.msg = f"❌ {record.msg}"
            elif level == 'CRITICAL':
                record.msg = f"🔥 {record.msg}"
            else:
                record.msg = f"➡️ {record.msg}"
            return super().format(record)

    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(SmileyFormatter(
        '%(asctime)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    ))

    if not logger.hasHandlers():
        logger.addHandler(console_handler)

    return logger


logger = setup_logger()

# Константы
API_ID = 2040
API_HASH = "b18441a1ff607e10a989891a5462e627"
OUTPUT_DIR = Path("converted_2040_sessions")
OUTPUT_DIR.mkdir(exist_ok=True, parents=True)


class SessionMetrics:
    def __init__(self):
        self.total = 0
        self.success = 0
        self.failed = 0
        self.start_time = time.time()
        self.lock = asyncio.Lock()

    async def add_success(self):
        async with self.lock:
            self.success += 1

    async def add_failed(self):
        async with self.lock:
            self.failed += 1

    def get_stats(self) -> Dict:
        duration = time.time() - self.start_time
        return {
            "total": self.total,
            "success": self.success,
            "failed": self.failed,
            "success_rate": (self.success / self.total) * 100 if self.total > 0 else 0,
            "duration_sec": duration,
            "sessions_per_min": (self.success / duration) * 60 if duration > 0 else 0
        }


metrics = SessionMetrics()


def print_banner():
    logger.info("")
    logger.info("▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄")
    logger.info("█                                                               █")
    logger.info("█                🚀 TData to Session Converter 🚀               █")
    logger.info("█                                                               █")
    logger.info("▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀")
    logger.info("")


async def convert_tdata_to_session(tdata_path: str, proxy_config: Dict, session_number: int):
    client = None
    temp_client = None
    try:
        metrics.total += 1
        logger.info(f"[#{session_number}] 🛠️ Начинаю обработку tdata: {tdata_path}")

        # Загружаем tdata с повторными попытками
        max_attempts = 3
        tdata = None

        for attempt in range(max_attempts):
            try:
                tdata = TDesktop(tdata_path)
                if tdata.isLoaded():
                    logger.info(f"[#{session_number}] ✅ Успешно загрузили tdata (попытка {attempt + 1})")
                    break
                else:
                    logger.warning(f"[#{session_number}] ⚠️ Попытка {attempt + 1}: не удалось загрузить tdata")
            except sqlite3.OperationalError as e:
                logger.warning(f"[#{session_number}] ⚠️ Попытка {attempt + 1}: Ошибка SQLite: {e}")
                if "database is locked" in str(e):
                    await asyncio.sleep(1 + attempt)  # Увеличиваем задержку с каждой попыткой
                else:
                    raise
            except Exception as e:
                logger.error(f"[#{session_number}] ❌ Ошибка при загрузке tdata: {e}")
                raise

            if attempt == max_attempts - 1:
                logger.error(f"[#{session_number}] 🔥 Не удалось загрузить tdata после {max_attempts} попыток")
                await metrics.add_failed()
                return False

        # Создаем временного клиента для получения информации
        temp_session_path = f"temp_session_{session_number}"
        logger.info(f"[#{session_number}] 🔄 Создаю временную сессию...")

        temp_client = await tdata.ToTelethon(
            session=temp_session_path,
            flag=UseCurrentSession,
            api=API.TelegramDesktop.Generate()
        )

        # Подключаемся и получаем информацию об аккаунте
        await temp_client.connect()
        me = await temp_client.get_me()
        phone_number = me.phone
        telegram_id = me.id
        await temp_client.disconnect()

        # Удаляем временную сессию
        temp_file = Path(f"{temp_session_path}.session")
        if temp_file.exists():
            temp_file.unlink()

        logger.info(f"[#{session_number}] 📱 Аккаунт: ID {telegram_id}, номер {phone_number}")

        # Создаем имя для новой сессии
        session_name = f"{telegram_id}_{phone_number}"
        session_path = OUTPUT_DIR / f"{session_name}.session"

        # Удаляем старый файл сессии, если существует
        if session_path.exists():
            session_path.unlink()

        # Конвертируем в новую сессию с нужными API данными
        logger.info(f"[#{session_number}] 🔄 Создаю новую сессию с API 2040...")
        client = OTClient(
            session=str(session_path),
            api_id=API_ID,
            api_hash=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
        )

        # Подключаемся и проверяем авторизацию
        await client.connect()

        if await client.is_user_authorized():
            logger.info(f"[#{session_number}] 🎉 Сессия успешно создана и сохранена: {session_path}")
            await metrics.add_success()

            # Выводим красивую информацию об аккаунте
            logger.info(f"[#{session_number}] 👤 Информация об аккаунте:")
            logger.info(f"[#{session_number}] ├─ ID: {telegram_id}")
            logger.info(f"[#{session_number}] ├─ Номер: {phone_number}")
            logger.info(f"[#{session_number}] └─ Файл сессии: {session_path}")

            return True
        else:
            logger.error(f"[#{session_number}] ❌ Не удалось авторизовать новую сессию")
            await metrics.add_failed()
            return False

    except Exception as e:
        logger.error(f"[#{session_number}] 🔥 Критическая ошибка: {str(e)}", exc_info=True)
        await metrics.add_failed()
        return False
    finally:
        if client is not None and client.is_connected():
            await client.disconnect()
        if temp_client is not None and temp_client.is_connected():
            await temp_client.disconnect()
        # Удаляем временный файл, если он остался
        temp_file = Path(f"temp_session_{session_number}.session")
        if temp_file.exists():
            temp_file.unlink()


def load_proxies(filename: str) -> List[Dict]:
    """Загружает прокси из файла в формате host:port:username:password"""
    logger.info(f"🔍 Загружаю прокси из файла {filename}...")
    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]
                        }
                        proxies.append(proxy)

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


async def process_with_semaphore(semaphore, tdata_path, proxy, session_number):
    """Обрабатывает tdata с семафором для ограничения одновременных задач"""
    async with semaphore:
        return await convert_tdata_to_session(tdata_path, proxy, session_number)


def find_tdata_paths(base_dir: str = "all_tdatas") -> List[str]:
    """Находит все папки tdata внутри поддиректорий указанной директории"""
    logger.info(f"🔍 Поиск tdata в директории {base_dir}...")
    base_path = Path(base_dir)
    if not base_path.exists():
        raise FileNotFoundError(f"❌ Директория {base_dir} не найдена")

    tdata_paths = []

    # Ищем во всех поддиректориях первого уровня
    for entry in base_path.iterdir():
        if entry.is_dir():
            # Проверяем есть ли внутри папка tdata
            tdata_path = entry / "tdata"
            if tdata_path.exists() and tdata_path.is_dir():
                tdata_paths.append(str(tdata_path))

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

    logger.info(f"✅ Найдено {len(tdata_paths)} tdata для обработки")
    return tdata_paths


async def process_all_tdata(tdata_paths: List[str], proxies: List[Dict]):
    """Обрабатывает все tdata с использованием прокси"""
    logger.info("🔄 Начинаю обработку всех tdata...")
    semaphore = asyncio.Semaphore(3)  # Ограничиваем количество одновременных задач
    tasks = []

    for i, tdata_path in enumerate(tdata_paths, 1):
        proxy = proxies[(i - 1) % len(proxies)]  # Циклически распределяем прокси
        task = asyncio.create_task(
            process_with_semaphore(semaphore, tdata_path, proxy, i)
        )
        tasks.append(task)
        await asyncio.sleep(1)  # Добавляем задержку между запуском задач

    results = await asyncio.gather(*tasks, return_exceptions=True)

    successful = 0
    for i, result in enumerate(results, 1):
        if isinstance(result, Exception):
            logger.error(f"[#{i}] ❌ Ошибка при обработке: {result}")
        elif result:
            successful += 1

    logger.info(f"📊 Итоговые результаты:")
    logger.info(f"├─ Всего обработано: {len(tdata_paths)}")
    logger.info(f"├─ Успешно: {successful} ✅")
    logger.info(f"└─ Неудачно: {len(tdata_paths) - successful} ❌")


def print_final_stats():
    stats = metrics.get_stats()
    logger.info("")
    logger.info("📈 📊 📉 ИТОГОВАЯ СТАТИСТИКА 📈 📊 📉")
    logger.info("▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔")
    logger.info(f"├─ Всего сессий: {stats['total']}")
    logger.info(f"├─ Успешно: {stats['success']} ✅")
    logger.info(f"├─ Неудачно: {stats['failed']} ❌")
    logger.info(f"├─ Процент успеха: {stats['success_rate']:.2f}%")
    logger.info(f"├─ Время работы: {stats['duration_sec']:.2f} сек")
    logger.info(f"└─ Скорость: {stats['sessions_per_min']:.2f} сессий/мин")
    logger.info("▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔")
    logger.info("")


async def main():
    print_banner()
    start_time = time.time()

    try:
        # Загружаем прокси
        proxies = load_proxies("proxies.txt")

        # Находим все tdata для обработки
        tdata_paths = find_tdata_paths()

        # Запускаем обработку
        await process_all_tdata(tdata_paths, proxies)

    except Exception as e:
        logger.error(f"🔥 Критическая ошибка в main: {e}", exc_info=True)
    finally:
        # Завершаем все задачи
        pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
        for task in pending:
            task.cancel()
        await asyncio.gather(*pending, return_exceptions=True)

        # Выводим итоговую статистику
        print_final_stats()

        duration = time.time() - start_time
        logger.info(f"⌛ Общее время работы: {duration:.2f} секунд")
        logger.info("✨ Скрипт завершил работу")


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


Этот скрипт не создаёт новые сессии, он создает лишь файлы сессий, но они не авторизованы. Как минимум, ты должен написать код, который авторизовывает новую сессию при помощи api id и api hash, получить код от телеграма (из цифр), ввести его и именно так создаётся сессия.

Сейчас это не так.


Вот пример создания новой сессии из моего старого скрипта, возьми из него основную логику:

async def create_new_session(phone_number: str, api_id: int, api_hash: str,
                             proxy_config: Dict, session_number: int, code_client: OTClient):
    global successful_sessions, failed_sessions

    max_attempts = 3
    last_error = None
    session_name = NEW_SESSION_DIR / f"new_session_{phone_number}"

    if not NEW_SESSION_DIR.exists():
        NEW_SESSION_DIR.mkdir(parents=True, exist_ok=True)

    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=str(session_name),
                api_id=api_id,
                api_hash=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(code_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} успешно создана!")

            # Сохраняем данные сессии в JSON
            try:
                me = await client.get_me()
                session_data = {
                    "phone_number": phone_number,
                    "telegram_id": me.id,
                    "api_id": api_id,
                    "api_hash": api_hash,
                    "proxy": {
                        "host": proxy_config["host"],
                        "port": proxy_config["port"],
                        "username": proxy_config["username"],
                        "password": proxy_config["password"]
                    }
                }
                json_path = NEW_SESSION_DIR / "all_api_id_phones.json"

                if json_path.exists():
                    with open(json_path, "r", encoding="utf-8") as f:
                        try:
                            all_data = json.load(f)
                        except json.JSONDecodeError:
                            all_data = {}
                else:
                    all_data = {}

                composite_key = f"{phone_number}|{me.id}"
                all_data[composite_key] = session_data

                with open(json_path, "w", encoding="utf-8") as f:
                    json.dump(all_data, f, ensure_ascii=False, indent=4)

                logger.info(f"[{session_number}] Информация о сессии обновлена в {json_path}")
            except Exception as e:
                logger.warning(f"[{session_number}] Не удалось сохранить JSON сессии: {e}")

            successful_sessions += 1
            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} попыток")
    failed_sessions += 1

    raise Exception(f"[{session_number}] Последняя ошибка: {last_error}")


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}")
                code = extract_code_from_text(message.text, code_type)
                if code:
                    logger.info(f"Найден {code_type} код: {code}")
                    return code

        await asyncio.sleep(random.uniform(0.5, 1.5))

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