goon/app/extractors/tubes/vjav.py

"""vjav.com — TXXX-network JAV tube. HLS stream extractor.

vjav to SPA na silniku TXXX (ktk_player + MSE/blob src). Stream NIE jest w HTML
strony (player renderuje się JS-owo). Zamiast tego AJAX endpoint zwraca zaciemniony
URL pliku:

  GET /api/videofile.php?video_id=<id>&lifetime=8640000
    -> [{"format":"_hq.mp4","video_url":"<base64+cyrylica>","is_default":1, ...}]

`video_url` to base64 w DWÓCH warstwach zaciemnienia: (1) wielkie łacińskie litery
podmienione na cyrylicę-homoglif (М->M, С->C, А->A, Е->E), (2) custom alfabet base64
gdzie `,`->`/`, `~`->`=`, `-`->`+`. Po odkręceniu obu + b64decode dostajemy get_file:

  # stare wideo (shared txxx pool) — URL absolutny:
  https://videotxxx.com/ext/get_file/9/<hash>/<floor>/<mid>/<mid>_hq.mp4/?d=..&br=..&ti=..
  # nowe wideo (własna infra vjav, server N) — URL RELATYWNY (prepend https://vjav.com):
  /get_file/3/<hash>/<floor>/<mid>/<mid>_hq.mp4/?d=..&br=..&ti=..

Dopisanie `&f=video.m3u8` -> 302 chain -> finalny HLS na *.ahcdn.com
(`key=..,end=..,limit=3` — time-bound, NIE IP-bound, `referer=none` na whiteliscie
-> portable, gra z residential IP telefonu bez sesji vjav).

Zwracamy get_file (`?f=video.m3u8`) jako type='m3u8' + `mobile_direct_ok` -> playback.py
owija w `/proxy/hls/<token>/play.m3u8` (passthrough manifestu ~1KB przez VPS, segmenty
direct z telefonu; patrz stream_proxy.proxy_hls_manifest). Media id w get_file (318577)
!= page id (5197) — dlatego MUSIMY przejść przez videofile.php. RE 2026-07-10.
"""
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__)

_BASE = "https://vjav.com"
_VIDEO_ID_RE = re.compile(r"/videos/(\d+)/")

# KVS video_url: base64 z cyrylica-homoglifami zamiast łacińskich liter (wielkie +
# część małych). Odkręcamy je z powrotem na łacinę przed b64decode.
_HOMOGLYPHS = str.maketrans(
    {
        "А": "A", "В": "B", "С": "C", "Е": "E", "Н": "H", "К": "K", "М": "M",
        "О": "O", "Р": "P", "Т": "T", "Х": "X", "У": "Y",
        "а": "a", "с": "c", "е": "e", "о": "o", "р": "p", "х": "x", "у": "y",
    }
)


def _decode_video_url(obfuscated: str) -> str | None:
    # (1) cyrylica-homoglif -> łacina, (2) custom alfabet base64 -> standardowy.
    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'. Trailer (`_tr`) filtrowany osobno."""
    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(page_url: str, *, timeout: float = 60.0) -> list[StreamSource] | None:
    m = _VIDEO_ID_RE.search(page_url)
    if not m:
        log.info("vjav: brak video id w %s", page_url)
        return None
    vid = m.group(1)
    api = f"{_BASE}/api/videofile.php?video_id={vid}&lifetime=8640000"
    try:
        res = browser_get(
            api,
            timeout=timeout,
            headers={"Referer": page_url, "X-Requested-With": "XMLHttpRequest"},
        )
        body = res.text if hasattr(res, "text") else res
        data = json.loads(body)
    except Exception as e:
        log.info("vjav: videofile.php fetch/parse fail %s: %s", 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` = trailer preview, NIE pełne wideo — pomijamy.
        if fmt.strip("_").replace(".mp4", "").lower() == "tr":
            continue
        vu = entry.get("video_url")
        if not vu:
            continue
        get_file = _decode_video_url(vu)
        if not get_file or "/get_file/" not in get_file:
            continue
        # Nowe wideo daje URL relatywny (/get_file/...) — prepend host vjav.
        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 (txxx.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("vjav: brak dekodowalnego video_url dla %s", page_url)
        return None
    return out