#!/usr/bin/env python3
# check_sessions.py — فحص جلسات pyrogram دون أي تعديل أو حذف.
# يفحص كل ملف .session في مجلد sessions/ ويخبرك أيها يعمل وأيها ميت.
# آمن تماماً: للقراءة فقط — لا يحذف ولا يحظر ولا يلمس قاعدة البيانات.
#
# التشغيل من داخل مجلد المشروع:
#     python3 check_sessions.py
#
# خيارات:
#     python3 check_sessions.py --concurrent 10    عدد الفحوص المتزامنة (افتراضي 5)
#     python3 check_sessions.py --timeout 30       مهلة كل حساب بالثواني (افتراضي 25)

import argparse
import asyncio
import glob
import os
import sys
from datetime import datetime

try:
    import config
except Exception as e:
    print(f"تعذّر استيراد config.py — شغّل السكربت من داخل مجلد المشروع. ({e})")
    sys.exit(1)

from pyrogram import Client
from pyrogram.errors import (
    AuthKeyUnregistered, AuthKeyDuplicated, SessionExpired,
    UserDeactivated, UserDeactivatedBan, FloodWait,
)
try:
    from pyrogram.errors import SessionRevoked
except Exception:
    class SessionRevoked(Exception):
        pass

SESSIONS_DIR = getattr(config, "SESSIONS_DIR", "sessions")

# تصنيفات النتيجة
OK = "ok"            # الجلسة تعمل
DEAD = "dead"        # جلسة منتهية/ملغاة (auth key)
BANNED = "banned"    # الحساب محظور/معطّل من تيليجرام
FLOOD = "flood"      # FloodWait — الحساب يعمل لكن مقيّد مؤقتاً
ERROR = "error"      # خطأ آخر (شبكة/غير معروف)


def list_sessions() -> list:
    paths = glob.glob(os.path.join(SESSIONS_DIR, "*.session"))
    # اسم الجلسة = اسم الملف بدون .session (يطابق ما يستخدمه gather_engine)
    return sorted(os.path.splitext(os.path.basename(p))[0] for p in paths)


async def check_one(name: str, timeout: int) -> tuple:
    """يعيد (name, category, detail). لا يحذف ولا يعدّل أي شيء.
    يستخدم connect() بدل start() لتجنّب الدخول في تدفّق تسجيل الدخول
    التفاعلي (طلب رقم الهاتف) عند الجلسات الملغاة/المنتهية."""
    c = Client(
        os.path.join(SESSIONS_DIR, name),
        api_id=config.API_ID,
        api_hash=config.API_HASH,
        no_updates=True,
    )
    try:
        # connect() يفتح الاتصال فقط دون أي مطالبة بتسجيل دخول.
        await asyncio.wait_for(c.connect(), timeout=timeout)
        try:
            # get_me() هو ما يكشف صلاحية التفويض فعلياً.
            me = await asyncio.wait_for(c.get_me(), timeout=timeout)
            who = f"@{me.username}" if me.username else (me.first_name or "")
            return (name, OK, who)
        finally:
            try:
                await asyncio.wait_for(c.disconnect(), timeout=10)
            except Exception:
                pass
    except (AuthKeyUnregistered, AuthKeyDuplicated, SessionExpired,
            SessionRevoked) as e:
        await _safe_close(c)
        return (name, DEAD, type(e).__name__)
    except (UserDeactivated, UserDeactivatedBan) as e:
        await _safe_close(c)
        return (name, BANNED, type(e).__name__)
    except FloodWait as e:
        await _safe_close(c)
        return (name, FLOOD, f"{getattr(e, 'value', '?')}s")
    except asyncio.TimeoutError:
        await _safe_close(c)
        return (name, ERROR, "timeout")
    except Exception as e:
        await _safe_close(c)
        return (name, ERROR, f"{type(e).__name__}: {e}")


async def _safe_close(c):
    try:
        await asyncio.wait_for(c.disconnect(), timeout=10)
    except Exception:
        pass


async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--concurrent", type=int, default=5)
    ap.add_argument("--timeout", type=int, default=25)
    args = ap.parse_args()

    names = list_sessions()
    if not names:
        print(f"لا توجد ملفات جلسات في: {SESSIONS_DIR}/")
        return

    total = len(names)
    print(f"بدء فحص {total} جلسة (تزامن={args.concurrent}, مهلة={args.timeout}s)\n")

    sem = asyncio.Semaphore(args.concurrent)
    results = []
    done = 0

    async def worker(name):
        nonlocal done
        async with sem:
            res = await check_one(name, args.timeout)
        done += 1
        cat = res[1]
        mark = {OK: "✓", DEAD: "منتهية", BANNED: "✗ محظور",
                FLOOD: "⏳ flood", ERROR: "؟ خطأ"}.get(cat, "?")
        print(f"[{done}/{total}] {res[0]}  →  {mark}  {res[2]}")
        results.append(res)

    await asyncio.gather(*(worker(n) for n in names))

    # ملخص
    buckets = {OK: [], DEAD: [], BANNED: [], FLOOD: [], ERROR: []}
    for name, cat, _ in results:
        buckets[cat].append(name)

    print("\n" + "=" * 40)
    print("الملخص:")
    print(f"  ✓ تعمل:        {len(buckets[OK])}")
    print(f"  - منتهية:      {len(buckets[DEAD])}")
    print(f"  ✗ محظورة:      {len(buckets[BANNED])}")
    print(f"  ⏳ مقيّدة flood: {len(buckets[FLOOD])}")
    print(f"  ؟ أخطاء:       {len(buckets[ERROR])}")
    print(f"  الإجمالي:      {total}")
    print("=" * 40)

    # حفظ القوائم في ملف نصي للرجوع إليها
    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    out = f"session_report_{stamp}.txt"
    with open(out, "w", encoding="utf-8") as f:
        for cat, label in [(OK, "تعمل"), (DEAD, "منتهية"),
                            (BANNED, "محظورة"), (FLOOD, "مقيدة"),
                            (ERROR, "أخطاء")]:
            f.write(f"# {label} ({len(buckets[cat])})\n")
            for name in buckets[cat]:
                f.write(f"{name}\n")
            f.write("\n")
    print(f"\nحُفظ التقرير الكامل في: {out}")
    print("ملاحظة: هذا الفحص للقراءة فقط — لم يُحذف أو يُعدّل أي ملف أو حساب.")


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