df9f384d57
Type yggreborn : login form classique (identifier=EMAIL, password, csrf_token), ratio lu dans l'en-tête (Ratio : X.XX). Pas de up/down ni jetons (ligne envoyé/ reçu rendue conditionnelle sur up/down>0).
270 lines
11 KiB
Python
270 lines
11 KiB
Python
"""Ratio des trackers torrent privés (envoi / réception / ratio du compte).
|
|
|
|
Conçu pour PLUSIEURS trackers : chaque `TrackerSpec` a un `type` qui sélectionne le
|
|
fetcher (`_FETCHERS`). v1 implémente `unit3d_nuxt` — frontend Nuxt au-dessus de UNIT3D
|
|
(ex. c411). Le ratio n'y est PAS lisible avec une clé API : il faut une session (login).
|
|
Flux :
|
|
1. GET {base}/login -> pose le cookie `__csrf` + un meta `<csrf-token>`
|
|
2. POST {base}/api/auth/login -> header `csrf-token` + {username,password} ;
|
|
pose le cookie de session (~7 j)
|
|
3. GET {base}/api/users/{user} -> profil JSON avec ratio/uploaded/downloaded
|
|
La session (cookies) est réutilisée entre rendus ; on ne relogue que si elle a expiré
|
|
(le profil revient non authentifié). Le résultat est caché (`tracker_ttl_seconds`) car
|
|
le ratio bouge lentement et le login est coûteux.
|
|
|
|
Pour ajouter un tracker d'une autre techno : écrire `async def _fetch_xxx(spec)` qui
|
|
renvoie un `TrackerStat`, et l'enregistrer dans `_FETCHERS` sous sa clé de `type`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
import httpx
|
|
|
|
from config import TrackerSpec, config
|
|
|
|
_UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/126.0 Safari/537.36")
|
|
_CSRF_META = re.compile(r'name="csrf-token"\s+content="([^"]+)"')
|
|
_CSRF_INPUT = re.compile(r'name="csrf_token"[^>]*value="([^"]+)"')
|
|
_YGG_RATIO = re.compile(r"Ratio\s*:\s*([\d.]+)")
|
|
|
|
|
|
@dataclass
|
|
class TrackerStat:
|
|
key: str
|
|
label: str
|
|
ok: bool
|
|
error: str | None = None
|
|
ratio: float = 0.0
|
|
up_bytes: int = 0
|
|
down_bytes: int = 0
|
|
tokens: int | None = None # jetons/points de seed (None = le tracker n'en a pas)
|
|
tokens_label: str = "jetons"
|
|
|
|
@property
|
|
def tokens_h(self) -> str:
|
|
return f"{self.tokens:,}".replace(",", " ") if self.tokens is not None else ""
|
|
|
|
@staticmethod
|
|
def _human(b: int) -> str:
|
|
gb = b / 1e9
|
|
if gb >= 1000:
|
|
return f"{gb / 1000:.2f}".replace(".", ",") + " To"
|
|
return f"{gb:.0f} Go"
|
|
|
|
@property
|
|
def ratio_h(self) -> str:
|
|
return f"{self.ratio:.2f}".replace(".", ",")
|
|
|
|
@property
|
|
def up_h(self) -> str:
|
|
return self._human(self.up_bytes)
|
|
|
|
@property
|
|
def down_h(self) -> str:
|
|
return self._human(self.down_bytes)
|
|
|
|
|
|
# Sessions réutilisées (cookies httpx) et derniers résultats connus, par tracker.
|
|
_sessions: dict[str, httpx.Cookies] = {}
|
|
_cache: dict[str, dict] = {}
|
|
|
|
|
|
class _AuthError(Exception):
|
|
pass
|
|
|
|
|
|
async def _login_unit3d(client: httpx.AsyncClient, spec: TrackerSpec) -> None:
|
|
"""Récupère le token CSRF de /login puis poste les identifiants ; les cookies de
|
|
session restent dans le jar du `client`."""
|
|
page = await client.get(f"{spec.base_url}/login")
|
|
m = _CSRF_META.search(page.text)
|
|
if not m:
|
|
raise _AuthError("csrf introuvable")
|
|
resp = await client.post(
|
|
f"{spec.base_url}/api/auth/login",
|
|
headers={"csrf-token": m.group(1), "Content-Type": "application/json"},
|
|
json={"username": spec.username, "password": spec.password},
|
|
)
|
|
try:
|
|
data = resp.json()
|
|
except ValueError:
|
|
raise _AuthError(f"login HTTP {resp.status_code}")
|
|
if data.get("mfaRequired"):
|
|
raise _AuthError("2FA requise")
|
|
if not data.get("success"):
|
|
raise _AuthError(str(data.get("message") or "login refusé"))
|
|
|
|
|
|
async def _fetch_unit3d(spec: TrackerSpec) -> TrackerStat:
|
|
if not (spec.base_url and spec.username and spec.password):
|
|
return TrackerStat(spec.key, spec.label, ok=False, error="non configuré")
|
|
|
|
cookies = _sessions.get(spec.key) or httpx.Cookies()
|
|
async with httpx.AsyncClient(
|
|
timeout=20, follow_redirects=True, cookies=cookies,
|
|
headers={"User-Agent": _UA, "Accept": "application/json"},
|
|
) as client:
|
|
url = f"{spec.base_url}/api/users/{spec.username}"
|
|
resp = await client.get(url)
|
|
data = resp.json() if resp.status_code == 200 else {}
|
|
if "ratio" not in data: # session absente/expirée (route 404-masque sans auth)
|
|
await _login_unit3d(client, spec)
|
|
resp = await client.get(url)
|
|
if resp.status_code != 200 or "ratio" not in (data := resp.json()):
|
|
raise _AuthError(f"profil indisponible (HTTP {resp.status_code})")
|
|
_sessions[spec.key] = client.cookies
|
|
|
|
return TrackerStat(
|
|
spec.key, spec.label, ok=True,
|
|
ratio=float(data.get("ratio", 0) or 0),
|
|
up_bytes=int(data.get("uploaded", 0) or 0),
|
|
down_bytes=int(data.get("downloaded", 0) or 0),
|
|
)
|
|
|
|
|
|
async def _fetch_torr9(spec: TrackerSpec) -> TrackerStat:
|
|
"""torr9 : API Go dédiée (`api.torr9.net`). Login JWT puis `/api/v1/users/me`.
|
|
Le profil n'a pas de champ `ratio` ; on le calcule comme le frontend :
|
|
(total_uploaded_bytes + bonus_uploaded) / (total_downloaded_bytes + bonus_downloaded).
|
|
NB : le passkey du compte ne sert qu'aux flux RSS, pas à ce profil."""
|
|
if not (spec.base_url and spec.username and spec.password):
|
|
return TrackerStat(spec.key, spec.label, ok=False, error="non configuré")
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=20, follow_redirects=True, headers={"User-Agent": _UA},
|
|
) as client:
|
|
r = await client.post(
|
|
f"{spec.base_url}/api/v1/auth/login",
|
|
json={"username": spec.username, "password": spec.password},
|
|
)
|
|
try:
|
|
d = r.json()
|
|
except ValueError:
|
|
raise _AuthError(f"login HTTP {r.status_code}")
|
|
token = d.get("token")
|
|
if not token:
|
|
raise _AuthError(str(d.get("error") or d.get("message") or "login refusé"))
|
|
me = await client.get(
|
|
f"{spec.base_url}/api/v1/users/me",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
if me.status_code != 200:
|
|
raise _AuthError(f"users/me HTTP {me.status_code}")
|
|
u = me.json()
|
|
|
|
up = int(u.get("total_uploaded_bytes", 0) or 0) + int(u.get("bonus_uploaded", 0) or 0)
|
|
down = int(u.get("total_downloaded_bytes", 0) or 0) + int(u.get("bonus_downloaded", 0) or 0)
|
|
jetons = u.get("jeton_balance")
|
|
return TrackerStat(
|
|
spec.key, spec.label, ok=True,
|
|
ratio=(up / down if down else 0.0), up_bytes=up, down_bytes=down,
|
|
tokens=int(jetons) if jetons is not None else None,
|
|
)
|
|
|
|
|
|
async def _fetch_tr4ker(spec: TrackerSpec) -> TrackerStat:
|
|
"""tr4ker : SPA React, API même origine. Login cookie (JWT `TR4KER_session`) via
|
|
`POST /api/auth/login` body `{identifier,password}` -> `{ok:true}`, puis `GET /api/me`.
|
|
Ratio comme le site : (uploaded + bonus_upload) / (downloaded + bonus_download) ;
|
|
si rien n'a été téléchargé, on compte 1 Go (le site fait pareil via bonus_download,
|
|
ça évite un ratio infini). Système de jetons = champ `money` (libellé `token_currency_name`)."""
|
|
if not (spec.base_url and spec.username and spec.password):
|
|
return TrackerStat(spec.key, spec.label, ok=False, error="non configuré")
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=20, follow_redirects=True, headers={"User-Agent": _UA},
|
|
) as client:
|
|
r = await client.post(
|
|
f"{spec.base_url}/api/auth/login",
|
|
json={"identifier": spec.username, "password": spec.password},
|
|
)
|
|
try:
|
|
d = r.json()
|
|
except ValueError:
|
|
raise _AuthError(f"login HTTP {r.status_code}")
|
|
if not d.get("ok"):
|
|
raise _AuthError(str(d.get("error") or d.get("message") or "login refusé"))
|
|
me = await client.get(f"{spec.base_url}/api/me") # cookie de session porté par le client
|
|
if me.status_code != 200:
|
|
raise _AuthError(f"me HTTP {me.status_code}")
|
|
u = me.json()
|
|
|
|
up = int(u.get("uploaded", 0) or 0) + int(u.get("bonus_upload", 0) or 0)
|
|
down = int(u.get("downloaded", 0) or 0) + int(u.get("bonus_download", 0) or 0)
|
|
if down <= 0:
|
|
down = 10 ** 9 # rien téléchargé -> on compte 1 Go (pas de ratio infini)
|
|
money = u.get("money")
|
|
return TrackerStat(
|
|
spec.key, spec.label, ok=True,
|
|
ratio=up / down, up_bytes=up, down_bytes=down,
|
|
tokens=int(money) if money is not None else None,
|
|
tokens_label=str(u.get("token_currency_name") or "crédits"),
|
|
)
|
|
|
|
|
|
async def _fetch_yggreborn(spec: TrackerSpec) -> TrackerStat:
|
|
"""yggreborn (YggTorrent) : site rendu serveur (Flask), login form classique avec
|
|
CSRF. ⚠️ l'`identifier` est l'EMAIL (champ `type=email`), pas le pseudo. Pas de token
|
|
API ; le ratio est affiché tel quel dans l'en-tête (« Ratio : 7.63 ») -> on le lit
|
|
directement (pas de up/down exposé simplement ; l'user veut juste le ratio)."""
|
|
if not (spec.base_url and spec.username and spec.password):
|
|
return TrackerStat(spec.key, spec.label, ok=False, error="non configuré")
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=20, follow_redirects=True, headers={"User-Agent": _UA},
|
|
) as client:
|
|
page = await client.get(f"{spec.base_url}/login")
|
|
m = _CSRF_INPUT.search(page.text)
|
|
if not m:
|
|
raise _AuthError("csrf introuvable")
|
|
resp = await client.post(f"{spec.base_url}/login", data={
|
|
"csrf_token": m.group(1), "identifier": spec.username, "password": spec.password,
|
|
})
|
|
if "incorrect" in resp.text.lower():
|
|
raise _AuthError("identifiants refusés (identifier = email)")
|
|
rm = _YGG_RATIO.search(resp.text)
|
|
if not rm:
|
|
raise _AuthError("ratio introuvable")
|
|
|
|
return TrackerStat(spec.key, spec.label, ok=True, ratio=float(rm.group(1)))
|
|
|
|
|
|
_FETCHERS = {
|
|
"unit3d_nuxt": _fetch_unit3d, "torr9": _fetch_torr9,
|
|
"tr4ker": _fetch_tr4ker, "yggreborn": _fetch_yggreborn,
|
|
}
|
|
|
|
|
|
async def _fetch_one(spec: TrackerSpec) -> TrackerStat:
|
|
now = time.time()
|
|
cached = _cache.get(spec.key)
|
|
if cached and cached["value"].ok and now - cached["ts"] < config.tracker_ttl_seconds:
|
|
return cached["value"]
|
|
|
|
fetcher = _FETCHERS.get(spec.type)
|
|
if fetcher is None:
|
|
return TrackerStat(spec.key, spec.label, ok=False, error=f"type {spec.type} inconnu")
|
|
|
|
try:
|
|
value = await fetcher(spec)
|
|
except (_AuthError, httpx.HTTPError, ValueError) as exc:
|
|
_sessions.pop(spec.key, None) # session douteuse -> repartir propre au prochain coup
|
|
if cached and cached["value"].ok:
|
|
return cached["value"] # repli sur la dernière valeur connue
|
|
return TrackerStat(spec.key, spec.label, ok=False, error=str(exc) or "injoignable")
|
|
|
|
_cache[spec.key] = {"value": value, "ts": now}
|
|
return value
|
|
|
|
|
|
async def fetch_all() -> list[TrackerStat]:
|
|
specs = config.trackers
|
|
if not specs:
|
|
return []
|
|
return list(await asyncio.gather(*(_fetch_one(s) for s in specs)))
|