feat(dedup): backfill phasha z zapisanych miniaturek

Root cause zgloszenia 9d272101 ("5 duplikatow, czemu nie zmergowane?"):
phash mialo tylko 37,8% grywalnych scen. Licza go WYLACZNIE niektore
konektory w momencie ingestu, wiec najwieksze zrodla maja miniaturke dla
~100% scen, a odcisk dla 1-2% (xvideos 1,8%, xnxx 1,4%, eporner 1,0%).
Material jest zapisany, brakowalo tylko obliczenia.

W zglaszanym przypadku 7 wierszy o identycznej dlugosci 2324 s, a dwa
kafelki z IDENTYCZNA okladka nie mialy phasha wcale, wiec jedyny sygnal,
ktory by je sklail, nigdy nie powstal.

Zakres CELOWANY, nie caly katalog: sceny w klastrze (performer, dokladna
dlugosc) o licznosci >1, bez phasha, ale z zapisana miniaturka = 238 704
pozycje (wobec ~1,4 mln scen bez phasha). Reszta nie ma sie z czym skleic.

Kolejka jako TABELA, nie zapytanie w locie: wyznaczenie kandydatow to
ciezki agregat po calym katalogu, ktorego nie chcemy powtarzac przy kazdej
paczce, a kolejka przezywa restart workera i widac z niej postep.

Weryfikacja end-to-end na zglaszanej scenie: ce81a110 dostal phash
85c1bac2f8c60ecf, czyli DOKLADNIE ten sam, ktory mial juz 8373bbb5 (dwa
z kafelkow uzytkownika) -> istniejacy phash_exact je teraz sklei.

Nie zapisujemy wartosci z phash_blacklist (zaslepki dzielone przez setki
scen). Throttling swiadomy: 800 co 15 min, ~3 dni na nadrobienie.

Dwie pulapki SQL zapisane w kodzie: `referer` ma 255 znakow wiec origin
wycinamy w SQL (pelny page_url wywalal planowanie), a rzutowanie musi isc
przez CAST() bo text() SQLAlchemy nie podstawia parametru za ktorym stoi
`::`.

alembic/versions/20260813_0028_phash_backfill_queue.py

@@ -0,0 +1,47 @@
+"""phash_backfill_queue: kolejka scen do policzenia phasha z zapisanej miniaturki
+
+Uzasadnienie i liczby: `app/models/phash_backfill_queue.py`.
+
+Revision ID: 0028_phash_backfill_queue
+Revises: 0027_scenes_title_trgm
+Create Date: 2026-08-13
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0028_phash_backfill_queue"
+down_revision: str | None = "0027_scenes_title_trgm"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "phash_backfill_queue",
+        sa.Column("scene_id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("thumbnail_url", sa.String(length=1024), nullable=False),
+        sa.Column("referer", sa.String(length=255), nullable=True),
+        sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("done_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.ForeignKeyConstraint(["scene_id"], ["scenes.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("scene_id"),
+    )
+    # Jedyne zapytanie ścieżki gorącej: „weź N nieprzetworzonych". Indeks częściowy,
+    # bo przetworzone wiersze zostają w tabeli jako ślad, ale nigdy ich nie czytamy.
+    op.create_index(
+        "ix_phash_backfill_pending",
+        "phash_backfill_queue",
+        ["attempts"],
+        postgresql_where=sa.text("done_at IS NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_phash_backfill_pending", table_name="phash_backfill_queue")
+    op.drop_table("phash_backfill_queue")

app/config.py

@@ -123,6 +123,21 @@ class Settings(BaseSettings):
     sched_phash_blacklist_hours: int = Field(
         default=24, validation_alias="GOON_SCHED_PHASH_BLACKLIST_HOURS"
     )
+    # Doliczanie phasha scenom, które mają zapisaną miniaturkę, ale nie mają odcisku.
+    # Audyt 2026-08-13: phash miało tylko 37,8% grywalnych scen, bo liczą go WYŁĄCZNIE
+    # niektóre konektory przy ingeście (xvideos 1,8%, xnxx 1,4%, eporner 1,0%, przy
+    # ~100% pokryciu miniaturką). Bez odcisku dedup nie ma czym skleić duplikatów, patrz
+    # zgłoszenie 9d272101. Job leci często i małymi paczkami, bo to pobrania z cudzych
+    # CDN-ów, patrz throttling w `app/scheduler/phash_backfill.py`. 0 = off.
+    # 800 co 15 min = ~3200/h, czyli ~3 dni na nadrobienie 239 tys. zaległości. Paczka
+    # 400 zajmuje ~22 s (4 wątki), więc realnie worker śpi prawie cały ten czas, a
+    # pobrania rozkładają się na kilkanaście różnych CDN-ów, nie na jeden host.
+    sched_phash_backfill_minutes: int = Field(
+        default=15, validation_alias="GOON_SCHED_PHASH_BACKFILL_MINUTES"
+    )
+    sched_phash_backfill_batch: int = Field(
+        default=800, validation_alias="GOON_SCHED_PHASH_BACKFILL_BATCH"
+    )
     # Próg ciszy dla konektorów FILMOWYCH. Wcześniej filmy nie były pokryte watchdogiem
     # w ogóle, streamporn.vip stał 23 dni i nic nie krzyknęło. Filmy ingestujemy
     # codziennie, więc 72h to bezpieczny margines na jeden nieudany cykl.

app/models/__init__.py

@@ -20,6 +20,7 @@ from app.models.performer import Performer, PerformerAlias, PerformerExternalRef
 from app.models.play_progress import MoviePlayProgress, ScenePlayProgress
 from app.models.playback_event import PlaybackEvent
 from app.models.playback_source import PlaybackSource
+from app.models.phash_backfill_queue import PhashBackfillQueue
 from app.models.phash_blacklist import PhashBlacklist
 from app.models.saved_search import SavedSearch
 from app.models.source_stats import SourceStats

app/models/phash_backfill_queue.py

@@ -0,0 +1,55 @@
+"""Kolejka scen do policzenia phasha z ZAPISANEJ już miniaturki.
+
+Phash liczą wyłącznie niektóre konektory, w momencie ingestu. Reszta zapisuje
+`thumbnail_url`, ale odcisku nie wylicza, więc audyt 2026-08-13 pokazał **37,8%**
+pokrycia wśród grywalnych scen. Największe źródła mają miniaturkę dla ~100% scen,
+a phash dla 1-2% (xvideos 1,8%, xnxx 1,4%, eporner 1,0%). Materiał jest, brakuje
+tylko obliczenia.
+
+Skutek widać w zgłoszeniu `9d272101` (strona Tiffani Time): siedem wierszy o
+identycznej długości 2324 s, ta sama scena. Dwa kafelki miały IDENTYCZNĄ okładkę,
+ale **żaden z nich nie miał phasha**, więc jedyny sygnał, który by je skleił, nigdy
+nie powstał.
+
+**Dlaczego kolejka jest tabelą, a nie zapytaniem liczonym w locie.** Wyznaczenie
+kandydatów (klaster performer + dokładna długość, liczność > 1) to ciężki agregat po
+całym katalogu, którego nie chcemy powtarzać przy każdej paczce. Liczymy go raz przy
+`plan`, a `process` bierze potem tanie `LIMIT` po indeksie. Przy okazji kolejka
+przeżywa restart workera i widać z niej postęp.
+
+**Zakres celowany, nie cały katalog.** Kandydaci to sceny, które w ogóle mają z czym
+się skleić: 241 763 pozycje wobec ~1,4 mln scen bez phasha. Reszta może go nie mieć,
+bo i tak nie ma pary.
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base
+
+
+class PhashBackfillQueue(Base):
+    __tablename__ = "phash_backfill_queue"
+
+    scene_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("scenes.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    #: Miniaturka zapisana przy ingeście; nie odpytujemy o nią strony ponownie.
+    thumbnail_url: Mapped[str] = mapped_column(String(1024), nullable=False)
+    #: Bez Referera część CDN-ów oddaje 403 (ta sama przyczyna co przy `/proxy/img`).
+    referer: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    #: Rośnie przy każdej nieudanej próbie; po `_MAX_ATTEMPTS` wiersz jest pomijany,
+    #: żeby martwy CDN nie blokował kolejki w nieskończoność.
+    attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0")
+    #: Ustawione po sukcesie ALBO po wyczerpaniu prób (wtedy `phash` został pusty).
+    done_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )

app/scheduler/jobs.py

@@ -276,6 +276,22 @@ def _job_thumb_asset_dedup() -> None:
         log.exception("[scheduler] thumb-asset dedup failed")
 
 
+def _job_phash_backfill(batch: int) -> None:
+    """Dolicza phash scenom, które mają miniaturkę, ale nie mają odcisku. Kolejkę
+    planujemy raz (`plan_phash_backfill`), tu tylko mielimy paczkami, bo to pobrania
+    z cudzych CDN-ów. Gdy kolejka pusta, job jest no-opem i nic nie kosztuje."""
+    try:
+        from app.scheduler.phash_backfill import run_phash_backfill
+
+        out = _run_with_timeout(
+            lambda: run_phash_backfill(batch=batch), label="phash-backfill"
+        )
+        if out and out.get("taken"):
+            log.info("[scheduler] phash backfill %s", out)
+    except Exception:
+        log.exception("[scheduler] phash backfill failed")
+
+
 def _job_phash_blacklist() -> None:
     """Wyłapuje phashe, które przestały cokolwiek znaczyć (zaślepka / czarna klatka /
     intro studia) i kasuje je z fingerprintów. Audyt 2026-07-27: 73 takie wartości przy
@@ -516,6 +532,24 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler:
         )
         log.info("scheduler: phash blacklist every %dh", cfg["phash_blacklist_hours"])
 
+    if cfg.get("phash_backfill_minutes"):
+        _batch = int(cfg.get("phash_backfill_batch") or 400)
+        sched.add_job(
+            _job_phash_backfill,
+            IntervalTrigger(
+                minutes=cfg["phash_backfill_minutes"], start_date=INTERVAL_ANCHOR
+            ),
+            args=[_batch],
+            id="phash_backfill",
+            replace_existing=True,
+            max_instances=1,
+            coalesce=True,
+        )
+        log.info(
+            "scheduler: phash backfill every %dm, batch %d",
+            cfg["phash_backfill_minutes"], _batch,
+        )
+
     if cfg.get("junk_performers_hours"):
         sched.add_job(
             _job_junk_performers,

app/scheduler/phash_backfill.py

@@ -0,0 +1,197 @@
+"""Dolicza phash scenom, które mają zapisaną miniaturkę, ale nie mają odcisku.
+
+Kontekst, liczby i powód istnienia kolejki: `app/models/phash_backfill_queue.py`.
+
+Dwa kroki, świadomie rozdzielone:
+
+- `plan_phash_backfill()` — wyznacza kandydatów (ciężki agregat po katalogu, ~minuty)
+  i wrzuca ich do kolejki. Idempotentny, `ON CONFLICT DO NOTHING`, więc można go
+  puszczać ponownie, gdy dojdą nowe sceny.
+- `run_phash_backfill()` — bierze paczkę z kolejki, pobiera miniaturki, liczy phash.
+  To jedyna część, która chodzi cyklicznie.
+
+**Throttling jest tu istotny, nie kosmetyczny.** To 242 tysiące pobrań z cudzych CDN-ów.
+Idziemy wąskim strumieniem (`_CONCURRENCY`) z przerwą między paczkami, żeby nie wyglądać
+jak atak i nie ściągnąć bana na IP VPS-a, z którego lecą też scrapery. Domyślny budżet
+paczki jest mały celowo: lepiej, żeby to trwało dniami w tle, niż żeby zablokowało worker.
+
+**Nie zapisujemy wartości z `phash_blacklist`.** To zaślepki i czarne klatki dzielone
+przez setki niepowiązanych scen (rekordzistka: 892 sceny o 892 różnych tytułach).
+Wpisanie ich tutaj wprowadzałoby dokładnie ten szum, który tamta tabela wycina.
+"""
+from __future__ import annotations
+
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor
+from urllib.parse import urlsplit
+
+from sqlalchemy import text
+
+from app.connectors.direct_scrapers._browse_base import compute_thumbnail_phash
+
+log = logging.getLogger(__name__)
+
+#: Ile miniaturek naraz. Nisko celowo, patrz throttling w docstringu.
+_CONCURRENCY = 4
+#: Domyślny rozmiar paczki na jedno uruchomienie.
+_DEFAULT_BATCH = 500
+#: Po tylu nieudanych próbach odpuszczamy scenę (martwy CDN, 404, format nie do odczytu).
+_MAX_ATTEMPTS = 3
+
+#: Kandydaci: scena w klastrze (performer, DOKŁADNA długość) o liczności > 1, czyli
+#: taka, która w ogóle ma z czym się skleić, bez phasha, ale z zapisaną miniaturką.
+_PLAN_SQL = text(
+    """
+    INSERT INTO phash_backfill_queue (scene_id, thumbnail_url, referer)
+    SELECT DISTINCT ON (s.id)
+           s.id,
+           ps.thumbnail_url,
+           -- SAM origin, nie pełny `page_url`: kolumna ma 255 znaków, a URL-e scen
+           -- bywają dłuższe (pierwsza wersja wywalała się na tym przy planowaniu).
+           -- CDN i tak sprawdza domenę odsyłającą, nie ścieżkę.
+           substring(ps.page_url from '^https?://[^/]+') || '/'
+    FROM scenes s
+    JOIN scene_performers sp ON sp.scene_id = s.id
+    JOIN (
+        SELECT sp2.performer_id, s2.duration_sec
+        FROM scenes s2
+        JOIN scene_performers sp2 ON sp2.scene_id = s2.id
+        WHERE s2.duration_sec >= 60
+          AND EXISTS (
+              SELECT 1 FROM playback_sources p2
+              WHERE p2.scene_id = s2.id AND p2.dead_at IS NULL
+          )
+        GROUP BY 1, 2
+        HAVING count(*) > 1
+    ) k ON k.performer_id = sp.performer_id AND k.duration_sec = s.duration_sec
+    JOIN playback_sources ps ON ps.scene_id = s.id
+                            AND ps.dead_at IS NULL
+                            AND ps.thumbnail_url IS NOT NULL
+    WHERE NOT EXISTS (
+        SELECT 1 FROM scene_fingerprints f
+        WHERE f.scene_id = s.id AND f.kind = 'phash'
+    )
+    ON CONFLICT (scene_id) DO NOTHING
+    """
+)
+
+_TAKE_SQL = text(
+    """
+    SELECT scene_id, thumbnail_url, referer
+    FROM phash_backfill_queue
+    WHERE done_at IS NULL AND attempts < :max_attempts
+    ORDER BY attempts, scene_id
+    LIMIT :limit
+    """
+)
+
+#: Wstawiamy odcisk tylko, gdy scena wciąż go nie ma (mogła dostać w międzyczasie
+#: z ingestu) i gdy wartość nie jest zablacklistowana.
+_INSERT_SQL = text(
+    """
+    INSERT INTO scene_fingerprints (scene_id, kind, value)
+    -- `kind` to enum `fingerprint_kind`, surowy string wymaga rzutowania. `:value`
+    -- też: występuje i w SELECT, i w porównaniu z blacklistą, a bez rzutowania Postgres
+    -- wnioskuje dla niego dwa różne typy (text vs varchar) i odmawia wykonania.
+    -- Rzutujemy przez CAST(), nie `::`, bo `text()` SQLAlchemy nie podstawia parametru,
+    -- za którym stoi `::` (leci wtedy surowe `:value` i Postgres wywala składnię).
+    SELECT CAST(:scene_id AS uuid), 'phash'::fingerprint_kind, CAST(:value AS varchar)
+    WHERE NOT EXISTS (
+        SELECT 1 FROM scene_fingerprints f
+        WHERE f.scene_id = CAST(:scene_id AS uuid) AND f.kind = 'phash'
+    )
+    AND NOT EXISTS (
+        SELECT 1 FROM phash_blacklist b WHERE b.value = CAST(:value AS varchar)
+    )
+    ON CONFLICT DO NOTHING
+    """
+)
+
+
+def _referer_origin(page_url: str | None) -> str | None:
+    """Sam origin. Kolejka trzyma już origin (wycięty w `_PLAN_SQL`), więc to zabezpieczenie
+    na wiersze wstawione inaczej; dla gotowego originu zwraca go bez zmian."""
+    if not page_url:
+        return None
+    try:
+        parts = urlsplit(page_url)
+        if parts.scheme and parts.netloc:
+            return f"{parts.scheme}://{parts.netloc}/"
+    except ValueError:
+        pass
+    return None
+
+
+def plan_phash_backfill() -> int:
+    """Dopisz kandydatów do kolejki. Zwraca rozmiar kolejki po planowaniu."""
+    from app.db import session_scope
+
+    with session_scope() as session:
+        session.execute(_PLAN_SQL)
+        total = session.execute(
+            text("SELECT count(*) FROM phash_backfill_queue WHERE done_at IS NULL")
+        ).scalar_one()
+    log.info("phash-backfill: w kolejce %d scen", total)
+    return int(total)
+
+
+def run_phash_backfill(*, batch: int = _DEFAULT_BATCH, pause_sec: float = 0.0) -> dict[str, int]:
+    """Przetwórz jedną paczkę. Zwraca liczniki: wzięte/policzone/zapisane/nieudane."""
+    from app.db import session_scope
+
+    with session_scope() as session:
+        rows = session.execute(
+            _TAKE_SQL, {"limit": batch, "max_attempts": _MAX_ATTEMPTS}
+        ).all()
+
+    if not rows:
+        return {"taken": 0, "hashed": 0, "stored": 0, "failed": 0}
+
+    def _work(row) -> tuple:
+        scene_id, thumb, page_url = row
+        value = compute_thumbnail_phash(thumb, referer=_referer_origin(page_url))
+        return scene_id, value
+
+    results: list[tuple] = []
+    with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
+        for res in pool.map(_work, rows):
+            results.append(res)
+
+    hashed = stored = failed = 0
+    with session_scope() as session:
+        for scene_id, value in results:
+            if value:
+                hashed += 1
+                stored += int(
+                    session.execute(
+                        _INSERT_SQL, {"scene_id": scene_id, "value": value}
+                    ).rowcount
+                    or 0
+                )
+                session.execute(
+                    text(
+                        "UPDATE phash_backfill_queue SET done_at = now() WHERE scene_id = :sid"
+                    ),
+                    {"sid": scene_id},
+                )
+            else:
+                failed += 1
+                # Wyczerpane próby → domknij wiersz, żeby nie wracał w kółko.
+                session.execute(
+                    text(
+                        """
+                        UPDATE phash_backfill_queue
+                        SET attempts = attempts + 1,
+                            done_at = CASE WHEN attempts + 1 >= :max THEN now() ELSE NULL END
+                        WHERE scene_id = :sid
+                        """
+                    ),
+                    {"sid": scene_id, "max": _MAX_ATTEMPTS},
+                )
+
+    out = {"taken": len(rows), "hashed": hashed, "stored": stored, "failed": failed}
+    log.info("phash-backfill: %s", out)
+    if pause_sec:
+        time.sleep(pause_sec)
+    return out

app/scheduler/worker.py

@@ -219,6 +219,8 @@ def run_forever() -> int:
         # Blacklista zdegenerowanych phashy, zaślepki dzielone przez setki scen (audyt
         # 2026-07-27). Bez tego zaślepka jest zawsze najbliższym trafieniem w Path 3.
         "phash_blacklist_hours": getattr(settings, "sched_phash_blacklist_hours", 24) or None,
+        "phash_backfill_minutes": getattr(settings, "sched_phash_backfill_minutes", 20) or None,
+        "phash_backfill_batch": getattr(settings, "sched_phash_backfill_batch", 400),
         # Śmieciowi performerzy, kategorie/studia udające osoby (audyt 2026-07-27).
         # Kryterium wąskie celowo: szerokie kasuje realnych ludzi bez refa w TPDB.
         "junk_performers_hours": getattr(settings, "sched_junk_performers_hours", 24) or None,