"""GET /scenes, lista i szczegóły scen z bazy kanonicznej."""
from __future__ import annotations

import logging
import re
import hashlib
import uuid
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel
from sqlalchemy import Date, distinct, exists, false, func, literal_column, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.auth import require_api_key

from app.api.schemas import (
    ExternalRefOut,
    PerformerOut,
    PlaybackSourceOut,
    SceneListOut,
    SceneOut,
    StudioOut,
    TagOut,
)
from app.db import get_session
from app.api.device import LEGACY_DEVICE, get_device_id
from app.models.favorite_scene import FavoriteScene
from app.models.performer import Performer, PerformerExternalRef
from app.models.play_progress import ScenePlayProgress
from app.models.playback_source import PlaybackSource
from app import verticals
from app.models.scene import Scene, SceneExternalRef, ScenePerformer, SceneTag
from app.models.source import Source, SourceKind
from app.models.studio import Studio
from app.models.tag import Tag

log = logging.getLogger(__name__)

router = APIRouter(prefix="/scenes", tags=["scenes"], dependencies=[Depends(require_api_key)])


_VALID_SORTS = {
    "created_at",
    "release_date",
    "title",
    "studio",
    "random",
    "duration_desc",
    "duration_asc",
}


def _effective_date_desc():
    """`release_date`, a gdy go brak, dzien trafienia sceny do katalogu, malejaco.

    MUSI byc identyczne z wyrazeniem w `ix_scenes_effective_date_desc`, inaczej planner
    indeksu nie uzyje i wroci do pelnego skanu `scenes`. `created_at::date` NIE nadaje
    sie do indeksu, bo zalezy od strefy czasowej sesji (stable, nie immutable), stad
    jawne AT TIME ZONE 'UTC'.
    """
    return func.coalesce(
        Scene.release_date,
        func.cast(func.timezone("UTC", Scene.created_at), Date),
    ).desc()

# TTL-cache dla count'u scen-z-żywym-playback (default lista bez filtra). Full-scan
# 1.69M scen + EXISTS ~950ms; liczba zmienia się wolno i jest przybliżona (header
# paginacji), więc 10-min cache w pamięci procesu API jest akceptowalny trade-off.
_DEFAULT_COUNT_CACHE: dict = {"ts": 0.0, "val": 0}
_DEFAULT_COUNT_TTL = 600.0

def _default_scene_count(session: Session) -> int:
    import time as _time
    now = _time.monotonic()
    if _DEFAULT_COUNT_CACHE["val"] and (now - _DEFAULT_COUNT_CACHE["ts"]) < _DEFAULT_COUNT_TTL:
        return _DEFAULT_COUNT_CACHE["val"]
    count_query = select(func.count()).select_from(
        select(Scene.id).where(
            exists(
                select(1).where(
                    PlaybackSource.scene_id == Scene.id,
                    PlaybackSource.dead_at.is_(None),
                )
            ),
            # Domyślny feed wyklucza wertykały (JAV, animacja) → licznik też.
            ~_vertical_source_exists(),
        ).subquery()
    )
    total = session.execute(count_query).scalar_one()
    _DEFAULT_COUNT_CACHE["ts"] = now
    _DEFAULT_COUNT_CACHE["val"] = total
    return total


# Wertykały (JAV, animacja) mają własne sekcje w apce i NIE wchodzą do głównego feedu.
# Definicje i uzasadnienie: `app/verticals.py`. Tu zostaje tylko budowa predykatu.
# `JAV_ORIGINS` re-eksportowane, bo importują je inne moduły i skrypty.
JAV_ORIGINS = verticals.JAV_ORIGINS


def _vertical_source_exists(vertical: str | None = None):
    """EXISTS żywego źródła z originem wertykału. `None` → DOWOLNY wertykał, czyli
    predykat, którym główny feed odcina je wszystkie naraz."""
    return exists(
        select(1).where(
            PlaybackSource.scene_id == Scene.id,
            PlaybackSource.dead_at.is_(None),
            PlaybackSource.origin.in_(verticals.origins_for(vertical)),
        )
    )


def _jav_source_exists():
    """Zachowane pod starą nazwą, używane przez `/scenes/{id}/similar` i testy."""
    return _vertical_source_exists("jav")


# Blacklisty (performer/studio/tag) są zwykle PUSTE (self-hosted, single-user). Mimo to
# 3 NOT EXISTS klauzule doklejały się do KAŻDEJ filtrowanej listy scen i były ewaluowane
# per-row, przy filtrze typu duży-tag/has_playback planer chodzi po ~176k scen, więc te
# puste-zawsze klauzule kosztowały ~3.4s (mega-tag „anal": 6.7s→3.3s po pominięciu).
# Cache'ujemy emptiness (TTL 5 min); gdy ktoś doda blacklist-wpis, w ciągu 5 min klauzule
# wracają. Patrz reference_scenes_list_perf / task #22.
# Cache per device_id (blacklisty są teraz device-scoped, bug 2026-06-08).
_BLACKLIST_EMPTY_CACHE: dict[str, tuple[float, bool]] = {}
_BLACKLIST_EMPTY_TTL = 300.0


def _blacklists_empty(session: Session, device_id: str) -> bool:
    """True gdy WSZYSTKIE 3 blacklisty TEGO device puste → pomiń NOT EXISTS klauzule."""
    import time as _time
    from app.models.blacklist import (
        BlacklistedPerformer,
        BlacklistedStudio,
        BlacklistedTag,
    )
    now = _time.monotonic()
    cached = _BLACKLIST_EMPTY_CACHE.get(device_id)
    if cached and (now - cached[0]) < _BLACKLIST_EMPTY_TTL:
        return cached[1]
    has_any = session.execute(
        select(
            exists(select(1).select_from(BlacklistedPerformer).where(BlacklistedPerformer.device_id == device_id))
            | exists(select(1).select_from(BlacklistedStudio).where(BlacklistedStudio.device_id == device_id))
            | exists(select(1).select_from(BlacklistedTag).where(BlacklistedTag.device_id == device_id))
        )
    ).scalar_one()
    _BLACKLIST_EMPTY_CACHE[device_id] = (now, not has_any)
    return not has_any


def _split_csv(raw: str | None) -> list[str]:
    if not raw:
        return []
    return [s.strip() for s in raw.split(",") if s.strip()]


# ---- Współdzielone klauzule widoczności sceny ----------------------------------
# Definicja "co user widzi na liście" żyje TU i jest reużywana przez list_scenes ORAZ
# licznik "+N nowych" w app/api/favorites.py. Wcześniej favorites miał ręczną kopię tych
# klauzul (dryf: zmiana filtra listy nie trafiała do licznika → "+6 a nic nowego").


def live_playback_exists():
    """EXISTS: scena ma choć jeden żywy playback_source."""
    return exists(
        select(1).where(
            PlaybackSource.scene_id == Scene.id,
            PlaybackSource.dead_at.is_(None),
        )
    )


def blacklist_clauses(session: Session, device_id: str) -> list:
    """NOT-EXISTS klauzule blacklist device (performer/studio/tag). [] gdy wszystkie puste."""
    if _blacklists_empty(session, device_id):
        return []
    from app.models.blacklist import (
        BlacklistedPerformer,
        BlacklistedStudio,
        BlacklistedTag,
    )

    return [
        ~exists(
            select(1)
            .select_from(ScenePerformer)
            .join(
                BlacklistedPerformer,
                (BlacklistedPerformer.performer_id == ScenePerformer.performer_id)
                & (BlacklistedPerformer.device_id == device_id),
            )
            .where(ScenePerformer.scene_id == Scene.id)
        ),
        ~Scene.studio_id.in_(
            select(BlacklistedStudio.studio_id).where(BlacklistedStudio.device_id == device_id)
        ),
        ~exists(
            select(1)
            .select_from(SceneTag)
            .join(
                BlacklistedTag,
                (BlacklistedTag.tag_id == SceneTag.tag_id)
                & (BlacklistedTag.device_id == device_id),
            )
            .where(SceneTag.scene_id == Scene.id)
        ),
    ]


def stub_exclusion_clause():
    """Odsiew stub-scen: tube-only bez release_date AND bez canonical (TPDB/StashDB) AND
    bez performera. NOT stub gdy ma release_date OR canonical OR performera."""
    canonical_exists = exists(
        select(1)
        .select_from(SceneExternalRef)
        .join(Source, Source.id == SceneExternalRef.source_id)
        .where(SceneExternalRef.scene_id == Scene.id)
        .where(Source.kind.in_([SourceKind.tpdb, SourceKind.stashdb]))
    )
    has_performer = exists(select(1).where(ScenePerformer.scene_id == Scene.id))
    return Scene.release_date.is_not(None) | canonical_exists | has_performer


@router.get("", response_model=SceneListOut)
def list_scenes(
    session: Annotated[Session, Depends(get_session)],
    device_id: Annotated[str, Depends(get_device_id)],
    q: str | None = Query(default=None, description="Wyszukiwanie po title_normalized (trgm)"),
    studio_slug: str | None = Query(default=None, description="DEPRECATED — użyj studio_slugs"),
    studio_slugs: str | None = Query(
        default=None, description="Comma-separated studio slugs (OR)"
    ),
    tags: str | None = Query(
        default=None,
        description="Comma-separated tag slugs (AND — scena musi mieć wszystkie wybrane tagi)",
    ),
    performer_ids: str | None = Query(
        default=None,
        description="Comma-separated performer UUIDs (AND — scena musi mieć wszystkich wybranych performerów)",
    ),
    has_playback: bool | None = Query(
        default=None, description="True: tylko sceny z ≥1 playback_source"
    ),
    min_duration_sec: int | None = Query(default=None, ge=0),
    max_duration_sec: int | None = Query(default=None, ge=0),
    released_within_days: int | None = Query(
        default=None, ge=1,
        description="Tylko sceny released w ostatnich N dniach",
    ),
    min_quality_p: int | None = Query(
        default=None, ge=1,
        description=(
            "Minimum quality (pixele wysokości — 2160 = 4K, 1080 = FullHD). Filtruje "
            "po PlaybackSource.quality (string typu '720p' / '1080p Full HD')."
        ),
    ),
    origin: str | None = Query(
        default=None,
        description=(
            "Filtruj po playback origin (np. 'tube:hqpornercom'). Substring match — "
            "'hqporner' złapie tube:hqpornercom. Diagnostyka per-hoster."
        ),
    ),
    include_stubs: bool = Query(
        default=False,
        description=(
            "False (default): ukrywa sceny-szkielety bez release_date, < 10min, "
            "z jedynym playback z hqporner (~7-min Brazzers trailer clipy zalewają katalog)."
        ),
    ),
    vertical: str | None = Query(
        default=None,
        description=(
            "Wertykał do pokazania: 'jav' albo 'animation'. Brak → główny feed, "
            "czyli WSZYSTKIE wertykały wykluczone. Zastępuje starsze `jav=true`, "
            "które zostaje dla instalacji sprzed sierpnia 2026."
        ),
    ),
    jav: bool = Query(
        default=False,
        description=(
            "False (default): ukrywa sceny JAV (osobny vertical). True: TYLKO sceny JAV. "
            "JAV to osobna sekcja w apce, nie zalewa głównego feedu (origin javflix/javguru/vjav/supjav)."
        ),
    ),
    sort: str = Query(
        default="created_at",
        description="created_at|release_date|title|studio|random|duration_desc|duration_asc",
    ),
    seed: str | None = Query(
        default=None,
        max_length=64,
        description="Dla sort=random: ten sam seed daje ten sam porzadek na kolejnych stronach.",
    ),
    page: int = Query(default=1, ge=1),
    per_page: int = Query(default=50, ge=1, le=200),
) -> SceneListOut:
    if sort not in _VALID_SORTS:
        raise HTTPException(status_code=400, detail=f"sort must be one of {sorted(_VALID_SORTS)}")

    base = select(Scene)

    if q:
        base = base.where(Scene.title_normalized.ilike(f"%{q.lower()}%"))

    # Cap rozmiarów filtrów. Bez tego pojedynczy request z setkami studio_slugs +
    # dziesiątkami tagów (każdy tag = osobny correlated EXISTS) + ILIKE budował zapytanie,
    # które OOM-killer ubijał → PG crash-recovery = ~1s globalnej przerwy (GOON-1M,
    # 2026-06-26: 194 studios + 23 tagi). Realny UI nigdy nie wysyła tylu. 422 zamiast
    # wywalania bazy. Limity hojne (>> normalne użycie), ale ograniczają złożoność query.
    _MAX_STUDIOS, _MAX_TAGS, _MAX_PERFORMERS = 50, 15, 15

    studio_slug_list = _split_csv(studio_slugs)
    if studio_slug:
        studio_slug_list.append(studio_slug)
    if len(studio_slug_list) > _MAX_STUDIOS:
        raise HTTPException(status_code=422, detail=f"too many studio filters (max {_MAX_STUDIOS})")
    if studio_slug_list:
        base = base.where(
            Scene.studio_id.in_(
                select(Studio.id).where(Studio.slug.in_(studio_slug_list))
            )
        )

    tag_slug_list = _split_csv(tags)
    if len(tag_slug_list) > _MAX_TAGS:
        raise HTTPException(status_code=422, detail=f"too many tag filters (max {_MAX_TAGS})")
    # AND między tagami: scena musi mieć WSZYSTKIE zaznaczone tagi. Każdy slug → osobny
    # exists(), zaznaczanie kolejnych filtrów zawęża wyniki, jak intuicja użytkownika.
    #
    # PERF (2026-06-07): resolvujemy slug→tag_id w aplikacji i filtrujemy po LITERALNYM
    # tag_id (NIE JOIN po Tag.slug). Z literałem planner zna kardynalność tagu ze
    # statystyk (MCV) → dla popularnych tagów (blowjob ~273k scen) wybiera index-walk po
    # ix_scenes_created_at_desc zamiast materializować wszystkie scene_tags. Slug-JOIN
    # ukrywał tag_id przed plannerem → używał średniej (8.4M/11541≈726) → zły plan
    # (4-12s). Z literałem: ~20ms. Zob. też _build... light mode.
    if tag_slug_list:
        id_by_slug = dict(
            session.execute(
                select(Tag.slug, Tag.id).where(Tag.slug.in_(tag_slug_list))
            ).all()
        )
        for slug in tag_slug_list:
            tag_id = id_by_slug.get(slug)
            if tag_id is None:
                base = base.where(false())  # nieznany slug → brak wyników
                break
            base = base.where(
                exists(
                    select(1)
                    .select_from(SceneTag)
                    .where(SceneTag.scene_id == Scene.id, SceneTag.tag_id == tag_id)
                )
            )

    perf_id_strings = _split_csv(performer_ids)
    if len(perf_id_strings) > _MAX_PERFORMERS:
        raise HTTPException(status_code=422, detail=f"too many performer filters (max {_MAX_PERFORMERS})")
    if perf_id_strings:
        try:
            perf_ids = [uuid.UUID(s) for s in perf_id_strings]
        except ValueError as e:
            raise HTTPException(status_code=400, detail=f"invalid performer UUID: {e}") from e
        # AND między performerami (analogicznie do tagów).
        for pid in perf_ids:
            base = base.where(
                exists(
                    select(1)
                    .select_from(ScenePerformer)
                    .where(
                        ScenePerformer.scene_id == Scene.id,
                        ScenePerformer.performer_id == pid,
                    )
                )
            )

    if has_playback is True:
        base = base.where(live_playback_exists())
    elif has_playback is False:
        base = base.where(~live_playback_exists())

    if origin:
        # Substring match na origin, 'hqporner' złapie 'tube:hqpornercom'.
        base = base.where(
            exists(
                select(1).where(
                    PlaybackSource.scene_id == Scene.id,
                    PlaybackSource.dead_at.is_(None),
                    PlaybackSource.origin.ilike(f"%{origin}%"),
                )
            )
        )

    # Brama wertykałów: domyślnie WSZYSTKIE wykluczone z głównego feedu, `?vertical=X`
    # → TYLKO ten wertykał. ALE gdy user browse'uje KONKRETNY origin (Sites → javflix/
    # vjav/hentaigasm → SiteScenes), bramy nie stosujemy, bo jawnie chce ten origin
    # (regresja "No videos" na stronach JAV, bug-report 2026-07-19). Brama dotyczy tylko
    # głównego feedu (origin=None).
    wanted = vertical if vertical in verticals.VERTICALS else ("jav" if jav else None)
    if wanted:
        base = base.where(_vertical_source_exists(wanted))
    elif origin is None:
        base = base.where(~_vertical_source_exists())

    # Blacklisty device (performer/studio/tag), globalne wykluczenia, współdzielone z
    # licznikiem +N ulubionych. Puste blacklisty → [] (typowy single-user, zero kosztu).
    for _bl_clause in blacklist_clauses(session, device_id):
        base = base.where(_bl_clause)

    # `min_duration_sec=0` znaczy "bez progu", a NIE `duration_sec >= 0`. To rozroznienie
    # jest tu istotne, bo w SQL NULL nie przechodzi ZADNEGO porownania: przy `>= 0` scena
    # o NIEZNANEJ dlugosci wypadala tak samo jak dziesieciosekundowy smiec. Dla serwisow,
    # ktore dlugosci nie podaja w ogole (hentaimama, javflix, supjav, hentaigasm,
    # hentaiplay, xanimeporn maja 100% NULL), lista wychodzila PUSTA, mimo ze wejscie w
    # konkretny serwis jest jawna prosba usera o jego zawartosc. Prog > 0 dalej odcina
    # NULL-e swiadomie, bo tam filtrujemy smieci w glownym feedzie.
    if min_duration_sec:
        base = base.where(Scene.duration_sec >= min_duration_sec)
    if max_duration_sec is not None:
        base = base.where(Scene.duration_sec <= max_duration_sec)

    if released_within_days is not None:
        from datetime import date, timedelta
        cutoff = date.today() - timedelta(days=released_within_days)
        base = base.where(Scene.release_date >= cutoff)

    if min_quality_p is not None:
        # Skalarny filtr po zdenormalizowanej kolumnie. Wcześniejsze podejścia (EXISTS
        # na playback_sources, potem JOIN z DISTINCT) dawały 32-52 s na produkcji, bo
        # planer i tak przechodził `scenes` po `created_at` scena po scenie. Pomiary
        # i uzasadnienie: migracja 0029_scenes_max_quality.
        #
        # NULL (93,5% katalogu) NIE przechodzi, i tak ma być: filtr obiecuje jakość
        # potwierdzoną, a nie zgadywaną. UI mówi o tym wprost pod przełącznikiem.
        base = base.where(Scene.max_quality_p >= min_quality_p)

    if not include_stubs:
        # Stub scene heuristic (współdzielona z licznikiem +N ulubionych): tube-only scena
        # bez release_date AND bez canonical (TPDB/StashDB) AND bez performera. Continuous
        # worker dodaje ScenePerformer (search-by-name), więc per-performer wynik nie jest
        # stubem. Filtruje anonymous tube-only sceny z newUrl/categories ingestu.
        base = base.where(stub_exclusion_clause())

    _is_pure_default = (
        not include_stubs and not q and not studio_slug_list and not tag_slug_list
        and not perf_id_strings and origin is None and has_playback is None
        and not min_duration_sec
        and max_duration_sec is None and released_within_days is None
        and min_quality_p is None and not jav and vertical is None
    )
    # Count strategy:
    #  - PURE default: cached pełny licznik katalogu (TTL 10 min).
    #  - FILTROWANE: NIE liczymy dokładnie. Bounded-count nad EXISTS-filtrami był
    #    dominującym kosztem (~4s na has_playback / min_duration / duży tag) i plan
    #    był NIESTABILNY (literal LIMIT + count-nad-PK pomogły w części przypadków,
    #    ale planer i tak czasem skanuje cały zbiór zamiast urwać). Mobile paginuje
    #    po `has_more` (per_page+1 fetch), NIE po `total`, `total` to tylko licznik
    #    "N+" w UI. Wyprowadzamy go z has_more PO fetchu (patrz niżej): dolna granica
    #    + flaga "jest więcej". Eliminuje cały koszt count z każdej filtrowanej listy.
    total_capped = False
    total: int | None = _default_scene_count(session) if _is_pure_default else None

    # Sort: KAŻDY tryb kończy się `Scene.id.desc()`, unikalny PK jako ostateczny
    # tie-break. Bez tego przy OFFSET-paginacji sceny z IDENTYCZNYM created_at (masowy
    # ingest = ten sam timestamp co do sekundy) i release_date mają NIEustalony porządek,
    # więc kolejne strony nakładają się na siebie → feed "powtarza się" przy scrollu
    # (report 304c9258). created_at desc dalej pierwszy, więc index-walk zostaje.
    if sort == "release_date":
        # Data premiery LUB, gdy jej nie ma, data trafienia do katalogu. Czyste
        # `release_date DESC NULLS LAST` chowalo swiezosc, bo tuby daty premiery zwykle
        # NIE podaja: xhamster 99,2% scen bez daty, xnxx 97,7%, porntrex 95,8%, hqporner
        # 93,5%, sxyprn 85,7%. Wszystkie NULL-e ladowaly za datowanymi, wiec 82 z 91 scen
        # dodanych w hqporner przez tydzien siedzialo za 3,4 tys. starszych pozycji i
        # user widzial "brak nowych scen" (zgloszenie 1df50de3), mimo ze ingest dzialal.
        # Sceny z prawdziwa data (kanon z TPDB/StashDB) sortuja sie jak dotad.
        # Wyrazenie musi byc IMMUTABLE, zeby weszlo do indeksu: `created_at::date` zalezy
        # od strefy sesji, stad jawne AT TIME ZONE 'UTC'. Indeks:
        # ix_scenes_effective_date_desc (migracja 0033).
        ordered = base.order_by(
            _effective_date_desc(), Scene.created_at.desc(), Scene.id.desc()
        )
    elif sort == "title":
        ordered = base.order_by(
            Scene.title_normalized.asc(), Scene.created_at.desc(), Scene.id.desc()
        )
    elif sort == "studio":
        # Sceny bez studio na końcu; w obrębie studio, najświeższe pierwsze.
        ordered = (
            base.outerjoin(Studio, Studio.id == Scene.studio_id)
            .order_by(
                Studio.name_normalized.asc().nullslast(),
                _effective_date_desc(),
                Scene.created_at.desc(),
                Scene.id.desc(),
            )
        )
    elif sort in ("duration_desc", "duration_asc"):
        # Najdluzsze / najkrotsze (zgloszenie ae6a045b). Sceny bez dlugosci odpadaja,
        # dzieki czemu jeden indeks (duration_sec, created_at, id) WHERE duration_sec
        # IS NOT NULL obsluguje oba kierunki skanem w przod i w tyl (migracja 0035).
        dur = Scene.duration_sec.desc() if sort == "duration_desc" else Scene.duration_sec.asc()
        created = Scene.created_at.desc() if sort == "duration_desc" else Scene.created_at.asc()
        ident = Scene.id.desc() if sort == "duration_desc" else Scene.id.asc()
        # Gorny limit 10 h: tuby raportuja czasem absurdy (rekord 130 h), ktore
        # zasmiecalyby szczyt listy „najdluzsze".
        ordered = (
            base.where(Scene.duration_sec.isnot(None), Scene.duration_sec <= 36000)
            .order_by(dur, created, ident)
        )
    elif sort == "random":
        # Losowo bez sortowania milionow wierszy: id to losowy UUID, wiec porzadek po id
        # od losowego punktu startu JEST losowy i idzie po indeksie PK. Punkt startu
        # wynika z seeda (md5 -> uuid), wiec paginacja jest stabilna, a nowy seed
        # (pull-to-refresh w apce) daje nowe rozdanie. Za koncem zakresu lista sie konczy.
        start = uuid.UUID(hashlib.md5((seed or uuid.uuid4().hex).encode("utf-8")).hexdigest())
        # Okno kandydatow po PK z LIMIT, a filtry (EXISTS/NOT EXISTS) dopiero na nim.
        # Bez okna planer robil Merge Anti Join z indeksem playback_sources po scene_id
        # i przechodzil 2,3 mln wierszy (7,6 s), bo ORDER BY id pasowal do obu indeksow.
        # Z oknem 2000 id to nested loop po kilkudziesieciu wierszach (pomiar 2026-09-14).
        # Okno rosnie ze strona, zeby glebsze strony mialy z czego wybierac. Sekcje
        # wertykalne (JAV, Anime) to ~2% katalogu, wiec dostaja 4x szersze okno; z 2000 id
        # JAV wyciagal 27 scen i druga strona bylaby pusta. Szersze okno dla wszystkich
        # nie wchodzi w gre: z filtrem tagu 6000 id to 2,7 s (probe scene_tags per scena).
        base_window = 8000 if (vertical or jav) else 2000
        window = max(base_window, page * per_page * 4)
        cand = (
            select(Scene.id).where(Scene.id > start).order_by(Scene.id.asc()).limit(window).subquery()
        )
        ordered = base.where(Scene.id.in_(select(cand.c.id))).order_by(Scene.id.asc())
    else:  # created_at
        ordered = base.order_by(
            Scene.created_at.desc(), Scene.release_date.desc().nullslast(), Scene.id.desc()
        )

    # Fetch per_page+1, obecność (per_page+1)-szego wiersza = jest kolejna strona.
    # To źródło prawdy dla paginacji (mobile getNextPageParam), niezależne od bounded
    # `total`. Nadmiarowy wiersz odcinamy przed serializacją.
    # LIMIT/OFFSET literalne (NIE bound-param), patrz wyżej: sparametryzowany LIMIT
    # psuje early-termination i przy filtrach EXISTS planer robi gather-all+sort (sekundy)
    # zamiast limit-aware index-walk po `ix_scenes_created_at_desc`. page/per_page to
    # walidowane inty (Query ge=1, le=200), więc literal_column jest bezpieczne.
    _off = (page - 1) * per_page
    rows = (
        session.execute(
            ordered.offset(literal_column(str(_off))).limit(literal_column(str(per_page + 1)))
        )
        .scalars()
        .all()
    )
    has_more = len(rows) > per_page
    rows = rows[:per_page]

    # Filtrowane listy: total = dolna granica z dotychczas-widzianych wierszy, a
    # total_capped=has_more daje UI "N+" (jest kolejna strona). Bez osobnego count query.
    if total is None:
        total = (page - 1) * per_page + len(rows)
        total_capped = has_more

    items = _build_scenes_out_batch(
        session, list(rows), light=True, device_id=device_id, prefer_origin=origin
    )

    return SceneListOut(
        items=items,
        total=total,
        page=page,
        per_page=per_page,
        has_more=has_more,
        total_capped=total_capped,
    )


@router.get("/{scene_id}", response_model=SceneOut)
def get_scene(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
    device_id: Annotated[str, Depends(get_device_id)],
) -> SceneOut:
    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")
    return _build_scene_out(session, scene, device_id=device_id)


@router.get("/{scene_id}/similar", response_model=SceneListOut)
def similar_scenes(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
    device_id: Annotated[str, Depends(get_device_id)],
    limit: int = Query(default=12, ge=1, le=30),
) -> SceneListOut:
    """Sceny podobne do danej, ranking po WSPÓLNYCH performerach (mocny sygnał) i
    tagach (słabszy). Zwraca tylko grywalne sceny (żywy playback), zgodne z bramą JAV
    (JAV↔JAV, nie-JAV↔nie-JAV), z pominięciem blacklist/stubów/samej siebie.

    Wydajność: performerzy są selektywni, więc pula kandydatów z nich jest mała (cap
    _CAND_CAP). Tagi liczymy TYLKO w obrębie tej puli (tanio). Fill z tagów (bounded
    _TAG_FILL_CAP, HAVING >=2 wspólne) odpala się gdy scena ma mało/zero performerów
    (anonimowe tube sceny), inaczej szeroki tag typu 'blowjob' skanowałby setki tysięcy
    wierszy. Cały ranking i sort robimy w Pythonie nad OGRANICZONYMI zbiorami id."""
    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")

    _empty = SceneListOut(
        items=[], total=0, page=1, per_page=limit, has_more=False, total_capped=False
    )

    perf_ids = session.execute(
        select(ScenePerformer.performer_id).where(ScenePerformer.scene_id == scene_id)
    ).scalars().all()
    # Odsiej ŚMIECIOWYCH performerów: kategorie/studia podszywające się pod performera
    # (over-attribution z luźnego tube-searcha, np. "Big Tits", "Brazzers", "Ball Licking").
    # Sygnał: slug pokrywa się z tagiem I performer NIE ma kanonicznego refa (tpdb/stashdb).
    # Prawdziwi performerzy z tym samym tagiem (tuby tagują sceny nazwiskiem) mają kanoniczny
    # ref → zostają. Bez tego similar klastrował się po "Ball Licking" zamiast po treści.
    if perf_ids:
        junk = set(
            session.execute(
                select(Performer.id).where(
                    Performer.id.in_(perf_ids),
                    exists(select(1).select_from(Tag).where(Tag.slug == Performer.slug)),
                    ~exists(
                        select(1)
                        .select_from(PerformerExternalRef)
                        .join(Source, Source.id == PerformerExternalRef.source_id)
                        .where(
                            PerformerExternalRef.performer_id == Performer.id,
                            Source.kind.in_([SourceKind.tpdb, SourceKind.stashdb]),
                        )
                    ),
                )
            ).scalars().all()
        )
        if junk:
            perf_ids = [p for p in perf_ids if p not in junk]
    tag_ids = session.execute(
        select(SceneTag.tag_id).where(SceneTag.scene_id == scene_id)
    ).scalars().all()
    if not perf_ids and not tag_ids:
        return _empty

    _PERF_W, _TAG_W = 3, 1
    _CAND_CAP = 1500      # pula kandydatów z performerów (bezpiecznik na płodnego performera)
    _TAG_FILL_CAP = 400   # pula fill z tagów gdy brak performerów

    perf_shared: dict[uuid.UUID, int] = {}
    if perf_ids:
        rows = session.execute(
            select(ScenePerformer.scene_id, func.count().label("c"))
            .where(
                ScenePerformer.performer_id.in_(perf_ids),
                ScenePerformer.scene_id != scene_id,
            )
            .group_by(ScenePerformer.scene_id)
            .order_by(func.count().desc())
            .limit(_CAND_CAP)
        ).all()
        perf_shared = {r[0]: r[1] for r in rows}

    cand_ids: set[uuid.UUID] = set(perf_shared)
    tag_shared: dict[uuid.UUID, int] = {}
    if tag_ids and cand_ids:
        rows = session.execute(
            select(SceneTag.scene_id, func.count(distinct(SceneTag.tag_id)))
            .where(SceneTag.scene_id.in_(cand_ids), SceneTag.tag_id.in_(tag_ids))
            .group_by(SceneTag.scene_id)
        ).all()
        tag_shared = {r[0]: r[1] for r in rows}
    # Fill z tagów gdy performerów mało/brak (anonimowe tube sceny), inaczej pomijamy,
    # bo szeroki tag skanuje setki tysięcy scen.
    if tag_ids and len(cand_ids) < limit * 4:
        rows = session.execute(
            select(SceneTag.scene_id, func.count(distinct(SceneTag.tag_id)).label("c"))
            .where(SceneTag.tag_id.in_(tag_ids), SceneTag.scene_id != scene_id)
            .group_by(SceneTag.scene_id)
            .having(func.count(distinct(SceneTag.tag_id)) >= 2)
            .order_by(func.count(distinct(SceneTag.tag_id)).desc())
            .limit(_TAG_FILL_CAP)
        ).all()
        for sid, c in rows:
            tag_shared.setdefault(sid, c)
            cand_ids.add(sid)

    if not cand_ids:
        return _empty

    # Filtr widoczności (ta sama definicja co lista): grywalne + ten sam wertykał +
    # blacklist + nie-stub + nie-self.
    # Podobne muszą zostać w tym samym wertykale co scena-cel: dla sceny animowanej
    # nie proponujemy zachodnich i odwrotnie (inne namespace'y, zero dedupu).
    target_vertical: str | None = None
    for _name in verticals.VERTICALS:
        if session.execute(
            select(_vertical_source_exists(_name)).select_from(Scene).where(Scene.id == scene_id)
        ).scalar():
            target_vertical = _name
            break
    filt = select(Scene.id).where(
        Scene.id.in_(cand_ids),
        Scene.id != scene_id,
        live_playback_exists(),
        stub_exclusion_clause(),
        _vertical_source_exists(target_vertical) if target_vertical
        else ~_vertical_source_exists(),
    )
    for _bl in blacklist_clauses(session, device_id):
        filt = filt.where(_bl)
    valid_ids = set(session.execute(filt).scalars().all())
    if not valid_ids:
        return _empty

    def _score(sid: uuid.UUID) -> int:
        return _PERF_W * perf_shared.get(sid, 0) + _TAG_W * tag_shared.get(sid, 0)

    top_ids = sorted(valid_ids, key=_score, reverse=True)[: limit * 2]
    scenes = session.execute(select(Scene).where(Scene.id.in_(top_ids))).scalars().all()
    scenes.sort(key=lambda sc: (_score(sc.id), sc.created_at), reverse=True)
    scenes = scenes[:limit]

    items = _build_scenes_out_batch(session, list(scenes), light=True, device_id=device_id)
    return SceneListOut(
        items=items, total=len(items), page=1, per_page=limit,
        has_more=False, total_capped=False,
    )


_SXYPRN_POST_RE = re.compile(r"sxyprn\.com/post/([0-9a-f]{6,40})", re.IGNORECASE)


def _sxyprn_thumb_url(page_url: str | None) -> str | None:
    """Dla źródła sxyprn zwraca STABILNY endpoint on-demand resolvera
    (`/proxy/sxyprn-thumb/<id>`) zamiast martwego trafficdeposit URL, token żyje ~1h,
    więc poster resolvujemy przy serwowaniu (bug 2026-06-10)."""
    if not page_url:
        return None
    m = _SXYPRN_POST_RE.search(page_url)
    return f"/proxy/sxyprn-thumb/{m.group(1)}" if m else None


_YOUPORN_WATCH_RE = re.compile(r"youporn\.com/watch/(\d+)", re.IGNORECASE)


def _youporn_thumb_url(page_url: str | None, thumb: str | None) -> str | None:
    """Dla youporn z PODPISANYM posterem (imgproxy `?hash=...&validto=`) zwraca stabilny
    on-demand endpoint `/proxy/youporn-thumb/<video_id>`, zapisany podpis wygasa i CDN
    oddaje 410 (bug-report bebf8c92; 8666 z 62930 źródeł). Statycznych `.../original/N.jpg`
    NIE ruszamy (działają bezterminowo, zero dodatkowych fetchy)."""
    if not page_url or not thumb or "hash=" not in thumb:
        return None
    m = _YOUPORN_WATCH_RE.search(page_url)
    return f"/proxy/youporn-thumb/{m.group(1)}" if m else None


def _is_rotting_thumb(url: str) -> bool:
    """sxyprn/trafficdeposit miniaturki są czasowo podpisane i rotują (asset 404 po
    ~tygodniach, nie odświeżalne server-side; bug 2026-06-10). De-prioritize je w wyborze
    slim-thumbnaila, używamy tylko gdy scena nie ma żadnej innej miniaturki."""
    if "trafficdeposit.com" in url:
        return True
    # imgs.xiaoshenke.net (fullporner) było tu wpisane 2026-08-02, bo host oddawał 200
    # z ~153-bajtową atrapą zamiast obrazu (zgłoszenia a01dc891 / 2fbba822 / b437d5f7).
    # To był stan PRZEJŚCIOWY: pomiar 2026-08-11 na 300 losowych URL-ach z bazy dał
    # 298 prawdziwych JPEG-ów (5-25 kB), tak samo z VPS jak i z łącza domowego.
    # Blokada zdjęta, bo szkodziła bardziej niż pomagała: w SceneDetail (linia ~879)
    # rotting-thumb jest ZEROWANY, więc sceny mające tylko fullporner traciły miniaturę
    # całkowicie („Jakiś problem z miniaturkami z fullporner"). Dotyczyło 53 tys. źródeł.
    # Gdyby atrapy wróciły, filtruj po ROZMIARZE odpowiedzi w image-proxy, nie po hoście.
    # tn.sextu.com (fullvideosporn), CDN odrzuca ruch z IP datacenter (429), więc nasze
    # image-proxy NIGDY tej miniatury nie pobierze (znany problem tego źródła). Traktujemy
    # jak martwą, żeby scena po merge sięgnęła po działającą z innego źródła.
    if "tn.sextu.com" in url:
        return True
    # fastpic.org (galaxyporn i inne re-uploadery), hosting zrywa połączenie zamiast
    # oddać obraz (RemoteProtocolError), czyli martwy. Zerowanie ma tu dodatkowy zysk:
    # scena bez miniatury odpala w apce auto-enrich, który pobiera świeżą ze strony
    # tube'a. Zgłoszenie c2f1dd41; dotyczy 240 z 20213 źródeł galaxyporn.
    return "fastpic.org" in url


def _is_flaky_thumb(url: str) -> bool:
    """Host, który działa BYWA, ale zbyt często pada, żeby wygrywać wybór miniaturki.

    W odróżnieniu od `_is_rotting_thumb` NIE zerujemy takiej miniatury w SceneDetail:
    to nie jest martwy URL, tylko niepewny. Zerowanie powtórzyłoby wpadkę z fullpornerem
    z sierpnia 2026, gdzie sceny mające tylko to źródło traciły obrazek całkowicie.

    Problem był w kolejności wyboru: miniaturkę wygrywał pierwszy origin alfabetycznie,
    a `tube:fullmoviesxxx` sortuje się PRZED `tube:fullporner`, więc chwiejny host
    przykrywał zdrowy. 51 799 z 52 758 takich scen (98%) ma alternatywę, więc degradacja
    niemal zawsze ma czym zastąpić.

    Pomiar 2026-08-13, po 20-25 losowych URL-ach na host:

        img.fullmovies.xxx    12/20 ok   (5× ReadTimeout, 3× 502)   54 019 źródeł
        img.hdporn.gg         16/20 ok   (3× 502, 1× ReadTimeout)   54 267 źródeł
        imgs.xiaoshenke.net   20/20 ok   (fullporner)               68 406 źródeł
        freshporno.org        20/20 ok

    `img.hdporn.gg` dopisany po sprawdzeniu efektu pierwszej wersji: degradacja samego
    fullmovies przesunęła wybór na hdporn (44 z 60 scen), czyli przeniosłaby problem
    zamiast go rozwiązać. Po dopisaniu obu wygrywają hosty ze 100% skutecznością.
    """
    return "img.fullmovies.xxx" in url or "img.hdporn.gg" in url


#: Im niżej, tym gorzej. Używane TYLKO wewnątrz tieru fallback, czyli gdy scena nie ma
#: żadnej zdrowej miniatury i trzeba wybrać mniejsze zło. Bez tego decydowała kolejność
#: alfabetyczna originu, przez co scena mająca fullmovies + hdporn dostawała fullmovies,
#: czyli host o 40% awaryjności zamiast tego o 20%.
_FLAKY_RANK = (
    ("img.hdporn.gg", 1),        # 16/20 ok
    ("img.fullmovies.xxx", 2),   # 12/20 ok
)


def _flaky_rank(url: str) -> int:
    for host, rank in _FLAKY_RANK:
        if host in url:
            return rank
    return 0  # nie-chwiejny (martwy rotting-thumb) — bez preferencji


def _is_preferred_origin(origin: str | None, prefer_origin: str | None) -> bool:
    """Czy to origin, ktory user wlasnie przeglada (`/scenes?origin=`).

    Porownanie substringiem, bo parametr API jest substringowy: `hqporner` ma zlapac
    `tube:hqpornercom`.
    """
    if not prefer_origin or not origin:
        return False
    return prefer_origin.lower() in origin.lower()


def _needs_proxy(url: str) -> bool:
    """Wszystkie thumbnaile z playback_sources są proxowane przez backend.
    Większość CDN-ów porn-tube'ów wymaga Refera (hqporner, mypornerleak/58img,
    inne sxyprn/eporner CDN-y), expo-image nie wysyła Referera.
    Self-hosted lub backend-internal URL-e (zaczynające się od `/`) skipujemy."""
    return url.startswith("http") and not url.startswith("/proxy/")


def _wrap_image_proxy(url: str, referer: str) -> str:
    """Wraps a thumbnail URL through /proxy/img/{token}/img.jpg. Klient nie musi
    znać sekretu Referer, backend wstawi sam. Long TTL (30d) bo thumby
    są stabilne, krótkie ttl by tylko niepotrzebnie zaśmiecało cache."""
    from app.api.stream_proxy import make_token
    # stable_bucket_sec=7d → proxied URL identyczny przez tydzień → expo-image
    # disk-cache hit zamiast re-download miniatur przy każdym fetchu listy / starcie apki.
    token = make_token(url, referer, ttl_sec=30 * 24 * 3600, stable_bucket_sec=7 * 24 * 3600)
    # Path zachowuje rozszerzenie żeby HTTP Content-Type był rozpoznany.
    import os as _os
    ext = _os.path.splitext(url.split("?")[0])[1].lstrip(".") or "jpg"
    return f"/proxy/img/{token}/img.{ext}"


def _build_scenes_out_batch(
    session: Session,
    scenes: list[Scene],
    *,
    light: bool = False,
    device_id: str = LEGACY_DEVICE,
    prefer_origin: str | None = None,
) -> list[SceneOut]:
    """Batch-fetch wszystkich relacji dla N scen w 7 zapytaniach (zamiast 7×N).

    Eliminuje N+1 z `_build_scene_out` w listach scen, `/scenes?per_page=24` szło
    z ~9.6s do <500ms. Pojedyncza scena (`/scenes/{id}`) nadal używa `_build_scene_out`
    bo overhead na batch nie ma sensu dla N=1.

    `light=True` (listy/grid): pomija `tags` i `external_refs` (kafelek SceneTile ich
    NIE używa, a SceneDetail re-fetchuje pełną scenę osobno) i ślimaczy `playback_sources`
    do 1 wpisu z samą miniaturką (kafelek czyta tylko thumbnail_url/animated_thumbnail_url).
    Mniej DB + mniej payloadu + szybszy parse na kliencie (perf 2026-06-07).
    """
    from collections import defaultdict
    if not scenes:
        return []

    scene_ids = [s.id for s in scenes]
    studio_ids = list({s.studio_id for s in scenes if s.studio_id is not None})

    # 1) Studios
    studios_by_id: dict = {}
    if studio_ids:
        for st in session.execute(
            select(Studio).where(Studio.id.in_(studio_ids))
        ).scalars():
            studios_by_id[st.id] = st

    # 2) Performers
    perf_rows = session.execute(
        select(ScenePerformer, Performer)
        .join(Performer, Performer.id == ScenePerformer.performer_id)
        .where(ScenePerformer.scene_id.in_(scene_ids))
        .order_by(ScenePerformer.position.asc().nullslast())
    ).all()
    performers_by_scene: dict = defaultdict(list)
    for sp, p in perf_rows:
        performers_by_scene[sp.scene_id].append(
            PerformerOut(
                id=p.id,
                canonical_name=p.canonical_name,
                slug=p.slug,
                gender=p.gender.value if p.gender else None,
                as_alias=sp.as_alias,
            )
        )

    # 3) Tags + 4) External refs, kafelek listy ich nie używa; w light mode pomijamy
    # (SceneDetail re-fetchuje pełną scenę przez /scenes/{id}).
    tags_by_scene: dict = defaultdict(list)
    refs_by_scene: dict = defaultdict(list)
    if not light:
        tag_rows = session.execute(
            select(SceneTag.scene_id, Tag)
            .join(Tag, Tag.id == SceneTag.tag_id)
            .where(SceneTag.scene_id.in_(scene_ids))
        ).all()
        for sid, t in tag_rows:
            tags_by_scene[sid].append(TagOut.model_validate(t))

        ref_rows = session.execute(
            select(SceneExternalRef, Source)
            .join(Source, Source.id == SceneExternalRef.source_id)
            .where(SceneExternalRef.scene_id.in_(scene_ids))
        ).all()
        for ref, src in ref_rows:
            refs_by_scene[ref.scene_id].append(
                ExternalRefOut(
                    source=src.name,
                    external_id=ref.external_id,
                    url=ref.url,
                    last_seen=ref.last_seen,
                )
            )

    # 5) Playback sources. Light mode: tylko miniaturka (jedna na scenę), kafelek
    # czyta wyłącznie playback_sources[].thumbnail_url / animated_thumbnail_url.
    pb_by_scene: dict = defaultdict(list)
    # origin per scena, wyłącznie do policzenia `vertical` (obie ścieżki go wypełniają).
    origins_by_scene: dict = {}
    if light:
        pb_light = session.execute(
            select(
                PlaybackSource.scene_id,
                PlaybackSource.thumbnail_url,
                PlaybackSource.animated_thumbnail_url,
                PlaybackSource.page_url,
                # origin dociągamy wyłącznie po to, żeby policzyć `vertical` (tania kolumna
                # z tego samego wiersza, zero dodatkowych zapytań). Do samego kafelka
                # nieużywany, slim wpis niżej dostaje origin="".
                PlaybackSource.origin,
            )
            .where(
                PlaybackSource.scene_id.in_(scene_ids),
                PlaybackSource.dead_at.is_(None),
            )
            .order_by(PlaybackSource.origin.asc())
        ).all()
        # Pierwsza miniaturka + pierwszy animated per scena (1 slim wpis). De-prioritize
        # sxyprn/trafficdeposit thumbnaile, są podpisane czasowo i ROTUJĄ (asset 404 po
        # ~tygodniach, nie da się re-signować; bug 2026-06-10). Wolimy miniaturkę z innego
        # źródła gdy istnieje; sxyprn bierzemy tylko gdy nic innego nie ma (świeże jeszcze
        # działają, martwe → mobile pokazuje placeholder zamiast broken-image).
        thumb_by_scene: dict = {}
        thumb_fallback: dict = {}
        anim_by_scene: dict = {}
        for sid, thumb, anim, page_url, origin in pb_light:
            origins_by_scene.setdefault(sid, []).append(origin)
            sxy = _sxyprn_thumb_url(page_url)
            yp = _youporn_thumb_url(page_url, thumb)
            if sxy:
                # sxyprn → żywy on-demand resolver (martwy stored URL ignorujemy),
                # tier fallback: użyty tylko gdy scena nie ma stabilniejszej miniatury.
                thumb_fallback.setdefault(sid, (sxy, page_url, origin))
            elif yp:
                # youporn z podpisanym (wygasającym) posterem → on-demand resolver.
                # Też tier fallback: statyczna miniatura z innego źródła jest tańsza.
                thumb_fallback.setdefault(sid, (yp, page_url, origin))
            elif thumb:
                if _is_rotting_thumb(thumb) or _is_flaky_thumb(thumb):
                    # Martwe ORAZ chwiejne lądują w tierze fallback: użyte tylko wtedy,
                    # gdy scena nie ma stabilniejszej miniatury z innego źródła. W obrębie
                    # samego fallbacku wybieramy mniejsze zło wg `_FLAKY_RANK`, bo inaczej
                    # decydowałaby kolejność alfabetyczna originu i scena z fullmovies +
                    # hdporn dostawała gorszy z dwóch (pomiar: 38 z 60 takich scen).
                    prev = thumb_fallback.get(sid)
                    if prev is None or _flaky_rank(thumb) < _flaky_rank(prev[0]):
                        thumb_fallback[sid] = (thumb, page_url, origin)
                elif sid not in thumb_by_scene or _is_preferred_origin(origin, prefer_origin):
                    # Gdy user przeglada KONKRETNY serwis, jego miniaturka wygrywa.
                    # Wczesniej decydowala kolejnosc alfabetyczna originu, wiec na liscie
                    # hqporner kafelki scen wspoldzielonych z fullpornerem bralo
                    # `tube:fullporner` (< `tube:hqpornercom`), a czesc tych miniaturek to
                    # 153-bajtowa atrapa z imgs.xiaoshenke.net oddawana z kodem 200. Efekt:
                    # puste kafelki podpisane FULLPORNER na liscie hqporner, mimo ze ta
                    # sama scena miala obok zdrowa miniaturke hqporner (zgloszenie 1df50de3).
                    thumb_by_scene[sid] = (thumb, page_url, origin)
            if anim and (sid not in anim_by_scene or _is_preferred_origin(origin, prefer_origin)):
                anim_by_scene[sid] = (anim, page_url, origin)
        # Uzupełnij scenami które mają TYLKO rotting thumbnail (sxyprn-only).
        for sid, val in thumb_fallback.items():
            thumb_by_scene.setdefault(sid, val)
        for sid in scene_ids:
            t = thumb_by_scene.get(sid)
            a = anim_by_scene.get(sid)
            if not t and not a:
                continue
            t_url = t[0] if t else None
            a_url = a[0] if a else None
            ref = (t or a)[1]
            if t_url and _needs_proxy(t_url):
                t_url = _wrap_image_proxy(t_url, ref)
            if a_url and _needs_proxy(a_url):
                a_url = _wrap_image_proxy(a_url, ref)
            # `origin` niesiemy PRAWDZIWY (źródła wybranej miniatury): kafelek pokazuje
            # nazwę serwisu (element z makiety C, 2026-08-18). `id`/`page_url` zostają
            # sentinelem, bo SceneDetail i tak re-fetchuje pełne źródła.
            pb_by_scene[sid].append(
                PlaybackSourceOut(
                    id=uuid.UUID(int=0), origin=(t or a)[2] or "", page_url="",
                    thumbnail_url=t_url, animated_thumbnail_url=a_url,
                )
            )
    else:
        pb_rows = session.execute(
            select(PlaybackSource)
            .where(
                PlaybackSource.scene_id.in_(scene_ids),
                PlaybackSource.dead_at.is_(None),
            )
            .order_by(PlaybackSource.origin.asc())
        ).scalars().all()
        for p in pb_rows:
            origins_by_scene.setdefault(p.scene_id, []).append(p.origin)
            out = PlaybackSourceOut.model_validate(p)
            # sxyprn/trafficdeposit stored thumb rotuje (404 po ~tygodniach). SceneDetail
            # bierze PIERWSZE źródło z thumbem (origin ASC → sxyprncom często pierwsze), więc
            # bez tego pokazywał martwy obraz mimo że LISTA używała żywej wersji (report:
            # "na liście jest miniaturka, w scenie nie"). Podmieniamy na żywy resolver;
            # inne martwe rotting-thumby zerujemy → mobile bierze kolejne źródło / placeholder.
            sxy = _sxyprn_thumb_url(p.page_url)
            yp = _youporn_thumb_url(p.page_url, out.thumbnail_url)
            if sxy:
                out.thumbnail_url = sxy
            elif yp:
                out.thumbnail_url = yp
            elif out.thumbnail_url and _is_rotting_thumb(out.thumbnail_url):
                out.thumbnail_url = None
            if out.thumbnail_url and _needs_proxy(out.thumbnail_url):
                out.thumbnail_url = _wrap_image_proxy(out.thumbnail_url, p.page_url)
            if out.animated_thumbnail_url and _needs_proxy(out.animated_thumbnail_url):
                out.animated_thumbnail_url = _wrap_image_proxy(out.animated_thumbnail_url, p.page_url)
            pb_by_scene[p.scene_id].append(out)

    # 6) Progress (device-scoped)
    progress_by_scene: dict = {}
    for prog in session.execute(
        select(ScenePlayProgress).where(
            ScenePlayProgress.scene_id.in_(scene_ids),
            ScenePlayProgress.device_id == device_id,
        )
    ).scalars():
        progress_by_scene[prog.scene_id] = prog

    # 7) Favorites (device-scoped)
    fav_scene_ids: set = set(
        session.execute(
            select(FavoriteScene.scene_id).where(
                FavoriteScene.scene_id.in_(scene_ids),
                FavoriteScene.device_id == device_id,
            )
        ).scalars()
    )

    out: list[SceneOut] = []
    for scene in scenes:
        studio_out = None
        if scene.studio_id is not None and scene.studio_id in studios_by_id:
            studio_out = StudioOut.model_validate(studios_by_id[scene.studio_id])
        progress = progress_by_scene.get(scene.id)
        out.append(
            SceneOut(
                id=scene.id,
                title=scene.title,
                slug=scene.slug,
                release_date=scene.release_date,
                duration_sec=scene.duration_sec,
                description=scene.description,
                code=scene.code,
                director=scene.director,
                studio=studio_out,
                performers=performers_by_scene.get(scene.id, []),
                tags=tags_by_scene.get(scene.id, []),
                external_refs=refs_by_scene.get(scene.id, []),
                playback_sources=pb_by_scene.get(scene.id, []),
                vertical=verticals.vertical_for_origins(origins_by_scene.get(scene.id, ())),
                max_quality_p=scene.max_quality_p,
                created_at=scene.created_at,
                backfill=scene.backfill,
                last_played_at=progress.last_played_at if progress else None,
                finished=progress.finished if progress else False,
                position_sec=progress.position_sec if progress else 0,
                is_favorite=scene.id in fav_scene_ids,
            )
        )
    return out


def _build_scene_out(session: Session, scene: Scene, *, device_id: str = LEGACY_DEVICE) -> SceneOut:
    studio_out: StudioOut | None = None
    if scene.studio_id is not None:
        st = session.get(Studio, scene.studio_id)
        if st is not None:
            studio_out = StudioOut.model_validate(st)

    performer_rows = session.execute(
        select(ScenePerformer, Performer)
        .join(Performer, Performer.id == ScenePerformer.performer_id)
        .where(ScenePerformer.scene_id == scene.id)
        .order_by(ScenePerformer.position.asc().nullslast())
    ).all()
    performers_out: list[PerformerOut] = []
    for sp, performer in performer_rows:
        performers_out.append(
            PerformerOut(
                id=performer.id,
                canonical_name=performer.canonical_name,
                slug=performer.slug,
                gender=performer.gender.value if performer.gender else None,
                as_alias=sp.as_alias,
            )
        )

    tag_rows = (
        session.execute(
            select(Tag).join(SceneTag, SceneTag.tag_id == Tag.id).where(SceneTag.scene_id == scene.id)
        )
        .scalars()
        .all()
    )
    tags_out = [TagOut.model_validate(t) for t in tag_rows]

    ref_rows = session.execute(
        select(SceneExternalRef, Source)
        .join(Source, Source.id == SceneExternalRef.source_id)
        .where(SceneExternalRef.scene_id == scene.id)
    ).all()
    refs_out = [
        ExternalRefOut(
            source=src.name,
            external_id=ref.external_id,
            url=ref.url,
            last_seen=ref.last_seen,
        )
        for ref, src in ref_rows
    ]

    playback_rows = (
        session.execute(
            select(PlaybackSource)
            .where(
                PlaybackSource.scene_id == scene.id,
                PlaybackSource.dead_at.is_(None),  # ukryj martwe linki
            )
            .order_by(PlaybackSource.origin.asc())
        )
        .scalars()
        .all()
    )
    # Collapse źródła dzielące ten sam origin (hoster). Zmergowana scena często agreguje
    # kilka uploadów z JEDNEGO tube'a (re-enkody / wersje 4K: bug-report aa79a995 "2 linki,
    # oba do porntrex" = ta sama scena std+4K), w UI to nierozróżnialne linki do tego
    # samego hostera (resolvują tym samym extractorem). Zostawiamy jeden najlepszy per
    # origin: preferuj długość zgodną ze sceną (realny match) → jakąkolwiek długość →
    # pierwszy (stabilnie, query jest origin-asc). Martwe już odfiltrowane (dead_at).
    def _origin_pick_key(p: PlaybackSource) -> tuple[int, int]:
        dur_match = (
            0 if (scene.duration_sec and p.duration_sec
                  and abs(p.duration_sec - scene.duration_sec) <= 5) else 1
        )
        return (dur_match, 0 if p.duration_sec else 1)

    _best_by_origin: dict[str, PlaybackSource] = {}
    for p in playback_rows:
        key = p.origin or ""
        cur = _best_by_origin.get(key)
        if cur is None or _origin_pick_key(p) < _origin_pick_key(cur):
            _best_by_origin[key] = p
    playback_rows = list(_best_by_origin.values())

    playback_out: list[PlaybackSourceOut] = []
    for p in playback_rows:
        out = PlaybackSourceOut.model_validate(p)
        # sxyprn: podmień martwy trafficdeposit URL na żywy on-demand resolver
        # (/proxy/sxyprn-thumb/<id>), TAK JAK lista (_build_scenes_out_batch). Bez tego
        # detal sceny pokazywał martwą miniaturę sxyprn mimo że lista działała, i "refresh
        # thumbnail" nie pomagał (sxyprn IP-bound, enrich z VPS pada). Report 2026-07-22.
        sxy = _sxyprn_thumb_url(p.page_url)
        yp = _youporn_thumb_url(p.page_url, out.thumbnail_url)
        if sxy:
            out.thumbnail_url = sxy
        elif yp:
            out.thumbnail_url = yp
        elif out.thumbnail_url and _is_rotting_thumb(out.thumbnail_url):
            out.thumbnail_url = None
        # Wrap thumbnail URL-e przez backend image proxy gdy CDN wymaga Refera
        # (hqporner, fastporndelivery zwraca 403 bez Referer headera, expo-image
        # nie wysyła go domyślnie). Token ma 30-dniowy TTL bo thumby są stabilne.
        if out.thumbnail_url and _needs_proxy(out.thumbnail_url):
            out.thumbnail_url = _wrap_image_proxy(out.thumbnail_url, p.page_url)
        if out.animated_thumbnail_url and _needs_proxy(out.animated_thumbnail_url):
            out.animated_thumbnail_url = _wrap_image_proxy(out.animated_thumbnail_url, p.page_url)
        playback_out.append(out)

    # Rank natywne-resolve źródła PRZED WebView-fallback (IP-bound/ad-heavy: fpoxxx,
    # pornxpph, pornhub...). Query był alfabetyczny po origin, więc np. fpoxxx-WebView
    # pokazywał się przed działającym freshporno (bug-report 2026-06-07). Stabilny sort:
    # natywne (0) → fallback (1), tie-break po origin.
    from app.extractors import is_vps_blocked_fallback

    def _resolve_rank(origin: str | None) -> int:
        if not origin:
            return 1
        sitetag = origin.split(":", 1)[1] if ":" in origin else origin
        return 1 if is_vps_blocked_fallback(sitetag) else 0

    playback_out.sort(key=lambda o: (_resolve_rank(o.origin), o.origin or ""))

    progress = session.get(ScenePlayProgress, (device_id, scene.id))
    is_fav = session.get(FavoriteScene, (device_id, scene.id)) is not None

    return SceneOut(
        id=scene.id,
        title=scene.title,
        slug=scene.slug,
        release_date=scene.release_date,
        duration_sec=scene.duration_sec,
        description=scene.description,
        code=scene.code,
        director=scene.director,
        studio=studio_out,
        performers=performers_out,
        tags=tags_out,
        external_refs=refs_out,
        playback_sources=playback_out,
        vertical=verticals.vertical_for_origins(p.origin for p in playback_rows),
        max_quality_p=scene.max_quality_p,
        created_at=scene.created_at,
        backfill=scene.backfill,
        last_played_at=progress.last_played_at if progress else None,
        finished=progress.finished if progress else False,
        position_sec=progress.position_sec if progress else 0,
        is_favorite=is_fav,
    )


@router.delete("/{scene_id}/tags/{tag_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_tag_from_scene(
    scene_id: uuid.UUID,
    tag_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> None:
    """Usuwa relację scene↔tag (np. user uznał że tag jest błędny dla tej sceny).

    Idempotent: brak relacji = success. Nie kasuje samego Tag-a, inne sceny mogą
    z niego korzystać. Sam tag zostaje w słowniku tagów.
    """
    rel = session.execute(
        select(SceneTag).where(SceneTag.scene_id == scene_id, SceneTag.tag_id == tag_id)
    ).scalar_one_or_none()
    if rel is None:
        return
    session.delete(rel)
    session.commit()


@router.delete(
    "/{scene_id}/performers/{performer_id}", status_code=status.HTTP_204_NO_CONTENT
)
def remove_performer_from_scene(
    scene_id: uuid.UUID,
    performer_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> None:
    """Usuwa relację scene↔performer (false-match dedup zostawił nie tą osobę).

    Idempotent. Sama Performer zostaje. Użyteczne np. gdy fuzzy match aliasu
    "Bella" wciągnął Anna Bella sceny pod Bad Bella, lub Miss Teela na xnxx
    została przypisana do scen w których jej nie ma (zgłoszenia 2026-05-10).
    """
    from app.models.scene import ScenePerformer

    rel = session.execute(
        select(ScenePerformer).where(
            ScenePerformer.scene_id == scene_id,
            ScenePerformer.performer_id == performer_id,
        )
    ).scalar_one_or_none()
    if rel is None:
        return
    session.delete(rel)
    session.commit()


class SceneHideOut(BaseModel):
    scene_id: uuid.UUID
    playback_marked_dead: int


@router.post("/{scene_id}/hide", response_model=SceneHideOut)
def hide_scene(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> SceneHideOut:
    """Ukryj scenę (user long-press → „usuń"). Oznacza wszystkie playback_sources
    jako dead → scena wypada z list (has_playback=false). Odwracalne w DB (dead_at).
    Nie kasujemy wiersza sceny, zachowujemy refs/dedup, tylko znika z UI."""
    from datetime import UTC, datetime

    from app.models.playback_source import PlaybackSource

    if session.get(Scene, scene_id) is None:
        raise HTTPException(status_code=404, detail="scene not found")
    rows = session.execute(
        select(PlaybackSource).where(
            PlaybackSource.scene_id == scene_id,
            PlaybackSource.dead_at.is_(None),
        )
    ).scalars().all()
    now = datetime.now(UTC)
    for p in rows:
        p.dead_at = now
        p.dead_reason = "user hid scene (long-press)"
    session.commit()
    return SceneHideOut(scene_id=scene_id, playback_marked_dead=len(rows))


class SceneMergeOut(BaseModel):
    keep_id: uuid.UUID
    dropped_id: uuid.UUID


@router.post("/{keep_id}/merge/{drop_id}", response_model=SceneMergeOut)
def merge_duplicate_scene(
    keep_id: uuid.UUID,
    drop_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> SceneMergeOut:
    """Scal `drop_id` w `keep_id` (user long-press → „oznacz duplikat" → wybór drugiej
    sceny). Przenosi refs/performers/tags/fingerprints/playback (scene_merge), kasuje
    `drop`. keep = scena na której user trzyma (zostaje), drop = wskazany duplikat."""
    from app.resolve.scene_merge import MergeError, merge_scenes

    if keep_id == drop_id:
        raise HTTPException(status_code=400, detail="keep_id == drop_id")
    if session.get(Scene, keep_id) is None or session.get(Scene, drop_id) is None:
        raise HTTPException(status_code=404, detail="scene not found")
    try:
        merge_scenes(session, keep_id=keep_id, drop_id=drop_id, resolved_by="user_long_press_duplicate")
    except MergeError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    session.commit()
    return SceneMergeOut(keep_id=keep_id, dropped_id=drop_id)


class EnrichTagsOut(BaseModel):
    scene_id: uuid.UUID
    added: int
    tube_used: str | None
    tags: list[str]


@router.post("/{scene_id}/enrich-tags", response_model=EnrichTagsOut)
def enrich_tags_from_tube(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> EnrichTagsOut:
    """Pobiera page HTML z dowolnego tube playback_source dla tej sceny i scrape'uje
    tagi (categories/tags). Dodaje brakujące do scene_tags.

    Mobile wywołuje to przy otwarciu SceneDetail jeśli scena ma 0 tagów AND ma
    tube source z obsługiwanym extractorem (porntrex/youporn/xvideos/xnxx/redtube/
    xhamster/eporner).

    Idempotent: ponowne wywołanie z tymi samymi tagami nic nie robi (UNIQUE PK
    scene_tags). Konkretne tube źródło wybierane wg priority listy (mainstream
    bardziej rzetelne niż aggregator).
    """
    from app.extractors._fetch import browser_get
    from app.extractors._models import TubePageError
    from app.extractors.tag_extract import EXTRACTORS, extract_tags
    from app.models.playback_source import PlaybackSource
    from app.models.tag import Tag
    from app.normalize.scenes import NormalizedTag
    from app.normalize.text import slugify
    from app.resolve.tag_resolver import resolve_tag

    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")

    # Priority: mainstream tubes (bogate metadane) > niche (mniej tagów albo garbage).
    PRIORITY = ["xhamstercom", "porntrexcom", "epornercom", "youporncom",
                "xvideoscom", "xnxxcom"]
    sources = session.execute(
        select(PlaybackSource).where(
            PlaybackSource.scene_id == scene_id,
            PlaybackSource.dead_at.is_(None),
        )
    ).scalars().all()

    # Wybierz pierwsze źródło wg priority listy które ma supported extractor
    chosen: PlaybackSource | None = None
    for tag in PRIORITY:
        for src in sources:
            if src.origin == f"tube:{tag}":
                chosen = src
                break
        if chosen:
            break
    if chosen is None:
        # Fallback: dowolne źródło z extractorem
        for src in sources:
            if src.origin.startswith("tube:"):
                sitetag = src.origin.split(":", 1)[1]
                if sitetag in EXTRACTORS:
                    chosen = src
                    break

    if chosen is None:
        return EnrichTagsOut(scene_id=scene_id, added=0, tube_used=None, tags=[])

    sitetag = chosen.origin.split(":", 1)[1]
    try:
        r = browser_get(chosen.page_url, timeout=15.0, follow_redirects=True)
        r.raise_for_status()
    except (TubePageError, Exception) as e:
        log.warning("enrich-tags fetch failed for %s: %s", chosen.page_url, e)
        return EnrichTagsOut(scene_id=scene_id, added=0, tube_used=sitetag, tags=[])

    tag_names = extract_tags(sitetag, r.text)
    if not tag_names:
        return EnrichTagsOut(scene_id=scene_id, added=0, tube_used=sitetag, tags=[])

    # Upsert: dla każdego taga utwórz/znajdź Tag, dorzuć SceneTag idempotentnie.
    # Używamy PostgreSQL INSERT ... ON CONFLICT DO NOTHING zamiast ORM session.add()
    # bo `resolve_tag` robi session.flush() w pętli, emitując pending SceneTag INSERT
    # z poprzednich iteracji, gdy 2 concurrent enrich-tags collide na tym samym
    # (scene_id, tag_id), drugi flush dostaje UniqueViolation (GOON-H, 4 events
    # w 10h mimo wcześniejszego seen_tag_ids fix). ON CONFLICT skip'uje silently.
    from sqlalchemy.dialects.postgresql import insert as pg_insert
    added = 0
    seen_tag_ids: set = set()
    for name in tag_names:
        norm = NormalizedTag(name=name, slug=slugify(name), external_id=None)
        tag = resolve_tag(session, norm=norm)
        if tag is None or tag.id in seen_tag_ids:
            continue
        seen_tag_ids.add(tag.id)
        stmt = (
            pg_insert(SceneTag.__table__)
            .values(scene_id=scene_id, tag_id=tag.id, source_id=None)
            .on_conflict_do_nothing(index_elements=["scene_id", "tag_id"])
        )
        result = session.execute(stmt)
        # rowcount == 1 gdy faktycznie wstawiony, 0 gdy ON CONFLICT skip
        if result.rowcount and result.rowcount > 0:
            added += 1
    session.commit()
    return EnrichTagsOut(scene_id=scene_id, added=added, tube_used=sitetag, tags=tag_names)


class EnrichDurationOut(BaseModel):
    scene_id: uuid.UUID
    duration_sec: int | None
    tube_used: str | None


@router.post("/{scene_id}/enrich-duration", response_model=EnrichDurationOut)
def enrich_duration_from_tube(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> EnrichDurationOut:
    """Wyciąga duration z dowolnego tube playback_source, wszystkie znane tube'y
    udostępniają duration na detail page (og:video:duration lub LD-JSON ISO 8601).

    Mobile wywołuje to przy otwarciu SceneDetail gdy scene.duration_sec jest null
    AND ma tube source. Dla dedupu duration to najsilniejszy single signal, bez
    niego sceny z weak title-only score są capowane na 0.85 (review queue).

    Idempotent: zwraca aktualne duration_sec jeśli już ustawione.
    """
    from app.extractors._fetch import browser_get
    from app.extractors._models import TubePageError
    from app.extractors.duration_extract import extract_duration_sec
    from app.models.playback_source import PlaybackSource

    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")

    if scene.duration_sec is not None:
        return EnrichDurationOut(
            scene_id=scene_id, duration_sec=scene.duration_sec, tube_used=None
        )

    sources = session.execute(
        select(PlaybackSource).where(
            PlaybackSource.scene_id == scene_id,
            PlaybackSource.dead_at.is_(None),
            PlaybackSource.origin.like("tube:%"),
        )
    ).scalars().all()

    for src in sources:
        try:
            r = browser_get(src.page_url, timeout=15.0, follow_redirects=True)
            r.raise_for_status()
        except (TubePageError, Exception) as e:
            log.debug("enrich-duration fetch failed for %s: %s", src.page_url, e)
            continue
        d = extract_duration_sec(r.text)
        if d is not None and d > 0:
            scene.duration_sec = d
            # Zapisz też na poziomie playback_source dla parity (przyda się jeśli
            # potem dorobimy per-source duration mismatch detection).
            if src.duration_sec is None:
                src.duration_sec = d
            session.commit()
            return EnrichDurationOut(
                scene_id=scene_id,
                duration_sec=d,
                tube_used=src.origin.split(":", 1)[1] if ":" in src.origin else None,
            )

    return EnrichDurationOut(scene_id=scene_id, duration_sec=None, tube_used=None)


class EnrichStudioOut(BaseModel):
    scene_id: uuid.UUID
    studio_id: uuid.UUID | None
    studio_name: str | None
    tube_used: str | None


@router.post("/{scene_id}/enrich-studio", response_model=EnrichStudioOut)
def enrich_studio_from_tube(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> EnrichStudioOut:
    """Wyciąga studio (DVD/series) z pornhat scene page'a.

    Pornhat ma `class="info-video js-ajax-dvd" data-setup='{"title": "Adult Time", ...}'`
    dla studio. Inne tube'y obsługiwane będą gdy znajdziemy ich pattern, na razie
    tylko pornhat (najczystsze studio metadata wśród free tubes).
    """
    import json as _json

    from app.extractors._fetch import browser_get
    from app.extractors._models import TubePageError
    from app.models.playback_source import PlaybackSource
    from app.models.studio import Studio
    from app.normalize.text import slugify

    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")

    if scene.studio_id is not None:
        existing = session.get(Studio, scene.studio_id)
        return EnrichStudioOut(
            scene_id=scene_id,
            studio_id=scene.studio_id,
            studio_name=existing.name if existing else None,
            tube_used=None,
        )

    chosen = session.execute(
        select(PlaybackSource).where(
            PlaybackSource.scene_id == scene_id,
            PlaybackSource.dead_at.is_(None),
            PlaybackSource.origin == "tube:pornhatcom",
        )
    ).scalars().first()
    if chosen is None:
        return EnrichStudioOut(scene_id=scene_id, studio_id=None, studio_name=None, tube_used=None)

    try:
        r = browser_get(chosen.page_url, timeout=15.0, follow_redirects=True)
        r.raise_for_status()
    except (TubePageError, Exception) as e:
        log.warning("enrich-studio fetch failed for %s: %s", chosen.page_url, e)
        return EnrichStudioOut(scene_id=scene_id, studio_id=None, studio_name=None, tube_used="pornhatcom")

    m = re.search(
        r"class=\"info-video js-ajax-dvd[^\"]*\"[^>]*data-setup='([^']+)'",
        r.text, re.IGNORECASE,
    )
    if m is None:
        return EnrichStudioOut(scene_id=scene_id, studio_id=None, studio_name=None, tube_used="pornhatcom")
    try:
        data = _json.loads(m.group(1))
    except _json.JSONDecodeError:
        return EnrichStudioOut(scene_id=scene_id, studio_id=None, studio_name=None, tube_used="pornhatcom")

    name = (data.get("title") or "").strip()
    if not name:
        return EnrichStudioOut(scene_id=scene_id, studio_id=None, studio_name=None, tube_used="pornhatcom")
    slug = (data.get("dir") or "").strip() or slugify(name)

    studio = session.execute(
        select(Studio).where(Studio.slug == slug)
    ).scalar_one_or_none()
    if studio is None:
        studio = session.execute(
            select(Studio).where(Studio.name == name)
        ).scalar_one_or_none()
    if studio is None:
        studio = Studio(name=name, slug=slug)
        session.add(studio)
        session.flush()
    scene.studio_id = studio.id
    session.commit()
    return EnrichStudioOut(
        scene_id=scene_id, studio_id=studio.id, studio_name=studio.name, tube_used="pornhatcom"
    )


class EnrichThumbOut(BaseModel):
    scene_id: uuid.UUID
    thumbnail_url: str | None
    tube_used: str | None
    sources_updated: int


@router.post("/{scene_id}/enrich-thumbnail", response_model=EnrichThumbOut)
def enrich_thumbnail_from_tube(
    scene_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
    force: bool = False,
) -> EnrichThumbOut:
    """Pobiera detail page z dowolnego tube playback_source bez thumbnail_url
    i wyciąga miniaturkę (og:image / twitter:image / LD-JSON thumbnailUrl /
    KVS html5player).

    Update'uje WSZYSTKIE PlaybackSource'y dla tej sceny które nie mają thumb,
    żeby kolejne otwarcia listy widziały miniaturę niezależnie od source pick.
    Mobile auto-wywoła to przy otwarciu SceneDetail bez thumb (jak duration).

    `force=true` (przycisk "Refresh thumbnail" na SceneDetail, zgłoszenie d3376a71):
    NADPISUJE istniejącą miniaturę świeżą ze strony tube'a, dla zepsutych/stałych
    (rotting sxyprn/trafficdeposit, błędna grafika)."""
    from app.extractors._fetch import browser_get
    from app.extractors._models import TubePageError
    from app.extractors.thumb_extract import extract_thumbnail_url
    from app.models.playback_source import PlaybackSource

    scene = session.get(Scene, scene_id)
    if scene is None:
        raise HTTPException(status_code=404, detail="scene not found")

    sources = session.execute(
        select(PlaybackSource).where(
            PlaybackSource.scene_id == scene_id,
            PlaybackSource.dead_at.is_(None),
            PlaybackSource.origin.like("tube:%"),
        )
    ).scalars().all()

    sources_with_thumb = [s for s in sources if s.thumbnail_url]
    if sources_with_thumb and not force:
        # już mamy, idempotent return (force=true pomija, żeby odświeżyć).
        return EnrichThumbOut(
            scene_id=scene_id,
            thumbnail_url=sources_with_thumb[0].thumbnail_url,
            tube_used=None,
            sources_updated=0,
        )

    for src in sources:
        try:
            r = browser_get(src.page_url, timeout=15.0, follow_redirects=True)
            r.raise_for_status()
        except (TubePageError, Exception) as e:
            log.debug("enrich-thumbnail fetch failed for %s: %s", src.page_url, e)
            continue
        thumb = extract_thumbnail_url(r.text)
        if thumb:
            # Zapisz na wszystkich źródłach bez thumb (force → też nadpisz istniejące).
            updated = 0
            for s in sources:
                if force or not s.thumbnail_url:
                    s.thumbnail_url = thumb
                    updated += 1
            session.commit()
            return EnrichThumbOut(
                scene_id=scene_id,
                thumbnail_url=thumb,
                tube_used=src.origin.split(":", 1)[1] if ":" in src.origin else None,
                sources_updated=updated,
            )

    return EnrichThumbOut(
        scene_id=scene_id, thumbnail_url=None, tube_used=None, sources_updated=0
    )