"""
scheduler.py — планировщик задач и автозагрузка сессий.
Каждая задача теперь запускается через per-user worker pool для изоляции.
"""

import asyncio
import json
import time
from typing import Optional

from telethon import errors

from bot_ui.logger_setup import logger
from config import USERS_DIR
from models import Account, JoinJob, SpamJob, OwnerCtx
from core.database import store, build_output_json
from core.client_builder import build_client, safe_disconnect
from core.account_helpers import refresh_known_ids
from core.join import run_join_job, deduplicate_links
from core.spam import run_spam_job, collect_writable_groups
from bot_ui.ui import get_ctx
import state


async def scheduler_loop():
    """
    Главный цикл планировщика.
    Каждые 10 секунд проверяет задачи и запускает их через per-user пул.
    Каждые 5 минут чистит память от завершённых задач и idle-контекстов.
    """
    await asyncio.sleep(1.0)
    logger.info("[scheduler] Планировщик запущен")

    _last_cleanup = time.time()
    _last_idle_check = time.time()
    _last_job_save = time.time()
    _last_heartbeat = time.time()
    _last_monitor_watch = time.time()
    _CLEANUP_INTERVAL = 300  # 5 минут
    _IDLE_CHECK_INTERVAL = 60  # 1 минута
    _JOB_SAVE_INTERVAL = 60  # 1 минута — сохраняем задачи часто (защита от force kill)
    _HEARTBEAT_INTERVAL = 300  # 5 минут
    _MONITOR_WATCH_INTERVAL = 120  # 2 минуты — проверяем не помер ли мониторинг

    while not getattr(state, "shutdown_flag", False):
        try:
            _loop = asyncio.get_event_loop()
            rows = await _loop.run_in_executor(None, store.pop_due_schedules)
            for sched_id, user_id, phone, kind, payload_raw in rows:
                try:
                    payload = json.loads(payload_raw or "{}")
                    await _dispatch_schedule(sched_id, user_id, phone, kind, payload)
                except Exception as row_err:
                    logger.error(
                        "[scheduler] ❌ Ошибка обработки задачи id=%s uid=%s: %s",
                        sched_id,
                        user_id,
                        row_err,
                        exc_info=True,
                    )
                    try:
                        await _loop.run_in_executor(
                            None, store.finish_schedule, sched_id, "error"
                        )
                    except Exception:
                        pass
        except Exception as loop_err:
            logger.error(
                "[scheduler] ❌ Ошибка в scheduler_loop: %s", loop_err, exc_info=True
            )

        # Авто-разблокировка аккаунтов у которых прошёл флудвейт
        try:
            _check_floodwait_recovery()
        except Exception as _fw_err:
            logger.debug("[scheduler] Ошибка _check_floodwait_recovery: %s", _fw_err)

        # Авто-экспорт отчётов по завершённым задачам (подписка/рассылка/DM).
        # Раз в 10 секунд смотрит все ctx, при status in done/stopped
        # и отсутствии флага auto_exported — пишет в Экспорт/<модуль>/<stamp>/.
        try:
            _auto_export_finished_jobs()
        except Exception as _ae_err:
            logger.debug("[scheduler] Ошибка auto_export: %s", _ae_err)


        # Авто-сон отключён (1.4.55). Ранее аккаунты отваливались после
        # 3 мин простоя и парсер/подписка/рассылка получали «не авторизован»
        # при пробуждении. Теперь коннект живёт пока жив процесс; отключаемся
        # только при явной остановке задач или закрытии программы.

        _now = time.time()
        # Периодическая очистка памяти (каждые 5 минут)
        if _now - _last_cleanup > _CLEANUP_INTERVAL:
            _last_cleanup = _now
            try:
                _periodic_memory_cleanup()
            except Exception as _mc_err:
                logger.debug("[scheduler] Ошибка memory_cleanup: %s", _mc_err)
        # Периодическое сохранение задач (каждую минуту — защита от force kill)
        if _now - _last_job_save > _JOB_SAVE_INTERVAL:
            _last_job_save = _now
            try:
                from core.job_persistence import save_pending_jobs
                await asyncio.get_event_loop().run_in_executor(None, save_pending_jobs)
            except Exception as _sp_err:
                logger.debug("[scheduler] Ошибка save_pending: %s", _sp_err)

        # Watchdog мониторинга: если у клиента включён autostart, а
        # ctx.monitor_job отсутствует или status="stopped" — поднимаем заново.
        # Раньше мониторинг молча умирал на сутки (упал коннект, упал acc.client),
        # клиент видел «Старт» вместо «Стоп» и думал что он остановил руками.
        if _now - _last_monitor_watch > _MONITOR_WATCH_INTERVAL:
            _last_monitor_watch = _now
            try:
                _restart_dead_monitors()
            except Exception as _mw_err:
                logger.debug("[scheduler] monitor watchdog error: %s", _mw_err)

        # Heartbeat
        if _now - _last_heartbeat > _HEARTBEAT_INTERVAL:
            _last_heartbeat = _now
            _ctx_count = len(state.owners)
            _acc_count = sum(len(c.accounts) for c in state.owners.values())
            _active_jobs = sum(
                (1 if c.job and c.job.status == "running" else 0) +
                (1 if c.spam_job and c.spam_job.status == "running" else 0) +
                (1 if getattr(c, "user_spam_job", None) and c.user_spam_job.status == "running" else 0)
                for c in state.owners.values()
            )
            logger.debug("[scheduler] heartbeat: users=%d accounts=%d active_jobs=%d", _ctx_count, _acc_count, _active_jobs)

        await asyncio.sleep(10.0)


IDLE_TIMEOUT = 180  # 3 минуты без использования -> сон


async def _idle_accounts_check():
    """
    Эконом-режим: отключает аккаунты неактивные > IDLE_TIMEOUT секунд.
    Экономит RAM и TCP-соединения. Аккаунт автоматически проснётся
    при следующем использовании через ensure_connected().
    """
    now = time.time()
    disconnected = 0

    for uid, ctx in list(state.owners.items()):
        # Не трогаем аккаунты с активными задачами
        has_active_job = (
            (ctx.job and ctx.job.status in ("running", "paused"))
            or (ctx.spam_job and ctx.spam_job.status in ("running", "paused"))
            or (getattr(ctx, "user_spam_job", None) and ctx.user_spam_job.status in ("running", "paused"))
        )

        # Собираем телефоны занятые в мульти-задачах
        _multi_busy_phones = set()
        _mt = ctx.temp.get("_multi_join_task")
        _ms = ctx.temp.get("_multi_spam_task")
        if (_mt is not None and not _mt.done()) or (_ms is not None and not _ms.done()):
            _multi_phones = (getattr(ctx, "multi", {}) or {}).get("phones") or []
            _multi_busy_phones = set(_multi_phones)

        for phone, acc in list(ctx.accounts.items()):
            # Пропускаем уже спящие, мёртвые, замороженные
            if acc.idle or acc.dead or acc.frozen:
                continue

            # Не отключаем аккаунты в активном прогреве
            if getattr(ctx, "warmup_job", None) and ctx.warmup_job.status == "running":
                if getattr(acc, "warmup", False):
                    continue

            # Не отключаем аккаунты задействованные в мульти-задачах
            if phone in _multi_busy_phones:
                continue

            # Не отключаем аккаунты задействованные в задачах
            if has_active_job:
                # Проверяем конкретно этот аккаунт
                if ctx.job and ctx.job.status in ("running", "paused") and ctx.job.acc.phone == phone:
                    continue
                if ctx.spam_job and ctx.spam_job.status in ("running", "paused") and ctx.spam_job.acc.phone == phone:
                    continue

            # Проверяем время бездействия
            last_used = getattr(acc, "last_used_at", 0) or acc.connected_at
            idle_sec = now - last_used

            if idle_sec > IDLE_TIMEOUT:
                # Double-check: re-read last_used_at to avoid race with active job
                last_used_recheck = getattr(acc, "last_used_at", 0) or acc.connected_at
                if (now - last_used_recheck) <= IDLE_TIMEOUT:
                    continue
                try:
                    await safe_disconnect(acc.client)
                    acc.idle = True
                    acc.idle_since = now
                    acc.client = None
                    disconnected += 1
                    logger.info(
                        "[idle] 💤 Аккаунт %s uid=%s уснул (неактивен %d мин)",
                        phone, uid, int(idle_sec / 60),
                    )
                except Exception as e:
                    logger.debug("[idle] Ошибка отключения %s: %s", phone, e)

    if disconnected:
        import gc
        gc.collect()
        logger.info("[idle] 💤 Усыплено %d аккаунтов для экономии ресурсов", disconnected)


def _periodic_memory_cleanup():
    """
    Очищает завершённые задачи, обрезает лог-линии, удаляет пустые контексты.
    Вызывается раз в 5 минут для предотвращения утечек памяти.
    """
    import gc
    cleaned = 0
    now = time.time()
    MAX_KNOWN_IDS = 5000  # Ограничение known_ids на аккаунт
    JOB_TTL = 1800  # Удаляем завершённые задачи старше 30 минут

    for uid, ctx in list(state.owners.items()):
        # Обрезаем known_ids у аккаунтов (основной потребитель RAM)
        for phone, acc in list(ctx.accounts.items()):
            if hasattr(acc, 'known_ids') and len(acc.known_ids) > MAX_KNOWN_IDS:
                # Оставляем только последние MAX_KNOWN_IDS
                ids_list = list(acc.known_ids)
                acc.known_ids = set(ids_list[-MAX_KNOWN_IDS:])
                cleaned += 1

        # Очищаем завершённые job'ы
        job = getattr(ctx, 'job', None)
        if job and getattr(job, 'status', '') in ('finished', 'stopped'):
            age = now - getattr(job, 'started_at', now)
            if age > JOB_TTL:
                # Полностью удаляем старую задачу
                ctx.job = None
                cleaned += 1
            else:
                # Обрезаем тяжёлые данные
                if hasattr(job, 'results') and len(job.results) > 500:
                    job.results = dict(list(job.results.items())[-200:])
                if hasattr(job, 'log_lines') and len(job.log_lines) > 20:
                    job.log_lines = job.log_lines[-20:]
                cleaned += 1

        spam_job = getattr(ctx, 'spam_job', None)
        if spam_job and getattr(spam_job, 'status', '') in ('finished', 'stopped'):
            age = now - getattr(spam_job, 'started_at', now)
            if age > JOB_TTL:
                ctx.spam_job = None
                cleaned += 1
            else:
                if hasattr(spam_job, 'results') and len(spam_job.results) > 500:
                    spam_job.results = dict(list(spam_job.results.items())[-200:])
                if hasattr(spam_job, 'log_lines') and len(spam_job.log_lines) > 20:
                    spam_job.log_lines = spam_job.log_lines[-20:]
                cleaned += 1

        # Чистим user_spam_job (DM рассылка)
        us_job = getattr(ctx, 'user_spam_job', None)
        if us_job and getattr(us_job, 'status', '') in ('finished', 'stopped'):
            age = now - getattr(us_job, 'started_at', now)
            if age > JOB_TTL:
                ctx.user_spam_job = None
                cleaned += 1
            else:
                if hasattr(us_job, 'results') and len(us_job.results) > 500:
                    us_job.results = dict(list(us_job.results.items())[-200:])
                if hasattr(us_job, 'log_lines') and len(us_job.log_lines) > 20:
                    us_job.log_lines = us_job.log_lines[-20:]
                cleaned += 1

        # Чистим temp от старых завершённых мульти-задач
        _mt = ctx.temp.get("_multi_join_task")
        if _mt is not None and _mt.done():
            ctx.temp.pop("_multi_join_task", None)
            cleaned += 1

        # Удаляем пустые контексты (нет аккаунтов, нет задач)
        if (
            not ctx.accounts
            and not ctx.job
            and not ctx.spam_job
            and (now - getattr(ctx.stats, 'last_active_at', now)) > 3600
        ):
            state.owners.pop(uid, None)
            cleaned += 1

    # Очистка глобальных кэшей (утечки памяти)
    try:
        from core.join import _join_session_count as _jsc1
        from core.account_helpers import _join_session_count as _jsc2, _known_ids_refresh_ts
        # Удаляем записи для телефонов которых больше нет в активных аккаунтах
        active_phones = set()
        for ctx in state.owners.values():
            active_phones.update(ctx.accounts.keys())
        for cache in (_jsc1, _jsc2, _known_ids_refresh_ts):
            stale = [k for k in cache if k not in active_phones]
            for k in stale:
                cache.pop(k, None)
            if stale:
                cleaned += len(stale)
    except Exception:
        pass

    # Очистка старых staging-файлов (>24 часа)
    try:
        from config import STAGING_DIR
        _staging_sessions = STAGING_DIR / "sessions"
        if _staging_sessions.exists():
            _cutoff = now - 86400  # 24 часа
            for f in _staging_sessions.iterdir():
                try:
                    if f.stat().st_mtime < _cutoff:
                        f.unlink()
                        cleaned += 1
                except Exception:
                    pass
    except Exception:
        pass

    # Очистка SESSION_LOCKS от неиспользуемых сессий
    try:
        from core.client_builder import SESSION_LOCKS
        active_sessions = set()
        for ctx in state.owners.values():
            for phone, acc in list(ctx.accounts.items()):
                if hasattr(acc, 'session_path') and acc.session_path:
                    active_sessions.add(str(acc.session_path))
        stale_locks = [k for k in SESSION_LOCKS if k not in active_sessions]
        for k in stale_locks:
            SESSION_LOCKS.pop(k, None)
            cleaned += 1
    except Exception:
        pass

    # Синхронизируем папку Аккаунты/
    try:
        from core.accounts_folder import sync_accounts_folder
        for ctx in state.owners.values():
            if ctx.accounts:
                sync_accounts_folder(ctx.accounts)
    except Exception:
        pass

    # Принудительная сборка мусора если что-то почистили
    if cleaned > 0:
        gc.collect()
        logger.debug("[scheduler] 🧹 Memory cleanup: %d объектов обрезано", cleaned)


def _check_floodwait_recovery():
    """
    Проверяет аккаунты с истёкшим флудвейтом и разблокирует их.
    Уведомляет пользователя.
    """
    now = time.time()
    for uid, ctx in list(state.owners.items()):
        for phone, acc in list(ctx.accounts.items()):
            try:
                if (
                    getattr(acc, "disabled", False)
                    and getattr(acc, "floodwait_until", 0) > 0
                    and acc.floodwait_until <= now
                    and not getattr(acc, "dead", False)
                    and not getattr(acc, "frozen", False)
                ):
                    # Разблокируем
                    acc.disabled = False
                    acc.disabled_reason = None
                    acc.floodwait_until = 0.0
                    acc.floodwait_seconds = 0

                    logger.info(
                        "[scheduler] 🔓 Аккаунт %s разблокирован (FloodWait истёк), uid=%s",
                        phone,
                        uid,
                    )

                    # Уведомляем пользователя
                    asyncio.create_task(_notify_account_recovered(uid, phone))
            except Exception as _e:
                logger.debug("[scheduler] Ошибка проверки аккаунта %s: %s", phone, _e)


async def _notify_account_recovered(uid: int, phone: str):
    """Логирует восстановление акка. UI прочитает статус из ctx — отдельное
    уведомление не шлём, в standalone бота нет."""
    logger.info("[scheduler] ✅ Аккаунт %s снова доступен (FloodWait истёк)", phone)
    try:
        from core.notifications import add_notification
        add_notification(uid, "system", f"Аккаунт {phone} восстановлен",
                         "FloodWait истёк, можно продолжать работу")
    except Exception:
        pass


async def _dispatch_schedule(
    sched_id: int, user_id: int, phone: str, kind: str, payload: dict
):
    """Диспетчер запуска задачи через per-user worker pool."""
    from core.worker_pool import submit_for_user

    ctx = get_ctx(user_id)
    acc = ctx.accounts.get(phone)
    if not acc:
        logger.warning(
            "[scheduler] Аккаунт %s не найден для uid=%s, задача %s пропущена",
            phone,
            user_id,
            sched_id,
        )
        store.finish_schedule(sched_id, "skipped:no_account")
        return

    if not acc.is_available:
        logger.warning(
            "[scheduler] Аккаунт %s недоступен (status=%s) для uid=%s, задача %s отложена",
            phone,
            acc.status_text(),
            user_id,
            sched_id,
        )
        store.finish_schedule(sched_id, f"skipped:{acc.status_text()}")
        return

    if kind == "spam":
        text = payload.get("text") or "Ping"
        targets, titles = await collect_writable_groups(acc)
        job = SpamJob(
            owner_id=user_id,
            acc=acc,
            text=text,
            targets=targets,
            titles=titles,
            total=len(targets),
        )

        async def _run():
            try:
                await run_spam_job(job)
                store.finish_schedule(sched_id, "done")
            except Exception as _e:
                logger.error(
                    "[scheduler] spam job error sched_id=%s: %s",
                    sched_id,
                    _e,
                    exc_info=True,
                )
                store.finish_schedule(sched_id, "error")

        await submit_for_user(
            user_id,
            lambda: _run(),
            job_id=f"sched_spam_{sched_id}",
            name=f"spam_{phone}",
        )

    elif kind == "join":
        links = payload.get("links") or []
        if not links:
            store.finish_schedule(sched_id, "skipped:no_links")
            return
        links, dup = deduplicate_links(links)
        job = JoinJob(
            owner_id=user_id, acc=acc, queue=links, total=len(links), dup_removed=dup
        )

        async def _run():
            try:
                await run_join_job(job)
                store.finish_schedule(sched_id, "done")
            except Exception as _e:
                logger.error(
                    "[scheduler] join job error sched_id=%s: %s",
                    sched_id,
                    _e,
                    exc_info=True,
                )
                store.finish_schedule(sched_id, "error")

        await submit_for_user(
            user_id,
            lambda: _run(),
            job_id=f"sched_join_{sched_id}",
            name=f"join_{phone}",
        )

    else:
        logger.warning("[scheduler] Неизвестный тип задачи: %s", kind)
        store.finish_schedule(sched_id, "skipped:unknown_kind")


_autoload_sem: Optional[asyncio.Semaphore] = None


def _get_autoload_sem() -> asyncio.Semaphore:
    global _autoload_sem
    if _autoload_sem is None:
        _autoload_sem = asyncio.Semaphore(5)
    return _autoload_sem


async def autoload_existing_accounts_for(owner_id: int):
    """Идемпотентная автозагрузка сессий — ищет .session файлы во всех вложенных папках."""
    ctx = get_ctx(owner_id)

    sem = _get_autoload_sem()

    # Собираем .session файлы из всех папок Аккаунты/ и корня клиента
    sessions_to_load = []
    from config import ACCOUNTS_DIR, DATA_DIR

    search_dirs = []
    # Все подпапки Аккаунты/ (В работе, Заморозка, и тд)
    if ACCOUNTS_DIR.exists():
        for d in ACCOUNTS_DIR.iterdir():
            if d.is_dir():
                search_dirs.append(d)
    # Fallback: старая английская Accounts/<Active|Frozen|…>/ — если миграция
    # в config.py что-то пропустила (например прав не хватило). Лучше загрузить
    # и показать клиенту, чем потерять молча.
    try:
        _old_accs = ACCOUNTS_DIR.parent / "Accounts"
        if _old_accs.exists() and _old_accs.is_dir():
            for d in _old_accs.iterdir():
                if d.is_dir() and d not in search_dirs:
                    search_dirs.append(d)
    except Exception:
        pass
    # Не ищем в корне клиента — аккаунты должны быть только в Аккаунты/

    # Миграция с WAL на journal: удаляем старые спутники -wal и -shm из всех
    # папок аккаунтов. В новом journal-mode эти файлы не создаются, но если
    # они остались от предыдущих запусков в WAL — SQLite попытается переключиться
    # обратно в WAL. rglob — рекурсивно, чтобы покрыть вложенные папки
    # типа «После инвайта/Спамблок/».
    _wal_cleaned = 0
    for _d in search_dirs:
        try:
            for pat in ("*.session-wal", "*.session-shm"):
                for _f in _d.rglob(pat):
                    try:
                        _f.unlink()
                        _wal_cleaned += 1
                    except Exception:
                        pass
        except Exception:
            pass
    if _wal_cleaned:
        logger.info("[autoload] очищено %d остаточных .session-wal/-shm файлов (миграция с WAL на journal)", _wal_cleaned)

    # Авто-нормализация имён .session/.json файлов. Клиенты часто кладут файлы
    # с пробелами/плюсами/скобками ("+998 99 443 45 83.session") — autoload их
    # не видел. Теперь автоматически переименовываем к digit-only + грузим.
    # rglob чтобы найти на любом уровне вложенности.
    import re as _re
    _renamed = 0
    for search_dir in search_dirs:
        try:
            for f in list(search_dir.rglob("*.session")):
                if not f.is_file() or f.stem.isdigit():
                    continue
                _digits = _re.sub(r'\D', '', f.stem)
                if not _digits or len(_digits) < 7:
                    continue  # не похоже на номер телефона — не трогаем
                _new_session = f.with_name(f"{_digits}.session")
                if _new_session.exists():
                    continue  # целевое имя занято — не перезаписываем
                try:
                    f.rename(_new_session)
                    # Вместе с .session переименовываем .json и -journal
                    for _ext in ('.json', '.session-journal'):
                        _old = f.with_suffix(_ext) if _ext == '.json' else f.with_name(f.name + '-journal')
                        if _old.exists():
                            _new = _new_session.with_suffix(_ext) if _ext == '.json' else _new_session.with_name(_new_session.name + '-journal')
                            try: _old.rename(_new)
                            except Exception: pass
                    _renamed += 1
                    logger.info("[autoload] Переименован %s → %s (нормализация имени)", f.name, _new_session.name)
                except Exception as _re_e:
                    logger.debug("[autoload] не удалось переименовать %s: %s", f.name, _re_e)
        except Exception:
            pass
    if _renamed:
        logger.info("[autoload] Автоматически переименовано %d .session-файлов с пробелами/плюсами", _renamed)

    seen_phones = set()
    # Рекурсивный обход всех уровней. До этого был только верхний уровень
    # подпапки — файлы в «После инвайта/Спамблок/79xxx.session» (2 уровня
    # вложенности под ACCOUNTS_DIR) НЕ загружались.
    for search_dir in search_dirs:
        try:
            for f in search_dir.rglob("*.session"):
                if f.is_file() and f.stem.isdigit():
                    phone = f.stem
                    if phone in seen_phones:
                        continue
                    seen_phones.add(phone)
                    json_path = f.with_suffix(".json")
                    sessions_to_load.append((f, json_path if json_path.exists() else None))
        except Exception as e:
            logger.warning("[autoload] Error scanning %s: %s", search_dir, e)

    logger.info("[autoload] Найдено %d сессий в %d папках: %s",
                len(sessions_to_load), len(search_dirs),
                ", ".join(d.name for d in search_dirs))

    # Диагностика: что-то осталось с неверным именем после нормализации
    # (например слишком короткие имена — меньше 7 цифр). rglob чтобы покрыть
    # вложенные папки.
    non_digit_sessions = []
    for search_dir in search_dirs:
        try:
            for f in search_dir.rglob("*.session"):
                if f.is_file() and not f.stem.isdigit():
                    non_digit_sessions.append(str(f))
        except Exception:
            pass
    if non_digit_sessions:
        logger.warning(
            "[autoload] Найдено %d .session файлов с неверным именем: %s. "
            "Переименуйте их в ЧИСЛО.session (без +, пробелов, @).",
            len(non_digit_sessions), non_digit_sessions[:5],
        )

    # Если вообще 0 сессий и папки пустые — клиент должен увидеть что искали
    if len(sessions_to_load) == 0:
        total_files = 0
        for d in search_dirs:
            try:
                total_files += sum(1 for _ in d.iterdir())
            except Exception:
                pass
        logger.warning(
            "[autoload] 0 аккаунтов загружено. Всего файлов в папках Аккаунты/: %d. "
            "Проверьте путь: %s. Если аккаунты есть — нажмите «Обновить» в разделе «Добавить аккаунт → Из папки».",
            total_files, ACCOUNTS_DIR,
        )

    for sess, json_path in sessions_to_load:
        phone_digits = sess.stem
        try:
            if phone_digits in ctx.accounts:
                continue

            # Читаем metadata из json если есть
            username = None
            first_name = None
            last_name = None
            is_premium = False
            spamblock = None
            session_created = ""
            register_time = 0
            dev_from_json = None
            if json_path and json_path.exists():
                try:
                    # Пробуем UTF-8, UTF-8-BOM, CP1251 — разные софты пишут по-разному
                    _raw = None
                    for _enc in ("utf-8-sig", "utf-8", "cp1251", "latin-1"):
                        try:
                            _raw = json_path.read_text(encoding=_enc)
                            break
                        except UnicodeDecodeError:
                            continue
                    meta = json.loads(_raw) if _raw else {}

                    # Унифицированный чтец разных форматов .json (Kraken/Raptor/Nekobox/Tdesktop/Telethon).
                    # Ключи отличаются:
                    #   Kraken/Raptor:  device, sdk, app_version, lang_pack, system_lang_pack
                    #   Nekobox/Modern: device_model, system_version, app_version, lang_code, system_lang_code
                    #   TgHub/2FAhub:   device_type, system, version, language, system_language
                    #   Pyrogram-like:  app_name, app_version, device
                    def _first(*keys, default=""):
                        for k in keys:
                            v = meta.get(k)
                            if v not in (None, ""):
                                return v
                        return default

                    username = _first("username", "user_name", default=None) or None
                    first_name = _first("first_name", "firstname", "first", default=None) or None
                    last_name = _first("last_name", "lastname", "last", default=None) or None
                    is_premium = bool(_first("is_premium", "premium", default=False))
                    spamblock = _first("spamblock", "spam_block", "spam_filter", default=None) or None
                    session_created = _first("session_created_date", "created", "created_at", "reg_time_str", default="")
                    try:
                        register_time = int(_first("register_time", "reg_time", "registered_at", default=0) or 0)
                    except (ValueError, TypeError):
                        register_time = 0

                    # Device info — пытаемся собрать из любого формата
                    _device = _first("device", "device_model", "device_type", default="")
                    _system = _first("sdk", "system_version", "system", "os", default="")
                    _appver = _first("app_version", "version", "app_ver", default="")
                    _lang = _first("lang_pack", "lang_code", "language", "lang", default="")
                    _syslang = _first("system_lang_pack", "system_lang_code", "system_language", default="")

                    if _device or _system or _appver:
                        from models import DeviceInfo
                        dev_from_json = DeviceInfo(
                            device_model=_device or "PC 64bit",
                            system_version=_system or "Windows 10",
                            app_version=_appver or "5.8.3 x64",
                            lang_code=_lang or "en",
                            system_lang_code=_syslang or "en-US",
                        )
                except Exception as _je:
                    logger.warning("[autoload] %s: не смог прочитать .json (%s) — запустим с device по-умолчанию",
                                   phone_digits, _je)

            # Use device from JSON if available, otherwise generate new
            if dev_from_json:
                acc_dev = dev_from_json
            else:
                from core.client_builder import load_device
                acc_dev = load_device(phone_digits)

            acc = Account(
                phone_digits,
                username,
                sess,
                json_path,
                None,  # client=None — ленивое подключение
                acc_dev,
                state.proxy_res,
            )
            acc.first_name = first_name
            acc.last_name = last_name
            acc.is_premium = is_premium
            acc.spamblock_json = spamblock
            acc.session_created = session_created or ""
            acc.register_time = register_time
            acc.folder_name = sess.parent.name
            if acc.folder_name in ("\u041f\u0440\u043e\u0433\u0440\u0435\u0432", "\u041e\u0442\u043b\u0451\u0436\u043a\u0430", "Warmup"):
                acc.warmup = True
            # \u041c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433: \u0441\u043f\u0435\u0446-\u043f\u0430\u043f\u043a\u0430 \u2014 \u0440\u043e\u043b\u044c \u043d\u0430\u0437\u043d\u0430\u0447\u0430\u0435\u0442\u0441\u044f \u0440\u0443\u043a\u0430\u043c\u0438 \u043a\u043b\u0438\u0435\u043d\u0442\u0430.
            if acc.folder_name in ("\u041c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433", "Monitoring"):
                acc.monitor = True
            # \u0415\u0441\u043b\u0438 \u0430\u043a\u043a \u0432 \u0441\u043f\u0435\u0446-\u043f\u0430\u043f\u043a\u0435 \u2014 \u0432\u044b\u0441\u0442\u0430\u0432\u043b\u044f\u0435\u043c \u0444\u043b\u0430\u0433\u0438, \u0447\u0442\u043e\u0431\u044b UI \u0438 \u043b\u043e\u0433\u0438\u043a\u0430
            # \u0441\u0440\u0430\u0437\u0443 \u0432\u0438\u0434\u0435\u043b\u0438 \u043f\u0440\u0430\u0432\u0438\u043b\u044c\u043d\u044b\u0439 \u0441\u0442\u0430\u0442\u0443\u0441 (\u0431\u0435\u0437 \u043e\u0436\u0438\u0434\u0430\u043d\u0438\u044f \u043f\u0435\u0440\u0432\u043e\u0439 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438):
            #   \u041c\u0451\u0440\u0442\u0432\u044b\u0435 \u2192 acc.dead = True
            #   \u0417\u0430\u043c\u043e\u0440\u043e\u0437\u043a\u0430 \u2192 acc.frozen = True
            #   \u0412 \u0441\u043f\u0430\u043c / \u0412\u0435\u0447\u043d\u044b\u0439 \u0441\u043f\u0430\u043c \u2192 acc.spam_blocked = True
            _fname = acc.folder_name
            if _fname in ("\u041c\u0451\u0440\u0442\u0432\u044b\u0435", "Dead"):
                acc.dead = True
                if not getattr(acc, "dead_reason", ""):
                    acc.dead_reason = "from_folder"
            elif _fname in ("\u0417\u0430\u043c\u043e\u0440\u043e\u0437\u043a\u0430", "Frozen"):
                acc.frozen = True
                if not getattr(acc, "frozen_reason", ""):
                    acc.frozen_reason = "from_folder"
            elif _fname in ("\u0412 \u0441\u043f\u0430\u043c", "\u0412\u0435\u0447\u043d\u044b\u0439 \u0441\u043f\u0430\u043c", "Spam", "Permanent_spam"):
                acc.spam_blocked = True
                if _fname in ("\u0412\u0435\u0447\u043d\u044b\u0439 \u0441\u043f\u0430\u043c", "Permanent_spam"):
                    setattr(acc, "spam_permanent", True)
            # \u041f\u043e\u0432\u0435\u0440\u0445 folder-based detection \u043d\u0430\u043a\u043b\u0430\u0434\u044b\u0432\u0430\u0435\u043c persisted state.
            # \u042d\u0442\u043e\u0442 \u0444\u0430\u0439\u043b (account_states.json) \u0445\u0440\u0430\u043d\u0438\u0442 \u0442\u043e\u0447\u043d\u044b\u0439 \u0441\u0442\u0430\u0442\u0443\u0441
            # \u043d\u0435\u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e \u043e\u0442 \u0442\u043e\u0433\u043e \u043f\u0435\u0440\u0435\u0435\u0445\u0430\u043b \u0444\u0430\u0439\u043b \u0432 \u043d\u0443\u0436\u043d\u0443\u044e \u043f\u0430\u043f\u043a\u0443 \u0438\u043b\u0438 \u043d\u0435\u0442
            # (sync_accounts_folder \u043c\u043e\u0436\u0435\u0442 \u043f\u0430\u0441\u0442\u044c \u0438\u0437-\u0437\u0430 file-lock, \u043d\u043e state
            # \u0443\u0436\u0435 \u0441\u043e\u0445\u0440\u0430\u043d\u0451\u043d \u0432 \u043c\u043e\u043c\u0435\u043d\u0442 check_spam / invite-retire).
            try:
                from core.account_state import apply_state_to_account
                apply_state_to_account(acc)
            except Exception as _se:
                logger.debug("[autoload] apply_state %s: %s", phone_digits, _se)
            acc.idle = True
            acc.idle_since = time.time()

            ctx.accounts[phone_digits] = acc
            if not ctx.active:
                ctx.active = phone_digits
        except Exception as e:
            logger.debug("Autoload metadata failed for %s: %s", sess, e)


async def autoload_all_users():
    """Автозагрузка аккаунтов при старте.

    Ищет .session файлы рекурсивно в USERS_DIR и загружает для всех найденных uid.
    """
    from bot_ui.ui import get_ctx

    from config import ACCOUNTS_DIR
    if not USERS_DIR.exists() and not ACCOUNTS_DIR.exists():
        logger.warning("[startup] No accounts directory found: %s / %s", USERS_DIR, ACCOUNTS_DIR)
        return

    # Определяем uid из лицензии
    try:
        from utils.license import LicenseManager
        lm = LicenseManager()
        uid = lm.get_uid()
    except Exception:
        # Fallback: ищем цифровые папки как uid
        uid = None
        for d in USERS_DIR.iterdir():
            if d.is_dir() and d.name.isdigit():
                uid = int(d.name)
                break
        if uid is None:
            uid = 0  # default

    logger.info("[startup] 🔄 Автозагрузка аккаунтов (uid=%s)...", uid)

    errors_count = 0
    try:
        get_ctx(uid)
        await autoload_existing_accounts_for(uid)
    except Exception as e:
        errors_count += 1
        logger.warning("[startup] autoload: %s", e)

    # Миграция: поднять multi_max_accounts с 3 до 10 (старый дефолт был 3)
    for ctx in state.owners.values():
        cur = ctx.user_settings.get("multi_max_accounts")
        if cur is not None and cur <= 3:
            ctx.user_settings["multi_max_accounts"] = 10

    loaded_users = sum(1 for u in state.owners.values() if u.accounts)
    total_accs = sum(len(u.accounts) for u in state.owners.values())
    logger.info("[startup] ✅ Загружено: %d польз., %d аккаунтов (ошибок: %d)",
                loaded_users, total_accs, errors_count)

    # Self-heal: снимаем ложно-мёртвые пометки (dead_reason="unauthorized").
    # Раньше 5 неудач is_user_authorized() автоматом ставили dead=True,
    # но это чаще false-positive (сеть моргнула). Настоящая смерть — только
    # session_revoked / frozen. Даём таким аккам шанс подняться заново.
    _healed = 0
    for _ctx in state.owners.values():
        for _acc in _ctx.accounts.values():
            if getattr(_acc, "dead", False) and getattr(_acc, "dead_reason", "") == "unauthorized":
                _acc.dead = False
                _acc.dead_reason = ""
                _acc.unauth_count = 0
                _healed += 1
    if _healed:
        logger.info("[startup] ♻️ Сняли ложную пометку «мёртвый» с %d аккаунтов (dead_reason=unauthorized)", _healed)

    # Восстанавливаем незавершённые задачи
    try:
        from core.job_persistence import load_pending_jobs, restore_jobs_for_user, clear_pending_jobs
        pending = load_pending_jobs()
        if pending:
            unique_uids = set(j.get("owner_id") for j in pending)
            restored_total = 0
            for uid in unique_uids:
                if uid is None:
                    continue
                try:
                    count = await restore_jobs_for_user(uid)
                    restored_total += count
                except Exception as e:
                    logger.warning("[startup] restore uid=%s: %s", uid, e)
            if restored_total:
                logger.info("[startup] ▶️ Восстановлено %d задач", restored_total)
    except Exception as e:
        logger.warning("[startup] Ошибка восстановления задач: %s", e)

    # Синхронизируем папку Аккаунты/
    try:
        from core.accounts_folder import sync_accounts_folder
        for ctx in state.owners.values():
            if ctx.accounts:
                sync_accounts_folder(ctx.accounts)
    except Exception:
        pass

    # Авто-возобновление прогрева после рестарта (напр. автообновление софта).
    # Если был снимок запуска с running-статусом — поднимаем заново с текущего дня.
    try:
        from core.warmup_resume import load_snapshot, clear_snapshot
        snap = load_snapshot()
        if snap:
            s_uid = int(snap.get("uid") or 0)
            payload = dict(snap.get("payload") or {})
            cur_day = int(snap.get("current_day") or 1)
            payload["start_day"] = cur_day
            ctx = state.owners.get(s_uid)
            if ctx is None:
                logger.warning("[warmup-resume] ctx uid=%s не найден — очищаю снимок", s_uid)
                clear_snapshot()
            else:
                try:
                    from core.warmup import (
                        build_warmup_pairs, run_warmup_job, get_day_limits, save_pairs,
                    )
                    from models import WarmupJob, WarmupSchedule
                    warmup_accs = {
                        phone: acc for phone, acc in list(ctx.accounts.items())
                        if getattr(acc, "folder_name", "") in ("Прогрев", "Warmup")
                        and not acc.dead and not acc.frozen
                    }
                    if len(warmup_accs) >= 2:
                        total_days = int(payload.get("total_days", 21))
                        start_day = max(1, min(cur_day, total_days))
                        peers, msgs, stage = get_day_limits(start_day)
                        schedule = WarmupSchedule(
                            total_days=total_days,
                            current_day=start_day,
                            started_at=time.time() - (start_day - 1) * 86400,
                            stage=stage,
                            max_peers_today=peers,
                            msgs_per_day=msgs,
                            auto_move_when_done=bool(payload.get("auto_move", True)),
                        )
                        pairs = build_warmup_pairs(ctx.accounts, max_peers=peers)
                        save_pairs(pairs)
                        scenarios_in = payload.get("scenarios") or {}
                        job = WarmupJob(
                            owner_id=s_uid,
                            pairs=pairs,
                            schedule=schedule,
                            config={
                                "active_hours_start": int(payload.get("active_hours_start", 9)),
                                "active_hours_end": int(payload.get("active_hours_end", 22)),
                                "min_delay": int(payload.get("min_delay", 60)),
                                "max_delay": int(payload.get("max_delay", 300)),
                                "burst_size": int(payload.get("burst_size", 3)),
                                "burst_pause_min": int(payload.get("burst_pause_min", 600)),
                                "burst_pause_max": int(payload.get("burst_pause_max", 1800)),
                                "gen_mode": payload.get("gen_mode", "auto"),
                                "scenarios": {
                                    "dm": bool(scenarios_in.get("dm", True)),
                                    "channels": bool(scenarios_in.get("channels", False)),
                                    "reactions": bool(scenarios_in.get("reactions", False)),
                                    "chats": bool(scenarios_in.get("chats", False)),
                                },
                                "channels_list": [s.strip().lstrip("@") for s in (payload.get("channels_list") or []) if s and s.strip()],
                                "chats_list": [s.strip().lstrip("@") for s in (payload.get("chats_list") or []) if s and s.strip()],
                                "channels_per_day": int(payload.get("channels_per_day", 3)),
                                "chat_messages_per_day": int(payload.get("chat_messages_per_day", 2)),
                                "reactions_per_day": int(payload.get("reactions_per_day", 5)),
                                "reply_to_incoming": bool(payload.get("reply_to_incoming", True)),
                                "threads": int(payload.get("threads", 1)),
                                "work_until_midnight": bool(payload.get("work_until_midnight", False)),
                                "manual_next_day": bool(payload.get("manual_next_day", False)),
                                "per_account_day": payload.get("per_account_day") or {},
                                "warmup_mode": str(payload.get("warmup_mode", "classic")),
                                "hub_phone": str(payload.get("hub_phone", "")),
                                "id_format": bool(payload.get("id_format", False)),
                                "duplicate_dm_after_reply": bool(payload.get("duplicate_dm_after_reply", True)),
                            },
                        )
                        job.pause_event.set()
                        ctx.warmup_job = job
                        asyncio.create_task(run_warmup_job(job, ctx))
                        logger.info(
                            "[autostart] Возобновляю прогрев аккаунтов — день %d/%d, %d пар. "
                            "Был в процессе на момент прошлого запуска, сохранено в warmup_resume.json.",
                            start_day, total_days, len(pairs),
                        )
                    else:
                        logger.warning("[warmup-resume] акков < 2 — возобновление отменено")
                        clear_snapshot()
                except Exception as e:
                    logger.error("[warmup-resume] restart failed: %s", e)
    except Exception as e:
        logger.debug("[warmup-resume] load_snapshot error: %s", e)

    # Авто-возобновление мониторинга (config в user_settings с autostart=True)
    try:
        from core.monitor import MonitorJob, run_monitor_job
        from core.client_builder import ensure_connected
        import re as _re
        for _uid, _ctx in list(state.owners.items()):
            _mon = (_ctx.user_settings.get("monitor") or {})
            if not _mon.get("autostart"):
                continue
            # Multi-acc: собираем все акки из настроек + папки Мониторинг
            _accs_alive = []
            _seen_phones = set()
            # Из папки Аккаунты/Мониторинг/ (приоритет)
            for _ph, _a in _ctx.accounts.items():
                if getattr(_a, "monitor", False) and not _a.dead and not getattr(_a, "frozen", False):
                    _accs_alive.append(_a)
                    _seen_phones.add(_a.phone)
            # Из сохранённого списка phones
            for _ph in (_mon.get("phones") or []):
                _ph = (_ph or "").strip()
                if not _ph or _ph in _seen_phones: continue
                _a = _ctx.accounts.get(_ph)
                if _a and not _a.dead and not getattr(_a, "frozen", False):
                    _accs_alive.append(_a)
                    _seen_phones.add(_ph)
            # Fallback: одиночный phone
            if not _accs_alive:
                _ph = (_mon.get("phone") or "").strip() or (_ctx.active or "").strip()
                _a = _ctx.accounts.get(_ph) if _ph else None
                if _a and not _a.dead and not getattr(_a, "frozen", False):
                    _accs_alive.append(_a)
            if not _accs_alive:
                continue
            _raw_chats = _mon.get("chats") or ""
            _raw_kws = _mon.get("keywords") or ""
            _raw_iws = _mon.get("ignore_words") or ""
            _chats = [x.strip() for x in _re.split(r"[\n\r]+", _raw_chats) if x.strip()]
            _kws = [x.strip() for x in _re.split(r"[\n\r,;]+", _raw_kws) if x.strip()]
            _iws = [x.strip() for x in _re.split(r"[\n\r,;]+", _raw_iws) if x.strip()]
            if not _chats or not _kws:
                continue
            _bot_token = ""
            try:
                _enc = _mon.get("bot_token_enc") or ""
                if _enc:
                    from utils.secrets_storage import decrypt_secret
                    _bot_token = decrypt_secret(_enc)
            except Exception:
                pass
            _multi_mode = (_mon.get("multi_mode") or "redundant").strip()
            if _multi_mode == "split" and len(_accs_alive) > 1:
                _chunks = [_chats[i::len(_accs_alive)] for i in range(len(_accs_alive))]
            else:
                _chunks = [list(_chats) for _ in _accs_alive]
            _shared_msgs = set()
            _shared_text_hashes = {}
            _new_pool = []
            for _acc, _my_chats in zip(_accs_alive, _chunks):
                if not _my_chats: continue
                _job = MonitorJob(
                    owner_id=_uid, acc=_acc,
                    chat_inputs=_my_chats, keywords=_kws,
                    ignore_words=_iws,
                    target=_mon.get("target") or "me",
                    auto_dm_text=_mon.get("auto_dm_text") or "",
                    bot_token=_bot_token,
                    bot_chat_id=_mon.get("bot_chat_id") or "",
                )
                _job._shared_msgs = _shared_msgs
                _job._shared_text_hashes = _shared_text_hashes
                _new_pool.append(_job)
            _ctx.monitor_jobs = _new_pool
            _ctx.monitor_job = _new_pool[0] if _new_pool else None

            def _make_runner(job, acc):
                async def _run():
                    if not await ensure_connected(acc):
                        job.status = "stopped"
                        return
                    try:
                        await run_monitor_job(job)
                    except Exception as e:
                        logger.error("[monitor-resume:%s] crashed: %s", acc.phone, e)
                        job.status = "stopped"
                return _run

            for _job in _new_pool:
                asyncio.create_task(_make_runner(_job, _job.acc)())
            try:
                _mon["phones"] = [j.acc.phone for j in _new_pool]
                _mon["phone"] = _mon["phones"][0] if _mon["phones"] else _mon.get("phone", "")
                _ctx.user_settings["monitor"] = _mon
                from web.app import _save_user_settings as _sus
                _sus(_ctx.user_settings)
            except Exception:
                pass
            logger.info(
                "[autostart] Возобновляю мониторинг — пул из %d акк(ов), mode=%s, %d чатов, %d ключей.",
                len(_new_pool), _multi_mode, len(_chats), len(_kws),
            )
    except Exception as e:
        logger.debug("[monitor-resume] skip: %s", e)

    # Прогрев соединений: после рестарта/обновы все акки в idle. Клиент жалуется
    # что в UI они показывают «Сон», и приходится жать «Обновить»/«Проверить»
    # чтобы они стали зелёными. Поднимаем их сами фоном — пачками по 5,
    # пауза между пачками 3с (не выглядит как DDOS на Telegram).
    try:
        from core.client_builder import ensure_connected as _ec_warmup
        _to_wake = []
        for _uid2, _ctx2 in list(state.owners.items()):
            for _ph, _ac in list(_ctx2.accounts.items()):
                if _ac.dead or getattr(_ac, "frozen", False):
                    continue
                if not getattr(_ac, "idle", False):
                    continue
                _to_wake.append(_ac)
        if _to_wake:
            logger.info("[startup] 🌅 Бужу %d аккаунтов после рестарта (пачками по 5)", len(_to_wake))
            async def _wake_all(accs=_to_wake):
                _BATCH = 5
                _ok = 0
                _fail = 0
                for i in range(0, len(accs), _BATCH):
                    batch = accs[i:i+_BATCH]
                    res = await asyncio.gather(
                        *[asyncio.wait_for(_ec_warmup(a), timeout=15) for a in batch],
                        return_exceptions=True,
                    )
                    for r in res:
                        if r is True:
                            _ok += 1
                        else:
                            _fail += 1
                    if i + _BATCH < len(accs):
                        await asyncio.sleep(3.0)
                logger.info("[startup] 🌅 Пробуждение завершено: %d живых, %d не подняли", _ok, _fail)
            asyncio.create_task(_wake_all())
    except Exception as e:
        logger.debug("[startup] warmup-connections skip: %s", e)


def _restart_dead_monitors():
    """Если у клиента autostart=True, а ctx.monitor_job помер (None или
    status='stopped') — поднимаем заново. Зовётся scheduler-loop'ом раз в
    2 минуты. Защищает от случая когда мониторинг тихо умер из-за дисконнекта
    и клиент только спустя сутки замечает что нет алертов."""
    try:
        from core.monitor import MonitorJob, run_monitor_job
        from core.client_builder import ensure_connected
        import re as _re
        for _uid, _ctx in list(state.owners.items()):
            _mon = (_ctx.user_settings.get("monitor") or {})
            if not _mon.get("autostart"):
                continue
            _existing = getattr(_ctx, "monitor_job", None)
            if _existing and getattr(_existing, "status", "") == "running":
                continue
            # Берём акк из папки Мониторинг (если есть), иначе сохранённый phone,
            # иначе ctx.active. Та же логика что в monitor-resume — иначе watchdog
            # после смерти исходного джоба может стартовать на чужом акке.
            _acc = None
            for _ph, _a in _ctx.accounts.items():
                if getattr(_a, "monitor", False) and not _a.dead and not getattr(_a, "frozen", False):
                    _acc = _a
                    break
            if _acc is None:
                _phone = (_mon.get("phone") or "").strip()
                if _phone:
                    _candidate = _ctx.accounts.get(_phone)
                    if _candidate and not _candidate.dead and not getattr(_candidate, "frozen", False):
                        _acc = _candidate
            if _acc is None:
                _phone = (_ctx.active or "").strip()
                _acc = _ctx.accounts.get(_phone) if _phone else None
            if not _acc or _acc.dead or getattr(_acc, "frozen", False):
                continue
            _phone = _acc.phone
            _raw_chats = _mon.get("chats") or ""
            _raw_kws = _mon.get("keywords") or ""
            _raw_iws = _mon.get("ignore_words") or ""
            _chats = [x.strip() for x in _re.split(r"[\n\r]+", _raw_chats) if x.strip()]
            _kws = [x.strip() for x in _re.split(r"[\n\r,;]+", _raw_kws) if x.strip()]
            _iws = [x.strip() for x in _re.split(r"[\n\r,;]+", _raw_iws) if x.strip()]
            if not _chats or not _kws:
                continue
            _bot_token = ""
            try:
                _enc = _mon.get("bot_token_enc") or ""
                if _enc:
                    from utils.secrets_storage import decrypt_secret
                    _bot_token = decrypt_secret(_enc)
            except Exception:
                pass
            _job = MonitorJob(
                owner_id=_uid, acc=_acc,
                chat_inputs=_chats, keywords=_kws,
                ignore_words=_iws,
                target=_mon.get("target") or "me",
                auto_dm_text=_mon.get("auto_dm_text") or "",
                bot_token=_bot_token,
                bot_chat_id=_mon.get("bot_chat_id") or "",
            )
            _ctx.monitor_job = _job
            async def _mon_run(acc=_acc, job=_job):
                if not await ensure_connected(acc):
                    job.status = "stopped"
                    return
                try:
                    await run_monitor_job(job)
                except Exception as e:
                    logger.error("[monitor-watchdog] crashed: %s", e)
                    job.status = "stopped"
            asyncio.create_task(_mon_run())
            logger.warning(
                "[monitor-watchdog] 🔁 мониторинг помер на акке %s%s — поднимаю заново "
                "(%d чатов, %d ключей)",
                _phone,
                " (из папки Мониторинг)" if getattr(_acc, "monitor", False) else "",
                len(_chats), len(_kws),
            )
    except Exception as e:
        logger.debug("[monitor-watchdog] skip: %s", e)


# ── Авто-экспорт отчётов при завершении задач ───────────────────────────────

def _auto_export_finished_jobs():
    """При каждом тике scheduler смотрит все ctx и для завершённых join/spam/dm
    задач пишет отчёт в Экспорт/<модуль>/<stamp>/. Ставит флаг `_auto_exported`
    на самом job'e чтобы не дублировать."""
    for _uid, _ctx in list(state.owners.items()):
        # Подписка (ctx.job)
        _job = getattr(_ctx, "job", None)
        if _job and getattr(_job, "status", "") in ("done", "stopped", "error"):
            if not getattr(_job, "_auto_exported", False):
                try:
                    from web.app import _do_join_export
                    _res = _do_join_export(_ctx, open_folder=False)
                    if _res:
                        logger.info(
                            "[auto_export] 📂 Подписка: отчёт сохранён — %s (✅%d / ❌%d)",
                            _res["folder"], _res["ok_count"], _res["fail_count"],
                        )
                except Exception as e:
                    logger.debug("[auto_export] join failed: %s", e)
                try:
                    _job._auto_exported = True
                except Exception:
                    pass

        # Рассылка по чатам (ctx.spam_job)
        _sj = getattr(_ctx, "spam_job", None)
        if _sj and getattr(_sj, "status", "") in ("done", "stopped", "error"):
            if not getattr(_sj, "_auto_exported", False):
                try:
                    from web.app import _do_spam_export
                    _res = _do_spam_export(_ctx, open_folder=False)
                    if _res:
                        logger.info(
                            "[auto_export] 📂 Рассылка: отчёт сохранён — %s (✅%d / ❌%d)",
                            _res["folder"], _res["ok_count"], _res["fail_count"],
                        )
                except Exception as e:
                    logger.debug("[auto_export] spam failed: %s", e)
                try:
                    _sj._auto_exported = True
                except Exception:
                    pass

        # DM (ctx.user_spam_job)
        _dj = getattr(_ctx, "user_spam_job", None)
        if _dj and getattr(_dj, "status", "") in ("done", "stopped", "error"):
            if not getattr(_dj, "_auto_exported", False):
                try:
                    from web.app import _do_dm_export
                    _res = _do_dm_export(_ctx, open_folder=False)
                    if _res:
                        logger.info(
                            "[auto_export] 📂 DM: отчёт сохранён — %s (✅%d / ❌%d)",
                            _res["folder"], _res["ok_count"], _res["fail_count"],
                        )
                except Exception as e:
                    logger.debug("[auto_export] dm failed: %s", e)
                try:
                    _dj._auto_exported = True
                except Exception:
                    pass

        # Мульти-задачи: когда ВСЕ multi_jobs завершены — делаем один сводный экспорт
        # по типу. Флаг ставим на сам список multi_jobs.
        _mjobs = (_ctx.temp.get("multi_jobs") or [])
        if _mjobs and not _ctx.temp.get("_multi_auto_exported", False):
            _all_done = all(
                getattr(mj, "status", "") in ("done", "stopped", "error") for mj in _mjobs
            )
            if _all_done:
                _types = {getattr(mj, "type", "") for mj in _mjobs}
                try:
                    if "join" in _types:
                        from web.app import _do_join_export
                        _r = _do_join_export(_ctx, open_folder=False)
                        if _r:
                            logger.info(
                                "[auto_export] 📂 Мульти-подписка: %s", _r["folder"]
                            )
                    if "spam" in _types:
                        from web.app import _do_spam_export
                        _r = _do_spam_export(_ctx, open_folder=False)
                        if _r:
                            logger.info(
                                "[auto_export] 📂 Мульти-рассылка: %s", _r["folder"]
                            )
                    if "dm" in _types:
                        from web.app import _do_dm_export
                        _r = _do_dm_export(_ctx, open_folder=False)
                        if _r:
                            logger.info(
                                "[auto_export] 📂 Мульти-DM: %s", _r["folder"]
                            )
                except Exception as e:
                    logger.debug("[auto_export] multi failed: %s", e)
                _ctx.temp["_multi_auto_exported"] = True

        # Сброс multi-флага при старте новой пачки (ctx.temp["multi_jobs"] очищается/заменяется)
        if not _mjobs and _ctx.temp.get("_multi_auto_exported"):
            _ctx.temp["_multi_auto_exported"] = False
