goon/app/api/movies.py

"""GET /movies, lista i szczegóły filmów."""
from __future__ import annotations

import uuid
from collections import defaultdict
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import exists, func, select
from sqlalchemy.orm import Session

from app.api.schemas import (
    ExternalRefOut,
    MovieChapterOut,
    MovieListOut,
    MovieOut,
    PerformerOut,
    PlaybackSourceOut,
    StudioOut,
    TagOut,
)
from app.auth import require_api_key
from app.db import get_session
from app.models.movie import (
    Movie,
    MovieChapter,
    MovieExternalRef,
    MoviePerformer,
    MovieTag,
)
from app.api.device import LEGACY_DEVICE, get_device_id
from app.models.favorite_movie import FavoriteMovie
from app.models.movie_playback_source import MoviePlaybackSource
from app.models.performer import Performer
from app.models.play_progress import MoviePlayProgress
from app.models.source import Source
from app.models.studio import Studio
from app.models.tag import Tag

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

_VALID_SORTS = {"created_at", "release_year", "release_date", "title", "rating"}


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


@router.get("", response_model=MovieListOut)
def list_movies(
    session: Annotated[Session, Depends(get_session)],
    device_id: Annotated[str, Depends(get_device_id)],
    q: str | None = Query(default=None, description="Title search (trgm)"),
    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)"),
    performer_ids: str | None = Query(default=None, description="Comma-separated performer UUIDs (AND)"),
    year_from: int | None = Query(default=None, ge=1900, le=2100),
    year_to: int | None = Query(default=None, ge=1900, le=2100),
    has_playback: bool | None = Query(default=None),
    sort: str = Query(default="created_at"),
    page: int = Query(default=1, ge=1),
    per_page: int = Query(default=50, ge=1, le=200),
) -> MovieListOut:
    if sort not in _VALID_SORTS:
        raise HTTPException(status_code=400, detail=f"sort must be one of {sorted(_VALID_SORTS)}")

    base = select(Movie)

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

    studio_slug_list = _split_csv(studio_slugs)
    if studio_slug_list:
        base = base.where(
            Movie.studio_id.in_(select(Studio.id).where(Studio.slug.in_(studio_slug_list)))
        )

    for slug in _split_csv(tags):
        base = base.where(
            exists(
                select(1).select_from(MovieTag).join(Tag, Tag.id == MovieTag.tag_id)
                .where(MovieTag.movie_id == Movie.id, Tag.slug == slug)
            )
        )

    perf_id_strings = _split_csv(performer_ids)
    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
        for pid in perf_ids:
            base = base.where(
                exists(
                    select(1).select_from(MoviePerformer).where(
                        MoviePerformer.movie_id == Movie.id,
                        MoviePerformer.performer_id == pid,
                    )
                )
            )

    if year_from is not None:
        base = base.where(Movie.release_year >= year_from)
    if year_to is not None:
        base = base.where(Movie.release_year <= year_to)

    if has_playback is True:
        base = base.where(
            exists(
                select(1).where(
                    MoviePlaybackSource.movie_id == Movie.id,
                    MoviePlaybackSource.dead_at.is_(None),
                )
            )
        )

    total = session.execute(
        select(func.count()).select_from(base.subquery())
    ).scalar_one()

    if sort == "created_at":
        base = base.order_by(Movie.created_at.desc())
    elif sort == "release_year":
        base = base.order_by(Movie.release_year.desc().nulls_last(), Movie.created_at.desc())
    elif sort == "release_date":
        base = base.order_by(Movie.release_date.desc().nulls_last(), Movie.created_at.desc())
    elif sort == "title":
        base = base.order_by(Movie.title_normalized.asc())
    elif sort == "rating":
        base = base.order_by(Movie.rating.desc().nulls_last(), Movie.created_at.desc())

    base = base.limit(per_page).offset((page - 1) * per_page)

    movies = list(session.execute(base).scalars().all())
    items = _movies_to_out(session, movies, device_id=device_id)

    return MovieListOut(items=items, total=total, page=page, per_page=per_page)


# Movie playback origin policy, module-level (kiedyś było inline per-request
# definition, code-review #19, perf hit + dorosły kod).
# Ranking ustalony ad-hoc 2026-05-09 (extract_stream_from_hoster na 5 sample
# random per origin).
_MOVIE_PREFERRED_ORIGINS = (
    "mangoporn:luluvid",  # KVS, działa
    "mangoporn:mixdrop",  # po domain fix może działać
    "mangoporn:voe",      # czasem yt-dlp łapie
    "mangoporn",
    "streamporn",
    "pandamovies",
)
# File hosters które NIGDY nie dadzą się stream-extract bez premium account,
# odfiltrowywane całkowicie (zaśmiecały listę watch options, bug-report
# 2026-05-15). Streamtape przywrócony 2026-05-15, ma dedicated extractor,
# ~5% URLów żyje.
_MOVIE_DROP_ORIGINS = frozenset({
    "mangoporn:rapidgator",
    "mangoporn:nitroflare",
    "mangoporn:frdl",
})
# Raw landing origins ukrywane gdy są sub-hosters (zob. komentarz w get_movie).
_MOVIE_LANDING_HIDE = frozenset({"mangoporn", "pandamovies", "streamporn"})


# Rodzina seekplayer (embed `#<hash>`, wspólny silnik) na sam koniec listy.
# Pomiar 2026-08-24 na 250 losowych źródłach: 3 miały jeszcze strumień, 88 oddało
# HTTP 404, a 79 odpowiedź bez pola `source`, po której ICH WŁASNY player pisze
# „Video is not ready yet" (sprawdzone w przeglądarce z łącza domowego, z VPS i przez
# brytyjskie proxy, więc to nie blokada po IP). Zgłoszenia `d59fc396`, `ccbc75da`,
# `7bfc2e8d` i sąsiednie z tego dnia to dokładnie te trafienia.
#
# Nie wyrzucamy ich z listy (te ~2% działa i bywa jedynym źródłem), ale nie mogą stać
# przed hosterami, które grają. Filmów zależnych WYŁĄCZNIE od tej rodziny jest w bazie
# jeden, więc de-ranking nikomu nie zabiera dostępu.
_MOVIE_DEPRIORITIZED_HOSTERS = frozenset({
    "easyvidplayer", "embedseek", "seekplayer", "upns", "player4me", "rpmplay",
    "4meplayer", "seekplays", "ezplayer", "seeks",
})


# Rodzina DoodStream (playmogo, doply, myvidplay i klony) wymaga KLIKANEGO Cloudflare
# Turnstile: widżet pojawia sie dopiero po tapie w play i przeladowaniu, wiec
# `pass_md5` nie ma w HTML i telefon nie rozwiaze tego sam (memory 2026-08-26, INJECTED_JS
# omija te hosty). Skutek w telemetrii: ZERO zdarzen odtwarzania w 30 dniach, bo droga
# WebView niczego nie raportuje, podczas gdy mixdrop/voe/luluvid/vidara graja w ~99%.
#
# Dotad te originy mialy priorytet neutralny (500) i sortowaly sie z dzialajacymi po dacie
# dodania, a doply/playmogo bywaja dopisane jako OSTATNIE, wiec ladowaly na GORZE listy
# "Watch on". Zgloszenia "Playmogo nie dziala tutaj" (1278eb49, 2026-09-04) i seria
# "Myvidplay nie dziala" (3b75cb54, d3af0813, 11d7b17b) to dokladnie ten mechanizm:
# zgloszony film mial obok zywy mixdrop, ktory gra.
#
# Pomiar 2026-09-05: rodzina siedzi w 54 446 filmach, z czego 54 320 (99,8%) ma dzialajaca
# alternatywe. Tylko 126 filmow zalezy od niej wylacznie, dla nich zostaje jako ostatnia
# deska ratunku, wiec de-ranking nikomu nie zabiera dostepu. Lista hostow = doodstream.ts.
_MOVIE_INTERACTIVE_HOSTERS = frozenset({
    "playmogo", "myvidplay", "doply", "doodstream", "dood", "d0000d", "dooood",
    "do0od", "do7go", "ds2play", "vikingfile", "vide0",
})


def _movie_origin_priority(origin: str) -> int:
    host = origin.rpartition(":")[2]
    if host in _MOVIE_DEPRIORITIZED_HOSTERS:
        return 900  # za wszystkim, co nieklasyfikowane
    if host in _MOVIE_INTERACTIVE_HOSTERS:
        return 800  # gra, ale tylko po recznym kliku w Turnstile; za wszystkim, co gra samo
    try:
        return _MOVIE_PREFERRED_ORIGINS.index(origin)
    except ValueError:
        return 500  # neutralne (paradisehill, mangoporn:* nieklasyfikowane)


@router.get("/{movie_id}", response_model=MovieOut)
def get_movie(
    movie_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
    device_id: Annotated[str, Depends(get_device_id)],
) -> MovieOut:
    movie = session.get(Movie, movie_id)
    if movie is None:
        raise HTTPException(status_code=404, detail="movie not found")
    return _movie_to_out(session, movie, device_id=device_id)


def _movie_to_out(session: Session, movie: Movie, *, device_id: str = LEGACY_DEVICE) -> MovieOut:
    return _movies_to_out(session, [movie], device_id=device_id)[0]


def _movies_to_out(
    session: Session, movies: list[Movie], *, device_id: str = LEGACY_DEVICE
) -> list[MovieOut]:
    """Serializacja wsadowa: po jednym zapytaniu na relacje dla calej strony listy.

    Wczesniej kazdy film robil 7 zapytan (studio, obsada, tagi, rozdzialy, refy,
    zrodla, ulubione + postep), czyli ~210 na strone 30 filmow i 350 ms na
    GET /movies (Sentry: N+1 Query list_movies, pomiar 2026-09-08). Logika per film
    jest ta sama co w starym `_movie_to_out`, tylko dane przychodza z map.
    """
    if not movies:
        return []
    ids = [m.id for m in movies]

    studio_ids = {m.studio_id for m in movies if m.studio_id}
    studios: dict = {}
    if studio_ids:
        studios = {
            st.id: st for st in session.execute(select(Studio).where(Studio.id.in_(studio_ids))).scalars()
        }

    perf_by: dict = defaultdict(list)
    for mid, perf, alias in session.execute(
        select(MoviePerformer.movie_id, Performer, MoviePerformer.as_alias)
        .join(Performer, MoviePerformer.performer_id == Performer.id)
        .where(MoviePerformer.movie_id.in_(ids))
        .order_by(MoviePerformer.position.asc().nulls_last())
    ).all():
        perf_by[mid].append((perf, alias))

    tags_by: dict = defaultdict(list)
    for mid, tag in session.execute(
        select(MovieTag.movie_id, Tag)
        .join(Tag, MovieTag.tag_id == Tag.id)
        .where(MovieTag.movie_id.in_(ids))
        .order_by(Tag.name.asc())
    ).all():
        tags_by[mid].append(tag)

    chapters_by: dict = defaultdict(list)
    for ch in session.execute(
        select(MovieChapter).where(MovieChapter.movie_id.in_(ids)).order_by(MovieChapter.chapter_index.asc())
    ).scalars():
        chapters_by[ch.movie_id].append(ch)

    refs_by: dict = defaultdict(list)
    for ref, name in session.execute(
        select(MovieExternalRef, Source.name)
        .join(Source, Source.id == MovieExternalRef.source_id)
        .where(MovieExternalRef.movie_id.in_(ids))
    ).all():
        refs_by[ref.movie_id].append((ref, name))

    pb_by: dict = defaultdict(list)
    for pb in session.execute(
        select(MoviePlaybackSource)
        .where(MoviePlaybackSource.movie_id.in_(ids))
        .where(MoviePlaybackSource.dead_at.is_(None))
        .order_by(MoviePlaybackSource.created_at.desc())
    ).scalars():
        pb_by[pb.movie_id].append(pb)

    favs = set(
        session.execute(
            select(FavoriteMovie.movie_id)
            .where(FavoriteMovie.device_id == device_id)
            .where(FavoriteMovie.movie_id.in_(ids))
        ).scalars()
    )
    progress_by = {
        pr.movie_id: pr
        for pr in session.execute(
            select(MoviePlayProgress)
            .where(MoviePlayProgress.device_id == device_id)
            .where(MoviePlayProgress.movie_id.in_(ids))
        ).scalars()
    }

    out: list[MovieOut] = []
    for movie in movies:
        studio = studios.get(movie.studio_id) if movie.studio_id else None
        studio_out = StudioOut.model_validate(studio) if studio is not None else None
        performers = [
            PerformerOut(
                id=perf.id,
                canonical_name=perf.canonical_name,
                slug=perf.slug,
                gender=perf.gender.value if perf.gender else None,
                as_alias=alias,
            )
            for perf, alias in perf_by.get(movie.id, [])
        ]
        tags = [TagOut.model_validate(t) for t in tags_by.get(movie.id, [])]
        chapters = [MovieChapterOut.model_validate(c) for c in chapters_by.get(movie.id, [])]
        external_refs = [
            ExternalRefOut(source=name, external_id=ref.external_id, url=ref.url, last_seen=ref.last_seen)
            for ref, name in refs_by.get(movie.id, [])
        ]
        pb_rows = list(pb_by.get(movie.id, []))
        pb_rows = [p for p in pb_rows if p.origin not in _MOVIE_DROP_ORIGINS]
        # Bug-report 2026-05-16: raw landing origins (`mangoporn`/`pandamovies`/
        # `streamporn` BEZ `:host`) otwierały WebView z reklamami pełnoekranowymi
        # i myliły usera. Ukrywamy raw landing GDY ten sam movie ma co najmniej
        # jeden sub-host entry (origin zawiera `:`). Jeśli movie nie ma sub-hosters
        # (bo theme HTML się zmienił lub regex nie złapał), zostawiamy landing jako
        # last-resort.
        has_subhost = any(":" in p.origin for p in pb_rows)
        if has_subhost:
            pb_rows = [p for p in pb_rows if p.origin not in _MOVIE_LANDING_HIDE]
        pb_rows = sorted(pb_rows, key=lambda p: _movie_origin_priority(p.origin))
        # Dedup po realnym celu: 3 mirrory dooplay (mangoporn/pandamovies/streamporn)
        # zapisują TEN SAM embed hostera jako osobne wiersze (np. luluvid/e/X z każdego
        # mirrora) → film miał ~100 linków, z czego ~połowa to ten sam film (bug-report
        # 41ca1fa4). Zostawiamy pierwszy = najwyższy priorytet (sort wyżej).
        seen_targets: set[str] = set()
        deduped: list[MoviePlaybackSource] = []
        for p in pb_rows:
            target = (p.embed_url or p.stream_url or p.page_url or "").strip()
            if target and target in seen_targets:
                continue
            if target:
                seen_targets.add(target)
            deduped.append(p)
        pb_rows = deduped
        playback_sources = [PlaybackSourceOut.model_validate(p) for p in pb_rows]

        progress = progress_by.get(movie.id)
        out.append(
            MovieOut(
                id=movie.id,
                title=movie.title,
                slug=movie.slug,
                release_year=movie.release_year,
                release_date=movie.release_date,
                duration_sec=movie.duration_sec,
                description=movie.description,
                director=movie.director,
                country=movie.country,
                rating=movie.rating,
                poster_url=movie.poster_url,
                backdrop_url=movie.backdrop_url,
                studio=studio_out,
                performers=performers,
                tags=tags,
                chapters=chapters,
                external_refs=external_refs,
                playback_sources=playback_sources,
                created_at=movie.created_at,
                is_favorite=movie.id in favs,
                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,
            )
        )
    return out


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


@router.post("/{keep_id}/merge/{drop_id}", response_model=MovieMergeOut)
def merge_duplicate_movie(
    keep_id: uuid.UUID,
    drop_id: uuid.UUID,
    session: Annotated[Session, Depends(get_session)],
) -> MovieMergeOut:
    """Scal `drop_id` w `keep_id` z apki (przytrzymanie filmu, "merge into this one"),
    jak przy scenach. Przenosi linki, obsade, tagi, rozdzialy i dane uzytkownikow
    (movie_merge), kasuje `drop`. Zgloszenie f8c9486e (2026-10-16): te same filmy z dwoch
    zrodel, ktorych automat nie zlaczyl ("Hardcore Gangbangs 8" i "... VIII")."""
    from app.resolve.movie_merge import MovieMergeError, merge_movies

    if keep_id == drop_id:
        raise HTTPException(status_code=400, detail="keep_id == drop_id")
    if session.get(Movie, keep_id) is None or session.get(Movie, drop_id) is None:
        raise HTTPException(status_code=404, detail="movie not found")
    try:
        merge_movies(session, keep_id=keep_id, drop_id=drop_id, resolved_by="user_long_press_duplicate")
    except MovieMergeError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    session.commit()
    return MovieMergeOut(keep_id=keep_id, dropped_id=drop_id)