goon/app/extractors/tubes/_txxx.py

"""Wspólne elementy sieci TXXX (vjav, tubepornclassic, fullvideosporn/sextu, …).

Silnik TXXX oddaje URL pliku przez `GET /api/videofile.php?video_id=<id>&lifetime=N`
w polu `video_url`, zaciemnionym DWIEMA warstwami:
  1. wielkie/małe litery łacińskie podmienione na cyrylickie homoglify (М→M, С→C, …),
  2. custom alfabet base64: `,`→`/`, `~`→`=`, `-`→`+`.

Po odkręceniu obu i b64decode dostajemy `/get_file/...` (czasem absolutny URL).
Wyniesione tutaj, żeby vjav i fullvideosporn nie trzymały dwóch kopii dekodera.

**Wspólny ekstraktor HLS** (`extract_hls`, vjav + tubepornclassic). Cała rodzina oddaje
strumień tak samo: po dekodzie `/get_file/...`, dopisanie `f=video.m3u8` daje 302 na HLS
w *.ahcdn.com (klucz time-bound, NIE IP-bound, `referer=none` na whiteliście). Zwracamy
get_file jako `m3u8` + `mobile_direct_ok`, a `playback.py` owija go w `/proxy/hls`
(przez VPS idzie ~1 KB manifestu, segmenty telefon ciągnie sam; `/proxy/hls` zna
wyzwanie 429 tej sieci). Historia i pułapki w docstringu `tubes/vjav.py`.

Dlaczego HLS, a nie mp4 z tego samego get_file: mp4 ma w URL-u CDN `speed=...`, czyli
limit ~1,2 MB/s po pierwszych 2 MB (pomiar 2026-09-07 na tubepornclassic), a segmenty
HLS lecą ~3,6 MB/s. Przy plikach HD 2,8 Mb/s mp4 ledwo nadąża, HLS ma zapas.
"""
from __future__ import annotations

import base64
import json
import logging
import re

from app.extractors import browser_get
from app.extractors._models import StreamSource

log = logging.getLogger(__name__)

# Cyrylickie homoglify → łacina. Bez tego b64decode dostaje śmieci.
HOMOGLYPHS = str.maketrans(
    {
        "А": "A", "В": "B", "С": "C", "Е": "E", "Н": "H", "К": "K", "М": "M",
        "О": "O", "Р": "P", "Т": "T", "Х": "X", "У": "Y",
        "а": "a", "с": "c", "е": "e", "о": "o", "р": "p", "х": "x", "у": "y",
    }
)

#: Wyzwanie anty-botowe sieci TXXX: 429 i `document.cookie = 'NAZWA=WARTOSC; max-age=...'`
#: w ciele odpowiedzi. Wartość stoi wprost, wystarczy ją odczytać i powtórzyć zapytanie,
#: bez wykonywania JS. Dotyczy ruchu z serwerowni, łącze domowe przechodzi bez tego.
CHALLENGE_COOKIE_RE = re.compile(r"document\.cookie\s*=\s*'([^=]+)=([^;']+)")


def decode_video_url(obfuscated: str) -> str | None:
    """Zaciemniony `video_url` → ścieżka/URL `get_file`. None gdy dekod padnie."""
    if not obfuscated:
        return None
    clean = (
        obfuscated.translate(HOMOGLYPHS)
        .replace(",", "/")
        .replace("~", "=")
        .replace("-", "+")
    )
    clean += "=" * (-len(clean) % 4)  # padding do wielokrotności 4
    try:
        return base64.b64decode(clean).decode("utf-8", "ignore")
    except Exception:
        return None


def quality_label(fmt: str | None) -> str | None:
    """`_hq.mp4` -> 'HD', `_sd.mp4` -> 'SD', `.mp4` (bez sufiksu) -> None."""
    if not fmt:
        return None
    tok = fmt.strip("_").replace(".mp4", "").lower()
    return {"hq": "HD", "sd": "SD", "lq": "LOW", "hd": "HD"}.get(tok, tok.upper() or None)


def extract_hls(
    base: str, page_url: str, vid: str, *, tag: str, timeout: float = 60.0
) -> list[StreamSource] | None:
    """videofile.php → get_file (`?f=video.m3u8`) dla tuba TXXX pod `base`."""
    api = f"{base}/api/videofile.php?video_id={vid}&lifetime=8640000"
    hdr = {"Referer": page_url, "X-Requested-With": "XMLHttpRequest"}
    try:
        res = browser_get(api, timeout=timeout, headers=hdr)
        body = res.text if hasattr(res, "text") else res
        if (ck := CHALLENGE_COOKIE_RE.search(body)) is not None:
            hdr = dict(hdr, Cookie=f"{ck.group(1)}={ck.group(2)}")
            res = browser_get(api, timeout=timeout, headers=hdr)
            body = res.text if hasattr(res, "text") else res
        data = json.loads(body)
    except Exception as e:
        log.info("%s: videofile.php fetch/parse fail %s: %s", tag, api, e)
        return None
    if not isinstance(data, list):
        return None

    seen: set[str] = set()
    out: list[StreamSource] = []
    for entry in data:
        if not isinstance(entry, dict):
            continue
        fmt = (entry.get("format") or "").strip()
        # `_tr.mp4` = zwiastun, NIE pełne wideo.
        if fmt.strip("_").replace(".mp4", "").lower() == "tr":
            continue
        get_file = decode_video_url(entry.get("video_url") or "")
        if not get_file or "/get_file/" not in get_file:
            continue
        # Nowe wideo daje URL relatywny (/get_file/...), prepend host tuba.
        if get_file.startswith("/"):
            get_file = base + get_file
        m3u8 = get_file + ("&" if "?" in get_file else "?") + "f=video.m3u8"
        if m3u8 in seen:
            continue
        seen.add(m3u8)
        out.append(
            StreamSource(
                link=m3u8,
                type="m3u8",
                quality=quality_label(fmt),
                referer=page_url,
                # Finalny HLS (*.ahcdn.com) time-bound + referer=none whitelisted ->
                # portable. mobile_direct_ok -> playback.py owija w /proxy/hls passthrough.
                raw={"mobile_direct_ok": True},
            )
        )
    if not out:
        log.info("%s: brak dekodowalnego video_url dla %s", tag, page_url)
        return None
    return out