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 json
import logging
import re

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

log = logging.getLogger(__name__)

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

# Dekoder video_url wyniesiony do `_txxx.py` — ten sam silnik TXXX obsługuje też
# fullvideosporn/sextu, więc trzymamy jedną implementację. Alias zachowany dla
# czytelności modułu i ewentualnych importów.
_decode_video_url = _txxx.decode_video_url


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