Update renewer.py

renewer.py

@@ -2,6 +2,7 @@ import requests
 import json
 import os
 import logging
+import time
 from datetime import datetime, timedelta
 from bs4 import BeautifulSoup
 
@@ -13,58 +14,77 @@ logger = logging.getLogger(__name__)
 
 STATE_FILE = "renewal_state.json"
 
+HOSTS = ["www.pythonanywhere.com", "eu.pythonanywhere.com"]
 
-def load_state():
+MAX_RETRIES   = 3
+RETRY_DELAY   = 5   # ثواني بين المحاولات
+REQUEST_TIMEOUT = 20
+
+
+# ─────────────────────────────────────────
+# State helpers
+# ─────────────────────────────────────────
+
+def load_state() -> dict:
     if os.path.exists(STATE_FILE):
-        with open(STATE_FILE, "r") as f:
-            return json.load(f)
+        try:
+            with open(STATE_FILE, "r") as f:
+                return json.load(f)
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning(f"تعذّر قراءة ملف الحالة: {e} — سيبدأ من الصفر")
     return {}
 
 
-def save_state(state):
-    with open(STATE_FILE, "w") as f:
-        json.dump(state, f, indent=2)
+def save_state(state: dict):
+    try:
+        with open(STATE_FILE, "w") as f:
+            json.dump(state, f, indent=2, ensure_ascii=False)
+    except OSError as e:
+        logger.error(f"تعذّر حفظ الحالة: {e}")
+
 
+# ─────────────────────────────────────────
+# Account loading
+# ─────────────────────────────────────────
 
-def load_accounts():
+def load_accounts() -> list[dict]:
     """
-    يقرأ الحسابات من Environment Variables بأي من الصيغ التالية:
+    يقرأ الحسابات من متغيرات البيئة بأي من الصيغتين:
 
-    الصيغة 1 — متغيرات مرقّمة (الموصى بها):
+    الصيغة 1 — مرقّمة (موصى بها):
         ACCOUNTS_1 = username1:password1
         ACCOUNTS_2 = username2:password2
-        ACCOUNTS_3 = username3:password3
 
-    الصيغة 2 — متغير واحد يحتوي الكل (مفصولة بفاصلة):
+    الصيغة 2 — متغير واحد مفصول بفاصلة:
         ACCOUNTS = username1:password1,username2:password2
-
-    يمكن الجمع بين الصيغتين — سيتم دمج الكل بدون تكرار.
     """
     state    = load_state()
     accounts = []
     seen     = set()
 
-    def add_account(username, password):
+    def add_account(username: str, password: str):
+        username = username.strip()
+        password = password.strip()
         if not username or not password or username in seen:
             return
         seen.add(username)
-        user_state = state.get(username, {})
+        s = state.get(username, {})
         accounts.append({
             "username":     username,
             "password":     password,
-            "host":         user_state.get("host", "www.pythonanywhere.com"),
-            "last_renewed": user_state.get("last_renewed"),
-            "next_renewal": user_state.get("next_renewal"),
-            "last_results": user_state.get("last_results", []),
+            "host":         s.get("host", "www.pythonanywhere.com"),
+            "last_renewed": s.get("last_renewed"),
+            "next_renewal": s.get("next_renewal"),
+            "last_results": s.get("last_results", []),
         })
 
-    def parse_entry(entry):
+    def parse_entry(entry: str):
         entry = entry.strip()
         if ":" in entry:
             username, password = entry.split(":", 1)
-            add_account(username.strip(), password.strip())
+            add_account(username, password)
 
-    # ── الصيغة 1: ACCOUNTS_1, ACCOUNTS_2, ACCOUNTS_3 ... ──
+    # الصيغة 1
     i = 1
     while True:
         entry = os.environ.get(f"ACCOUNTS_{i}", "").strip()
@@ -73,10 +93,10 @@ def load_accounts():
         parse_entry(entry)
         i += 1
 
-    # ── الصيغة 2: ACCOUNTS = user1:pass1,user2:pass2 ──
-    accounts_env = os.environ.get("ACCOUNTS", "").strip()
-    if accounts_env:
-        for entry in accounts_env.split(","):
+    # الصيغة 2
+    bulk = os.environ.get("ACCOUNTS", "").strip()
+    if bulk:
+        for entry in bulk.split(","):
             parse_entry(entry)
 
     if accounts:
@@ -87,147 +107,278 @@ def load_accounts():
     return accounts
 
 
-def save_accounts(accounts):
-    """يحفظ حالة التجديد فقط — كلمات المرور تبقى في Secrets"""
-    state = {}
-    for a in accounts:
-        state[a["username"]] = {
+def save_accounts(accounts: list[dict]):
+    """يحفظ الحالة فقط — كلمات المرور لا تُكتب على القرص."""
+    state = {
+        a["username"]: {
             "host":         a.get("host", "www.pythonanywhere.com"),
             "last_renewed": a.get("last_renewed"),
             "next_renewal": a.get("next_renewal"),
             "last_results": a.get("last_results", []),
         }
+        for a in accounts
+    }
     save_state(state)
 
 
-def get_host(username, token):
-    for host in ["www.pythonanywhere.com", "eu.pythonanywhere.com"]:
-        try:
-            r = requests.get(
-                f"https://{host}/api/v0/user/{username}/cpu/",
-                headers={"Authorization": f"Token {token}"},
-                timeout=10
-            )
-            if r.status_code == 200:
-                return host
-        except Exception:
-            continue
-    return None
-
+# ─────────────────────────────────────────
+# Session builder
+# ─────────────────────────────────────────
 
-def extend_webapp(username, password, host="www.pythonanywhere.com"):
+def _build_session() -> requests.Session:
     session = requests.Session()
-    results = []
-
     session.headers.update({
-        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
-        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+        "User-Agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/124.0.0.0 Safari/537.36"
+        ),
+        "Accept":          "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
         "Accept-Language": "en-US,en;q=0.9",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Connection":      "keep-alive",
     })
+    return session
 
-    try:
-        # الخطوة 1: صفحة الدخول
-        login_url = f"https://{host}/login/"
-        login_page = session.get(login_url, timeout=15)
-        logger.info(f"[{username}] صفحة الدخول: {login_page.status_code}")
-
-        soup = BeautifulSoup(login_page.text, "html.parser")
-        form = soup.find("form")
-        if not form:
-            logger.error(f"[{username}] لم يتم العثور على form")
-            return []
-
-        form_data = {}
-        for inp in form.find_all("input"):
-            name = inp.get("name")
-            if name:
-                form_data[name] = inp.get("value", "")
-
-        form_data["auth-username"] = username
-        form_data["auth-password"] = password
-        if "login_view-current_step" not in form_data:
+
+# ─────────────────────────────────────────
+# Login
+# ─────────────────────────────────────────
+
+def _login(session: requests.Session, username: str, password: str, host: str) -> bool:
+    """
+    يسجّل الدخول بطريقة موثوقة:
+    1. يجلب الـ CSRF من صفحة اللوجن
+    2. يرسل البيانات
+    3. يتحقق من وجود sessionid في الكوكيز
+    4. يتحقق من وجود "logout" في نص الصفحة
+    """
+    login_url = f"https://{host}/login/"
+
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            login_page = session.get(login_url, timeout=REQUEST_TIMEOUT)
+            login_page.raise_for_status()
+
+            soup = BeautifulSoup(login_page.text, "html.parser")
+            form = soup.find("form")
+            if not form:
+                logger.error(f"[{username}] لم يتم العثور على form في صفحة الدخول")
+                return False
+
+            form_data: dict = {}
+            for inp in form.find_all("input"):
+                name = inp.get("name")
+                if name:
+                    form_data[name] = inp.get("value", "")
+
+            form_data["auth-username"]          = username
+            form_data["auth-password"]          = password
             form_data["login_view-current_step"] = "auth"
 
-        # الخطوة 2: تسجيل الدخول
-        form_action = form.get("action") or login_url
-        if not form_action.startswith("http"):
-            form_action = f"https://{host}{form_action}"
-
-        login_resp = session.post(
-            form_action,
-            data=form_data,
-            headers={
-                "Referer": login_url,
-                "Content-Type": "application/x-www-form-urlencoded",
-                "Origin": f"https://{host}",
-            },
-            timeout=15,
-            allow_redirects=True
-        )
-        logger.info(f"[{username}] نتيجة الدخول: {login_resp.status_code} | {login_resp.url}")
-
-        if "sessionid" not in session.cookies.get_dict():
-            logger.error(f"[{username}] فشل تسجيل الدخول — لا يوجد sessionid")
-            return []
-
-        logger.info(f"[{username}] ✅ تم تسجيل الدخول")
-
-        # ��لخطوة 3: صفحة الويب آب
-        webapps_url = f"https://{host}/user/{username}/webapps/"
-        page = session.get(webapps_url, timeout=15)
-
-        soup = BeautifulSoup(page.text, "html.parser")
-        csrf_input = soup.find("input", {"name": "csrfmiddlewaretoken"})
-        csrf = csrf_input["value"] if csrf_input else ""
-        if not csrf:
-            logger.error(f"[{username}] لم يتم العثور على CSRF")
-            return []
-
-        # الخطوة 4: اكتشاف النطاقات
-        domains = []
-        for link in soup.find_all("a", href=True):
-            href = link["href"]
-            if f"/user/{username}/webapps/" in href:
-                part = href.split(f"/user/{username}/webapps/")[-1].strip("/")
-                if part and "." in part and part not in domains:
-                    domains.append(part)
-
-        if not domains:
-            domains = [f"{username}.pythonanywhere.com"]
-
-        # الخطوة 5: التمديد
-        for domain in domains:
-            extend_url = f"https://{host}/user/{username}/webapps/{domain}/extend"
+            action = form.get("action") or login_url
+            if not action.startswith("http"):
+                action = f"https://{host}{action}"
+
             resp = session.post(
-                extend_url,
-                data={"csrfmiddlewaretoken": csrf},
+                action,
+                data=form_data,
                 headers={
-                    "Referer": webapps_url,
+                    "Referer":      login_url,
+                    "Origin":       f"https://{host}",
                     "Content-Type": "application/x-www-form-urlencoded",
-                    "Origin": f"https://{host}",
                 },
-                timeout=15
+                timeout=REQUEST_TIMEOUT,
+                allow_redirects=True,
             )
-            logger.info(f"[{username}] Extend {domain}: HTTP {resp.status_code}")
-            status = "✅ تم التمديد" if resp.status_code in (200, 302) else f"❌ فشل ({resp.status_code})"
-            results.append({"domain": domain, "status": status})
 
-    except Exception as e:
-        logger.error(f"[{username}] خطأ: {e}")
+            cookies = session.cookies.get_dict()
+
+            # التحقق المزدوج: كوكي sessionid + نص يحتوي logout
+            has_session = "sessionid" in cookies
+            has_logout  = (
+                "log out" in resp.text.lower()
+                or "logout" in resp.text.lower()
+            )
+
+            if has_session and has_logout:
+                logger.info(f"[{username}] ✅ تم تسجيل الدخول على {host}")
+                return True
+
+            if not has_session:
+                logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — لا يوجد sessionid")
+            elif not has_logout:
+                logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — 'logout' غير موجود في الصفحة")
+
+        except requests.Timeout:
+            logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — انتهت المهلة")
+        except requests.RequestException as e:
+            logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — خطأ شبكة: {e}")
+
+        if attempt < MAX_RETRIES:
+            time.sleep(RETRY_DELAY)
+
+    logger.error(f"[{username}] ❌ فشل تسجيل الدخول بعد {MAX_RETRIES} محاولات")
+    return False
+
+
+# ─────────────────────────────────────────
+# Host discovery
+# ─────────────────────────────────────────
+
+def _detect_host(session: requests.Session, username: str) -> str | None:
+    """يكتشف الخادم الصحيح (www أو eu) بعد تسجيل الدخول."""
+    for host in HOSTS:
+        try:
+            r = session.get(
+                f"https://{host}/user/{username}/webapps/",
+                timeout=REQUEST_TIMEOUT,
+                allow_redirects=True,
+            )
+            if r.status_code == 200 and username.lower() in r.url.lower():
+                logger.info(f"[{username}] الخادم: {host}")
+                return host
+        except requests.RequestException:
+            continue
+    return None
+
+
+# ─────────────────────────────────────────
+# Webapp extension
+# ─────────────────────────────────────────
+
+def _get_domains(soup: BeautifulSoup, username: str, host: str) -> list[str]:
+    """
+    يكتشف النطاقات بطريقتين:
+    1. روابط <a> تحتوي /webapps/ (الأكثر دقة)
+    2. نماذج <form> تحتوي /extend في action (احتياطي)
+    """
+    domains: list[str] = []
+
+    # الطريقة 1 — روابط الـ webapps
+    for tag in soup.find_all("a", href=True):
+        href: str = tag["href"]
+        marker = f"/user/{username}/webapps/"
+        if marker in href:
+            part = href.split(marker)[-1].strip("/")
+            if part and "." in part and part not in domains:
+                domains.append(part)
+
+    # الطريقة 2 — نماذج extend (احتياطي)
+    if not domains:
+        for form in soup.find_all("form", action=True):
+            action: str = form.get("action", "")
+            if "/extend" in action.lower():
+                domain = action.split("/webapps/")[-1].split("/extend")[0].strip("/")
+                if domain and domain not in domains:
+                    domains.append(domain)
+
+    # الافتراضي
+    if not domains:
+        domains = [f"{username}.pythonanywhere.com"]
+        logger.warning(f"[{username}] لم يتم اكتشاف أي نطاق — سيستخدم الافتراضي: {domains[0]}")
+
+    return domains
+
+
+def extend_webapp(username: str, password: str, host: str = "www.pythonanywhere.com") -> list[dict]:
+    session = _build_session()
+    results: list[dict] = []
+
+    # ── تسجيل الدخول ──
+    if not _login(session, username, password, host):
+        return []
+
+    # ── صفحة الـ webapps ──
+    webapps_url = f"https://{host}/user/{username}/webapps/"
+    try:
+        page = session.get(webapps_url, timeout=REQUEST_TIMEOUT)
+        page.raise_for_status()
+    except requests.RequestException as e:
+        logger.error(f"[{username}] تعذّر الوصول لصفحة webapps: {e}")
+        return []
+
+    soup = BeautifulSoup(page.text, "html.parser")
+
+    # ── CSRF ──
+    csrf_input = soup.find("input", {"name": "csrfmiddlewaretoken"})
+    csrf = csrf_input["value"] if csrf_input else session.cookies.get("csrftoken", "")
+    if not csrf:
+        logger.error(f"[{username}] لم يتم العثور على CSRF token")
+        return []
+
+    # ── النطاقات ──
+    domains = _get_domains(soup, username, host)
+    logger.info(f"[{username}] النطاقات المكتشفة: {domains}")
+
+    # ── التمديد ──
+    for domain in domains:
+        extend_url = f"https://{host}/user/{username}/webapps/{domain}/extend"
+        success    = False
+
+        for attempt in range(1, MAX_RETRIES + 1):
+            try:
+                resp = session.post(
+                    extend_url,
+                    data={"csrfmiddlewaretoken": csrf},
+                    headers={
+                        "Referer":      webapps_url,
+                        "Origin":       f"https://{host}",
+                        "Content-Type": "application/x-www-form-urlencoded",
+                    },
+                    timeout=REQUEST_TIMEOUT,
+                    allow_redirects=True,
+                )
+                logger.info(f"[{username}] Extend {domain}: HTTP {resp.status_code} → {resp.url}")
+
+                if resp.status_code in (200, 302):
+                    success = True
+                    break
+                else:
+                    logger.warning(
+                        f"[{username}] محاولة {attempt}/{MAX_RETRIES} — "
+                        f"extend {domain} فشل ({resp.status_code})"
+                    )
+
+            except requests.Timeout:
+                logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — انتهت المهلة أثناء extend {domain}")
+            except requests.RequestException as e:
+                logger.warning(f"[{username}] محاولة {attempt}/{MAX_RETRIES} — خطأ: {e}")
+
+            if attempt < MAX_RETRIES:
+                time.sleep(RETRY_DELAY)
+
+        status = "✅ تم التمديد" if success else f"❌ فشل التمديد بعد {MAX_RETRIES} محاولات"
+        results.append({"domain": domain, "status": status})
 
     return results
 
 
-def renew_account(account):
+# ─────────────────────────────────────────
+# Renew single account
+# ─────────────────────────────────────────
+
+def renew_account(account: dict) -> tuple[bool, list[dict]]:
     username = account["username"]
     password = account.get("password", "").strip()
-    host     = account.get("host", "www.pythonanywhere.com")
+    host     = account.get("host") or "www.pythonanywhere.com"
 
     if not password:
         logger.error(f"[{username}] لا توجد كلمة مرور")
         return False, []
 
+    # محاولة الاتصال — اكتشاف الخادم إذا لزم
     results = extend_webapp(username, password, host)
+
+    # إذا فشل الخادم المحفوظ، جرّب الآخر
+    if not results:
+        other_host = next((h for h in HOSTS if h != host), None)
+        if other_host:
+            logger.info(f"[{username}] جرب الخادم الآخر: {other_host}")
+            results = extend_webapp(username, password, other_host)
+            if results:
+                host = other_host
+
     success = any("✅" in r.get("status", "") for r in results) if results else False
 
     account["host"]         = host
@@ -238,31 +389,40 @@ def renew_account(account):
     return success, results
 
 
-def run_renewals():
+# ─────────────────────────────────────────
+# Batch renewal
+# ─────────────────────────────────────────
+
+def run_renewals() -> list[dict] | None:
     accounts = load_accounts()
     if not accounts:
-        logger.info("لا توجد حسابات — أضف ACCOUNTS_1 أو ACCOUNTS في Secrets.")
-        return
+        logger.info("لا توجد حسابات.")
+        return None
 
-    now = datetime.utcnow()
+    now           = datetime.utcnow()
     renewed_count = 0
 
     for account in accounts:
-        username         = account["username"]
-        next_renewal_str = account.get("next_renewal")
-
-        if next_renewal_str:
-            next_renewal = datetime.fromisoformat(next_renewal_str)
-            if now < next_renewal:
-                remaining = (next_renewal - now).days
-                logger.info(f"[{username}] التجديد بعد {remaining} يوم")
-                continue
-
-        logger.info(f"[{username}] جاري التمديد...")
-        success, results = renew_account(account)
+        username     = account["username"]
+        next_renewal = account.get("next_renewal")
+
+        if next_renewal:
+            try:
+                nxt = datetime.fromisoformat(next_renewal)
+                if now < nxt:
+                    remaining = (nxt - now).days
+                    logger.info(f"[{username}] التجديد القادم بعد {remaining} يوم — لا حاجة الآن")
+                    continue
+            except ValueError:
+                logger.warning(f"[{username}] next_renewal غير صالح — سيُجدَّد الآن")
+
+        logger.info(f"[{username}] جاري التجديد...")
+        success, _ = renew_account(account)
         if success:
             renewed_count += 1
+        else:
+            logger.error(f"[{username}] ❌ فشل التجديد")
 
     save_accounts(accounts)
-    logger.info(f"اكتمل. تم تمديد {renewed_count}/{len(accounts)} حساب.")
+    logger.info(f"اكتمل: تم تجديد {renewed_count}/{len(accounts)} حساب.")
     return accounts