diff --git a/alembic/versions/20260820_0031_duration_backfill_queue.py b/alembic/versions/20260820_0031_duration_backfill_queue.py new file mode 100644 index 0000000..1f46e08 --- /dev/null +++ b/alembic/versions/20260820_0031_duration_backfill_queue.py @@ -0,0 +1,55 @@ +"""duration_backfill_queue: kolejka scen do doliczenia dlugosci ze strony tuba + +Uzasadnienie i liczby: `app/models/duration_backfill_queue.py`. + +Revision ID: 0031_duration_backfill +Revises: 0030_scenes_quality_idx +Create Date: 2026-08-20 +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# Uwaga: `alembic_version.version_num` to varchar(32), dluzsze id nie wejdzie. +revision: str = "0031_duration_backfill" +down_revision: str | None = "0030_scenes_quality_idx" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "duration_backfill_queue", + sa.Column("scene_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("page_url", sa.String(length=1024), nullable=False), + sa.Column("origin", sa.String(length=64), nullable=False), + 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", + name="fk_duration_backfill_queue_scene_id_scenes", + ), + sa.PrimaryKeyConstraint("scene_id", name="pk_duration_backfill_queue"), + ) + # Paczka bierze wiersze niezrobione, po originie (zeby nie zawalic jednego hosta). + op.create_index( + "ix_duration_backfill_queue_pending", + "duration_backfill_queue", + ["origin", "attempts"], + postgresql_where=sa.text("done_at IS NULL"), + ) + + +def downgrade() -> None: + op.drop_index( + "ix_duration_backfill_queue_pending", table_name="duration_backfill_queue" + ) + op.drop_table("duration_backfill_queue") diff --git a/app/config.py b/app/config.py index 6d89214..25c5471 100644 --- a/app/config.py +++ b/app/config.py @@ -138,6 +138,16 @@ class Settings(BaseSettings): sched_phash_backfill_batch: int = Field( default=800, validation_alias="GOON_SCHED_PHASH_BACKFILL_BATCH" ) + # Doliczanie brakujacej dlugosci scen (patrz app/models/duration_backfill_queue.py). + # 500 stron co 5 min to ~144 tys. dziennie, wiec kolejka 846 tys. schodzi w ok. 6 dni. + # To ~1,7 zapytania na sekunde rozlozone na trzy duze tuby, czyli ruch, ktorego one + # nie zauwaza. Job ma 30-minutowy limit, a paczka schodzi w ok. 3 min, wiec z zapasem. + sched_duration_backfill_minutes: int = Field( + default=5, validation_alias="GOON_SCHED_DURATION_BACKFILL_MINUTES" + ) + sched_duration_backfill_batch: int = Field( + default=500, validation_alias="GOON_SCHED_DURATION_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/extractors/duration_extract.py b/app/extractors/duration_extract.py index 880c4c8..6fe8ded 100644 --- a/app/extractors/duration_extract.py +++ b/app/extractors/duration_extract.py @@ -89,3 +89,68 @@ def extract_duration_sec(html: str) -> int | None: return total return None + + +# --- Wzorce punktowe, per origin -------------------------------------------------- +# +# `extract_duration_sec` jest generyczny i tym samym łatwowierny. Na stronach, które +# wypisują długości filmów z paska "powiązane", trafia w cudzą wartość, a na porntreksie +# w `duration: '5'` reklamy. Backfill katalogu (`app/scheduler/duration_backfill.py`) +# używa więc wyłącznie wzorców z potwierdzoną, punktową kotwicą. + +# xhamster trzyma prawdziwą długość w `videoEntity`, obok id zgodnego z URL-em. Sprawdzone +# 2026-08-20: dla `.../macy-gray-i-try-pmv-10506445` videoEntity daje 240 s, a naiwne +# szukanie najbliższego `"duration"` dawało 198 s z filmu obok. +_XH_ENTITY_RE = re.compile( + r'"videoEntity"\s*:\s*\{(?P.{0,600}?)\}', re.DOTALL +) +_XH_ID_RE = re.compile(r'"id"\s*:\s*(\d+)') +_XH_DUR_RE = re.compile(r'"duration"\s*:\s*(\d+)') +# id filmu jest na końcu sluga: `/videos/-`. +_XH_URL_ID_RE = re.compile(r"-(\d{4,})(?:[/?#]|$)") + +# Górna granica zdrowego rozsądku: 6 h. Wyżej to prawie na pewno zła wartość +# (sklejony live albo pomyłka jednostek), a długość idzie potem do dedupu. +_MAX_SANE_DURATION_SEC = 6 * 3600 + + +def _extract_xhamster(html: str, page_url: str) -> int | None: + want = _XH_URL_ID_RE.search(page_url) + for m in _XH_ENTITY_RE.finditer(html): + body = m.group("body") + vid = _XH_ID_RE.search(body) + dur = _XH_DUR_RE.search(body) + if not dur: + continue + # Bez zgodności id wolimy nie zapisać nic, niż zapisać długość obcego filmu. + if want and vid and vid.group(1) != want.group(1): + continue + n = int(dur.group(1)) + if 0 < n <= _MAX_SANE_DURATION_SEC: + return n + return None + + +def _extract_ldjson_only(html: str, page_url: str) -> int | None: + """xvideos i xnxx: `VideoObject` z ISO 8601. Sprawdzone 5/5 na losowej próbce.""" + if (m := _LD_DURATION_RE.search(html)) and ( + parsed := _parse_iso8601(m.group(1)) + ) is not None: + if 0 < parsed <= _MAX_SANE_DURATION_SEC: + return parsed + return None + + +#: Origin → wzorzec. Dopisuj TYLKO to, co sprawdziłeś na próbce prawdziwych stron; +#: porntrex i mypornerleak celowo tu nie ma, patrz `app/models/duration_backfill_queue.py`. +ORIGIN_DURATION_RESOLVERS = { + "tube:xhamstercom": _extract_xhamster, + "tube:xvideoscom": _extract_ldjson_only, + "tube:xnxxcom": _extract_ldjson_only, +} + + +def extract_duration_for_origin(origin: str, html: str, page_url: str) -> int | None: + """Długość dla originu z potwierdzonym wzorcem; None gdy origin nieobsługiwany.""" + fn = ORIGIN_DURATION_RESOLVERS.get(origin) + return fn(html, page_url) if fn else None diff --git a/app/models/__init__.py b/app/models/__init__.py index c0c406d..1b360dc 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.duration_backfill_queue import DurationBackfillQueue from app.models.phash_backfill_queue import PhashBackfillQueue from app.models.phash_blacklist import PhashBlacklist from app.models.saved_search import SavedSearch diff --git a/app/models/duration_backfill_queue.py b/app/models/duration_backfill_queue.py new file mode 100644 index 0000000..7047f7c --- /dev/null +++ b/app/models/duration_backfill_queue.py @@ -0,0 +1,67 @@ +"""Kolejka scen, którym trzeba doliczyć długość ze strony tuba. + +Filtr `min_duration_sec=60` odcina śmieciowe klipy i tak ma zostać, ale w SQL-u +`duration_sec >= 60` NIE jest prawdą dla NULL, więc scena o NIEZNANEJ długości znika +z feedu tak samo jak dziesięciosekundowy śmieć. Audyt 2026-08-20: **1 755 421 scen** +nie ma długości ani na scenie, ani na źródle, czyli scraper nigdy jej nie zebrał. +Największe zaległości to porntrex (676 tys.), xhamster (361 tys.), xvideos (259 tys.) +i xnxx (226 tys.). + +**Dlaczego nie jeden uniwersalny regex po stronie szczegółów.** Bo trafia w śmieci. +Na próbce: porntrex ma w HTML `duration: '5'` (długość reklamy), mypornerleak +`disclaimer_cookie_duration":"720"` (ciastko), a xhamster i porntrex wypisują długości +WSZYSTKICH filmów powiązanych z paska obok. Naiwne dopasowanie przy jednym xhamsterze +dało 198 s zamiast prawdziwych 240 s. Zła długość jest gorsza niż jej brak, bo dedup +traktuje długość jako najmocniejszy pojedynczy sygnał i skleiłby obce sceny. + +Dlatego backfill obsługuje tylko originy z POTWIERDZONYM, punktowym wzorcem +(`app/extractors/duration_extract.py`), a nie wszystko, co się nawinie: + +* xvideos, xnxx: `VideoObject` LD-JSON z ISO 8601, sprawdzone 5/5 +* xhamster: `"videoEntity":{"id":,…,"duration":}`, z kontrolą + zgodności id, bo bez niej łapie film z paska obok + +porntrex świadomie NIE jest tu obsługiwany: jego strona szczegółów w ogóle nie podaje +własnej długości (`video_duration` w flashvars nie występuje), a jedyne `durations` +w HTML należą do filmów powiązanych. Dla niego drogą jest listing, bo kafelek długość +pokazuje i `porntrex_browse.py` już ją stamtąd czyta. + +Kolejka jest tabelą z tego samego powodu co przy phashu: wyznaczenie kandydatów to +ciężki przegląd całego katalogu, a mielenie ma przeżyć restart workera i pokazywać +postęp. Patrz [[phash_backfill_queue]] dla bliźniaczego wzorca. +""" +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 DurationBackfillQueue(Base): + __tablename__ = "duration_backfill_queue" + + scene_id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), + ForeignKey("scenes.id", ondelete="CASCADE"), + primary_key=True, + ) + #: Strona tuba, z której czytamy długość. Zapisana przy planowaniu, żeby mielenie + #: nie musiało co paczkę wracać do `playback_sources`. + page_url: Mapped[str] = mapped_column(String(1024), nullable=False) + #: Decyduje, KTÓRY punktowy wzorzec zostanie użyty. Bez tego wracamy do zgadywania. + origin: Mapped[str] = mapped_column(String(64), nullable=False) + #: Rośnie przy każdej nieudanej próbie; po wyczerpaniu limitu wiersz jest odkładany, + #: żeby skasowana strona nie blokowała kolejki w nieskończoność. + attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0") + #: Ustawione po sukcesie ALBO po wyczerpaniu prób (wtedy długość została pusta). + 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/duration_backfill.py b/app/scheduler/duration_backfill.py new file mode 100644 index 0000000..b5f2953 --- /dev/null +++ b/app/scheduler/duration_backfill.py @@ -0,0 +1,148 @@ +"""Dolicza brakującą długość scenom ze stron tubów, paczkami. + +Po co i dlaczego tylko wybrane originy: `app/models/duration_backfill_queue.py`. +W skrócie: bez długości scena wypada z feedu, bo `duration_sec >= 60` nie jest prawdą +dla NULL, a zła długość jest gorsza niż brak, bo dedup opiera się na niej najmocniej. +Stąd wyłącznie wzorce punktowe z `ORIGIN_DURATION_RESOLVERS`. +""" +from __future__ import annotations + +import logging +from concurrent.futures import ThreadPoolExecutor + +from sqlalchemy import text + +from app.db import session_scope +from app.extractors.duration_extract import ( + ORIGIN_DURATION_RESOLVERS, + extract_duration_for_origin, +) + +log = logging.getLogger(__name__) + +# Po tylu nieudanych próbach odkładamy wiersz. Strony bywają skasowane i nie ma sensu +# wracać do nich w kółko, blokując kolejkę. +_MAX_ATTEMPTS = 3 +_DEFAULT_BATCH = 500 +# Sześć wątków na paczkę, a paczki są rozłożone w czasie przez scheduler. Pobrania +# rozkładają się na trzy różne serwisy, więc żaden nie dostaje serii pod rząd. +_CONCURRENCY = 6 +_FETCH_TIMEOUT = 20.0 + + +def plan_duration_backfill(limit: int | None = None) -> int: + """Wrzuca do kolejki sceny bez długości, mające źródło na obsługiwanym originie. + + Idempotentne: `ON CONFLICT DO NOTHING`, więc można wołać wielokrotnie, także po + dopisaniu nowego originu do `ORIGIN_DURATION_RESOLVERS`. + """ + origins = sorted(ORIGIN_DURATION_RESOLVERS) + if not origins: + return 0 + sql = """ + INSERT INTO duration_backfill_queue (scene_id, page_url, origin) + SELECT DISTINCT ON (s.id) s.id, ps.page_url, ps.origin + FROM scenes s + JOIN playback_sources ps ON ps.scene_id = s.id + WHERE s.duration_sec IS NULL + AND ps.origin = ANY(:origins) + AND ps.page_url IS NOT NULL + AND ps.dead_at IS NULL + ORDER BY s.id, ps.origin + ON CONFLICT (scene_id) DO NOTHING + """ + if limit is not None: + # LIMIT musi objąć SELECT, nie INSERT, stąd podzapytanie przy ograniczonym planie. + sql = f""" + INSERT INTO duration_backfill_queue (scene_id, page_url, origin) + SELECT * FROM ( + SELECT DISTINCT ON (s.id) s.id, ps.page_url, ps.origin + FROM scenes s + JOIN playback_sources ps ON ps.scene_id = s.id + WHERE s.duration_sec IS NULL + AND ps.origin = ANY(:origins) + AND ps.page_url IS NOT NULL + AND ps.dead_at IS NULL + ORDER BY s.id, ps.origin + ) q LIMIT {int(limit)} + ON CONFLICT (scene_id) DO NOTHING + """ + with session_scope() as s: + n = s.execute(text(sql), {"origins": origins}).rowcount + log.info("duration-backfill: zaplanowano %d scen (originy: %s)", n, ",".join(origins)) + return n + + +def _fetch_one(row: tuple) -> tuple: + """(scene_id, duration|None). Sieć i parsowanie, bez dotykania bazy.""" + scene_id, page_url, origin = row + from app.extractors._fetch import browser_get + + try: + r = browser_get(page_url, timeout=_FETCH_TIMEOUT, follow_redirects=True) + if r.status_code >= 400: + return (scene_id, None) + return (scene_id, extract_duration_for_origin(origin, r.text, page_url)) + except Exception as e: # pojedyncza strona nie może wywrócić paczki + log.debug("duration-backfill: %s padlo (%s)", page_url, e) + return (scene_id, None) + + +def run_duration_backfill(*, batch: int = _DEFAULT_BATCH) -> dict[str, int]: + """Bierze paczkę z kolejki, dolicza długość, zapisuje. Pusta kolejka = no-op.""" + with session_scope() as s: + rows = s.execute( + text( + """ + SELECT scene_id, page_url, origin + FROM duration_backfill_queue + WHERE done_at IS NULL AND attempts < :max_attempts + ORDER BY attempts, created_at + LIMIT :batch + """ + ), + {"max_attempts": _MAX_ATTEMPTS, "batch": batch}, + ).fetchall() + + if not rows: + return {"taken": 0, "filled": 0, "missed": 0} + + with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool: + results = list(pool.map(_fetch_one, [tuple(r) for r in rows])) + + filled = [(sid, dur) for sid, dur in results if dur] + missed = [sid for sid, dur in results if not dur] + + with session_scope() as s: + for scene_id, dur in filled: + # Warunek na NULL, bo między planem a zapisem scena mogła zostać scalona + # albo wzbogacona z kanonu, a ten jest wiarygodniejszy niż tube. + s.execute( + text( + "UPDATE scenes SET duration_sec = :d " + " WHERE id = :id AND duration_sec IS NULL" + ), + {"d": dur, "id": scene_id}, + ) + s.execute( + text( + "UPDATE duration_backfill_queue SET done_at = now() WHERE scene_id = :id" + ), + {"id": scene_id}, + ) + if missed: + # Wyczerpane próby zamykamy, żeby kolejka nie kręciła się w kółko na martwych. + s.execute( + text( + """ + UPDATE duration_backfill_queue + SET attempts = attempts + 1, + done_at = CASE WHEN attempts + 1 >= :max_attempts + THEN now() ELSE NULL END + WHERE scene_id = ANY(:ids) + """ + ), + {"ids": missed, "max_attempts": _MAX_ATTEMPTS}, + ) + + return {"taken": len(rows), "filled": len(filled), "missed": len(missed)} diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 83c4770..3ae89c6 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -292,6 +292,38 @@ def _job_phash_backfill(batch: int) -> None: log.exception("[scheduler] phash backfill failed") +def _job_duration_plan() -> None: + """Dopisuje do kolejki sceny, ktore od wczoraj przyszly bez dlugosci. Scrapery + czesci tubow jej nie zbieraja przy ingescie, wiec bez tego zaleglosc odrastalaby + z kazda doba. `ON CONFLICT DO NOTHING`, wiec powtorzenie nic nie psuje.""" + try: + from app.scheduler.duration_backfill import plan_duration_backfill + + added = _run_with_timeout( + lambda: plan_duration_backfill(), label="duration-plan" + ) + log.info("[scheduler] duration plan done (%s)", added) + except Exception: + log.exception("[scheduler] duration plan failed") + + +def _job_duration_backfill(batch: int) -> None: + """Dolicza dlugosc scenom, ktore jej nie maja. Bez dlugosci scena wypada z feedu, + bo `duration_sec >= 60` nie jest prawda dla NULL. Kolejke planujemy raz + (`plan_duration_backfill`), tu tylko mielimy paczkami, bo to pobrania cudzych stron. + Pusta kolejka = no-op, wiec nic nie kosztuje.""" + try: + from app.scheduler.duration_backfill import run_duration_backfill + + out = _run_with_timeout( + lambda: run_duration_backfill(batch=batch), label="duration-backfill" + ) + if out and out.get("taken"): + log.info("[scheduler] duration backfill %s", out) + except Exception: + log.exception("[scheduler] duration 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 @@ -547,6 +579,35 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: ) log.info("scheduler: phash blacklist every %dh", cfg["phash_blacklist_hours"]) + if cfg.get("duration_backfill_minutes"): + sched.add_job( + _job_duration_plan, + IntervalTrigger(hours=24, start_date=INTERVAL_ANCHOR), + id="duration_plan", + replace_existing=True, + max_instances=1, + coalesce=True, + ) + log.info("scheduler: duration plan every 24h") + if cfg.get("duration_backfill_minutes"): + _dur_batch = int(cfg.get("duration_backfill_batch") or 300) + sched.add_job( + _job_duration_backfill, + IntervalTrigger( + minutes=cfg["duration_backfill_minutes"], start_date=INTERVAL_ANCHOR + ), + args=[_dur_batch], + id="duration_backfill", + replace_existing=True, + max_instances=1, + coalesce=True, + ) + log.info( + "scheduler: duration backfill every %dmin (batch=%d)", + cfg["duration_backfill_minutes"], + _dur_batch, + ) + if cfg.get("phash_backfill_minutes"): _batch = int(cfg.get("phash_backfill_batch") or 400) sched.add_job( diff --git a/app/scheduler/worker.py b/app/scheduler/worker.py index c1c189c..de272cb 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, + "duration_backfill_minutes": getattr(settings, "sched_duration_backfill_minutes", 5) or None, + "duration_backfill_batch": getattr(settings, "sched_duration_backfill_batch", 500), "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).