# full_leave.py version: 1.2
# ============================================================
# سكربت المغادرة الشاملة (تشغيل يدوي آمن ومحكَم)
# المرحلة أ: مغادرة كل قنوات الاشتراك الإجباري لكل البوتات.
# المرحلة ب: مغادرة كل القنوات المسجّلة في القاعدة.
#
# ضوابط الأمان:
#  1) يرفض العمل إذا كان التجميع مفعلاً (is_active=1) إلا مع --force.
#  2) قفل PID يمنع تشغيل نسختين في نفس الوقت.
#  3) Watchdog: FloodWait >= 300s أو تعليق بلا تقدّم (>300s) يوقف تلقائياً.
#  4) إيقاف طوارئ يدوي: أنشئ ملف logs/full_leave.stop لأي إيقاف فوري.
#  5) كل عملية على حساب داخل مهلة زمنية محددّة (25s) — لا تجمد الأبد.
#
# التشغيل:  tele/bin/python3 full_leave.py --dry-run   (استعراض فقط)
#            tele/bin/python3 full_leave.py            (تنفيذ كامل)
# ============================================================

import argparse
import asyncio
import logging
import os
import re
import signal
import time

import config
import shared as _shared
from shared import (
    get_setting, get_bots, get_accounts, get_account_lock,
    get_force_sub_channels, get_channels, del_force_sub_channels,
    flood_map, dead_sessions, cleanup_client,
)
import gather_engine as engine

BASE = os.path.dirname(os.path.abspath(__file__))
LOG_DIR = os.path.join(BASE, "logs")
LOG_FILE = os.path.join(LOG_DIR, "full_leave.log")
PID_FILE = os.path.join(LOG_DIR, "full_leave.pid")
STOP_FILE = os.path.join(LOG_DIR, "full_leave.stop")

FLOOD_STOP_SEC = 300
HANG_CHECK_SEC = 300
PHASE_MAX_SEC = 3 * 3600
FORCE_SUB_CONC = 10
ACCOUNT_TIMEOUT = 25
WARM_CONC = 15
WARM_GAP = 0.3

stop_required = asyncio.Event()
_state = {"phase": "ready", "last_progress": 0.0}

log = logging.getLogger("full_leave")


def setup_logging():
    os.makedirs(LOG_DIR, exist_ok=True)
    fmt = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    console = logging.StreamHandler()
    console.setFormatter(fmt)
    file_h = logging.FileHandler(LOG_FILE, encoding="utf-8")
    file_h.setFormatter(fmt)
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.handlers[:] = [console, file_h]
    logging.getLogger("pyrogram").setLevel(logging.WARNING)
    logging.getLogger("aiohttp").setLevel(logging.WARNING)


def acquire_pid_lock() -> bool:
    os.makedirs(LOG_DIR, exist_ok=True)
    if os.path.exists(PID_FILE):
        try:
            with open(PID_FILE) as f:
                old = int(f.read().strip() or 0)
            if old and old != os.getpid():
                os.kill(old, 0)
                return False
        except (ValueError, OSError):
            pass
    with open(PID_FILE, "w") as f:
        f.write(str(os.getpid()))
    return True


def release_pid_lock():
    try:
        if os.path.exists(PID_FILE):
            os.remove(PID_FILE)
    except OSError:
        pass


def request_stop(reason: str):
    if stop_required.is_set():
        return
    log.error(f"🛑 [STOP] إيقاف العملية: {reason}")
    stop_required.set()
    _shared.leave_all_on = False
    try:
        with open(STOP_FILE, "w") as f:
            f.write(reason)
    except OSError:
        pass


def install_signal_handlers(loop):
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(
                sig, lambda s=sig: request_stop(f"استقبلت إشارة {s.name}")
            )
        except (NotImplementedError, RuntimeError):
            pass


def build_ch_map(rows, urls):
    ch_map = {}
    seen = set()
    for row in rows:
        url = row["url"]
        if url in seen:
            continue
        seen.add(url)
        un = url.rstrip("/").split("/")[-1]
        ch_map[url] = {
            "id": row.get("channel_id"),
            "username": None if un.startswith("+") else un,
            "url": url,
        }
    for url in urls:
        if url not in ch_map:
            un = url.rstrip("/").split("/")[-1]
            ch_map[url] = {
                "id": None,
                "username": None if un.startswith("+") else un,
                "url": url,
            }
    return ch_map


async def warm_sessions(accounts) -> int:
    to_warm = [
        a["phone"] for a in accounts
        if a["phone"] not in _shared.clients and a["phone"] not in dead_sessions
    ]
    if not to_warm:
        return 0
    log.info(f"⏳ تسخين {len(to_warm)} جلسة (متدرّج)...")
    opened = [0]
    sem = asyncio.Semaphore(WARM_CONC)

    async def _one(phone):
        async with sem:
            try:
                await asyncio.wait_for(
                    engine.get_client(phone), timeout=ACCOUNT_TIMEOUT
                )
                opened[0] += 1
            except Exception:
                pass
            await asyncio.sleep(WARM_GAP)

    await asyncio.gather(*(_one(p) for p in to_warm), return_exceptions=True)
    log.info(f"✅ اكتمل التسخين: {opened[0]}/{len(to_warm)} جلسة")
    return opened[0]


async def _leave_one_fs(phone, ch_obj, sem) -> bool:
    if stop_required.is_set():
        return False
    async with sem:
        if stop_required.is_set() or phone in dead_sessions:
            return False
        _state["last_progress"] = time.time()
        try:
            async with get_account_lock(phone):
                c = await asyncio.wait_for(
                    engine.get_client(phone), timeout=ACCOUNT_TIMEOUT
                )
            await asyncio.wait_for(
                engine.leave_ch(c, phone, ch_obj, ctx="FS FULL LEAVE"),
                timeout=ACCOUNT_TIMEOUT,
            )
            _state["last_progress"] = time.time()
            return True
        except Exception as e:
            err = str(e).lower()
            if "flood_wait" in err or "420" in err:
                sec = 0
                for m in re.findall(r"\d+", err):
                    sec = max(sec, int(m))
                sec = sec or 3600
                flood_map[phone] = time.time() + sec + 60
                log.warning(
                    f"[FS] {phone}: FloodWait {sec}s — {ch_obj.get('url')}"
                )
                if sec >= FLOOD_STOP_SEC:
                    request_stop(f"FloodWait ضخم ({sec}s) على الحساب {phone}")
            elif any(k in err for k in (
                "not found", "invalid", "not participant",
                "user not participant", "channel_private",
                "peer_id_invalid", "usernotparticipant",
                "chat_not_found", "channel_invalid",
            )):
                return True
            elif any(k in err for k in (
                "auth_key", "unregistered", "revoked",
                "malformed", "disk image", "eof when reading",
            )):
                dead_sessions.add(phone)
                log.warning(f"[FS] {phone}: جلسة تالفة/ميتة — تُتخطّى")
            else:
                log.warning(f"[FS] {phone} | {ch_obj.get('url')}: {e}")
            return False


async def phase_force_sub_all_bots(accounts):
    bots = await get_bots()
    unames = []
    seen = set()
    for bot in bots:
        un = (bot.get("username") or "").strip("@").lower()
        if un and un not in seen:
            seen.add(un)
            unames.append(un)
    try:
        async with _shared._pg_pool.acquire() as con:
            rows = await con.fetch(
                "SELECT DISTINCT bot_username FROM force_sub_channels"
            )
        for r in rows:
            un = str(r["bot_username"] or "").strip("@").lower()
            if un and un not in seen:
                seen.add(un)
                unames.append(un)
    except Exception as e:
        log.warning(f"تعذّر جلب القائمة الكاملة للاشتراكات: {e}")
    if not unames:
        log.info("لا توجد بوتات ولا اشتراكات إجباري — تُتخطّى المرحلة أ.")
        return 0, 0
    sem = asyncio.Semaphore(FORCE_SUB_CONC)
    total_urls = 0
    total_done = 0
    for bot_un in unames:
        if stop_required.is_set():
            break
        rows = await get_force_sub_channels(bot_un)
        urls = [r["url"] for r in rows]
        if not urls:
            log.info(f"🤖 @{bot_un}: لا توجد قنوات اشتراك إجباري.")
            continue
        ch_map = build_ch_map(rows, urls)
        log.info(
            f"🤖 @{bot_un}: بدء مغادرة {len(urls)} قناة × {len(accounts)} حساب"
        )
        bot_done = 0
        bot_failed = 0
        for url in urls:
            if stop_required.is_set():
                break
            ch_obj = ch_map[url]
            results = await asyncio.gather(
                *(_leave_one_fs(a["phone"], ch_obj, sem) for a in accounts),
                return_exceptions=True,
            )
            for r in results:
                if r is True:
                    bot_done += 1
                else:
                    bot_failed += 1
            await asyncio.sleep(0.5)
        total_urls += len(urls)
        total_done += bot_done
        if not stop_required.is_set():
            await del_force_sub_channels(bot_un)
            log.info(
                f"🤖 @{bot_un}: انتهت — نجح {bot_done} | فشل {bot_failed} | "
                "حُذفت قنواته من القاعدة"
            )
        else:
            log.warning(
                f"🤖 @{bot_un}: توقفت منتصف الطريق — نجح {bot_done} | "
                "أُبقيت بقية قنواته في القاعدة"
            )
            break
    return total_urls, total_done


async def phase_leave_registered():
    log.info("▶️ بدء المرحلة ب: مغادرة القنوات المسجّلة...")
    _shared.leave_all_on = True
    try:
        await engine.do_leave_all(0)
    finally:
        _shared.leave_all_on = False
    log.info("🏁 انتهت المرحلة ب: مغادرة القنوات المسجّلة.")


async def watchdog(task, name, hang_monitor):
    started = time.time()
    while True:
        if task.done():
            return
        await asyncio.sleep(4)
        now = time.time()
        if os.path.exists(STOP_FILE):
            request_stop("تم إنشاء ملف الإيقاف اليدوي (logs/full_leave.stop)")
        for phone, until in list(flood_map.items()):
            wait = until - now
            if wait >= FLOOD_STOP_SEC:
                request_stop(f"FloodWait {int(wait)}s على الحساب {phone}")
        if hang_monitor and (now - _state["last_progress"]) > HANG_CHECK_SEC:
            request_stop(
                f"لا تقدّم منذ {int(now - _state['last_progress'])}s — "
                f"المرحلة ({name}) يبدو أنها معلّقة"
            )
        if (now - started) > PHASE_MAX_SEC:
            request_stop(
                f"مرحلة ({name}) تجاوزت الحد الزمني ({PHASE_MAX_SEC // 60} دقيقة)"
            )
        if stop_required.is_set():
            log.info("⏳ في انتظار الإيقاف اللطيف للمرحلة (حتى 30 ثانية)...")
            try:
                await asyncio.wait_for(task, timeout=30)
                return
            except asyncio.TimeoutError:
                log.error(f"🛑 إلغاء قسري لمرحلة ({name})")
                task.cancel()
                try:
                    await task
                except Exception:
                    pass
                return


async def run_phase(coro, name, hang_monitor):
    task = asyncio.create_task(coro)
    wd = asyncio.create_task(watchdog(task, name, hang_monitor))
    try:
        res = await asyncio.shield(task)
    except asyncio.CancelledError:
        res = None
    wd.cancel()
    try:
        await wd
    except asyncio.CancelledError:
        pass
    if task.cancelled():
        return None
    exc = task.exception()
    if exc:
        log.exception(f"خطأ غير متوقع في مرحلة ({name})")
        request_stop(f"خطأ في مرحلة ({name}): {exc}")
        return None
    return res


async def main():
    parser = argparse.ArgumentParser(description="سكربت المغادرة الشاملة الآمن")
    parser.add_argument(
        "--force", action="store_true",
        help="تجاوز فحص التجميع النشط (خطر — استخدمه بحذر)",
    )
    parser.add_argument(
        "--dry-run", action="store_true",
        help="استعراض الأعداد فقط دون أي مغادرة",
    )
    args = parser.parse_args()

    setup_logging()
    if not acquire_pid_lock():
        log.error("🛑 يوجد تشغيل آخر للسكربت (قفل PID نشط) — أُجهض.")
        return
    log.info("=" * 60)
    log.info("🧹 بدء سكربت المغادرة الشاملة")
    result_a = (0, 0)
    try:
        await _shared.init_db()
        active = await get_setting("is_active", "0")
        if active == "1" and not args.force:
            log.error(
                "🛑 التجميع يعمل الآن (is_active=1). "
                "أوقفه من لوحة التحكم أولاً، أو أضِف --force."
            )
            return

        try:
            if os.path.exists(STOP_FILE):
                os.remove(STOP_FILE)
        except OSError:
            pass

        accounts = await get_accounts(active_only=True)
        bots = await get_bots()
        reg = await get_channels()
        log.info(
            f"الحسابات النشطة: {len(accounts)} | البوتات: {len(bots)} | "
            f"القنوات المسجّلة: {len(reg)}"
        )

        if args.dry_run:
            for bot in bots:
                un = (bot.get("username") or "").strip("@")
                rows = await get_force_sub_channels(un) if un else []
                log.info(f"🔍 @{un}: {len(rows)} قناة اشتراك إجباري")
            log.info(f"🔍 القنوات المسجّلة الكلية: {len(reg)}")
            log.info("👌 وضع الاستعراض انتهى — لم تتم أي مغادرة.")
            return

        if not accounts:
            log.error("🛑 لا توجد حسابات نشطة — لا يمكن المغادرة.")
            return

        install_signal_handlers(asyncio.get_running_loop())

        if not stop_required.is_set():
            await warm_sessions(accounts)
            _state["phase"] = "force_sub"
            _state["last_progress"] = time.time()
            total_urls, total_done = await run_phase(
                phase_force_sub_all_bots(accounts),
                "الاشتراك الإجباري",
                True,
            ) or (0, 0)
            result_a = (total_urls, total_done)

        if not stop_required.is_set():
            _state["phase"] = "registered"
            _state["last_progress"] = time.time()
            await run_phase(
                phase_leave_registered(),
                "القنوات المسجّلة",
                False,
            )

        if stop_required.is_set():
            log.error(
                "🛑 انتهى السكربت بحالة توقّف. "
                "راجع السجل logs/full_leave.log لمعرفة السبب."
            )
        else:
            total_urls, total_done = result_a
            log.info(
                f"المرحلة أ (اشتراك إجباري): {total_done} عملية من "
                f"{total_urls} قناة"
            )
            try:
                async with _shared._pg_pool.acquire() as con:
                    fc = await con.fetchval(
                        "SELECT count(*) FROM force_sub_channels"
                    )
                    chc = await con.fetchval(
                        "SELECT count(*) FROM channels"
                    )
            except Exception as e:
                log.warning(f"تعذّر التحقق النهائي من القاعدة: {e}")
                fc, chc = -1, -1
            if (fc or 0) > 0 or (chc or 0) > 0:
                log.warning(
                    f"⚠️ اكتمل التشغيل لكن تبقّى: "
                    f"اشتراك إجباري={fc} | قنوات مسجّلة={chc}"
                )
            else:
                log.info("🎉 اكتملت المغادرة الشاملة بنجاح — القاعدة نظيفة.")
    except Exception:
        log.exception("خطأ عام غير متوقع في السكربت")
    finally:
        for phone in list(_shared.clients):
            try:
                await asyncio.wait_for(cleanup_client(phone), timeout=15)
            except Exception:
                pass
        release_pid_lock()
        log.info("🧹 خروج نظيف من سكربت المغادرة الشاملة.")
        log.info("=" * 60)


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