"""
join.py — логика задачи подписки (run_join_job) и check-only.
Редактируй поведение при вступлении, обработку ошибок и прогресс здесь.
"""
import asyncio
import json
import random
import re
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Tuple, List, Dict, Set

from telethon import errors
from telethon.tl.functions.channels import JoinChannelRequest
from telethon.tl.functions.messages import ImportChatInviteRequest, CheckChatInviteRequest

from config import (
    JOIN_MIN_INTERVAL_SEC, JOIN_JITTER_SEC, JOIN_MAX_PER_SESSION,
    JOIN_SESSION_BREAK_MIN, JOIN_SESSION_BREAK_MAX,
    JOIN_INVITE_PENALTY_MIN, JOIN_INVITE_PENALTY_MAX,
    FLOODWAIT_PENALTY_FACTOR,
)
from bot_ui.logger_setup import (
    log_join_start, log_join_result, log_join_finish,
)
import logging
# Свой логгер «telepush.join» — UI-консоль роутит на вкладку «Подписка»,
# а не «Системные» как раньше (импортировался общий «telepush»).
logger = logging.getLogger("telepush.join")
from models import Account, JoinJob
from core.database import store
from core.account_helpers import safe_join
from bot_ui.ui import (
    progress_bar, progress_bar_detailed,
    _eta_text, _status_label, _md_code, _job_log_add, _sleep_with_job_ui,
)
from bot_ui.utils import classify_link, canonical_key, get_user_timing
from core.database import ensure_user_folder
from core.client_builder import safe_disconnect
from telethon.tl import types as tltypes


# Счётчик «сессий» подписок для длинных антизаморозок пауз (phone -> int)
_join_session_count: dict = {}


def deduplicate_links(raw_links: List[str]) -> Tuple[List[str], int]:
    seen: Set[str] = set()
    out: List[str] = []
    for s in (raw_links or []):
        s = (s or "").strip()
        if not s:
            continue
        # Парсим ссылку в dict, потом canonical_key
        link_info = classify_link(s)
        if link_info:
            key = canonical_key(link_info)
        else:
            key = s.lower().strip("/")
        if key in seen:
            continue
        seen.add(key)
        out.append(s)
    dup_removed = max(0, len(raw_links) - len(out))
    return out, dup_removed

def summarize_counts(c: Dict[str, int]) -> str:
    """Краткая сводка счётчиков подписок на русском."""
    return (
        f"✅ Вступил: {c.get('joined', 0)}  "
        f"⏭️  Уже был: {c.get('already', 0)}  "
        f"📨 Заявка: {c.get('request', 0)}  "
        f"🚫 Пропущено: {c.get('ignored', 0)}  "
        f"⏳ Флуд: {c.get('flood', 0)}  "
        f"❌ Ошибок: {c.get('failed', 0)}  "
        f"❓ Прочее: {c.get('other', 0)}  "
        f"🔐 Капча: {c.get('captcha_solved', 0)}"
    )


def job_progress_text(job: JoinJob) -> str:
    """Текст прогресса задачи подписки — красивый, понятный, весь на русском."""
    bar  = progress_bar_detailed(job.processed, job.total)
    eta  = _eta_text(job.processed, job.total, job.started_at)
    status = _status_label(getattr(job, "status", "idle"))

    step = (getattr(job, "current_step", "") or "").strip()
    if not step:
        step = "Пауза — ждём команду ▶️" if job.status == "paused" else "—"

    cur      = getattr(job, "current_link",  None) or "—"
    last_res = getattr(job, "last_result",   None) or ""
    last_h   = humanize_join_status(last_res) if last_res else "—"

    # Последнее событие из кольцевого лога
    tail = ""
    try:
        logs = getattr(job, "log_lines", []) or []
        if logs:
            tail = logs[-1]
    except Exception:
        pass

    last_event = f"\n🕐 Последнее: {_md_code(tail, 140)}" if tail else ""

    # Скорость (ссылок/мин)
    speed_str = ""
    try:
        elapsed = time.time() - float(job.started_at or 0)
        if elapsed > 5 and job.processed > 0:
            rpm = job.processed / elapsed * 60
            speed_str = f"  💨 {rpm:.1f}/мин"
    except Exception:
        pass

    # Тайминги
    timing_str = ""
    try:
        from bot_ui.utils import get_user_timing
        from config import JOIN_MIN_INTERVAL_SEC, JOIN_JITTER_SEC, JOIN_MAX_PER_SESSION, JOIN_SESSION_BREAK_MIN, JOIN_SESSION_BREAK_MAX
        _interval = get_user_timing(job.owner_id, "join", "min_interval", JOIN_MIN_INTERVAL_SEC, phone=job.acc.phone)
        _jitter = get_user_timing(job.owner_id, "join", "jitter", JOIN_JITTER_SEC, phone=job.acc.phone)
        _max_sess = int(get_user_timing(job.owner_id, "join", "max_per_session", JOIN_MAX_PER_SESSION, phone=job.acc.phone))
        _brk_min = get_user_timing(job.owner_id, "join", "session_break_min", JOIN_SESSION_BREAK_MIN, phone=job.acc.phone)
        timing_str = f"⏳ Тайминги: {_interval:.0f}±{_jitter:.0f}с, пауза кожні {_max_sess} шт ({_brk_min:.0f}с)\n"
    except Exception:
        pass

    return (
        f"📥 **Автоподписки** — `{job.acc.phone}`\n"
        f"{'─' * 28}\n"
        f"📊 Прогресс: {bar}{speed_str}\n"
        f"⏱ Осталось: {eta}\n"
        f"🔄 Статус: {status}\n"
        f"⚙️  Шаг: {step}\n"
        f"{timing_str}"
        f"🔗 Ссылка: {_md_code(cur, 120)}\n"
        f"📋 Результат: {last_h}"
        f"{last_event}\n"
        f"{'─' * 28}\n"
        f"🗑 Дубликатов убрано: {job.dup_removed}\n"
        f"{summarize_counts(job.counts)}"
    )

def job_report_text(job: JoinJob, limit: int = 60) -> str:
    lines = [f"**Отчёт по подпискам** для `{job.acc.phone}`",
             f"Всего уникальных ссылок: {job.total} (дубликатов удалено: {job.dup_removed})",
             f"Счётчики: {summarize_counts(job.counts)}",
             ""]
    if not job.results:
        lines.append("_Пока нет результатов._")
    else:
        items = list(job.results.items())
        if len(items) > limit:
            head = items[:limit//2]
            tail = items[-limit//2:]
            body = head + [("…", "… (сокращено)")] + tail
        else:
            body = items
        lines.append("Подробности:")
        lines += [f"- {k} → {humanize_join_status(v)}" for k,v in body]
    if job.status in ("finished","stopped"):
        lines.append("")
        lines.append(f"Статус: **{job.status}**. Время: ~{int(time.time()-job.started_at)} сек.")
    return "\n".join(lines)



def humanize_join_status(code: str) -> str:
    c = str(code or "")
    # keep any tail details after colon
    def tail(prefix):
        return c[len(prefix):].strip()
    if c.startswith("joined:invite"): return "вступил по приватной ссылке"
    if c.startswith("joined:public"): return "вступил в публичный канал/группу"
    if c.startswith("already_member"): return "уже участник"
    if c.startswith("request_sent"): return "заявка на вступление отправлена (ждёт модерации)"
    if c.startswith("floodwait:"):
        t = tail("floodwait:")
        return f"слишком часто — FloodWait {t} c" if t else "слишком часто — временный лимит Telegram (FloodWait)"
    if c.startswith("failed:quota"):
        _parts = c.split("failed:quota", 1)
        t = _parts[1].strip(" :") if len(_parts) > 1 else ""
        return "превышен лимит (квота бота)" + (f" — {t}" if t else "")
    if c.startswith("failed:banned_in_channel"): return "забанены в этом канале/чате"
    if c.startswith("failed:invite_expired"): return "инвайт просрочен"
    if c.startswith("failed:invite_private"): return "приватная ссылка/канал недоступен"
    if c.startswith("ignored:not_a_chat"): return "это не чат/канал"
    if c.startswith("ignored:broadcast_no_write"): return "broadcast-канал или запрещено писать — пропущен"
    if c.startswith("ignored:username_not_found"): return "имя пользователя не существует"
    if c.startswith("ignored:username_invalid"): return "неверное имя пользователя"
    if c.startswith("ignored:cannot_resolve"): return "не удалось распознать ссылку"
    if "SessionRevokedError" in c or "AuthKeyUnregisteredError" in c or "session_invalid" in c:
        return "сессия недействительна — переподключите аккаунт"
    if "FrozenMethodInvalid" in c or "frozen" in c.lower():
        return "❄️ аккаунт заморожен Telegram (FrozenMethodInvalidError) — замените аккаунт"
    return c

def humanize_spam_status(code: str) -> str:
    c = str(code or "")
    def tail(prefix):
        return c[len(prefix):].strip()
    if c.startswith("sent"): return "отправлено"
    if c.startswith("spamfilter:peer_flood"): return "спам‑фильтр аккаунта (PeerFlood)"
    if c.startswith("floodwait:"):
        t = tail("floodwait:")
        return f"слишком часто — FloodWait {t} c" if t else "слишком часто — временный лимит Telegram (FloodWait)"
    if c.startswith("slowmode:"):
        t = tail("slowmode:")
        return f"в чате включён slow‑mode — ждать {t} c" if t else "в чате включён slow‑mode"
    if c.startswith("no_rights:forbidden"): return "нет права писать в этот чат"
    if c.startswith("no_rights:banned"): return "забанены в чате/канале"
    if c.startswith("no_rights:admin_required"): return "нужны права администратора"
    if c == "failed:not_in_chat": return "аккаунт не состоит в этом чате — сначала вступите"
    if c.startswith("failed:quota"):
        _parts = c.split("failed:quota", 1)
        t = _parts[1].strip(" :") if len(_parts) > 1 else ""
        return "превышен лимит (квота бота)" + (f" — {t}" if t else "")
    if c.startswith("failed:FrozenMethodInvalidError"):
        return "❄️ аккаунт заморожен Telegram"
    if c.startswith("failed:account_dead"):
        return "❌ аккаунт недоступен (сессия)"
    return c

def humanize_dm_status(code: str) -> str:
    c = str(code or "")
    def tail(prefix):
        return c[len(prefix):].strip()
    if c.startswith("sent"): return "отправлено"
    if c.startswith("privacy:restricted"): return "нельзя писать (403 Forbidden): приватность получателя или блокировка"
    if c.startswith("privacy:write_forbidden"): return "нет права писать (write forbidden)"
    if c.startswith("spamfilter:peer_flood"): return "спам‑фильтр аккаунта (PeerFlood)"
    if c.startswith("floodwait:"):
        t = tail("floodwait:")
        return f"слишком часто — FloodWait {t} c" if t else "слишком часто — временный лимит Telegram (FloodWait)"
    if c.startswith("blocked:trigger:"):
        w = tail("blocked:trigger:")
        return f"заблокировано: запрещённое слово «{w}»"
    if c.startswith("skipped:recently_contacted"): return "пропущено: недавно уже писали"
    if c.startswith("failed:quota"):
        _parts = c.split("failed:quota", 1)
        t = _parts[1].strip(" :") if len(_parts) > 1 else ""
        return "превышен лимит (квота бота)" + (f" — {t}" if t else "")
    if "SessionRevokedError" in c or "AuthKeyUnregisteredError" in c or "session_invalid" in c:
        return "сессия недействительна — переподключите аккаунт"
    if c.startswith("failed:"):
        return "ошибка: " + c.split("failed:",1)[1]
    return c

def save_job_snapshot(job: JoinJob):
    if not job.job_id:
        job.job_id = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    base_user = ensure_user_folder(job.owner_id, job.acc.username, job.acc.phone)
    snap = {
        "owner_id": job.owner_id,
        "phone": job.acc.phone,
        "queue": job.queue,
        "total": job.total,
        "dup_removed": job.dup_removed,
        "results": job.results,
        "processed": job.processed,
        "counts": job.counts,
        "status": job.status,
        "started_at": job.started_at,
    }
    (base_user / "queues").mkdir(exist_ok=True, parents=True)
    (base_user / "queues" / f"{job.job_id}.json").write_text(json.dumps(snap, ensure_ascii=False, indent=2), encoding="utf-8")


async def run_join_job(job: JoinJob):
    job.status = "running"
    job.pause_event.set()
    acc = job.acc

    # Пробуждаем аккаунт если он в idle
    from core.client_builder import ensure_connected
    if not await ensure_connected(acc):
        job.status = "stopped"
        job.current_step = "Аккаунт недоступен"
        return

    log_join_start(acc.phone, job.total, job.dup_removed, job.owner_id)

    try:
        job.current_step = "Запуск…"
        job.current_link = None
    except Exception:
        pass
    _job_log_add(job, f"Старт задачи подписок: {job.total} ссылок (дубликатов удалено: {job.dup_removed})")

    for idx, link in enumerate(job.queue, start=1):
        if job.stop_flag:
            break

        # Пауза/продолжить от пользователя
        await job.pause_event.wait()
        if job.stop_flag:
            break

        # Текущая ссылка в UI
        try:
            job.current_link = link
            job.current_step = "Подготовка…"
        except Exception:
            pass

        # --- Пейсинг: соблюдаем минимальный интервал между попытками (с учётом настроек пользователя) ---
        # Заранее классифицируем — invite-ссылки требуют более длинного интервала
        # (ImportChatInviteRequest у Telegram строже, чем JoinChannelRequest).
        _is_invite = False
        try:
            _kind_preview = classify_link(link)
            _is_invite = bool(_kind_preview and _kind_preview.get("kind") == "invite")
        except Exception:
            pass
        try:
            now = time.time()  # Unix timestamp, как last_join_at
            _min_interval = get_user_timing(job.owner_id, "join", "min_interval", JOIN_MIN_INTERVAL_SEC, phone=acc.phone)
            _jitter = get_user_timing(job.owner_id, "join", "jitter", JOIN_JITTER_SEC, phone=acc.phone)
            # Для invite-ссылок (t.me/+xxx, joinchat/) — увеличиваем базовый
            # интервал на 70%. У клиента в скриншоте было 4 FloodWait (115/88/238/1431с)
            # на пачке из 12 invite-ссылок с дефолтным 18с пейсингом.
            if _is_invite:
                _min_interval = float(_min_interval) * 1.7
            # Адаптивный пейсинг: каждый пойманный FloodWait в этой задаче
            # умножает базовый интервал — Telegram не любит когда после
            # «ой подожди 115с» сразу же лупим запросы дальше с тем же ритмом.
            # Множитель 1.4^N: 1 flood → ×1.4, 2 → ×1.96, 3 → ×2.74. На 4-5 повторах
            # ловим лимит срабатывания (см. floodwait_stop_sec) и job останавливается.
            _fw_count = int(getattr(job, "_fw_count", 0) or 0)
            if _fw_count > 0:
                _adaptive_mult = min(1.4 ** _fw_count, 4.0)
                _min_interval = float(_min_interval) * _adaptive_mult
            gap = (acc.last_join_at + float(_min_interval)) - now
            if gap > 0 and gap < 600:  # max 10 min wait, защита от бага с clock
                wait_sec = gap + random.uniform(0, _jitter)
                if wait_sec >= 8:
                    await _sleep_with_job_ui(job, wait_sec, reason="Пейсинг (лимит по времени)")
                else:
                    try:
                        job.current_step = f"Пейсинг: жду {int(round(wait_sec))}с"
                    except Exception:
                        pass
                    await asyncio.sleep(wait_sec)
        except Exception:
            pass

        # Проверка сессии: только на первой ссылке или если аккаунт был помечен dead
        try:
            if getattr(acc, 'dead', False):
                reason = acc.dead_reason or "dead"
                status_code = "failed:FrozenMethodInvalidError" if getattr(acc, 'frozen', False) else "failed:session_invalid"
                job.results[link] = status_code
                job.last_result = job.results[link]
                job.counts["failed"] += 1
                _job_log_add(job, f"{idx}/{job.total} {link} -> {status_code} (аккаунт {reason})")
                break
            # is_user_authorized только на первой ссылке — не бьём API каждый раз
            if idx == 1:
                auth_ok = False
                try:
                    if acc.client is None:
                        auth_ok = False
                    else:
                        auth_ok = await acc.client.is_user_authorized()
                except (errors.AuthKeyUnregisteredError, errors.SessionRevokedError):
                    auth_ok = False
                if not auth_ok:
                    try:
                        await safe_disconnect(acc.client)
                    except Exception:
                        pass
                    job.results[link] = "failed:session_invalid"
                    job.last_result = job.results[link]
                    job.counts["failed"] += 1
                    _job_log_add(job, f"session_invalid (не авторизован)")
                    break
        except Exception:
            pass

        # Классификация ссылки
        try:
            job.current_step = "Проверяю ссылку…"
        except Exception:
            pass
        t = classify_link(link)
        logger.debug("[classify] %r -> %r", link, t)

        if not t:
            job.results[link] = "ignored:unsupported_or_unsafe_link"
            job.last_result = job.results[link]
            job.counts["ignored"] += 1
            _job_log_add(job, f"{idx}/{job.total} {link} -> ignored (неподдерживаемая/небезопасная ссылка)")
        else:
            # ── Health-check соединения каждые 25 ссылок ────────────────────
            # Длинные задачи (100+ ссылок, часы работы) могут потерять TCP-соединение
            # из-за NAT/файрвол таймаутов. Проверяем и восстанавливаем без остановки задачи.
            if idx % 25 == 0:
                try:
                    cli = acc.client
                    if cli and not cli.is_connected():
                        job.current_step = "⚠️ Переподключение…"
                        _job_log_add(job, f"⚠️ Клиент {acc.phone} отключён — переподключаюсь…")
                        await asyncio.wait_for(cli.connect(), timeout=20.0)
                        _job_log_add(job, f"✅ Переподключён {acc.phone}")
                        await asyncio.sleep(1.0)
                except asyncio.TimeoutError:
                    _job_log_add(job, f"❌ Таймаут переподключения {acc.phone} — пропускаю ссылку")
                    job.results[link] = "failed:reconnect_timeout"
                    job.last_result = job.results[link]
                    job.counts["failed"] += 1
                    job.processed += 1
                    continue
                except Exception as _hc_err:
                    logger.warning("[health-check] %s: %s", acc.phone, _hc_err)
            # ────────────────────────────────────────────────────────────────
            try:
                job.current_step = "Вступаю…"
            except Exception:
                pass
            status, note = await safe_join(acc, job.owner_id, t)
            view = status if not note else f"{status} ({note})"
            job.results[link] = view
            job.last_result = view

            log_join_result(acc.phone, idx, job.total, link, view)
            _job_log_add(job, f"{idx}/{job.total} {link} -> {view}")

            # Anti-detect: после успешного вступления — попытка решить капчу
            # если в чате есть бот-капча (Shieldy/MissRose/Combot). Сразу же,
            # потому что многие чаты кикают за неактивность через 60-300 сек.
            try:
                ad_cfg = getattr(job, 'ad_config', None) or {}
                if ad_cfg.get('solve_captcha') and status.startswith('joined'):
                    from utils.anti_detect import solve_captcha
                    # Пауза перед попыткой — даём боту время прислать капчу
                    await asyncio.sleep(random.uniform(2, 5))
                    # Получаем entity чата для сканирования
                    try:
                        chat_entity = await acc.client.get_entity(t.get('peer') or link)
                        await solve_captcha(acc.client, chat_entity, module="join")
                    except Exception as _ce:
                        logger.debug("[capt] %s: %s", acc.phone, _ce)
            except Exception:
                pass

            try:
                acc.last_join_at = time.time()
                acc.last_used_at = time.time()
            except Exception:
                pass

            # Антизаморозка: invite-ссылки отслеживаются строже — доп. задержка
            try:
                if t and t.get("kind") == "invite" and status.startswith("joined"):
                    _inv_delay = random.uniform(JOIN_INVITE_PENALTY_MIN, JOIN_INVITE_PENALTY_MAX)
                    _job_log_add(job, f"🛡️ Доп. пауза для invite-ссылки: {int(_inv_delay)}с")
                    await asyncio.sleep(_inv_delay)
            except Exception:
                pass

            if note and "captcha:solved" in note:
                job.counts["captcha_solved"] += 1
            elif note and "captcha:not_seen" in note and status.startswith("joined"):
                # Капча не решена — уведомляем клиента
                _job_log_add(job, f"⚠️ Капча не решена в чате — возможно требуется ручная верификация")
                job.counts.setdefault("captcha_unsolved", 0)
                job.counts["captcha_unsolved"] += 1

            if status.startswith("joined"):
                job.counts["joined"] += 1
                acc.stats.joins_ok += 1
                # Антизаморозка: считаем подписки в текущей «сессии», при достижении лимита — длинная пауза (с учётом настроек)
                try:
                    phone_key = acc.phone
                    _max_per_sess = int(get_user_timing(job.owner_id, "join", "max_per_session", JOIN_MAX_PER_SESSION, phone=acc.phone))
                    _join_session_count[phone_key] = _join_session_count.get(phone_key, 0) + 1
                    if _join_session_count[phone_key] >= _max_per_sess:
                        _join_session_count[phone_key] = 0
                        _brk_min = get_user_timing(job.owner_id, "join", "session_break_min", JOIN_SESSION_BREAK_MIN, phone=acc.phone)
                        _brk_max = get_user_timing(job.owner_id, "join", "session_break_max", JOIN_SESSION_BREAK_MAX, phone=acc.phone)
                        _break = random.uniform(_brk_min, _brk_max)
                        _job_log_add(job, f"🛡️ Антизаморозка: пауза {int(_break)}с после {_max_per_sess} подписок подряд")
                        await _sleep_with_job_ui(job, _break, reason="🛡️ Пауза безопасности (антизаморозка)")
                except Exception:
                    pass
            elif status.startswith("already"):
                job.counts["already"] += 1
            elif status.startswith("request_sent"):
                job.counts["request"] += 1
            # Per-account лимит: суммарное число «занятых слотов» (joined + already +
            # request) — когда акк дошёл до настроенного потолка, останавливаем
            # задачу. 0 = без лимита (старое поведение — до конца очереди или
            # квоты Telegram).
            try:
                _join_cap = int(getattr(job, "max_per_acc", 0) or 0)
                if _join_cap > 0:
                    _used = (job.counts.get("joined", 0) + job.counts.get("already", 0)
                             + job.counts.get("request", 0))
                    if _used >= _join_cap:
                        _job_log_add(job, f"🛑 Достигнут лимит: {_join_cap} подписок с этого аккаунта. Останавливаюсь.")
                        try:
                            job.current_step = f"🛑 Стоп: лимит {_join_cap} подписок"
                        except Exception:
                            pass
                        job.stop_flag = True
            except Exception:
                pass
            if status.startswith("floodwait"):
                job.counts["flood"] += 1
                # Считаем накопленный FloodWait — пейсинг наверху по нему
                # экспоненциально растягивается, чтобы не плодить новые floods.
                try:
                    job._fw_count = int(getattr(job, "_fw_count", 0) or 0) + 1
                except Exception:
                    pass
                # Авто-уведомление о FloodWait
                try:
                    from core.auto_resume import handle_floodwait_join
                    _fw_sec = 0
                    try:
                        _fw_sec = int(status.split(":", 1)[1])
                    except Exception:
                        pass
                    if _fw_sec > 0:
                        await handle_floodwait_join(acc, job, _fw_sec)
                    # Большой FloodWait = Telegram плотно прижал акк. Нет смысла
                    # спать 9 часов — останавливаем, клиент запустит на другом акке.
                    try:
                        _fw_limit = int(get_user_timing(job.owner_id, "join", "floodwait_stop_sec", 1800, phone=acc.phone))
                    except Exception:
                        _fw_limit = 1800
                    if _fw_sec >= _fw_limit:
                        _job_log_add(job, f"🛑 FloodWait {_fw_sec}с ≥ порога {_fw_limit}с — останавливаю задачу. Акк устал, запусти остальное на другом аккаунте.")
                        logger.warning("[подписки:%s] FloodWait %ds превысил порог %ds — остановка задачи", acc.phone, _fw_sec, _fw_limit)
                        try:
                            job.current_step = f"🛑 Стоп: FloodWait {_fw_sec}с превысил лимит"
                        except Exception:
                            pass
                        job.stop_flag = True
                        break
                except Exception:
                    pass
            elif status.startswith("failed:quota"):
                job.counts["failed"] += 1
                # Квота на сутки — не бесконечный цикл, корректно завершаемся.
                try:
                    from core.auto_resume import handle_quota_limit_join
                    _reason = status.split("failed:quota", 1)[1].strip(" :()")
                    await handle_quota_limit_join(acc, job)
                except Exception:
                    pass
                _job_log_add(job, f"🛑 Квота подписок исчерпана на {acc.phone} — задача остановлена, продолжите на другом аккаунте.")
                try:
                    job.current_step = "🛑 Стоп: квота подписок исчерпана (24ч)"
                except Exception:
                    pass
                job.stop_flag = True
                break
            elif status.startswith("failed:SessionRevokedError") or status.startswith("failed:AuthKeyUnregisteredError") or status.startswith("failed:session_invalid"):
                job.counts["failed"] += 1
                job.stop_flag = True
                # Важно: дальше задача должна корректно завершиться
                break
            elif status.startswith("failed:FrozenMethodInvalidError"):
                # Аккаунт заморожен Telegram — нет смысла продолжать с ним
                job.counts["failed"] += 1
                _job_log_add(job, f"❄️ Аккаунт {acc.phone} заморожен Telegram — задача остановлена. Замените аккаунт.")
                # Синхронизируем папку: переносим файлы в Заморозка/ чтобы
                # после рестарта софта flag не потерялся. acc.frozen=True уже
                # выставлен в safe_join, sync вычитает категорию из лямбды.
                try:
                    from core.accounts_folder import sync_accounts_folder
                    sync_accounts_folder({acc.phone: acc})
                except Exception as _se:
                    logger.debug("[join] sync after frozen failed: %s", _se)
                job.stop_flag = True
                break
            elif status.startswith("failed"):
                job.counts["failed"] += 1
            elif status.startswith("ignored"):
                job.counts["ignored"] += 1
                if status == "ignored:broadcast_no_write":
                    _job_log_add(job, f"ℹ️ {link} — broadcast/нельзя писать, пропущен")
            else:
                job.counts["other"] += 1

            # FloodWait показываем в UI как ожидание
            if status.startswith("floodwait:"):
                try:
                    fw = int(status.split(":", 1)[1])
                except Exception:
                    fw = 0
                if fw > 0:
                    logger.info("⏳ [подписки:%s] FloodWait %d сек — ожидаю...", acc.phone, fw)
                    await _sleep_with_job_ui(job, fw, reason="FloodWait (Telegram попросил подождать)", update_every=25)

        job.processed += 1
        # Ограничиваем рост памяти
        if len(job.log_lines) > 200:
            job.log_lines = job.log_lines[-100:]
        if len(job.results) > 5000:
            job.results = dict(list(job.results.items())[-2000:])
        # Обновляем per-account статистику
        if status.startswith("failed"):
            acc.stats.joins_failed += 1
        # Снапшот каждые 10 элементов — не на каждой итерации (диск I/O)
        if job.processed % 10 == 0 or job.processed == job.total:
            save_job_snapshot(job)

        # Небольшая пауза между ссылками (видно в UI)
        delay = random.uniform(4.0, 9.0)
        try:
            job.current_step = f"Пауза перед следующей ссылкой: {int(round(delay))}с"
        except Exception:
            pass

        if job.stop_flag:
            break
        
        try:
            remaining = float(delay)
        except Exception:
            remaining = 0.0
        while remaining > 0:
            if job.stop_flag:
                break
            await job.pause_event.wait()
            step = 1.0 if remaining > 1.0 else remaining
            await asyncio.sleep(step)
            remaining -= step

    job.status = "stopped" if job.stop_flag else "finished"
    log_join_finish(acc.phone, job.status, job.counts)
    try:
        if getattr(acc, 'frozen', False):
            job.current_step = "❄️ Аккаунт заморожен Telegram. Замените аккаунт."
        elif job.stop_flag:
            job.current_step = "Остановлено пользователем."
        else:
            job.current_step = "Завершено."
    except Exception:
        pass
    save_job_snapshot(job)

    # Уведомляем что акк заморожен — через лог приложения, не через бота.
    if getattr(acc, 'frozen', False):
        try:
            job.log_lines.append(
                f"❄️ Аккаунт {acc.phone} заморожен Telegram (FrozenMethodInvalidError). "
                f"Удалите его через «Аккаунты» → «Удалить» и подключите новый."
            )
        except Exception:
            pass

    # Сгруппированный отчёт по результатам (пишется в job.log_lines).
    _emit_join_summary(job)

    # ── Уведомление о завершении ──
    try:
        from core.notifications import notify_task_finished, maybe_loop_join
        await notify_task_finished(
            None, job.owner_id, "join", acc.phone,
            dict(job.counts), job.status,
        )
    except Exception as _ne:
        logger.debug("[join] notify error: %s", _ne)

    # ── Loop-режим: авто-повтор (со свежими ссылками из контекста) ──
    if job.status == "finished" and not job.stop_flag:
        try:
            from core.notifications import maybe_loop_join
            _owner = job.owner_id
            _acc = acc

            async def _restart_join():
                from bot_ui.ui import get_ctx
                ctx = get_ctx(_owner)
                if not _acc.client or not _acc.client.is_connected():
                    return
                # Берём СВЕЖИЕ ссылки из контекста (пользователь мог обновить)
                raw_links = ctx.temp.get("loaded_links") or list(job.queue)
                new_links, _ = deduplicate_links(raw_links)
                if not new_links:
                    logger.info("[loop-join] нет ссылок для повтора")
                    return
                new_job = JoinJob(
                    owner_id=_owner, acc=_acc,
                    queue=new_links, total=len(new_links), dup_removed=0,
                )
                ctx.job = new_job
                from core.worker_pool import submit_for_user
                new_job.task = submit_for_user(
                    _owner, lambda: run_join_job(new_job),
                    job_id=f"loop_join_{_owner}", name="loop-join",
                )

            await maybe_loop_join(_owner, _restart_join)
        except Exception as _le:
            logger.debug("[join] loop error: %s", _le)


def _emit_join_summary(job):
    """Пишет в консоль приложения сгруппированный отчёт после задачи подписки.
    Группировка как в популярных join-ботах (AUSTOJOIN и др.) — клиенту нужно
    одним взглядом понять «куда подал, куда не дошёл, куда ждать модерацию»."""
    try:
        groups = {
            "joined":     [],   # ✅ вступил
            "request":    [],   # 📋 заявка отправлена
            "already":    [],   # 📌 уже состою
            "flood":      [],   # ⏳ FloodWait
            "ignored":    [],   # ❕ пропущено
            "failed":     [],   # ❌ ошибки
        }
        for link, status in job.results.items():
            s = (status or "").lower()
            if s.startswith("joined"):       groups["joined"].append((link, status))
            elif s.startswith("request"):    groups["request"].append((link, status))
            elif s.startswith("already"):    groups["already"].append((link, status))
            elif s.startswith("floodwait"):  groups["flood"].append((link, status))
            elif s.startswith("ignored"):    groups["ignored"].append((link, status))
            else:                            groups["failed"].append((link, status))

        _job_log_add(job, "")
        _job_log_add(job, "═══════ Итоги подписки ═══════")
        if groups["joined"]:
            _job_log_add(job, f"✅ Вступил: {len(groups['joined'])}")
            for l, _ in groups["joined"][:30]:
                _job_log_add(job, f"  • {l}")
            if len(groups["joined"]) > 30:
                _job_log_add(job, f"  … и ещё {len(groups['joined']) - 30}")
        if groups["request"]:
            _job_log_add(job, f"📋 Заявка отправлена (ждёт модерации): {len(groups['request'])}")
            for l, _ in groups["request"][:30]:
                _job_log_add(job, f"  • {l}")
            if len(groups["request"]) > 30:
                _job_log_add(job, f"  … и ещё {len(groups['request']) - 30}")
        if groups["already"]:
            _job_log_add(job, f"📌 Уже состою: {len(groups['already'])}")
        if groups["flood"]:
            _job_log_add(job, f"⏳ FloodWait (Telegram попросил подождать): {len(groups['flood'])}")
            for l, status in groups["flood"][:30]:
                _job_log_add(job, f"  • {l} — {humanize_join_status(status)}")
            if len(groups["flood"]) > 30:
                _job_log_add(job, f"  … и ещё {len(groups['flood']) - 30}")
        if groups["failed"]:
            _job_log_add(job, f"❌ Не удалось: {len(groups['failed'])}")
            for l, status in groups["failed"][:30]:
                _job_log_add(job, f"  • {l} — {humanize_join_status(status)}")
            if len(groups["failed"]) > 30:
                _job_log_add(job, f"  … и ещё {len(groups['failed']) - 30}")
        if groups["ignored"]:
            _job_log_add(job, f"❕ Пропущено (нерабочие/неподдерживаемые ссылки): {len(groups['ignored'])}")
        _job_log_add(job, "═══════════════════════════════")
    except Exception as e:
        logger.debug("[join] summary emit failed: %s", e)


async def check_link_only(acc: Account, link: str) -> str:
    t = classify_link(link)
    if not t:
        return "плохой формат"
    if acc.client is None:
        return "аккаунт не подключён"
    try:
        if t["kind"] == "invite":
            info = await acc.client(CheckChatInviteRequest(t["hash"]))
            # ChatInvite / ChatInviteAlready
            if isinstance(info, tltypes.ChatInviteAlready):
                return "уже участник (без вступления)"
            return "инвайт валиден (без вступления)"
        if t["kind"] == "public":
            try:
                await acc.client.get_entity(t["username"])
                return "публичная валидна (без вступления)"
            except errors.UsernameNotOccupiedError:
                return "не существует"
            except errors.UsernameInvalidError:
                return "невалидное имя"
        return "неизвестный тип"
    except errors.InviteHashExpiredError:
        return "инвайт протух (expired)"
    except errors.InviteHashInvalidError:
        return "инвайт невалиден"
    except errors.ChannelPrivateError:
        return "приват/по запросу"
    except errors.FloodWaitError as e:
        return f"проверить нельзя (FloodWait {e.seconds}s)"
    except Exception:
        return "проверить нельзя (ошибка)"


async def perform_check_only(uid: int, acc: Account, links: List[str]):
    """Возвращает список (link, status) для UI приложения. Раньше слал результат
    батчами через Telegram-бот; в standalone это уже не нужно — UI забирает
    результат как обычный ответ API."""
    unique, _ = deduplicate_links(links)
    out = []
    for ln in unique:
        try:
            st = await check_link_only(acc, ln)
        except Exception as e:
            st = f"ошибка проверки: {type(e).__name__}"
        out.append((ln, st))
        await asyncio.sleep(0.15)
    return out

# ---------------------- Spam helpers (β) ----------------------
