From a16883d2f35335e74ab60296058af72744f7902c Mon Sep 17 00:00:00 2001 From: goon-foss Date: Thu, 13 Aug 2026 10:49:12 +0200 Subject: [PATCH] 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 `::`. --- .../20260813_0028_phash_backfill_queue.py | 47 +++++ app/config.py | 15 ++ app/models/__init__.py | 1 + app/models/phash_backfill_queue.py | 55 +++++ app/scheduler/jobs.py | 34 +++ app/scheduler/phash_backfill.py | 197 ++++++++++++++++++ app/scheduler/worker.py | 2 + 7 files changed, 351 insertions(+) create mode 100644 alembic/versions/20260813_0028_phash_backfill_queue.py create mode 100644 app/models/phash_backfill_queue.py create mode 100644 app/scheduler/phash_backfill.py diff --git a/alembic/versions/20260813_0028_phash_backfill_queue.py b/alembic/versions/20260813_0028_phash_backfill_queue.py new file mode 100644 index 0000000..792259b --- /dev/null +++ b/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") diff --git a/app/config.py b/app/config.py index e40b6d7..221f0b0 100644 --- a/app/config.py +++ b/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. diff --git a/app/models/__init__.py b/app/models/__init__.py index a13f681..c0c406d 100644 --- a/app/models/__init__.py +++ b/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 diff --git a/app/models/phash_backfill_queue.py b/app/models/phash_backfill_queue.py new file mode 100644 index 0000000..9dd1e7d --- /dev/null +++ b/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 + ) diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 4caf10d..ff8bf8f 100644 --- a/app/scheduler/jobs.py +++ b/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, diff --git a/app/scheduler/phash_backfill.py b/app/scheduler/phash_backfill.py new file mode 100644 index 0000000..4835f68 --- /dev/null +++ b/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 diff --git a/app/scheduler/worker.py b/app/scheduler/worker.py index 503a5f8..ebd9c7e 100644 --- a/app/scheduler/worker.py +++ b/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,