"""Merge wysoko-pewnych missing-merge duplikatów (exact phash + same duration + shared
performer/title).

Problem (bug-report 2026-06-03 "ten sam czas, ta sama miniaturka, czemu się nie
mergują"): duplikaty scen nie zmergowane przy ingeście. Exact-phash sam w sobie jest
hałaśliwy (95% to kolizje na wspólnych miniaturkach/intro — różne sceny), więc bulk_dedup
scorer słusznie ich nie auto-merguje. ALE podzbiór z DODATKOWO tą samą długością (±3s)
ORAZ wspólnym performerem lub tytułem to niemal pewny ten sam scene.

Same-duration jest kluczowe: wyklucza wzorzec false-merge (tam długości się RÓŻNIĄ —
krótki klip vs pełna scena; patrz audit_false_merges). Tu długości są równe → realny dup.

UWAGA: merge KASUJE zdublowaną scenę (refs/sources/tags łączone do keepera) —
NIEODWRACALNE. Domyślnie dry-run; --yes wykonuje. Per-para re-fetch → obsługuje klastry
(scena znika po wcześniejszym merge'u w tym samym klastrze).

**Bezpiecznik na pary drugiego rzędu (dodany 2026-07-27).** Merge PRZENOSI odciski na
keepera, więc scalona scena zbiera phashe wszystkich wchłoniętych — jedna miała ich 14.
Taki konglomerat zaczyna potem kolidować z obcymi scenami przez odziedziczony phash,
który nie przedstawia nawet jego treści. Po przebiegu na 4647 parach zostało 7 nowych,
POWSTAŁYCH przez ten przebieg, i 5 z nich (71%) było fałszywkami: Brazzers vs Reality
Kings, AdultTime vs Teen Curves. W parach pierwszego rzędu było 0 fałszywek na 18.

Rozdziela je podobieństwo tytułu, i to z dużym marginesem — zmierzone na tych 7 parach:
prawdziwe duplikaty 92,1 i 84,8, wszystkie fałszywki 23,0-52,4. Stąd próg 0,70, który
leży w środku luki. Prefiks studia progu nie rusza, bo `token_set_ratio` traktuje
„Bursting The Bag" jako podzbiór „FakehubOriginals – Sophia Locke – Bursting The Bag".

Uruchomienie:
  python -m scripts.merge_phash_exact_dupes          # dry-run
  python -m scripts.merge_phash_exact_dupes --yes     # wykonaj
"""
from __future__ import annotations

import argparse

from rapidfuzz import fuzz
from sqlalchemy import text

from app.db import session_scope
from app.models.scene import Scene
from app.resolve.scene_merge import merge_scenes
from app.scheduler.bulk_dedup import _pick_keep_drop

# Patrz docstring: zmierzona luka to 52,4 (najgorsza fałszywka) ↔ 84,8 (najsłabszy
# prawdziwy duplikat). 0.70 siedzi w jej środku.
_MIN_TITLE_SIM = 0.70

# UWAGA (2026-07-27): bez wykluczenia `phash_blacklist` ten self-join jest NIEBEZPIECZNY.
# Jedna zaślepka była dzielona przez 892 sceny → sam ten klaster daje ~397 tys. par, a
# każde dwie z nich o zbliżonej długości i wspólnym performerze zostałyby scalone, mimo
# że to zupełnie różne sceny. Blacklista musi być odświeżona PRZED uruchomieniem:
#   python -c "from app.scheduler.phash_blacklist import run_phash_blacklist; run_phash_blacklist()"
_SAFE_PAIRS_SQL = """
SELECT DISTINCT a.scene_id AS sa, b.scene_id AS sb
FROM scene_fingerprints a
JOIN scene_fingerprints b ON a.value = b.value AND a.scene_id < b.scene_id
JOIN scenes sca ON sca.id = a.scene_id
JOIN scenes scb ON scb.id = b.scene_id
WHERE a.kind = 'phash' AND b.kind = 'phash'
  AND NOT EXISTS (SELECT 1 FROM phash_blacklist bl WHERE bl.value = a.value)
  AND sca.duration_sec IS NOT NULL AND scb.duration_sec IS NOT NULL
  AND abs(sca.duration_sec - scb.duration_sec) <= 3
  AND (
    lower(coalesce(sca.title, 'x')) = lower(coalesce(scb.title, 'y'))
    OR EXISTS (
      SELECT 1 FROM scene_performers pa
      JOIN scene_performers pb ON pa.performer_id = pb.performer_id
      WHERE pa.scene_id = a.scene_id AND pb.scene_id = b.scene_id
    )
  )
"""


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--yes", action="store_true", help="wykonaj merge (bez tego dry-run)")
    args = ap.parse_args()

    with session_scope() as s:
        pairs = [(r[0], r[1]) for r in s.execute(text(_SAFE_PAIRS_SQL))]
    print(f"safe high-confidence duplicate pairs: {len(pairs)}")

    merged = skipped = rejected = 0
    for sa, sb in pairs:
        with session_scope() as s:
            a = s.get(Scene, sa)
            b = s.get(Scene, sb)
            if a is None or b is None or a.id == b.id:
                skipped += 1  # już zmergowane w tym klastrze
                continue
            # Bezpiecznik na pary drugiego rzędu (patrz docstring): sam wspólny phash
            # nie wystarcza, gdy któraś strona jest konglomeratem po wcześniejszym
            # merge'u i niesie odziedziczone odciski obcych scen.
            sim = fuzz.token_set_ratio(a.title_normalized or "", b.title_normalized or "") / 100.0
            if sim < _MIN_TITLE_SIM:
                rejected += 1
                print(f"  SKIP (tytuł {sim:.2f}) '{(a.title or '')[:34]}' vs '{(b.title or '')[:34]}'")
                continue
            keep, drop = _pick_keep_drop(s, a, b)
            print(f"  {'MERGE' if args.yes else '[dry] MERGE'} keep={str(keep.id)[:8]} "
                  f"'{(keep.title or '')[:32]}' ({keep.duration_sec}s) <- drop={str(drop.id)[:8]} "
                  f"'{(drop.title or '')[:32]}'")
            if args.yes:
                merge_scenes(s, keep_id=keep.id, drop_id=drop.id, resolved_by="phash_exact_safe")
                s.commit()
                merged += 1

    if args.yes:
        print(f"\nAPPLIED: merged={merged} skipped(already-gone)={skipped} "
              f"rejected(tytuł<{_MIN_TITLE_SIM})={rejected}")
    else:
        print(f"\n(dry-run — {len(pairs)} par, z tego {rejected} odrzuconych na tytule; "
              f"uruchom z --yes aby scalić. NIEODWRACALNE)")


if __name__ == "__main__":
    main()