"""
client_builder.py — создание Telethon-клиентов, SQLiteSessionWAL, disconnect.
Можно редактировать параметры подключения или retry-логику независимо.
"""

import asyncio
import hashlib
import inspect
import json
import os
import random
import sqlite3
import time
import threading
from pathlib import Path
from typing import Optional, Dict, List

from telethon import TelegramClient
from telethon import errors
from telethon.sessions import SQLiteSession

from config import OFFICIAL_TDESKTOP_API_ID, OFFICIAL_TDESKTOP_APP_HASH, DATA_DIR
from bot_ui.logger_setup import logger
from models import DeviceInfo, ProxyResolved


# ── Platform-specific app versions ──────────────────────────────────────────

_DESKTOP_VERSIONS = [
    "6.5.0 x64", "6.4.1 x64", "6.4.0 x64", "6.3.2 x64",
    "6.3.0 x64", "6.2.2 x64", "6.2.0 x64", "6.1.2 x64",
    "6.1.0 x64", "6.0.3 x64", "6.0.2 x64", "6.0.0 x64",
]

_ANDROID_VERSIONS = [
    "11.7.2", "11.6.0", "11.5.4", "11.4.2", "11.3.1", "11.2.3",
    "11.1.0", "11.0.1", "10.15.2", "10.14.5", "10.13.2", "10.12.1",
    "10.11.1", "10.10.1", "10.9.1", "10.8.3", "10.7.0", "10.6.2",
]

_IOS_VERSIONS = [
    "11.7", "11.6.1", "11.5", "11.4.2", "11.3", "11.2",
    "11.1", "11.0", "10.15", "10.14.1", "10.13", "10.12.1",
    "10.11", "10.10", "10.9.1", "10.8", "10.7", "10.6.3",
]

# ── Platform-specific system versions ──────────────────────────────────────

_WINDOWS_VERSIONS = [
    "Windows 10", "Windows 10", "Windows 10", "Windows 10",
    "Windows 11", "Windows 11", "Windows 11",
]

_ANDROID_OS_VERSIONS = [
    "SDK 35", "SDK 35", "SDK 34", "SDK 34", "SDK 34",
    "SDK 33", "SDK 33", "SDK 32", "SDK 31", "SDK 30",
]

_IOS_OS_VERSIONS = [
    "18.4", "18.3.2", "18.3.1", "18.2", "18.1.1", "18.1", "18.0",
    "17.7.6", "17.7.5", "17.6.1", "17.5.1", "17.4.1", "17.3.1",
]

# ── Android device models ──────────────────────────────────────────────────

_ANDROID_DEVICES = [
    "Samsung Galaxy S24 Ultra", "Samsung Galaxy S24+", "Samsung Galaxy S24",
    "Samsung Galaxy S23 Ultra", "Samsung Galaxy S23+", "Samsung Galaxy S23",
    "Samsung Galaxy S22 Ultra", "Samsung Galaxy S22+", "Samsung Galaxy S22",
    "Samsung Galaxy S21 Ultra", "Samsung Galaxy S21+", "Samsung Galaxy S21",
    "Samsung Galaxy A55", "Samsung Galaxy A54", "Samsung Galaxy A53",
    "Samsung Galaxy A52", "Samsung Galaxy A34", "Samsung Galaxy A33",
    "Samsung Galaxy A25", "Samsung Galaxy A15", "Samsung Galaxy A14",
    "Samsung Galaxy Z Fold5", "Samsung Galaxy Z Fold4", "Samsung Galaxy Z Flip5",
    "Samsung Galaxy Z Flip4", "Samsung Galaxy Note 20 Ultra", "Samsung Galaxy M54",
    "Xiaomi 14 Ultra", "Xiaomi 14 Pro", "Xiaomi 14", "Xiaomi 13 Pro",
    "Xiaomi 13", "Xiaomi 12 Pro", "Xiaomi 12", "Xiaomi 11T Pro",
    "Xiaomi Redmi Note 13 Pro+", "Xiaomi Redmi Note 13 Pro", "Xiaomi Redmi Note 13",
    "Xiaomi Redmi Note 12 Pro+", "Xiaomi Redmi Note 12 Pro", "Xiaomi Redmi Note 12",
    "Xiaomi Redmi Note 11 Pro+", "Xiaomi Redmi Note 11", "Xiaomi Redmi 13C",
    "Xiaomi POCO F6 Pro", "Xiaomi POCO F5", "Xiaomi POCO X6 Pro",
    "OnePlus 12", "OnePlus 11", "OnePlus 10 Pro", "OnePlus 10T",
    "OnePlus Nord CE 3", "OnePlus Nord 3", "OnePlus 9 Pro", "OnePlus 9",
    "Google Pixel 9 Pro", "Google Pixel 9", "Google Pixel 8 Pro", "Google Pixel 8",
    "Google Pixel 7 Pro", "Google Pixel 7a", "Google Pixel 7", "Google Pixel 6a",
    "Huawei P60 Pro", "Huawei P50 Pro", "Huawei Mate 60 Pro", "Huawei Nova 12",
    "OPPO Find X7 Ultra", "OPPO Find X6 Pro", "OPPO Reno 11 Pro", "OPPO Reno 10 Pro+",
    "OPPO A79", "OPPO A58", "OPPO A38", "OPPO A18",
    "Vivo X100 Pro", "Vivo X90 Pro", "Vivo V30 Pro", "Vivo V29",
    "Vivo Y100", "Vivo Y36", "Vivo T2x",
    "Realme GT 5 Pro", "Realme GT Neo 5", "Realme 12 Pro+", "Realme 11 Pro+",
    "Realme C67", "Realme C55", "Realme Narzo 60 Pro",
    "Motorola Edge 40 Pro", "Motorola Razr 40 Ultra", "Motorola Moto G84",
    "Nothing Phone (2)", "Nothing Phone (1)",
    "Sony Xperia 1 V", "Sony Xperia 5 V", "Sony Xperia 10 V",
    "ASUS ROG Phone 8 Pro", "ASUS Zenfone 10",
    "Honor Magic6 Pro", "Honor 90", "Honor X9b", "Honor X8a",
    "Tecno Spark 20 Pro+", "Tecno Camon 20 Pro", "Tecno Pova 5 Pro",
    "Infinix Note 30 Pro", "Infinix Hot 30",
    "ZTE Blade V50", "ZTE Nubia Z60 Ultra",
]

# ── iOS device models ──────────────────────────────────────────────────────

_IOS_DEVICES = [
    "iPhone 16 Pro Max", "iPhone 16 Pro", "iPhone 16 Plus", "iPhone 16",
    "iPhone 15 Pro Max", "iPhone 15 Pro", "iPhone 15 Plus", "iPhone 15",
    "iPhone 14 Pro Max", "iPhone 14 Pro", "iPhone 14 Plus", "iPhone 14",
    "iPhone 13 Pro Max", "iPhone 13 Pro", "iPhone 13", "iPhone 13 mini",
    "iPhone 12 Pro Max", "iPhone 12 Pro", "iPhone 12", "iPhone 12 mini",
    "iPhone SE (3rd generation)", "iPhone 11 Pro Max", "iPhone 11 Pro", "iPhone 11",
    "iPad Pro 12.9-inch (6th generation)", "iPad Pro 11-inch (4th generation)",
    "iPad Air (5th generation)", "iPad (10th generation)",
    "iPad mini (6th generation)",
]

# ── Geo → Language mapping ─────────────────────────────────────────────────

_GEO_LANG = {
    "7":   ("ru", "ru-RU"),    # Russia
    "380": ("uk", "uk-UA"),    # Ukraine
    "375": ("ru", "ru-BY"),    # Belarus
    "998": ("uz", "uz-UZ"),    # Uzbekistan
    "996": ("ru", "ru-KG"),    # Kyrgyzstan
    "992": ("ru", "ru-TJ"),    # Tajikistan
    "993": ("ru", "ru-TM"),    # Turkmenistan
    "994": ("az", "az-AZ"),    # Azerbaijan
    "995": ("ka", "ka-GE"),    # Georgia
    "373": ("ro", "ro-MD"),    # Moldova
    "374": ("hy", "hy-AM"),    # Armenia
    "372": ("et", "et-EE"),    # Estonia
    "371": ("lv", "lv-LV"),    # Latvia
    "370": ("lt", "lt-LT"),    # Lithuania
    "77":  ("kk", "kk-KZ"),    # Kazakhstan (77x)
    "49":  ("de", "de-DE"),    # Germany
    "44":  ("en", "en-GB"),    # UK
    "33":  ("fr", "fr-FR"),    # France
    "34":  ("es", "es-ES"),    # Spain
    "39":  ("it", "it-IT"),    # Italy
    "31":  ("nl", "nl-NL"),    # Netherlands
    "48":  ("pl", "pl-PL"),    # Poland
    "40":  ("ro", "ro-RO"),    # Romania
    "420": ("cs", "cs-CZ"),    # Czech
    "36":  ("hu", "hu-HU"),    # Hungary
    "46":  ("sv", "sv-SE"),    # Sweden
    "47":  ("nb", "nb-NO"),    # Norway
    "45":  ("da", "da-DK"),    # Denmark
    "358": ("fi", "fi-FI"),    # Finland
    "43":  ("de", "de-AT"),    # Austria
    "41":  ("de", "de-CH"),    # Switzerland
    "32":  ("fr", "fr-BE"),    # Belgium
    "30":  ("el", "el-GR"),    # Greece
    "351": ("pt", "pt-PT"),    # Portugal
    "90":  ("tr", "tr-TR"),    # Turkey
    "1":   ("en", "en-US"),    # USA/Canada
    "55":  ("pt", "pt-BR"),    # Brazil
    "54":  ("es", "es-AR"),    # Argentina
    "52":  ("es", "es-MX"),    # Mexico
    "51":  ("es", "es-PE"),    # Peru
    "57":  ("es", "es-CO"),    # Colombia
    "56":  ("es", "es-CL"),    # Chile
    "91":  ("hi", "hi-IN"),    # India
    "92":  ("ur", "ur-PK"),    # Pakistan
    "93":  ("fa", "fa-AF"),    # Afghanistan
    "86":  ("zh", "zh-CN"),    # China
    "82":  ("ko", "ko-KR"),    # South Korea
    "81":  ("ja", "ja-JP"),    # Japan
    "66":  ("th", "th-TH"),    # Thailand
    "84":  ("vi", "vi-VN"),    # Vietnam
    "62":  ("id", "id-ID"),    # Indonesia
    "60":  ("ms", "ms-MY"),    # Malaysia
    "65":  ("en", "en-SG"),    # Singapore
    "971": ("ar", "ar-AE"),    # UAE
    "966": ("ar", "ar-SA"),    # Saudi Arabia
    "972": ("he", "he-IL"),    # Israel
    "974": ("ar", "ar-QA"),    # Qatar
    "961": ("ar", "ar-LB"),    # Lebanon
    "20":  ("ar", "ar-EG"),    # Egypt
    "27":  ("en", "en-ZA"),    # South Africa
    "234": ("en", "en-NG"),    # Nigeria
}


def _detect_lang(phone: str) -> tuple:
    """Returns (lang_code, system_lang_code) based on phone prefix."""
    p = phone.lstrip("+")
    for prefix in sorted(_GEO_LANG.keys(), key=len, reverse=True):
        if p.startswith(prefix):
            return _GEO_LANG[prefix]
    return ("en", "en-US")


def _get_platform() -> str:
    """Returns current platform setting from user_settings."""
    try:
        import state
        for attr in dir(state):
            obj = getattr(state, attr, None)
            if hasattr(obj, 'user_settings'):
                return obj.user_settings.get('json_device', 'desktop_2040')
    except Exception:
        pass
    return 'desktop_2040'


# ── Device model file loading ──────────────────────────────────────────────

_device_cache: Dict[str, List[str]] = {}

def _load_device_file(filename: str, fallback: List[str]) -> List[str]:
    if filename in _device_cache:
        return _device_cache[filename]
    for base in [DATA_DIR, Path(__file__).parent.parent / "data"]:
        path = base / filename
        if path.exists():
            lines = [l.strip() for l in path.read_text(encoding="utf-8").splitlines() if l.strip()]
            if lines:
                _device_cache[filename] = lines
                return lines
    _device_cache[filename] = fallback
    return fallback


# ── Per-account device persistence ─────────────────────────────────────────

_DEVICE_MAP_FILE = DATA_DIR / "device_map.json"
_device_map: Optional[Dict] = None
_device_map_lock = threading.Lock()

def _load_device_map() -> Dict:
    global _device_map
    with _device_map_lock:
        if _device_map is not None:
            return _device_map
        if _DEVICE_MAP_FILE.exists():
            try:
                _device_map = json.loads(_DEVICE_MAP_FILE.read_text(encoding="utf-8"))
            except Exception:
                _device_map = {}
        else:
            _device_map = {}
        return _device_map

def _save_device_map():
    """Атомарно пишет device_map.json через tmp+replace. Без атомарности
    параллельный autoload (5 акков одновременно) при kill процесса оставлял
    битый JSON → у клиента сбрасывались все device fingerprints, Telegram
    каждому акку присылал «новый вход с устройства»."""
    with _device_map_lock:
        try:
            _DEVICE_MAP_FILE.parent.mkdir(parents=True, exist_ok=True)
            tmp = _DEVICE_MAP_FILE.with_suffix(".json.tmp")
            tmp.write_text(json.dumps(_device_map, ensure_ascii=False, indent=2), encoding="utf-8")
            import os as _os
            _os.replace(str(tmp), str(_DEVICE_MAP_FILE))
        except Exception as e:
            logger.warning("[device] failed to save device_map: %s", e)


def forget_device(phone: str) -> bool:
    """Удаляет запись устройства для номера. При следующем load_device
    сгенерируется новый девайс с актуальной версией Telegram Desktop.

    Нужно: клиент удалил аккаунт → добавляет заново → хочет новое устройство.
    Раньше брали закэшированное, и Telegram видел те же старые device_model/
    app_version (6.0.1 EP45-UD3LR), даже если наш список версий обновился."""
    import re as _re
    phone_digits = _re.sub(r"\D", "", phone or "")
    if not phone_digits:
        return False
    dm = _load_device_map()
    # Убираем и digit-only ключ, и возможный старый с "+"
    removed = False
    for _k in [phone_digits, f"+{phone_digits}", phone or ""]:
        if _k and _k in dm:
            dm.pop(_k, None)
            removed = True
    if removed:
        _save_device_map()
        logger.info("[device] Забыли устройство для %s — при следующем добавлении сгенерируется новое", phone_digits)
    return removed


def _probe_session_auth_key(session_path):
    """Читает sqlite .session напрямую и говорит есть ли там auth_key.
    Нужно для диагностики «не авторизован»: отличаем «файл пустой/повреждён»
    от «auth_key есть, но Telegram не принимает» (сессия использовалась с другого IP,
    .session залочен другим процессом и т.п.)."""
    try:
        from pathlib import Path as _P
        _p = _P(session_path)
        if _p.suffix != '.session':
            _p = _p.with_suffix('.session')
        if not _p.exists():
            return None, False
        # Открываем read-only копию чтобы не конфликтовать с активным Telethon-клиентом
        import sqlite3 as _sq
        # URI-open в read-only режиме
        _uri = f"file:{str(_p).replace(chr(92), '/')}?mode=ro&immutable=1"
        try:
            _conn = _sq.connect(_uri, uri=True, timeout=1.5)
        except Exception:
            _conn = _sq.connect(str(_p), timeout=1.5)
        try:
            _cur = _conn.cursor()
            _cur.execute("SELECT dc_id, auth_key FROM sessions LIMIT 1")
            _row = _cur.fetchone()
            if not _row:
                return None, False
            _dc = _row[0]
            _key = _row[1]
            _has = bool(_key) and (len(_key) >= 8 if isinstance(_key, (bytes, bytearray)) else True)
            return _dc, _has
        finally:
            try: _conn.close()
            except Exception: pass
    except Exception as e:
        logger.debug("[probe_session] %s: %s", session_path, e)
        return None, False


def load_device(phone: str = "") -> DeviceInfo:
    """Unique device per account. Platform-aware. Persisted in device_map.json."""
    platform = _get_platform()

    # Нормализуем ключ — иначе "+79587317853" и "79587317853" создают
    # два разных device entry и у Telegram слетает auth key (разный fingerprint
    # на одном session-файле). Ключ — только цифры, так согласуется и со
    # scheduler.py (phone_digits) и с database.py (phone как есть).
    import re as _re
    phone = _re.sub(r"\D", "", phone or "")

    if not phone:
        return _generate_device("0000000000", platform)

    dm = _load_device_map()
    # Миграция старых ключей с "+": если есть "+79..." — переносим на "79..."
    for _old_key in [k for k in list(dm.keys()) if k and k.startswith("+")]:
        _clean = _re.sub(r"\D", "", _old_key)
        if _clean and _clean not in dm:
            dm[_clean] = dm[_old_key]
        dm.pop(_old_key, None)

    if phone in dm:
        d = dm[phone]
        # Check if platform changed — regenerate if so
        if d.get("platform") == platform:
            return DeviceInfo(
                device_model=d["device_model"],
                system_version=d["system_version"],
                app_version=d["app_version"],
                system_lang_code=d.get("system_lang_code", "en-US"),
                lang_code=d.get("lang_code", "en"),
            )

    dev = _generate_device(phone, platform)

    dm[phone] = {
        "platform": platform,
        "device_model": dev.device_model,
        "system_version": dev.system_version,
        "app_version": dev.app_version,
        "lang_code": dev.lang_code,
        "system_lang_code": dev.system_lang_code,
    }
    _save_device_map()
    logger.info("[device] %s -> %s, %s, %s (%s)", phone, dev.device_model, dev.system_version, dev.app_version, platform)
    return dev


def _generate_device(phone: str, platform: str) -> DeviceInfo:
    """Generate device fingerprint based on platform and phone geo."""
    seed = int(hashlib.md5(phone.encode()).hexdigest(), 16)
    rng = random.Random(seed)

    lang_code, sys_lang_code = _detect_lang(phone)

    if platform == "android":
        models = _ANDROID_DEVICES
        device_model = rng.choice(models)
        system_version = rng.choice(_ANDROID_OS_VERSIONS)
        app_version = rng.choice(_ANDROID_VERSIONS)
    elif platform == "ios":
        models = _IOS_DEVICES
        device_model = rng.choice(models)
        system_version = rng.choice(_IOS_OS_VERSIONS)
        app_version = rng.choice(_IOS_VERSIONS)
    else:
        # desktop_2040
        models = _load_device_file("devices_2040.txt", ["PC 64bit"])
        device_model = rng.choice(models)
        system_version = rng.choice(_WINDOWS_VERSIONS)
        app_version = rng.choice(_DESKTOP_VERSIONS)

    return DeviceInfo(
        device_model=device_model,
        system_version=system_version,
        app_version=app_version,
        lang_code=lang_code,
        system_lang_code=sys_lang_code,
    )


_proxy_counter = 0
_proxy_counter_lock = threading.Lock()

def load_all_alive_proxies() -> list:
    """Возвращает полный список alive-прокси (ProxyResolved, addr, proto).
    Используется для перебора прокси при неудаче подключения — если основной
    не работает, пробуем следующий из пула вместо того чтобы сдаться сразу."""
    try:
        from config import DATA_DIR
        import json
        proxy_file = DATA_DIR / "proxies.json"
        if not proxy_file.exists():
            return []
        proxies = json.loads(proxy_file.read_text(encoding="utf-8"))
        out = []
        for p in proxies:
            if not p.get("alive"):
                continue
            addr = p.get("resolved_url") or p.get("addr", "")
            if not addr:
                continue
            res = ProxyResolved.parse(addr)
            if res:
                out.append((res, addr, p.get("proto", "?")))
        return out
    except Exception as e:
        logger.debug("load_all_alive_proxies failed: %s", e)
        return []


def load_proxy(phone: str = "") -> Optional[ProxyResolved]:
    """Загружает прокси с round-robin распределением.

    - 1 прокси, 5 аккаунтов → все через 1
    - 5 прокси, 5 аккаунтов → каждый через свой
    - 3 прокси, 7 аккаунтов → 1→P1, 2→P2, 3→P3, 4→P1, 5→P2...
    """
    global _proxy_counter

    # 1. Из файла proxies.json — round-robin по alive
    try:
        from config import DATA_DIR
        import json
        proxy_file = DATA_DIR / "proxies.json"
        if proxy_file.exists():
            proxies = json.loads(proxy_file.read_text(encoding="utf-8"))
            alive_proxies = []
            for p in proxies:
                if p.get("alive"):
                    addr = p.get("resolved_url") or p.get("addr", "")
                    if addr:
                        res = ProxyResolved.parse(addr)
                        if res:
                            alive_proxies.append((res, addr, p.get("proto", "?")))

            if alive_proxies:
                with _proxy_counter_lock:
                    idx = _proxy_counter % len(alive_proxies)
                    _proxy_counter += 1
                chosen, addr, proto = alive_proxies[idx]
                # Показываем host:port (без кредов) — клиентам ясно через какой
                # прокси идёт трафик. Раньше резали всё до 40 символов и получался
                # обрезок "proxy.pro" который выглядел как другой сервер.
                _host_port = addr
                try:
                    if "://" in _host_port:
                        _host_port = _host_port.split("://", 1)[1]
                    if "@" in _host_port:
                        _host_port = _host_port.split("@", 1)[1]
                except Exception:
                    pass
                logger.info("[proxy] Аккаунт %s → прокси #%d/%d: %s → %s",
                            phone or "?", idx + 1, len(alive_proxies), proto, _host_port)
                return chosen
    except Exception as e:
        logger.debug("Не удалось загрузить прокси из файла: %s", e)

    # 2. Из ENV
    try:
        from config import DEFAULT_PROXY_URL
        if DEFAULT_PROXY_URL:
            return ProxyResolved.parse(DEFAULT_PROXY_URL)
    except Exception as e:
        logger.debug("Прокси не загружен из ENV: %s", e)
    return None


# Глобальные локи для сериализации connect() по session-файлам
SESSION_LOCKS: Dict[str, asyncio.Lock] = {}
SESSION_SQL_LOCKS_T: Dict[str, threading.RLock] = {}


def get_api_credentials() -> tuple:
    """Returns (api_id, api_hash) from user settings or defaults."""
    try:
        import state
        ctx = getattr(state, '_owner_ctx', None)
        if ctx is None:
            # Try to get from any existing context
            for attr in dir(state):
                obj = getattr(state, attr, None)
                if hasattr(obj, 'user_settings'):
                    ctx = obj
                    break
        if ctx and hasattr(ctx, 'user_settings'):
            s = ctx.user_settings
            aid = s.get('api_id')
            ahash = s.get('api_hash')
            if aid and ahash:
                return (int(aid), str(ahash))
    except Exception:
        pass
    return (OFFICIAL_TDESKTOP_API_ID, OFFICIAL_TDESKTOP_APP_HASH)


def display_device(dev: DeviceInfo) -> str:
    return f"{dev.device_model}, {dev.system_version}, {dev.app_version}"


async def build_client(
    session_path: Path, dev: DeviceInfo, proxy_res: Optional[ProxyResolved],
    phone: Optional[str] = None,
) -> TelegramClient:

    # TLS-fingerprint pool: каждый аккаунт получает СТАБИЛЬНУЮ пару API_ID/HASH
    # из пула (default 6 пар + опционально юзерские). Раньше все акки шли с
    # одной парой OFFICIAL_TDESKTOP — TG видел одинаковый handshake на всех.
    # Маппинг персистится в data/api_pool_map.json: один раз выбрано → не менять
    # (иначе Telegram revoke'нет сессию).
    api_id, api_hash = None, None
    if phone:
        try:
            from core.api_pool import get_mapping_for_existing, get_api_for_phone
            # Если сессия УЖЕ существует — используем существующий маппинг
            # (если был) или НЕ трогаем (старая сессия с OFFICIAL_TDESKTOP).
            session_exists = Path(str(session_path) + ".session").exists() or Path(str(session_path)).exists()
            if session_exists:
                pair = get_mapping_for_existing(phone)
            else:
                # Новая сессия — берём из пула и запоминаем
                pair = get_api_for_phone(phone)
            if pair:
                api_id, api_hash, _platform = pair
        except Exception as e:
            logger.debug("[api_pool] resolve %s: %s", phone, e)

    if not api_id or not api_hash:
        api_id, api_hash = get_api_credentials()

    # Ensure parent dir exists for session file (Windows-safe)
    try:
        Path(session_path).parent.mkdir(parents=True, exist_ok=True)
    except Exception:
        pass

    kwargs = dict(
        device_model=dev.device_model,
        system_version=dev.system_version,
        app_version=dev.app_version,
        lang_code=dev.lang_code,
        system_lang_code=dev.system_lang_code,
        proxy=(proxy_res.pysocks_tuple if proxy_res else None),
        request_retries=5,  # было 3 → увеличили для надёжности
        connection_retries=5,  # количество попыток переподключения внутри Telethon
        retry_delay=2,  # пауза между попытками (сек)
        auto_reconnect=True,  # автоматический реконнект при обрыве
    )
    if "update_workers" in inspect.signature(TelegramClient.__init__).parameters:
        kwargs["update_workers"] = 1

    # Use patched SQLite session with WAL + busy_timeout and serialize connect() per session file
    try:
        sess = SQLiteSessionWAL(str(session_path))
    except ValueError as _ve:
        # Smart-loader для несовместимых форматов (Kraken/Raptor/TgHub).
        # Telethon SQLiteSession.__init__ падает с «too many values to unpack»
        # если служебная таблица `version` имеет другую структуру.
        # Решение: читаем auth_key/dc_id/server/port из таблицы `sessions`
        # напрямую (она обычно идентична во всех Telethon-форках), потом
        # пересоздаём session-файл в нашем формате с теми же креденшелами.
        # Аккаунт продолжает работать БЕЗ переавторизации.
        if "unpack" in str(_ve).lower():
            sess = _load_session_via_smart_loader(session_path)
            if sess is None:
                raise ValueError(
                    "session-файл несовместим — не удалось извлечь auth_key. "
                    "Переавторизуй аккаунт через «Аккаунты → Добавить»."
                ) from _ve
            logger.info("[smart-loader] %s: session импортирована из несовместимого формата",
                        Path(session_path).stem)
        else:
            raise
    lock = SESSION_LOCKS.setdefault(str(session_path), asyncio.Lock())

    async with lock:
        cli = TelegramClient(
            sess, api_id, api_hash, **kwargs
        )
        # Suppresses "Fatal error handling updates" for frozen/banned accounts
        try:
            def _quiet_handler(loop, ctx_):
                msg = ctx_.get("message", "")
                # Suppress only Telethon update-loop noise, log everything else
                noise_kw = ("Frozen", "PersistentTimestamp", "SecurityError", "very old message",
                            "ignored consecutively", "wrong session ID", "AuthKey")
                if any(k in str(msg) for k in noise_kw):
                    return
                exc = ctx_.get("exception")
                if exc and any(k in type(exc).__name__ for k in noise_kw):
                    return
                loop.default_exception_handler(ctx_)
            cli.loop.set_exception_handler(_quiet_handler)
        except Exception:
            pass

        # ── Robust connect: retry на SQLite lock + TimeoutError + Security errors ──
        max_attempts = 5
        last_exc: Optional[Exception] = None
        for attempt in range(1, max_attempts + 1):
            try:
                await cli.connect()
                last_exc = None
                break
            except sqlite3.OperationalError as e:
                last_exc = e
                if "locked" in str(e).lower() and attempt < max_attempts:
                    await asyncio.sleep(0.5 * attempt)
                    continue
                raise
            except (TimeoutError, asyncio.TimeoutError, ConnectionError, OSError) as e:
                last_exc = e
                # TimeoutError / network error — ждём и пробуем снова
                logger.debug(
                    "build_client [%s] attempt %d/%d: %s — retry in %ds",
                    session_path.name,
                    attempt,
                    max_attempts,
                    type(e).__name__,
                    attempt * 2,
                )
                if attempt < max_attempts:
                    await asyncio.sleep(attempt * 2)
                    # Пересоздаём сессию на случай если она "застряла"
                    try:
                        await safe_disconnect(cli)
                    except Exception:
                        pass
                    cli = TelegramClient(
                        sess,
                        api_id,
                        api_hash,
                        **kwargs,
                    )
                    try:
                        cli.loop.set_exception_handler(_quiet_handler)
                    except Exception:
                        pass
                    continue
                raise
            except Exception as e:
                last_exc = e
                _emsg = str(e)
                # Security errors ("wrong session ID", "very old message") —
                # обычно решаются переподключением с небольшой паузой
                if any(
                    k in _emsg
                    for k in (
                        "session ID",
                        "old message",
                        "Security",
                        "ignored consecutively",
                    )
                ):
                    logger.debug(
                        "build_client [%s] attempt %d/%d: security noise — retry in %ds",
                        session_path.name,
                        attempt,
                        max_attempts,
                        attempt * 3,
                    )
                    if attempt < max_attempts:
                        await asyncio.sleep(attempt * 3)
                        try:
                            await safe_disconnect(cli)
                        except Exception:
                            pass
                        cli = TelegramClient(
                            sess,
                            OFFICIAL_TDESKTOP_API_ID,
                            OFFICIAL_TDESKTOP_APP_HASH,
                            **kwargs,
                        )
                        try:
                            cli.loop.set_exception_handler(_quiet_handler)
                        except Exception:
                            pass
                        continue
                # Non-retryable error — cleanup before raising
                try:
                    await safe_disconnect(cli)
                except Exception:
                    pass
                raise

        # Extra safeguard: re-apply PRAGMAs after connect (no harm if already applied)
        try:
            sess2 = getattr(cli, "session", None)
            conn = getattr(sess2, "_conn", None)
            if conn:
                try:
                    conn.execute("PRAGMA busy_timeout=5000")
                    conn.execute("PRAGMA journal_mode=MEMORY")
                    conn.execute("PRAGMA synchronous=NORMAL")
                except Exception:
                    pass
        except Exception:
            pass
    return cli


async def safe_disconnect(client):
    """Safely disconnect a Telethon client and close its SQLite connection to release file locks."""
    _lock_key = None
    _session_path = None
    try:
        sess = getattr(client, "session", None)
        _lock_key = getattr(sess, "_lock_key", None)
        _session_path = getattr(sess, "filename", None)
    except Exception:
        pass
    try:
        if client:
            # Зачищаем _handle_auto_reconnect-таску ДО disconnect — Telethon
            # запускает её внутри _handle_reconnect и при disconnect её только
            # отменяет, но никогда не await'ит. Без явного await + GC видит
            # «pending Task» → лог-спам «Task was destroyed but it is pending!»
            # ×30+ за рестарт у клиентов с 5+ акков.
            for attr in ("_reconnect_task", "_loop_handle", "_updates_handle",
                         "_keepalive_handle"):
                t = getattr(client, attr, None)
                if t is not None and hasattr(t, "cancel") and not getattr(t, "done", lambda: False)():
                    try:
                        t.cancel()
                        try:
                            await asyncio.wait_for(t, timeout=0.5)
                        except (asyncio.CancelledError, asyncio.TimeoutError, Exception):
                            pass
                    except Exception:
                        pass
            await client.disconnect()
            # Telethon disconnect() триггерит cancel() для send/recv-loops,
            # но не дожидается их завершения. Без этого они остаются «pending»
            # до GC, что даёт Python warning «Task was destroyed but it is pending».
            await asyncio.sleep(0.3)
            sender = getattr(client, "_sender", None)
            if sender:
                for tn in ("_send_loop_task", "_recv_loop_task", "_send_task", "_recv_task"):
                    t = getattr(sender, tn, None)
                    if t and not t.done():
                        try:
                            t.cancel()
                            await asyncio.wait_for(t, timeout=1.0)
                        except (asyncio.CancelledError, asyncio.TimeoutError, Exception):
                            pass
            # Connection-level loops (под sender'ом).
            conn_obj = getattr(sender, "_connection", None) if sender else None
            if conn_obj:
                for tn in ("_send_task", "_recv_task"):
                    t = getattr(conn_obj, tn, None)
                    if t and not t.done():
                        try:
                            t.cancel()
                            await asyncio.wait_for(t, timeout=0.5)
                        except (asyncio.CancelledError, asyncio.TimeoutError, Exception):
                            pass
    except Exception:
        pass
    try:
        sess = getattr(client, "session", None)
        conn = getattr(sess, "_conn", None)
        if conn:
            try:
                conn.close()
            except Exception:
                pass
    except Exception:
        pass
    # Зачистка спутников от старого WAL mode (-wal, -shm) + случайного journal
    # (DELETE mode должен удалять -journal сам, но на всякий — подчистим).
    if _session_path:
        for suffix in ("-wal", "-shm", "-journal"):
            try:
                p = Path(str(_session_path) + suffix)
                if p.exists():
                    p.unlink()
            except Exception:
                pass
    # Cleanup per-session locks to free memory
    if _lock_key:
        SESSION_LOCKS.pop(str(_lock_key), None)
        SESSION_SQL_LOCKS_T.pop(str(_lock_key), None)


# Отложенные задачи на дисконнект: phone -> asyncio.Task
# Нужны чтобы gloss-period (grace 45с) поглощал короткие паузы между задачами
# и не делать лишних handshake'ов при рапид Старт/Стоп/Старт.
_RELEASE_GRACE_SEC = 45
_pending_releases: Dict[str, "asyncio.Task"] = {}


def _release_key(acc) -> str:
    return str(getattr(acc, "phone", None) or id(acc))


def _cancel_pending_release(acc) -> bool:
    """Отменяет отложенный дисконнект. True если что-то отменили."""
    key = _release_key(acc)
    task = _pending_releases.pop(key, None)
    if task and not task.done():
        task.cancel()
        return True
    return False


async def release_after_task(acc, grace_sec: int = 0):
    """Отключает аккаунт от Telegram СРАЗУ после окончания задачи.
    Раньше был grace 45с (для переиспользования коннекта если следующая задача
    стартовала в течение 45с), но клиент не хотел чтобы коннект висел — теперь
    аккаунт ложится в idle моментально, без таймера.
    Параметр grace_sec оставлен для совместимости с вызывающим кодом, но игнорируется."""
    cli = getattr(acc, "client", None)
    if cli is None and not getattr(acc, "idle", False):
        return

    # Отменяем любой ранее запланированный pending release (от старых версий)
    _cancel_pending_release(acc)

    phone = getattr(acc, "phone", "?")
    try:
        if cli:
            await safe_disconnect(cli)
    except Exception as e:
        logger.debug("[release] disconnect %s: %s", phone, e)
    try:
        acc.client = None
        acc.idle = True
        acc.idle_since = time.time()
    except Exception:
        pass
    # Отключение — внутренняя операция. Клиенту показывать её в консоли нет смысла,
    # он и так знает что задача закончилась. Оставляем только в debug-файле.
    logger.debug("[release] %s отключён", phone)


async def ensure_connected(acc) -> bool:
    """
    Гарантирует что аккаунт подключён к Telegram.
    Если аккаунт в idle-режиме — автоматически переподключает.
    Мёртвой сессия помечается ТОЛЬКО при AuthKeyUnregistered / SessionRevoked.
    Остальные ошибки — временные, аккаунт просто не подключится сейчас.
    Возвращает True если аккаунт готов к работе.
    """
    # Уже frozen — не трогаем. Dead — тоже не трогаем, НО:
    # если это soft-dead (dead_reason="unauthorized"), даём шанс ожить —
    # такие dead-пометки часто ложные (сеть моргнула / прокси залип),
    # а сессия реально жива. Юзер через UI «Проверить» сможет поднять заново.
    if getattr(acc, "frozen", False):
        acc.last_fail_reason = "аккаунт в заморозке"
        return False
    if getattr(acc, "dead", False):
        reason = getattr(acc, "dead_reason", "") or ""
        if reason == "session_revoked":
            acc.last_fail_reason = "сессия отозвана Telegram — переавторизуй"
        else:
            acc.last_fail_reason = "аккаунт помечен мёртвым — проверь статус"
        return False

    # Обновляем время последнего использования
    if hasattr(acc, "touch"):
        acc.touch()

    # Если висит pending disconnect — отменим: задача стартует снова,
    # переиспользуем существующий коннект без handshake'а.
    try:
        if _cancel_pending_release(acc):
            logger.debug("[release] %s — отменили отложенный disconnect", getattr(acc, "phone", "?"))
    except Exception:
        pass

    cli = getattr(acc, "client", None)

    # Проверяем нужно ли переподключение
    needs_reconnect = False
    if getattr(acc, "idle", False):
        needs_reconnect = True
    elif cli is None:
        needs_reconnect = True
    elif hasattr(cli, "is_connected") and not cli.is_connected():
        needs_reconnect = True

    if not needs_reconnect:
        return True

    phone = getattr(acc, "phone", "?")

    # Берём настройки подключения из user_settings.
    # connect_retries = кол-во попыток (default 2), proxy_timeout = таймаут connect (default 30с).
    _user_settings = {}
    try:
        import state as _st
        for _c in _st.owners.values():
            if hasattr(_c, "user_settings"):
                _user_settings = _c.user_settings
                break
    except Exception:
        pass
    MAX_RETRIES = max(1, min(10, int(_user_settings.get("connect_retries", 2) or 2)))
    _proxy_timeout = max(5, min(120, int(_user_settings.get("proxy_timeout", 30) or 30)))

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            # Техническая строка «попытка 1/2» — только в debug-файл, не в UI-консоль.
            # В UI клиент видит только итоговый статус (✅ жив / 💀 мёртв / 🔌 прокси).
            logger.debug(
                "[ensure_connected] 🔌 Пробуждение аккаунта %s (попытка %d/%d)",
                phone, attempt, MAX_RETRIES,
            )

            # Отключаем старый клиент если есть
            if cli:
                try:
                    await safe_disconnect(cli)
                except Exception:
                    pass
                cli = None

            # Строим новый клиент
            dev = getattr(acc, "dev", None)
            proxy = getattr(acc, "proxy", None)
            session_path = getattr(acc, "session_path", None)

            if not session_path or not dev:
                logger.warning("⚠ %s Нет файла сессии или device_info — переподключи аккаунт", phone)
                acc.last_fail_reason = "нет файла сессии или device_info — переподключи аккаунт"
                return False

            # Если прокси не назначен — загружаем из списка (round-robin)
            if proxy is None:
                proxy = load_proxy(phone)
                if proxy:
                    acc.proxy = proxy

            # Таймаут прокси (user_settings.proxy_timeout) — чтобы не висеть вечно
            # если прокси мёртв. Раньше был hardcoded внутри build_client.
            try:
                new_cli = await asyncio.wait_for(
                    build_client(session_path, dev, proxy, phone=phone),
                    timeout=float(_proxy_timeout),
                )
            except asyncio.TimeoutError:
                if attempt < MAX_RETRIES:
                    logger.debug("[ensure_connected] %s: connect таймаут %dс, попытка %d", phone, _proxy_timeout, attempt)
                    await asyncio.sleep(2)
                    continue
                acc.last_fail_reason = f"Прокси не ответил за {_proxy_timeout} сек — проверь интернет/прокси"
                logger.warning("🔌 %s Прокси не отвечает (таймаут %d сек)", phone, _proxy_timeout)
                return False

            # Проверяем авторизацию — AuthKey ошибки ловим отдельно
            try:
                authorized = await new_cli.is_user_authorized()
            except (errors.AuthKeyUnregisteredError, errors.SessionRevokedError):
                await safe_disconnect(new_cli)
                acc.dead = True
                acc.dead_reason = "session_revoked"
                acc.last_fail_reason = "сессия отозвана Telegram — переавторизуй"
                logger.warning("💀 %s Аккаунт мёртвый (сессия отозвана Telegram)", phone)
                return False
            except errors.AuthKeyDuplicatedError:
                await safe_disconnect(new_cli)
                acc.dead = True
                acc.dead_reason = "auth_key_duplicated"
                acc.last_fail_reason = "сессия использовалась с двух IP — переавторизуй или смени прокси"
                logger.warning("💀 %s Аккаунт заблокирован (использовался с двух IP одновременно)", phone)
                return False

            if not authorized:
                # is_user_authorized() вернул False. Это НЕ всегда значит что сессия
                # мёртвая — часто Telethon просто не успел прочитать auth_key из
                # SQLite .session файла (блокировка другим процессом: Raptor, Telegram
                # Desktop, параллельный TelePush и т.п.). Двойная проверка через
                # get_me() — если он реально возвращает юзера, значит авторизация
                # валидна и первый False был ложным.
                real_me = None
                _me_err_type = ""
                _me_err_msg = ""
                try:
                    real_me = await asyncio.wait_for(new_cli.get_me(), timeout=10.0)
                    if real_me is None:
                        # get_me вернул None без исключения — Telethon так делает
                        # когда сервер не опознал auth_key (сессия де-факто отозвана,
                        # но RPC не кинул AuthKeyUnregistered). Трактуем как revoked.
                        _me_err_type = "AuthKeyRevokedSilent"
                        _me_err_msg = (
                            "сервер не признал auth_key (get_me вернул None) — "
                            "сессия отозвана, хотя Telegram не кинул явную ошибку"
                        )
                except errors.AuthKeyUnregisteredError as _e:
                    _me_err_type = "AuthKeyUnregistered"
                    _me_err_msg = "ключ отозван Telegram"
                except errors.SessionRevokedError as _e:
                    _me_err_type = "SessionRevoked"
                    _me_err_msg = "сессия отозвана Telegram"
                except errors.AuthKeyDuplicatedError as _e:
                    _me_err_type = "AuthKeyDuplicated"
                    _me_err_msg = "сессия использована с двух IP одновременно"
                except errors.UserDeactivatedError as _e:
                    _me_err_type = "UserDeactivated"
                    _me_err_msg = "аккаунт удалён пользователем"
                except errors.UserDeactivatedBanError as _e:
                    _me_err_type = "UserDeactivatedBan"
                    _me_err_msg = "аккаунт забанен Telegram"
                except errors.FloodWaitError as _e:
                    _me_err_type = f"FloodWait({_e.seconds}s)"
                    _me_err_msg = f"FloodWait {_e.seconds} сек"
                except asyncio.TimeoutError:
                    _me_err_type = "Timeout"
                    _me_err_msg = "get_me не ответил за 10 сек (прокси тормозит или DC недоступен)"
                except Exception as _me_e:
                    # FrozenMethodInvalidError — отдельный класс с 1.5.x Telethon.
                    # Не у всех билдов есть errors.FrozenMethodInvalidError, ловим
                    # по имени класса.
                    _me_err_type = type(_me_e).__name__
                    _me_err_msg = str(_me_e)[:120] or _me_err_type
                    if "Frozen" in _me_err_type:
                        # Аккаунт реально заморожен — get_me() прошёл бы, но
                        # сейчас Telegram кинул frozen → помечаем сразу.
                        acc.frozen = True
                        acc.frozen_reason = _me_err_type
                        try:
                            from core.account_state import save_account_state
                            save_account_state(phone, frozen=True, frozen_reason=_me_err_type)
                        except Exception:
                            pass
                        logger.warning("❄ %s Аккаунт заморожен (%s)", phone, _me_err_type)
                        return False

                # Дополнительно: если get_me вернул юзера, но он restricted
                # (Telegram сам помечает заморозкой через флаг) — тоже frozen.
                if real_me is not None and getattr(real_me, "restricted", False):
                    acc.frozen = True
                    acc.frozen_reason = "restricted_by_telegram"
                    try:
                        from core.account_state import save_account_state
                        save_account_state(phone, frozen=True, frozen_reason="restricted_by_telegram")
                    except Exception:
                        pass
                    logger.warning("❄ %s Аккаунт ограничен Telegram (restricted=True)", phone)
                    return False

                if real_me is not None:
                    logger.info(
                        "✅ %s Аккаунт жив (восстановлен после ложной блокировки сессии)",
                        phone,
                    )
                    # Продолжаем — акк реально жив
                    authorized = True
                    # Помечаем что аккаунт реально верифицирован — get_me вернул валидного юзера.
                    # Это переключает UI с «Не проверялся» на нормальный статус.
                    acc.verified = True
                    # Если был помечен dead с транзиентной причиной — снимаем флаг
                    acc.dead = False
                    acc.dead_reason = None
                else:
                    # Диагностика: проверяем сам session file — есть ли там auth_key вообще
                    _dc_id, _auth_key_present = _probe_session_auth_key(session_path)
                    await safe_disconnect(new_cli)
                    # Критические ошибки — сразу помечаем dead, без повторных попыток
                    if _me_err_type in ("AuthKeyUnregistered", "SessionRevoked", "AuthKeyRevokedSilent"):
                        acc.dead = True
                        acc.dead_reason = _me_err_type.lower()
                        acc.last_fail_reason = "сессия отозвана Telegram — переавторизуй"
                        logger.warning("💀 %s Аккаунт мёртвый (сессия отозвана Telegram)", phone)
                        # Тех. детали для саппорта — в debug-файл, не в UI
                        logger.debug("[ensure_connected] %s tech: auth_key_in_file=%s dc=%s get_me=%s",
                                     phone, _auth_key_present, _dc_id, _me_err_type)
                        return False
                    if _me_err_type == "UserDeactivated":
                        acc.dead = True
                        acc.dead_reason = "user_deactivated"
                        acc.last_fail_reason = "аккаунт удалён пользователем"
                        logger.warning("💀 %s Аккаунт удалён пользователем", phone)
                        return False
                    if _me_err_type == "UserDeactivatedBan":
                        acc.dead = True
                        acc.dead_reason = "user_banned"
                        acc.last_fail_reason = "аккаунт забанен Telegram"
                        logger.warning("💀 %s Аккаунт забанен Telegram", phone)
                        return False
                    if _me_err_type == "AuthKeyDuplicated":
                        acc.dead = True
                        acc.dead_reason = "auth_key_duplicated"
                        acc.last_fail_reason = "сессия использовалась с двух IP — переавторизуй или смени прокси"
                        logger.warning("💀 %s Аккаунт заблокирован (использовался с двух IP одновременно)", phone)
                        return False
                    if _me_err_type.startswith("FloodWait"):
                        import re as _re
                        _m = _re.search(r"(\d+)", _me_err_type)
                        _sec = int(_m.group(1)) if _m else 0
                        acc.floodwait_until = time.time() + _sec
                        acc.floodwait_seconds = _sec
                        acc.last_fail_reason = _me_err_msg
                        logger.warning("⏳ %s FloodWait %d сек", phone, _sec)
                        return False
                    if _me_err_type == "Timeout":
                        acc.last_fail_reason = "прокси не отвечает"
                        logger.warning("🔌 %s Прокси не отвечает (таймаут 10 сек)", phone)
                        return False

                    if attempt < MAX_RETRIES:
                        logger.debug(
                            "[ensure_connected] %s не авторизован, попытка %d (auth_key=%s dc=%s get_me=%s: %s)",
                            phone, attempt, _auth_key_present, _dc_id, _me_err_type or "False", _me_err_msg or "",
                        )
                        await asyncio.sleep(2)
                        continue
                    else:
                        acc.unauth_count = getattr(acc, "unauth_count", 0) + 1
                        if not _auth_key_present:
                            acc.last_fail_reason = (
                                "В файле сессии нет auth_key — сессия повреждена. "
                                "Скопируй .session+.json заново."
                            )
                            logger.warning("💀 %s Аккаунт мёртвый (файл сессии повреждён — нет auth_key)", phone)
                        else:
                            acc.last_fail_reason = (
                                "Telegram не подтвердил авторизацию. Возможно сессия открыта в другом софте "
                                "(закрой Raptor/TG Desktop), прокси моргнул или сессия с разных IP."
                            )
                            logger.warning("⚠ %s Не подключается (возможно сессия занята другим софтом или прокси тупит)", phone)
                        # Тех. детали для саппорта — в debug
                        logger.debug(
                            "[ensure_connected] %s tech: auth_key_in_file=%s dc=%s get_me=%s: %s unauth_count=%d",
                            phone, _auth_key_present, _dc_id,
                            _me_err_type or "False", _me_err_msg or "", acc.unauth_count,
                        )
                        return False

            # Успех — аккаунт подключён
            acc.client = new_cli
            acc.idle = False
            acc.idle_since = 0.0
            acc.unauth_count = 0  # Сбрасываем счётчик неудач
            acc.connected_at = time.time()
            # Одна короткая строка в UI — как в Raptor
            logger.info("✅ %s Аккаунт жив", phone)
            # Дубль в debug-файл для саппорта
            logger.debug("[ensure_connected] Аккаунт %s проснулся", phone)
            return True

        except errors.FloodWaitError as fw:
            logger.warning(
                "⏳ %s FloodWait %d сек",
                phone, fw.seconds,
            )
            acc.floodwait_until = time.time() + fw.seconds
            acc.floodwait_seconds = fw.seconds
            acc.last_fail_reason = f"FloodWait от Telegram: {fw.seconds} сек. Подожди и попробуй снова"
            return False

        except (errors.AuthKeyUnregisteredError, errors.SessionRevokedError):
            acc.dead = True
            acc.dead_reason = "session_revoked"
            acc.last_fail_reason = "сессия отозвана Telegram — переавторизуй"
            logger.warning("💀 %s Аккаунт мёртвый (сессия отозвана Telegram)", phone)
            return False

        except errors.AuthKeyDuplicatedError:
            acc.dead = True
            acc.dead_reason = "auth_key_duplicated"
            acc.last_fail_reason = "сессия использовалась с двух IP — переавторизуй или смени прокси"
            logger.warning("💀 %s Аккаунт заблокирован (использовался с двух IP одновременно)", phone)
            return False

        except (ConnectionError, OSError, TimeoutError, asyncio.TimeoutError) as e:
            # Сетевая ошибка — не мёртвая, просто не подключились
            if attempt < MAX_RETRIES:
                logger.debug("[ensure_connected] сеть %s: %s, повтор через 3 сек", phone, type(e).__name__)
                await asyncio.sleep(3)
                continue
            acc.last_fail_reason = (
                f"Сетевая ошибка ({type(e).__name__}) — проверь интернет или смени прокси"
            )
            logger.warning("🔌 %s Прокси не отвечает (%s)", phone, type(e).__name__)
            return False

        except Exception as e:
            # Несовместимый формат session-файла (от другого софта) — помечаем
            # акк как «требует переавторизации» а не просто disconnected.
            _es = str(e)
            if "session-файл несовместим" in _es or "unpack" in _es.lower():
                acc.dead = True
                acc.dead_reason = "session_format_incompatible"
                acc.last_fail_reason = "Session-файл несовместим с Telethon — переавторизуй акк (вкладка Аккаунты → Переавторизация)"
                logger.warning("❌ %s Несовместимый session-файл (формат от другого софта)", phone)
                return False
            acc.last_fail_reason = f"{type(e).__name__}: {str(e)[:120]}"
            logger.warning("❌ %s Ошибка подключения [%s]: %s", phone, type(e).__name__, e, exc_info=True)
            return False

    return False


def _load_session_via_smart_loader(session_path):
    """Smart-импорт session-файла от несовместимого софта (Kraken/Raptor/TgHub).

    Telethon SQLiteSession падает на init если служебная таблица `version`
    имеет другое количество столбцов. Но **таблица `sessions`** во всех
    Telethon-форках одинакова: dc_id, server_address, port, auth_key, takeout_id.
    Читаем auth_key напрямую через sqlite3 и пересоздаём session-файл в
    нашем формате — аккаунт работает без переавторизации.

    Возвращает SQLiteSessionWAL или None если не получилось.
    """
    import sqlite3 as _s
    from pathlib import Path
    p = Path(session_path)
    if not p.exists():
        return None

    auth_key_blob = None
    dc_id = server = port = None
    takeout_id = None
    try:
        conn = _s.connect(str(p), timeout=5)
        cur = conn.cursor()
        # Пробуем прочитать sessions — стандарт для Telethon-форков
        try:
            row = cur.execute(
                "SELECT dc_id, server_address, port, auth_key, takeout_id FROM sessions LIMIT 1"
            ).fetchone()
        except Exception:
            # 4-колоночный вариант (Telethon < 1.18)
            try:
                row = cur.execute(
                    "SELECT dc_id, server_address, port, auth_key FROM sessions LIMIT 1"
                ).fetchone()
                if row:
                    row = (*row, None)
            except Exception:
                row = None
        if row:
            dc_id, server, port, auth_key_blob, takeout_id = row
        conn.close()
    except Exception as _e:
        logger.warning("[smart-loader] %s: не открыть как SQLite: %s", p.stem, _e)
        return None

    if not auth_key_blob or not dc_id or not server:
        logger.warning("[smart-loader] %s: пустые поля в sessions (auth_key=%s)",
                       p.stem, "yes" if auth_key_blob else "no")
        return None

    # Бэкапим оригинальный файл и удаляем — Telethon создаст свежий.
    # 3 попытки с задержкой против WinError 32 (файл занят).
    bak = p.with_suffix(p.suffix + ".bak")
    if bak.exists():
        try: bak.unlink()
        except Exception: pass
    _last_be = None
    _renamed = False
    # Раньше было 3 попытки × 0/0.5/1с = 1.5с total — слишком мало для антивируса
    # (Defender держит файл 3-5с при первом scan'е). Расширили до 5 попыток с
    # экспоненциальным бэкофом 0/0.5/1/2/4 = 7.5с — у клиентов ушло ~80%
    # «backup не удался» из логов.
    for _attempt in range(5):
        if _attempt > 0:
            import time as _t
            _t.sleep(min(0.5 * (2 ** (_attempt - 1)), 4.0))
        try:
            p.rename(bak)
            _renamed = True
            break
        except Exception as _be:
            _last_be = _be
    if not _renamed:
        # Файл стабильно залочен (антивирус, открытая сессия).
        # Возвращаем None → код выше пойдёт через обычный Telethon-путь,
        # который сам разберётся (или покажет понятную ошибку «несовместимый формат»).
        logger.warning("[smart-loader] %s: backup не удался после 5 попыток: %s. Пытаемся загрузить через обычный путь.", p.stem, _last_be)
        return None
    # journal-файлы тоже почистим, иначе SQLite может прицепиться к старому
    for _suff in ("-journal", "-wal", "-shm"):
        _j = p.with_name(p.name + _suff)
        if _j.exists():
            try: _j.unlink()
            except Exception: pass
    logger.info("[smart-loader] %s: оригинал переименован в %s.bak", p.stem, p.name)

    # Создаём свежую session нашего формата с теми же auth_key + dc
    try:
        new_sess = SQLiteSessionWAL(str(p))
        new_sess.set_dc(int(dc_id), str(server), int(port))
        new_sess.auth_key = type(new_sess.auth_key)(data=auth_key_blob) \
            if hasattr(new_sess, "auth_key") and hasattr(new_sess.auth_key, "key") \
            else None
        # Альтернатива: используем raw присваивание через AuthKey
        try:
            from telethon.crypto import AuthKey
            new_sess.auth_key = AuthKey(data=auth_key_blob)
        except Exception:
            pass
        if takeout_id is not None:
            try:
                new_sess.takeout_id = int(takeout_id)
            except Exception:
                pass
        new_sess.save()
        return new_sess
    except Exception as _ce:
        logger.error("[smart-loader] %s: создание свежей session не удалось: %s",
                     p.stem, _ce)
        # Откатываем бэкап
        try:
            bak = p.with_suffix(p.suffix + ".bak")
            if bak.exists() and not p.exists():
                bak.rename(p)
        except Exception:
            pass
        return None


# and by serializing writes to the session DB per-session file across this process.
class SQLiteSessionWAL(SQLiteSession):
    def __init__(self, filename: str):
        # Important: call parent to initialize internal connection and structures
        super().__init__(str(filename))
        # Prepare a per-file lock key for this session
        try:
            self._lock_key = os.path.abspath(str(filename))
        except Exception:
            self._lock_key = str(filename)
        # Prepare a process-wide re-entrant lock for this session database path
        if self._lock_key not in SESSION_SQL_LOCKS_T:
            import threading as _thr

            SESSION_SQL_LOCKS_T[self._lock_key] = _thr.RLock()
        # Apply PRAGMAs right after session creation
        self._tune_pragmas()

    def _tune_pragmas(self) -> None:
        conn = getattr(self, "_conn", None)
        if not conn:
            return
        try:
            # journal_mode=DELETE — классический rollback journal. Во время активной
            # записи создаётся короткоживущий .session-journal, после коммита удаляется.
            # Раньше пробовали MEMORY (журнал в RAM, папка чище) — но при быстром
            # закрытии программы auth_key не успевал flush-нуться на диск и клиент
            # терял только что добавленный аккаунт. DELETE + synchronous=FULL
            # гарантируют что каждый commit дошёл до диска (fsync).
            conn.execute("PRAGMA journal_mode=DELETE")
        except Exception:
            pass
        try:
            # 10 сек вместо 5: дольше ждём пока другой поток отпустит lock,
            # вместо immediate OperationalError. Особенно важно на shutdown
            # path (keepalive_loop пытается save state одновременно с disconnect).
            conn.execute("PRAGMA busy_timeout=10000")
        except Exception:
            pass
        try:
            conn.execute("PRAGMA locking_mode=NORMAL")
        except Exception:
            pass
        try:
            # synchronous=FULL — fsync на каждый commit. Медленнее на микросекунды,
            # но гарантирует сохранность при внезапном завершении процесса.
            # Критично для auth_key после sign_in.
            conn.execute("PRAGMA synchronous=FULL")
        except Exception:
            pass

    # Telethon writes to the 'sessions' table during connect() via this method.
    def _execute(self, stmt: str, *values):
        """
        Wrapper around SQLite executes with per-file lock serialization.
        busy_timeout PRAGMA handles waiting + 1 manual retry на shutdown race
        (keepalive_loop save_state одновременно с disconnect → database is locked).
        """
        import time as _t
        lock = SESSION_SQL_LOCKS_T.get(getattr(self, "_lock_key", None))
        head = (stmt or "").lstrip().lower()
        is_read = head.startswith("select") or head.startswith("pragma")
        last_err = None
        for attempt in range(2):
            try:
                if lock:
                    with lock:
                        c = getattr(self, "_conn").cursor()
                        c.execute(stmt, values)
                else:
                    c = getattr(self, "_conn").cursor()
                    c.execute(stmt, values)
                if is_read:
                    return c.fetchone()
                try:
                    getattr(self, "_conn").commit()
                except Exception:
                    pass
                return None
            except sqlite3.OperationalError as _oe:
                last_err = _oe
                emsg = str(_oe).lower()
                # Только для transient lock errors retry. Для остальных —
                # сразу raise чтобы не маскировать настоящие баги.
                if "locked" not in emsg and "busy" not in emsg:
                    raise
                if attempt == 0:
                    _t.sleep(0.3)
                    continue
                # Второй attempt тоже не помог — для writes на shutdown path
                # лучше тихо проглотить (данные не критичные, Telethon заново
                # синхронизирует state при reconnect). Для reads — поднимаем.
                if is_read:
                    raise
                # NB: НЕ пишем sessions/auth_key через этот путь; auth_key
                # пишется через _update_session_table выше — там тот же retry.
                return None
        if last_err:
            raise last_err
        return None

    def _update_session_table(self, *a, **kw):
        """Single attempt — busy_timeout handles waiting."""
        lock = SESSION_SQL_LOCKS_T.get(getattr(self, "_lock_key", None))
        if lock:
            with lock:
                return super()._update_session_table(*a, **kw)
        return super()._update_session_table(*a, **kw)

    def save(self, *args, **kwargs):
        lock = SESSION_SQL_LOCKS_T.get(getattr(self, "_lock_key", None))
        if lock:
            with lock:
                return super().save(*args, **kwargs)
        return super().save(*args, **kwargs)

    def process_entities(self, tlo):
        """
        Catch 'database is locked' — entity cache loss is non-fatal.
        Single attempt, no retry/sleep to avoid blocking event loop.
        """
        lock = SESSION_SQL_LOCKS_T.get(getattr(self, "_lock_key", None))
        try:
            if lock:
                with lock:
                    return super().process_entities(tlo)
            else:
                return super().process_entities(tlo)
        except sqlite3.OperationalError as e:
            logger.debug("[SQLiteSessionWAL] process_entities failed: %s", e)
            return
        except Exception as e:
            logger.debug("[SQLiteSessionWAL] process_entities unexpected: %s", e)
            return

    def _get_connection_and_cursor(self):
        """Ensure WAL mode is active on every new connection Telethon opens."""
        result = (
            super()._get_connection_and_cursor()
            if hasattr(super(), "_get_connection_and_cursor")
            else None
        )
        self._tune_pragmas()
        return result


# Global, per-session file asyncio locks to serialize connect/open against the same .session file.
# Per-session-file locks for SQLite writes (sync context)
# ---------------------- QUEUE: join job ----------------------
