goon/app/resolve/movie_merge.py

"""Scalanie dwóch kanonicznych movies w jeden (dedup).

`keep_id` przejmuje od `drop_id`: external_refs, performers, tags, chapters,
playback_sources, z deduplikacją na kluczach złączeń. Potem `drop` Movie jest
usuwany. Mirror `scene_merge.merge_scenes` dla movies.

Używane przez TPDB enrichment: gdy dwa nasze filmy (mirrory paradisehill/dooplay)
mapują się na TEN SAM TPDB movie UUID, to ten sam film → merge. `MergeKind.movie`
już istnieje w merge_candidates.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime

from sqlalchemy import or_, select, text, update
from sqlalchemy.orm import Session

from app.models.merge_candidate import MergeCandidate, MergeKind, MergeStatus
from app.models.movie import (
    Movie,
    MovieChapter,
    MovieExternalRef,
    MoviePerformer,
    MovieTag,
)
from app.models.movie_playback_source import MoviePlaybackSource

log = logging.getLogger(__name__)


class MovieMergeError(Exception):
    pass


def merge_movies(
    session: Session,
    *,
    keep_id: uuid.UUID,
    drop_id: uuid.UUID,
    resolved_by: str | None = None,
) -> Movie:
    if keep_id == drop_id:
        raise MovieMergeError("cannot merge movie into itself")
    keep = session.get(Movie, keep_id)
    drop = session.get(Movie, drop_id)
    if keep is None or drop is None:
        raise MovieMergeError("movie not found")

    _move_external_refs(session, keep_id=keep_id, drop_id=drop_id)
    _move_performers(session, keep_id=keep_id, drop_id=drop_id)
    _move_tags(session, keep_id=keep_id, drop_id=drop_id)
    _move_chapters(session, keep_id=keep_id, drop_id=drop_id)
    _move_playback_sources(session, keep_id=keep_id, drop_id=drop_id)
    _move_user_data(session, keep_id=keep_id, drop_id=drop_id)
    _coalesce_canonical_fields(keep, drop)

    session.delete(drop)
    session.flush()
    _close_pending_candidates(session, movie_id=drop_id, resolved_by=resolved_by)
    log.info("merged movie %s ← %s", keep_id, drop_id)
    return keep


# ---- helpery --------------------------------------------------------------

def _move_external_refs(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    for ref in session.execute(
        select(MovieExternalRef).where(MovieExternalRef.movie_id == drop_id)
    ).scalars().all():
        clash = session.execute(
            select(MovieExternalRef).where(
                MovieExternalRef.source_id == ref.source_id,
                MovieExternalRef.external_id == ref.external_id,
                MovieExternalRef.movie_id == keep_id,
            )
        ).scalar_one_or_none()
        if clash is not None:
            session.delete(ref)
        else:
            ref.movie_id = keep_id


def _move_performers(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    for link in session.execute(
        select(MoviePerformer).where(MoviePerformer.movie_id == drop_id)
    ).scalars().all():
        clash = session.execute(
            select(MoviePerformer).where(
                MoviePerformer.movie_id == keep_id,
                MoviePerformer.performer_id == link.performer_id,
            )
        ).scalar_one_or_none()
        if clash is not None:
            if link.as_alias and not clash.as_alias:
                clash.as_alias = link.as_alias
            session.delete(link)
        else:
            link.movie_id = keep_id


def _move_tags(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    for link in session.execute(
        select(MovieTag).where(MovieTag.movie_id == drop_id)
    ).scalars().all():
        clash = session.execute(
            select(MovieTag).where(
                MovieTag.movie_id == keep_id, MovieTag.tag_id == link.tag_id
            )
        ).scalar_one_or_none()
        if clash is not None:
            session.delete(link)
        else:
            link.movie_id = keep_id


def _move_chapters(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    """Chaptery są lokalne per-movie (UQ movie_id, chapter_index). Kolizja indeksu z
    keep → keep swoje (kasujemy drop's); brak kolizji → przepięcie."""
    for ch in session.execute(
        select(MovieChapter).where(MovieChapter.movie_id == drop_id)
    ).scalars().all():
        clash = session.execute(
            select(MovieChapter).where(
                MovieChapter.movie_id == keep_id,
                MovieChapter.chapter_index == ch.chapter_index,
            )
        ).scalar_one_or_none()
        if clash is not None:
            session.delete(ch)
        else:
            ch.movie_id = keep_id


def _move_playback_sources(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    """Unique (origin, page_url) jest GLOBALNY → drop i keep nie mogą współdzielić
    tego samego źródła, więc samo przepięcie movie_id nie grozi kolizją."""
    session.execute(
        update(MoviePlaybackSource)
        .where(MoviePlaybackSource.movie_id == drop_id)
        .values(movie_id=keep_id)
    )


def _move_user_data(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
    """Ulubione i postep odtwarzania z `drop` na `keep`.

    Oba maja FK `ondelete=CASCADE` na movies.id, wiec bez tego kroku skasowanie `drop`
    po cichu wymazywalo userowi ulubione i miejsce, w ktorym skonczyl ogladac (code review
    2026-09-02, sprzatanie dubli po rozdzielaniu filmow). Gdy urzadzenie ma juz wpis na
    `keep`, ten na `drop` zostaje i idzie z kaskada: nie ma czego laczyc.
    """
    for tabela in ("favorite_movies", "movie_play_progress"):
        session.execute(
            text(
                f"UPDATE {tabela} SET movie_id = :keep WHERE movie_id = :drop "
                f" AND device_id NOT IN (SELECT device_id FROM {tabela} WHERE movie_id = :keep)"
            ),
            {"keep": str(keep_id), "drop": str(drop_id)},
        )


def _coalesce_canonical_fields(keep: Movie, drop: Movie) -> None:
    """Wypełnij braki w `keep` polami z `drop` (bez nadpisywania ustawionych)."""
    if not keep.description and drop.description:
        keep.description = drop.description
    if not keep.duration_sec and drop.duration_sec:
        keep.duration_sec = drop.duration_sec
    if not keep.director and drop.director:
        keep.director = drop.director
    if not keep.country and drop.country:
        keep.country = drop.country
    if not keep.release_date and drop.release_date:
        keep.release_date = drop.release_date
    if not keep.release_year and drop.release_year:
        keep.release_year = drop.release_year
    if not keep.studio_id and drop.studio_id:
        keep.studio_id = drop.studio_id
    if not keep.poster_url and drop.poster_url:
        keep.poster_url = drop.poster_url
    if not keep.backdrop_url and drop.backdrop_url:
        keep.backdrop_url = drop.backdrop_url
    if keep.rating is None and drop.rating is not None:
        keep.rating = drop.rating
    if drop.title and len(drop.title) > len(keep.title or ""):
        keep.title = drop.title
        keep.title_normalized = drop.title_normalized
    # created_at = najwcześniejsze "first seen" (NEW badge correctness, jak w scenach).
    if drop.created_at and keep.created_at and drop.created_at < keep.created_at:
        keep.created_at = drop.created_at


def _close_pending_candidates(
    session: Session, *, movie_id: uuid.UUID, resolved_by: str | None
) -> None:
    session.execute(
        update(MergeCandidate)
        .where(
            MergeCandidate.kind == MergeKind.movie,
            MergeCandidate.status == MergeStatus.pending,
            or_(MergeCandidate.left_id == movie_id, MergeCandidate.right_id == movie_id),
        )
        .values(
            status=MergeStatus.rejected,
            resolved_at=datetime.now(UTC),
            resolved_by=resolved_by or "auto:movie_dropped",
        )
    )