goon/scripts/remerge_movies_roman_studio.py

"""Jednorazowy dedup filmow po poprawkach oceny z 2026-10-16 (zgloszenie f8c9486e).

Ocena (movie_score) liczy teraz cyfry rzymskie jak arabskie ("Hardcore Gangbangs VIII"
= "... 8") i uznaje za zgodne studia o tej samej nazwie bez spacji ("EvilAngel" z xtapes
i "Evil Angel"). Istniejace pary, ktore ingest wtedy rozdzielil, zostaja rozdzielone,
wiec ten skrypt przechodzi tylko po parach, na ktore poprawka ma wplyw, i scala te z
ocena >= AUTO_THRESHOLD. Pary odrzucone przez czlowieka (merge_candidates.rejected)
pomija. Zostaje starszy film (jego tytul i id), mlodszy jest wlewany.

  python3 /srv/scripts/remerge_movies_roman_studio.py            # dry-run
  python3 /srv/scripts/remerge_movies_roman_studio.py --apply
"""
from __future__ import annotations

import argparse
import uuid

from sqlalchemy import select, text

from app.db import session_scope
from app.models.merge_candidate import MergeCandidate, MergeKind, MergeStatus
from app.models.movie import Movie, MoviePerformer
from app.resolve.movie_match import find_movie_candidates
from app.resolve.movie_merge import MovieMergeError, merge_movies
from app.resolve.movie_score import AUTO_THRESHOLD, _roman_to_arabic, score_movie_candidate


def _affected_ids(session) -> list[uuid.UUID]:
    """Filmy z cyfra rzymska w tytule albo ze studiem, ktore ma blizniaka bez spacji."""
    rows = session.execute(text("""
        WITH twins AS (
            SELECT a.id FROM studios a JOIN studios b
              ON a.id <> b.id
             AND replace(a.name_normalized, ' ', '') = replace(b.name_normalized, ' ', '')
             AND length(replace(a.name_normalized, ' ', '')) >= 4
        )
        SELECT m.id FROM movies m
        WHERE m.title_normalized ~ '(^| )(x{0,3})(ix|iv|v?i{0,3})( |$)'
           OR m.studio_id IN (SELECT id FROM twins)
        ORDER BY m.created_at DESC
    """)).scalars().all()
    return list(rows)


def _rejected_pairs(session) -> set[frozenset]:
    rows = session.execute(
        select(MergeCandidate.left_id, MergeCandidate.right_id).where(
            MergeCandidate.kind == MergeKind.movie,
            MergeCandidate.status == MergeStatus.rejected,
        )
    ).all()
    return {frozenset((a, b)) for a, b in rows}


def _keep_rank(session, movie: Movie) -> tuple:
    """Zostaje film z rokiem, potem z liczniejsza obsada, potem starszy. Sam wiek dawal
    np. "Hard Core Gangbangs 8" bez roku zamiast "Hardcore Gangbangs VIII" z TPDB."""
    cast = session.execute(
        text("SELECT count(*) FROM movie_performers WHERE movie_id = :m"), {"m": movie.id}
    ).scalar_one()
    return (movie.release_year is not None, cast, -movie.created_at.timestamp())


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--apply", action="store_true")
    ap.add_argument("--limit", type=int, default=0)
    args = ap.parse_args()

    with session_scope() as s:
        ids = _affected_ids(s)
        rejected = _rejected_pairs(s)
    if args.limit:
        ids = ids[: args.limit]
    print(f"filmy do sprawdzenia: {len(ids)}", flush=True)

    gone: set[uuid.UUID] = set()
    merged = 0
    for n, mid in enumerate(ids, 1):
        if n % 1000 == 0:
            print(f"... {n}/{len(ids)}, par: {merged}", flush=True)
        if mid in gone:
            continue
        with session_scope() as s:
            m = s.get(Movie, mid)
            if m is None or not m.title_normalized:
                continue
            perfs = [
                r.performer_id
                for r in s.execute(
                    MoviePerformer.__table__.select().where(MoviePerformer.movie_id == m.id)
                ).fetchall()
            ]
            best, best_b = None, None
            for c in find_movie_candidates(
                s, title_normalized=m.title_normalized, studio_id=m.studio_id, release_year=m.release_year
            ):
                if c.id == m.id or c.id in gone or frozenset((c.id, m.id)) in rejected:
                    continue
                roman = _roman_to_arabic(c.title_normalized or "") != (c.title_normalized or "") or (
                    _roman_to_arabic(m.title_normalized) != m.title_normalized
                )
                twin_studio = bool(c.studio_id and m.studio_id and c.studio_id != m.studio_id)
                if not (roman or twin_studio):
                    continue  # ta para wyglada dla oceny tak samo jak przy ingescie
                b = score_movie_candidate(
                    s,
                    candidate=c,
                    norm_title_normalized=m.title_normalized,
                    norm_release_year=m.release_year,
                    norm_studio_id=m.studio_id,
                    norm_performer_ids=perfs,
                )
                if b.composite >= AUTO_THRESHOLD and (best_b is None or b.composite > best_b.composite):
                    best, best_b = c, b
            if best is None:
                continue
            keep, drop = sorted((best, m), key=lambda x: _keep_rank(s, x), reverse=True)
            print(
                f"{best_b.composite:.2f}  ZOSTAJE {keep.title!r} ({keep.release_year})"
                f"  <=  {drop.title!r} ({drop.release_year})",
                flush=True,
            )
            if args.apply:
                try:
                    merge_movies(s, keep_id=keep.id, drop_id=drop.id, resolved_by="remerge_roman_studio_20261016")
                except MovieMergeError as e:
                    print("  blad:", e, flush=True)
                    continue
            gone.add(drop.id)
            merged += 1
    print(f"{'scalono' if args.apply else 'do scalenia'}: {merged}", flush=True)


if __name__ == "__main__":
    main()