pythonany / renewer.py
nopoh22's picture
Update renewer.py
e0ee4cb verified
Raw History Blame Contribute Delete
16.5 kB
import requests
import json
import os
import logging
import time
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
STATE_FILE = "renewal_state.json"
HOSTS = ["www.pythonanywhere.com", "eu.pythonanywhere.com"]
MAX_RETRIES = 3
RETRY_DELAY = 5 # ุซูˆุงู†ูŠ ุจูŠู† ุงู„ู…ุญุงูˆู„ุงุช
REQUEST_TIMEOUT = 20
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# State helpers
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def load_state() -> dict:
if os.path.exists(STATE_FILE):
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: 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() -> list[dict]:
"""
ูŠู‚ุฑุฃ ุงู„ุญุณุงุจุงุช ู…ู† ู…ุชุบูŠุฑุงุช ุงู„ุจูŠุฆุฉ ุจุฃูŠ ู…ู† ุงู„ุตูŠุบุชูŠู†:
ุงู„ุตูŠุบุฉ 1 โ€” ู…ุฑู‚ู‘ู…ุฉ (ู…ูˆุตู‰ ุจู‡ุง):
ACCOUNTS_1 = username1:password1
ACCOUNTS_2 = username2:password2
ุงู„ุตูŠุบุฉ 2 โ€” ู…ุชุบูŠุฑ ูˆุงุญุฏ ู…ูุตูˆู„ ุจูุงุตู„ุฉ:
ACCOUNTS = username1:password1,username2:password2
"""
state = load_state()
accounts = []
seen = set()
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)
s = state.get(username, {})
accounts.append({
"username": username,
"password": password,
"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: str):
entry = entry.strip()
if ":" in entry:
username, password = entry.split(":", 1)
add_account(username, password)
# ุงู„ุตูŠุบุฉ 1
i = 1
while True:
entry = os.environ.get(f"ACCOUNTS_{i}", "").strip()
if not entry:
break
parse_entry(entry)
i += 1
# ุงู„ุตูŠุบุฉ 2
bulk = os.environ.get("ACCOUNTS", "").strip()
if bulk:
for entry in bulk.split(","):
parse_entry(entry)
if accounts:
logger.info(f"ุชู… ุชุญู…ูŠู„ {len(accounts)} ุญุณุงุจ: {[a['username'] for a in accounts]}")
else:
logger.warning("ู„ู… ูŠุชู… ุงู„ุนุซูˆุฑ ุนู„ู‰ ุญุณุงุจุงุช โ€” ุฃุถู ACCOUNTS_1 ุฃูˆ ACCOUNTS ููŠ Secrets")
return accounts
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)
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Session builder
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def _build_session() -> requests.Session:
session = requests.Session()
session.headers.update({
"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
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# 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"
action = form.get("action") or login_url
if not action.startswith("http"):
action = f"https://{host}{action}"
resp = session.post(
action,
data=form_data,
headers={
"Referer": login_url,
"Origin": f"https://{host}",
"Content-Type": "application/x-www-form-urlencoded",
},
timeout=REQUEST_TIMEOUT,
allow_redirects=True,
)
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
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Renew single account
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def renew_account(account: dict) -> tuple[bool, list[dict]]:
username = account["username"]
password = account.get("password", "").strip()
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
account["last_renewed"] = datetime.utcnow().isoformat()
account["next_renewal"] = (datetime.utcnow() + timedelta(days=10)).isoformat()
account["last_results"] = results
return success, results
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Batch renewal
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def run_renewals() -> list[dict] | None:
accounts = load_accounts()
if not accounts:
logger.info("ู„ุง ุชูˆุฌุฏ ุญุณุงุจุงุช.")
return None
now = datetime.utcnow()
renewed_count = 0
for account in accounts:
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)} ุญุณุงุจ.")
return accounts