goon/app/scheduler/phash_backfill.py

"""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