feat(watchdog): jakosc zrodel, progi adaptacyjne, dlugosci perverzija i mypornerleak, poprawki parserow

- browse_runs (migracja 0056): jeden wiersz per scraper i przebieg browse-latest
- quality_watchdog: listing pusty albo bez nowych, skok lub zanik obsady, sceny bez
  dlugosci i miniatur, spadek udanych odtworzen per origin; fingerprint per zrodlo i rodzaj
- ingest_watchdog: prog swiezosci z mediany odstepow zrodla (3x, max 14 d) zamiast stalej
- duration backfill: resolvery perverzija (JSON-LD) i mypornerleak (itemprop), paczka
  po rowno z originow, 404/410 zamyka wiersz od razu
- parsery: galaxyporn data z JSON-LD, hentaimama dlugosc, javflix aktorki, fullmovies
  i hdporngg data z uploadDate

alembic/versions/20261014_0056_browse_runs.py

@@ -0,0 +1,41 @@
+"""Przebiegi browse-latest per scraper (watchdog listingu).
+
+Revision ID: 0056_browse_runs
+Revises: 0055_survey_responses
+Create Date: 2026-10-14
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import UUID
+
+revision = "0056_browse_runs"
+down_revision: str | None = "0055_survey_responses"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "browse_runs",
+        sa.Column("id", UUID(as_uuid=True), primary_key=True),
+        sa.Column("sitetag", sa.String(64), nullable=False),
+        sa.Column(
+            "started_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column("elapsed_s", sa.Float, nullable=True),
+        sa.Column("seen", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("new", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("updated", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("skipped", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("errors", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("failed", sa.Boolean, nullable=False, server_default=sa.false()),
+        sa.Column("note", sa.Text, nullable=True),
+    )
+    op.create_index("ix_browse_runs_sitetag_started", "browse_runs", ["sitetag", "started_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_browse_runs_sitetag_started", table_name="browse_runs")
+    op.drop_table("browse_runs")

app/connectors/direct_scrapers/fullmovies.py

@@ -7,6 +7,7 @@ Identyczny engine co hdporn.gg (KVS sponsor_groups stack): `/videos/<slug>/`,
 from __future__ import annotations
 
 import re
+from datetime import date
 
 from app.connectors.base import (
     RawFingerprint,
@@ -28,6 +29,7 @@ _SCENE_URL_RE = re.compile(r'href="(https://www\.fullmovies\.xxx/videos/[a-z0-9\
 _NETWORK_LINK_RE = re.compile(
     r'href="https://www\.fullmovies\.xxx/networks/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE
 )
+_UPLOAD_DATE_RE = re.compile(r'"uploadDate"\s*:\s*"(\d{4})-(\d{2})-(\d{2})')
 _MODEL_LINK_RE = re.compile(
     r'href="https://www\.fullmovies\.xxx/models/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE
 )
@@ -68,6 +70,14 @@ class FullmoviesScraper(BaseBrowseScraper):
         dur_meta = meta_content(detail_html, property="video:duration")
         if dur_meta and dur_meta.isdigit():
             duration_sec = int(dur_meta)
+        # Data z JSON-LD `"uploadDate": "2026-10-13T20:48:46Z"` (audyt 2026-10-14: 0 dat).
+        release_date = None
+        um = _UPLOAD_DATE_RE.search(detail_html)
+        if um:
+            try:
+                release_date = date(int(um.group(1)), int(um.group(2)), int(um.group(3)))
+            except ValueError:
+                release_date = None
 
         # Studio z PREFIKSU tytułu ("Studio - Scene Title"), nie z sidebara /networks/.
         # Sidebar listuje WSZYSTKIE sieci → `_NETWORK_LINK_RE.finditer().first()` zawsze
@@ -122,6 +132,7 @@ class FullmoviesScraper(BaseBrowseScraper):
             external_id=f"{self.sitetag}:{scene_url}",
             title=title,
             description=description,
+            release_date=release_date,
             duration_sec=duration_sec,
             url=scene_url,
             studio=studio,

app/connectors/direct_scrapers/galaxyporn.py

@@ -62,6 +62,9 @@ _IFRAME_RE = re.compile(r'<iframe[^>]+src="([^"]+)"', re.IGNORECASE)
 _TITLE_STUDIO_DATE_RE = re.compile(r"^(?P<studio>[A-Za-z0-9][A-Za-z0-9._-]{1,30})\s+(?P<y>\d{2})\s+(?P<m>\d{2})\s+(?P<d>\d{2})\s+(?P<rest>.+)$")
 # `[Onlyfans] Some Title` / `[Darkkotv] …`
 _TITLE_BRACKET_RE = re.compile(r"^\[(?P<studio>[^\]]{2,30})\]\s*(?P<rest>.+)$")
+# Data z JSON-LD strony, gdy tytul jej nie niesie (audyt 2026-10-14: 66% scen z 7 dni
+# bez daty i bez dlugosci; data jest w `"datePublished":"2026-10-13T18:40:00+03:00"`).
+_JSONLD_DATE_RE = re.compile(r'"datePublished"\s*:\s*"(\d{4})-(\d{2})-(\d{2})')
 # `… [2018-09-28]`
 _TITLE_ISO_DATE_RE = re.compile(r"\[(\d{4})-(\d{2})-(\d{2})\]")
 
@@ -178,6 +181,11 @@ class GalaxyPornScraper(BaseBrowseScraper):
                 release_date = date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
             except ValueError:
                 release_date = None
+        if release_date is None and (m := _JSONLD_DATE_RE.search(detail_html)):
+            try:
+                release_date = date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
+            except ValueError:
+                release_date = None
         # Guard: prefiks nie może być nazwiskiem performera (wtedy to nie studio).
         if studio_name and slugify(studio_name) not in seen_p:
             studio = RawStudio(

app/connectors/direct_scrapers/hdporngg.py

@@ -10,6 +10,7 @@ factory, wyłącz; analogia do shyfap).
 from __future__ import annotations
 
 import re
+from datetime import date
 from urllib.parse import urljoin
 
 from app.connectors.base import (
@@ -32,6 +33,7 @@ _SCENE_URL_RE = re.compile(r'href="(https://www\.hdporn\.gg/videos/[a-z0-9\-]+/)
 _NETWORK_LINK_RE = re.compile(
     r'href="https://www\.hdporn\.gg/networks/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE
 )
+_UPLOAD_DATE_RE = re.compile(r'"uploadDate"\s*:\s*"(\d{4})-(\d{2})-(\d{2})')
 _MODEL_LINK_RE = re.compile(
     r'href="https://www\.hdporn\.gg/models/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE
 )
@@ -75,6 +77,14 @@ class HDPornGGScraper(BaseBrowseScraper):
         dur_meta = meta_content(detail_html, property="video:duration")
         if dur_meta and dur_meta.isdigit():
             duration_sec = int(dur_meta)
+        # Data z JSON-LD `"uploadDate": "2026-10-13T16:51:32Z"` (audyt 2026-10-14: 0 dat).
+        release_date = None
+        um = _UPLOAD_DATE_RE.search(detail_html)
+        if um:
+            try:
+                release_date = date(int(um.group(1)), int(um.group(2)), int(um.group(3)))
+            except ValueError:
+                release_date = None
 
         # Studio z PREFIKSU tytułu ("Studio - Scene Title"), nie z sidebara /networks/.
         # Sidebar listuje WSZYSTKIE sieci → pierwszy match zawsze ten sam (Brazzers) dla
@@ -130,6 +140,7 @@ class HDPornGGScraper(BaseBrowseScraper):
             external_id=f"{self.sitetag}:{scene_url}",
             title=title,
             description=description,
+            release_date=release_date,
             duration_sec=duration_sec,
             url=scene_url,
             studio=studio,

app/connectors/direct_scrapers/hentaimama.py

@@ -49,6 +49,10 @@ _OG_TITLE_RE = re.compile(r"property=['\"]og:title['\"][^>]*content=['\"]([^'\"]
 _GENRES_BLOCK_RE = re.compile(r'<div class="sgeneros">(.*?)</div>', re.DOTALL)
 _ANCHOR_TEXT_RE = re.compile(r">\s*([^<]{2,40}?)\s*</a>")
 _DATE_RE = re.compile(r'<span class="date">\s*([A-Z][a-z]{2}\.?\s+\d{1,2},\s+\d{4})')
+# Dlugosc odcinka w sekundach siedzi w konfiguracji skryptu historii motywu dooplay:
+# `var dtHist = {..., "episode":"77667", "duration":"953", ...}`. Bez niej 100% scen
+# mialo NULL (audyt 2026-10-14), a apka filtruje min 60 s.
+_DT_DURATION_RE = re.compile(r'"duration"\s*:\s*"(\d{2,6})"')
 _SERIES_RE = re.compile(r'href="https://hentaimama\.io/tvshows/([a-z0-9-]+)/"')
 # Uwaga: atrybuty bywają w innej kolejności niż `<meta property=... content=...>`,
 # a wartość ma na końcu \r — stąd luźny wzorzec i strip() przy użyciu.
@@ -112,17 +116,21 @@ class HentaiMamaScraper(BaseBrowseScraper):
         dm = _DATE_RE.search(detail_html)
         om = _OG_IMAGE_RE.search(detail_html)
         slug = scene_url.rstrip("/").rsplit("/", 1)[-1]
+        dur_m = _DT_DURATION_RE.search(detail_html)
+        duration_sec = int(dur_m.group(1)) if dur_m and int(dur_m.group(1)) > 0 else None
 
         return RawScene(
             external_id=f"{self.sitetag}:{slug}",
             title=title,
             release_date=_parse_date(dm.group(1) if dm else None),
+            duration_sec=duration_sec,
             url=scene_url,
             tags=tags,
             playback_sources=[
                 RawPlaybackSource(
                     origin=f"tube:{self.sitetag}",
                     page_url=scene_url,
+                    duration_sec=duration_sec,
                     thumbnail_url=om.group(1).strip() if om else None,
                 )
             ],

app/connectors/direct_scrapers/javflix.py

@@ -18,6 +18,7 @@ import re
 
 from app.connectors.base import RawPlaybackSource, RawScene, RawTag
 from app.connectors.direct_scrapers._browse_base import BaseBrowseScraper
+from app.connectors.base import RawPerformer
 from app.connectors.direct_scrapers._playtube import _parse_iso_date
 
 _BASE = "https://javflix.cc"
@@ -30,6 +31,11 @@ _NON_POST = (
     "/terms", "/faq", "/policy", "/disclaimer", "/sitemap",
 )
 _CODE_RE = re.compile(r"^([a-z]+-?\d+[a-z]?)", re.IGNORECASE)
+# Aktorki: tylko z bloku `<div id="video-actors">Actress: <a href=".../actor/<slug>/">`,
+# jeden na scene zwykle. Do 2026-10-14 scraper ich nie bral (0 performerow na scene),
+# a JAV bez aktorki nie da sie laczyc z kanonem ani filtrowac po osobie.
+_ACTORS_BLOCK_RE = re.compile(r'<div id="video-actors">(.*?)</div>', re.IGNORECASE | re.DOTALL)
+_ACTOR_RE = re.compile(r'href="https://javflix\.cc/actor/([a-z0-9\-]+)/"[^>]*>([^<]+)</a>', re.IGNORECASE)
 
 
 def _itemprop(html: str, name: str) -> str | None:
@@ -96,11 +102,24 @@ class JavflixScraper(BaseBrowseScraper):
             code = cm.group(1).upper()
             tags.append(RawTag(external_id=f"javcode:{code}", name=code, slug=cm.group(1).lower()))
 
+        performers: list[RawPerformer] = []
+        ab = _ACTORS_BLOCK_RE.search(detail_html)
+        if ab:
+            seen_p: set[str] = set()
+            for am in _ACTOR_RE.finditer(ab.group(1)):
+                a_slug = am.group(1)
+                a_name = html.unescape(am.group(2)).strip()
+                if a_slug in seen_p or not a_name:
+                    continue
+                seen_p.add(a_slug)
+                performers.append(RawPerformer(external_id=f"{self.sitetag}:actor:{a_slug}", name=a_name))
+
         return RawScene(
             external_id=f"{self.sitetag}:{scene_url}",
             title=title,
             release_date=release_date,
             url=scene_url,
+            performers=performers,
             tags=tags,
             playback_sources=[
                 RawPlaybackSource(

app/connectors/direct_scrapers/watchporn.py

@@ -32,6 +32,21 @@ _SCENE_URL_RE = re.compile(r'href="(https://watchporn\.to/video/\d+/[a-z0-9\-]+/
 _MODEL_RE = re.compile(r'href="https://watchporn\.to/models/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE)
 _TAG_RE = re.compile(r'href="https://watchporn\.to/tags/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE)
 _CAT_RE = re.compile(r'href="https://watchporn\.to/categories/([a-z0-9\-]+)/"[^>]*>([^<]+)', re.IGNORECASE)
+# Nowy motyw od ok. 2026-10-10 (prefiks klas `wp-`). Obsada sceny siedzi TYLKO w bloku
+# `<div class="wp-castrow">`, nazwa w `<span class="wp-cast__n">` (nie w tekscie linku),
+# a karty "related" ponizej maja wlasne linki `/models/` z nazwa w tekscie. Stary globalny
+# regex gubil prawdziwa obsade (pusty tekst linku) i dokladal performerow z kart related:
+# srednio 12,5 na scene zamiast 2 (10 do 13.10, 193 sceny). Kategorie to teraz gatunki
+# (chipy `wp-vchip`), studio jest w linku `/sites/<slug>/` i w JSON-LD "publisher".
+_NEW_THEME_MARK = 'class="wp-vi__panel"'
+_CASTROW_RE = re.compile(r'<div class="wp-castrow">(.*?)</div>', re.IGNORECASE | re.DOTALL)
+_CAST_RE = re.compile(
+    r'href="https://watchporn\.to/models/([a-z0-9\-]+)/"[^>]*>.*?<span class="wp-cast__n">([^<]+)</span>',
+    re.IGNORECASE | re.DOTALL,
+)
+_CHIPS_RE = re.compile(r'<div class="wp-chips">(.*?)</div>', re.IGNORECASE | re.DOTALL)
+_SITE_RE = re.compile(r'href="https://watchporn\.to/sites/([a-z0-9\-]+)/"', re.IGNORECASE)
+_PUBLISHER_RE = re.compile(r'"publisher"\s*:\s*\{[^}]*?"name"\s*:\s*"([^"]+)"', re.DOTALL)
 _DUR_RE = re.compile(r'"duration"\s*:\s*"([^"]+)"')
 _UPLOAD_RE = re.compile(r'"uploadDate"\s*:\s*"([^"]+)"')
 _VIDEO_ID_RE = re.compile(r"/video/(\d+)/")
@@ -74,9 +89,18 @@ class WatchPornScraper(BaseBrowseScraper):
         um = _UPLOAD_RE.search(detail_html)
         release_date = _parse_iso_date(um.group(1)) if um else None
 
+        new_theme = _NEW_THEME_MARK in detail_html
+
         performers: list[RawPerformer] = []
         seen_p: set[str] = set()
-        for m in _MODEL_RE.finditer(detail_html):
+        if new_theme:
+            # Tylko blok obsady; brak bloku = scena bez obsady, NIE wracamy do globalnego
+            # regexu, bo ten zbiera karty related.
+            crm = _CASTROW_RE.search(detail_html)
+            cast_iter = _CAST_RE.finditer(crm.group(1)) if crm else iter(())
+        else:
+            cast_iter = _MODEL_RE.finditer(detail_html)
+        for m in cast_iter:
             slug = m.group(1)
             name = html.unescape(m.group(2)).strip()
             if slug.isdigit() or slug in seen_p or not name:
@@ -86,19 +110,38 @@ class WatchPornScraper(BaseBrowseScraper):
 
         tags: list[RawTag] = []
         seen_t: set[str] = set()
-        for m in _TAG_RE.finditer(detail_html):
-            slug = m.group(1)
-            name = html.unescape(m.group(2)).strip()
+
+        def _add_tag(slug: str, raw_name: str) -> None:
+            name = html.unescape(raw_name).strip().lstrip("#").strip()
             if slug in seen_t or not name:
-                continue
+                return
             seen_t.add(slug)
             tags.append(RawTag(external_id=f"{self.sitetag}:tag:{slug}", name=name, slug=slug))
 
+        for m in _TAG_RE.finditer(detail_html):
+            _add_tag(m.group(1), m.group(2))
+        if new_theme:
+            # Kategorie-chipy sceny (Blowjob, MILF...) to gatunki, ida do tagow.
+            chm = _CHIPS_RE.search(detail_html)
+            if chm:
+                for m in _CAT_RE.finditer(chm.group(1)):
+                    _add_tag(m.group(1), m.group(2))
+
         thumbnail_url = (
             meta_content(detail_html, property="og:image") or _derive_thumb(scene_url)
         )
 
         studio: RawStudio | None = None
+        if new_theme:
+            sm = _SITE_RE.search(detail_html)
+            if sm:
+                pm = _PUBLISHER_RE.search(detail_html)
+                cname = html.unescape(pm.group(1)).strip() if pm else sm.group(1).replace("-", " ").title()
+                if cname:
+                    studio = RawStudio(
+                        external_id=f"{self.sitetag}:studio:{slugify(cname)}", name=cname, slug=slugify(cname)
+                    )
+        else:
             cm = _CAT_RE.search(detail_html)
             if cm:
                 cname = html.unescape(cm.group(2)).strip()

app/extractors/duration_extract.py

@@ -141,12 +141,29 @@ def _extract_ldjson_only(html: str, page_url: str) -> int | None:
     return None
 
 
+def _extract_itemprop_only(html: str, page_url: str) -> int | None:
+    """mypornerleak: `<meta itemprop="duration" content="P0DT0H12M13S">` w <head> to
+    dlugosc TEJ sceny; karty powiazane maja tylko `<span class="duration">40:33</span>`,
+    ktorych tu nie czytamy. Sprawdzone 2026-10-14 na stronach z nowego listingu REST
+    (ingest przez wp-json nie niesie dlugosci, stad 76 tys. scen bez niej)."""
+    if (m := _ITEMPROP_DURATION_RE.search(html)):
+        v = m.group(1).strip()
+        if v.upper().startswith("P") and (parsed := _parse_iso8601(v)) 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`.
+#: porntrex celowo tu nie ma, patrz `app/models/duration_backfill_queue.py`.
+#: perverzija: ingest z wp-json nie ma dlugosci, strona ma JSON-LD VideoObject
+#: `"duration":"PT73M2S"` (2026-10-14, 21 tys. scen bez dlugosci).
 ORIGIN_DURATION_RESOLVERS = {
     "tube:xhamstercom": _extract_xhamster,
     "tube:xvideoscom": _extract_ldjson_only,
     "tube:xnxxcom": _extract_ldjson_only,
+    "tube:perverzijacom": _extract_ldjson_only,
+    "tube:mypornerleakcom": _extract_itemprop_only,
 }
 
 

app/models/__init__.py

@@ -1,5 +1,6 @@
 from app.models.base import Base
 from app.models.blacklist import BlacklistedPerformer, BlacklistedStudio, BlacklistedTag
+from app.models.browse_run import BrowseRun
 from app.models.bug_report import BugReport
 from app.models.clip import Clip, ClipTag
 from app.models.clip_feed import ClipAffinity, ClipBlacklist, ClipFavorite, ClipFeedEvent, ClipSeen
@@ -44,6 +45,7 @@ from app.models.tag_vertical_count import TagVerticalCount
 
 __all__ = [
     "Base",
+    "BrowseRun",
     "BlacklistedPerformer",
     "BlacklistedStudio",
     "BlacklistedTag",

app/models/browse_run.py

@@ -0,0 +1,34 @@
+"""Przebieg browse-latest per scraper (sitetag), zeby watchdog odroznial trzy rozne cisze:
+listing pusty (fetch albo parser listingu padl), listing pelny ale wszystko znane (zly
+endpoint, rotacja karuzeli) i zwykly brak nowych publikacji. ingest_runs tego nie daje,
+bo wszystkie tuby dziela jeden Source "tube-scraper".
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, Float, Index, Integer, String, Text, func
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base
+
+
+class BrowseRun(Base):
+    __tablename__ = "browse_runs"
+    __table_args__ = (Index("ix_browse_runs_sitetag_started", "sitetag", "started_at"),)
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    sitetag: Mapped[str] = mapped_column(String(64), nullable=False)
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    elapsed_s: Mapped[float | None] = mapped_column(Float, nullable=True)
+    seen: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    new: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    updated: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    skipped: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    errors: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    failed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    note: Mapped[str | None] = mapped_column(Text, nullable=True)

app/scheduler/browse_latest.py

@@ -21,12 +21,37 @@ import time
 from dataclasses import dataclass
 
 from app.connectors.direct_scrapers import ALL_BROWSE_SCRAPERS, SCRAPER_SOURCE_NAME
+from app.db import session_scope
+from app.models.browse_run import BrowseRun
 from app.models.source import SourceKind
 from app.scheduler.performer_driven import _ingest_iter_into_run
 
 log = logging.getLogger(__name__)
 
 
+def _record_run(sitetag: str, counters: dict | None, elapsed: float, failed: bool, note: str | None) -> None:
+    """Jeden wiersz browse_runs per scraper i przebieg; czyta go watchdog listingu.
+    Blad zapisu nie moze zatrzymac browse (log i dalej)."""
+    c = counters or {}
+    try:
+        with session_scope() as s:
+            s.add(
+                BrowseRun(
+                    sitetag=sitetag,
+                    elapsed_s=round(elapsed, 1),
+                    seen=int(c.get("seen", 0)),
+                    new=int(c.get("new", 0)),
+                    updated=int(c.get("updated", 0)),
+                    skipped=int(c.get("skipped", 0)),
+                    errors=int(c.get("errors", 0)),
+                    failed=failed,
+                    note=(note or None) and note[:500],
+                )
+            )
+    except Exception:  # pragma: no cover - tabela moze nie istniec przed migracja 0056
+        log.exception("browse-latest: zapis browse_runs dla %s nie powiodl sie", sitetag)
+
+
 @dataclass
 class BrowseCounters:
     scrapers_run: int = 0
@@ -62,9 +87,11 @@ def run_browse_latest(*, max_pages: int = 5) -> BrowseCounters:
                 "browse-latest: %s done in %.1fs counters=%s",
                 scraper.sitetag, elapsed, c,
             )
+            _record_run(scraper.sitetag, c, elapsed, failed=False, note=None)
         except Exception as e:
             counters.total_errors += 1
             log.exception("browse-latest: %s failed: %s", scraper.sitetag, e)
+            _record_run(scraper.sitetag, None, time.time() - t0, failed=True, note=f"{type(e).__name__}: {e}")
 
     log.info("browse-latest done: %s", counters)
     return counters

app/scheduler/duration_backfill.py

@@ -74,44 +74,89 @@ def plan_duration_backfill(limit: int | None = None) -> int:
 
 
 def _fetch_one(row: tuple) -> tuple:
-    """(scene_id, duration|None). Sieć i parsowanie, bez dotykania bazy."""
+    """(scene_id, duration|None, gone). Sieć i parsowanie, bez dotykania bazy.
+
+    `gone` = strona odpowiedziała 404/410, więc kolejne próby nic nie dadzą i wiersz
+    można zamknąć od razu zamiast odwiedzać martwy adres trzy razy. Błąd sieci, 5xx
+    albo 403 (limit, geo) zostają zwykłą nieudaną próbą.
+    """
     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 in (404, 410):
+            return (scene_id, None, True)
         if r.status_code >= 400:
-            return (scene_id, None)
-        return (scene_id, extract_duration_for_origin(origin, r.text, page_url))
+            return (scene_id, None, False)
+        return (scene_id, extract_duration_for_origin(origin, r.text, page_url), False)
     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)
+        return (scene_id, None, False)
+
+
+def _take_batch(session, *, batch: int) -> list[tuple]:
+    """Paczka po rowno z kazdego originu, ktory ma jeszcze cos w kolejce.
+
+    Kolejnosc dopisania (`created_at`) jako jedyny klucz oznaczalaby, ze origin dopisany
+    pozniej (perverzija, mypornerleak, 2026-10-14) czeka, az 900 tys. wierszy xhamster,
+    xvideos i xnxx sie wyczerpie. Zamiast tego limit dzielimy miedzy originy z otwartymi
+    wierszami (najmniejsze kolejki dostaja swoj udzial w calosci, reszta dzieli resztę),
+    a wewnatrz originu zostaje stary porzadek: najpierw nieprobowane, potem najstarsze.
+    """
+    remaining = session.execute(
+        text(
+            """
+            SELECT origin, count(*) AS n
+              FROM duration_backfill_queue
+             WHERE done_at IS NULL AND attempts < :max_attempts
+             GROUP BY origin
+            """
+        ),
+        {"max_attempts": _MAX_ATTEMPTS},
+    ).fetchall()
+    pending = sorted(((int(n), o) for o, n in remaining if n), key=lambda t: t[0])
+    rows: list[tuple] = []
+    left = batch
+    for i, (n, origin) in enumerate(pending):
+        quota = min(n, max(1, left // (len(pending) - i)))
+        if quota <= 0:
+            break
+        rows += [
+            tuple(r)
+            for r in session.execute(
+                text(
+                    """
+                    SELECT scene_id, page_url, origin
+                      FROM duration_backfill_queue
+                     WHERE origin = :origin AND done_at IS NULL AND attempts < :max_attempts
+                     ORDER BY attempts, created_at
+                     LIMIT :quota
+                    """
+                ),
+                {"origin": origin, "max_attempts": _MAX_ATTEMPTS, "quota": quota},
+            ).fetchall()
+        ]
+        left -= quota
+        if left <= 0:
+            break
+    return rows
 
 
 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()
+        rows = _take_batch(s, batch=batch)
 
     if not rows:
-        return {"taken": 0, "filled": 0, "missed": 0}
+        return {"taken": 0, "filled": 0, "missed": 0, "gone": 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]
+    filled = [(sid, dur) for sid, dur, _gone in results if dur]
+    gone = [sid for sid, dur, is_gone in results if not dur and is_gone]
+    missed = [sid for sid, dur, is_gone in results if not dur and not is_gone]
 
     with session_scope() as s:
         for scene_id, dur in filled:
@@ -130,6 +175,17 @@ def run_duration_backfill(*, batch: int = _DEFAULT_BATCH) -> dict[str, int]:
                 ),
                 {"id": scene_id},
             )
+        if gone:
+            s.execute(
+                text(
+                    """
+                    UPDATE duration_backfill_queue
+                       SET attempts = :max_attempts, done_at = now()
+                     WHERE scene_id = ANY(:ids)
+                    """
+                ),
+                {"ids": gone, "max_attempts": _MAX_ATTEMPTS},
+            )
         if missed:
             # Wyczerpane próby zamykamy, żeby kolejka nie kręciła się w kółko na martwych.
             s.execute(
@@ -145,4 +201,9 @@ def run_duration_backfill(*, batch: int = _DEFAULT_BATCH) -> dict[str, int]:
                 {"ids": missed, "max_attempts": _MAX_ATTEMPTS},
             )
 
-    return {"taken": len(rows), "filled": len(filled), "missed": len(missed)}
+    return {
+        "taken": len(rows),
+        "filled": len(filled),
+        "missed": len(missed),
+        "gone": len(gone),
+    }

app/scheduler/ingest_watchdog.py

@@ -61,12 +61,51 @@ def _bucket(age_h: float) -> tuple[str, str]:
 
 # Serwisy, ktore publikuja rzadziej niz co dwa dni: prog 48 h robil z nich staly alarm
 # (sprawdzone 2026-10-13: xanimeporn 1 do 3 wpisow na tydzien z RSS, hentaimama wydaje
-# odcinki raz w tygodniu, obie listy w calosci znane w bazie). Prog w godzinach.
+# odcinki raz w tygodniu, obie listy w calosci znane w bazie). Prog w godzinach, reczny
+# podlog; do tego kazde zrodlo dostaje prog z wlasnej kadencji (patrz _cadence_threshold_h).
 _BROWSE_MAX_AGE_OVERRIDE_H: dict[str, int] = {
     "xanimeporn": 168,
     "hentaimama": 240,
 }
 
+# Prog z kadencji: mediana odstepu (w dobach) miedzy dniami z nowymi pozycjami w ostatnich
+# 60 dniach, razy 3, nie mniej niz prog klasy i nie wiecej niz 14 dni. Zrodlo codzienne
+# zostaje przy 48 h, tygodniowe dostaje ok. 14 dni zamiast alarmu co dobe (2026-10-14).
+_ADAPTIVE_FACTOR = 3.0
+_ADAPTIVE_CAP_H = 14 * 24
+_ADAPTIVE_MIN_ACTIVE_DAYS = 5
+_ADAPTIVE_LOOKBACK_DAYS = 60
+
+
+def _cadence_threshold_h(session, *, origin: str, table: str, base_h: int) -> int:
+    import statistics
+
+    if table == "movie_playback_sources":
+        where = "(origin = :o OR origin LIKE :p)"
+        params: dict = {"o": origin, "p": f"{origin}:%", "lb": _ADAPTIVE_LOOKBACK_DAYS}
+    else:
+        where = "origin = :o"
+        params = {"o": origin, "lb": _ADAPTIVE_LOOKBACK_DAYS}
+    try:
+        rows = session.execute(
+            text(
+                f"SELECT DISTINCT created_at::date AS d FROM {table} WHERE {where} "
+                "AND created_at > now() - make_interval(days => :lb) ORDER BY d"
+            ),
+            params,
+        ).all()
+    except Exception:  # pragma: no cover
+        log.exception("ingest-watchdog: kadencja %s nieznana, prog klasy", origin)
+        return base_h
+    days = [r[0] for r in rows]
+    if len(days) < _ADAPTIVE_MIN_ACTIVE_DAYS:
+        return base_h
+    gaps = [(b - a).days for a, b in zip(days, days[1:])]
+    if not gaps:
+        return base_h
+    median_gap_h = statistics.median(gaps) * 24.0
+    return int(max(base_h, min(_ADAPTIVE_CAP_H, _ADAPTIVE_FACTOR * median_gap_h)))
+
 
 def run_ingest_freshness_watchdog(
     *,
@@ -132,6 +171,10 @@ def run_ingest_freshness_watchdog(
             if total < min_history or newest is None:
                 continue
             age_h = (now - newest).total_seconds() / 3600.0
+            if age_h >= threshold:
+                # Dopiero przy przekroczeniu progu klasy liczymy kadencje (jedno zapytanie
+                # per podejrzane zrodlo, nie per wszystkie).
+                threshold = _cadence_threshold_h(s, origin=origin, table=table, base_h=threshold)
             if age_h >= threshold:
                 label, level = _bucket(age_h)
                 stale.append(

app/scheduler/jobs.py

@@ -455,6 +455,15 @@ def _job_ingest_watchdog(
             log.warning("[scheduler] ingest-watchdog: %d stale origin(s)", len(res["stale"]))
     except Exception:
         log.exception("[scheduler] ingest-watchdog failed")
+    # Jakosc: listing (browse_runs), parser (performerzy, dlugosc, miniatury), odtwarzanie.
+    # Osobny try, zeby awaria jednego sprawdzenia nie zdjela drugiego.
+    try:
+        from app.scheduler.quality_watchdog import run_quality_watchdogs
+
+        q = run_quality_watchdogs()
+        log.info("[scheduler] quality-watchdog: %s", q["summary"])
+    except Exception:
+        log.exception("[scheduler] quality-watchdog failed")
 
 
 def _job_hetzner_monitor() -> None:

app/scheduler/quality_watchdog.py

@@ -0,0 +1,252 @@
+"""Watchdogi jakosci zrodel, uzupelnienie freshness z ingest_watchdog.py.
+
+Freshness widzi tylko "brak nowych pozycji". Cztery klasy awarii, ktore przez to przechodzily
+(wszystkie z 10.2026): watchporn zmienil motyw i parser dopinal 12 performerow na scene
+z kart "related" (ingest "zdrowy", new>0); galaxyporn, latestpornvideo i porndish oddaja
+sceny bez dlugosci (apka filtruje min 60 s, wiec sa niewidoczne, a licznik new rosnie);
+porndoe oddawal 31 tych samych scen (listing pelny, new=0, nie wiadomo czy to serwis stoi,
+czy czytamy zla strone); perverzija grala w 50% przy "zywych" zrodlach.
+
+Trzy sprawdzenia, kazde z wlasnym fingerprintem w Sentry (jedno issue na zrodlo i rodzaj,
+bez spamu co 6 h):
+
+- **listing** (browse_runs): ostatnie przebiegi scrapera z seen=0 albo wyjatkiem = listing
+  pusty lub blad; dlugi ciag seen>0 i new=0 u zrodla publikujacego codziennie = czytamy
+  zla strone (rotacja karuzeli, endpoint nie sortowany po dacie).
+- **parser** (playback_sources + scenes z 24 h wobec 14 dni wczesniej): skok lub zanik
+  performerow na scene, udzial scen bez dlugosci, bez miniatury. Lapie zmiane motywu
+  strony w pierwszej dobie, bez czekania na zgloszenie z apki.
+- **playback** (playback_events 24 h wobec 7 dni): spadek udanych odtworzen per origin.
+
+`to_sentry=False` daje sam odczyt (reczne sprawdzenie bez zasmiecania issue).
+"""
+from __future__ import annotations
+
+import logging
+import statistics
+from collections import defaultdict
+from typing import Any
+
+from sqlalchemy import text
+
+from app.db import session_scope
+
+log = logging.getLogger(__name__)
+
+# listing
+_LISTING_EMPTY_STREAK = 3        # tyle ostatnich przebiegow z seen=0/bledem z rzedu
+_LISTING_ALLKNOWN_STREAK = 8     # tyle przebiegow (48 h przy co 6 h) z seen>0 i new=0
+_LISTING_ALLKNOWN_MAX_GAP_DAYS = 2.0  # tylko zrodla publikujace co najmniej co 2 dni
+# parser
+_PARSER_MIN_RECENT = 10
+_PARSER_MIN_BASE = 30
+# playback
+_PLAYBACK_MIN_ATTEMPTS = 30
+_PLAYBACK_MIN_BASE_ATTEMPTS = 50
+
+
+def _cadence_gap_days(session, origin: str, lookback_days: int = 60) -> float | None:
+    """Mediana odstepu (w dobach) miedzy dniami, w ktorych zrodlo dalo nowe pozycje."""
+    rows = session.execute(
+        text(
+            "SELECT DISTINCT created_at::date AS d FROM playback_sources "
+            "WHERE origin = :o AND created_at > now() - make_interval(days => :lb) ORDER BY d"
+        ),
+        {"o": origin, "lb": lookback_days},
+    ).all()
+    days = [r[0] for r in rows]
+    if len(days) < 5:
+        return None
+    gaps = [(b - a).days for a, b in zip(days, days[1:])]
+    return float(statistics.median(gaps)) if gaps else None
+
+
+def check_listing_health(session) -> list[dict[str, Any]]:
+    rows = session.execute(
+        text(
+            "SELECT sitetag, started_at, seen, new, skipped, failed, note FROM browse_runs "
+            "WHERE started_at > now() - interval '60 hours' ORDER BY sitetag, started_at DESC"
+        )
+    ).all()
+    by_tag: dict[str, list] = defaultdict(list)
+    for r in rows:
+        by_tag[r.sitetag].append(r)
+    out: list[dict[str, Any]] = []
+    for sitetag, runs in by_tag.items():
+        last = runs[:_LISTING_EMPTY_STREAK]
+        if len(last) >= _LISTING_EMPTY_STREAK and all(r.failed or r.seen == 0 for r in last):
+            note = next((r.note for r in last if r.note), None)
+            out.append(
+                {
+                    "sitetag": sitetag,
+                    "problem": "empty",
+                    "runs": len(last),
+                    "note": note,
+                    "level": "warning",
+                    "message": (
+                        f"listing-watchdog: {sitetag} listing pusty albo blad w {len(last)} "
+                        f"przebiegach z rzedu" + (f" ({note[:120]})" if note else "")
+                    ),
+                }
+            )
+            continue
+        last = runs[:_LISTING_ALLKNOWN_STREAK]
+        if len(last) >= _LISTING_ALLKNOWN_STREAK and all(
+            (not r.failed) and r.seen > 0 and r.new == 0 for r in last
+        ):
+            gap = _cadence_gap_days(session, f"tube:{sitetag}")
+            if gap is not None and gap <= _LISTING_ALLKNOWN_MAX_GAP_DAYS:
+                seen = sum(r.seen for r in last)
+                out.append(
+                    {
+                        "sitetag": sitetag,
+                        "problem": "all_known",
+                        "runs": len(last),
+                        "seen": seen,
+                        "gap_days": gap,
+                        "level": "warning",
+                        "message": (
+                            f"listing-watchdog: {sitetag} {len(last)} przebiegi z rzedu, {seen} pozycji "
+                            f"i 0 nowych, a zrodlo publikuje co {gap:.0f} d: zla strona listingu "
+                            f"(rotacja, brak sortu po dacie) albo serwis stanal"
+                        ),
+                    }
+                )
+    return out
+
+
+def check_parser_quality(session) -> list[dict[str, Any]]:
+    rows = session.execute(
+        text(
+            """
+            WITH w AS (
+              SELECT ps.origin, ps.scene_id, ps.thumbnail_url,
+                     (ps.created_at > now() - interval '24 hours') AS recent
+              FROM playback_sources ps
+              WHERE ps.origin LIKE 'tube:%' AND ps.created_at > now() - interval '15 days'
+            )
+            SELECT w.origin, w.recent, count(*) AS n,
+                   avg((SELECT count(*) FROM scene_performers sp WHERE sp.scene_id = w.scene_id)) AS perf,
+                   avg(CASE WHEN x.duration_sec IS NULL THEN 1.0 ELSE 0.0 END) AS dur_null,
+                   avg(CASE WHEN w.thumbnail_url IS NULL THEN 1.0 ELSE 0.0 END) AS thumb_null
+            FROM w JOIN scenes x ON x.id = w.scene_id
+            GROUP BY 1, 2
+            """
+        )
+    ).all()
+    stats: dict[str, dict[bool, Any]] = defaultdict(dict)
+    for r in rows:
+        stats[r.origin][bool(r.recent)] = r
+    out: list[dict[str, Any]] = []
+    for origin, parts in stats.items():
+        rec, base = parts.get(True), parts.get(False)
+        if rec is None or base is None or rec.n < _PARSER_MIN_RECENT or base.n < _PARSER_MIN_BASE:
+            continue
+        pr, pb = float(rec.perf or 0), float(base.perf or 0)
+        dr, db = float(rec.dur_null or 0), float(base.dur_null or 0)
+        tr, tb = float(rec.thumb_null or 0), float(base.thumb_null or 0)
+        probs: list[tuple[str, str]] = []
+        if pb >= 0 and pr >= 2.5 * max(pb, 0.4) and pr - pb >= 3:
+            probs.append(("perf_inflated", f"obsada rozdeta: {pr:.1f} performerow na scene (wczesniej {pb:.1f}), karty related?"))
+        if pb >= 0.8 and pr <= 0.25 * pb:
+            probs.append(("perf_lost", f"obsada zniknela: {pr:.2f} na scene (wczesniej {pb:.2f})"))
+        if dr >= 0.5 and db <= 0.25:
+            probs.append(("duration_lost", f"{dr*100:.0f}% scen bez dlugosci (wczesniej {db*100:.0f}%), apka je ukrywa"))
+        if tr >= 0.5 and tb <= 0.25:
+            probs.append(("thumb_lost", f"{tr*100:.0f}% zrodel bez miniatury (wczesniej {tb*100:.0f}%)"))
+        for metric, desc in probs:
+            out.append(
+                {
+                    "origin": origin,
+                    "metric": metric,
+                    "recent_n": int(rec.n),
+                    "base_n": int(base.n),
+                    "level": "warning",
+                    "message": f"parser-watchdog: {origin} {desc} (24 h: {int(rec.n)} scen, baza {int(base.n)})",
+                }
+            )
+    return out
+
+
+def check_playback_health(session) -> list[dict[str, Any]]:
+    rows = session.execute(
+        text(
+            """
+            SELECT origin,
+                   count(*) FILTER (WHERE created_at > now() - interval '24 hours') AS att24,
+                   avg(CASE WHEN status = 'success' THEN 1.0 ELSE 0.0 END)
+                       FILTER (WHERE created_at > now() - interval '24 hours') AS ok24,
+                   count(*) FILTER (WHERE created_at <= now() - interval '24 hours') AS attb,
+                   avg(CASE WHEN status = 'success' THEN 1.0 ELSE 0.0 END)
+                       FILTER (WHERE created_at <= now() - interval '24 hours') AS okb
+            FROM playback_events
+            WHERE created_at > now() - interval '8 days'
+            GROUP BY 1
+            """
+        )
+    ).all()
+    out: list[dict[str, Any]] = []
+    for r in rows:
+        att24 = int(r.att24 or 0)
+        if att24 < _PLAYBACK_MIN_ATTEMPTS or r.ok24 is None:
+            continue
+        ok24 = float(r.ok24)
+        okb = float(r.okb) if r.okb is not None else None
+        attb = int(r.attb or 0)
+        bucket = None
+        if ok24 < 0.3:
+            bucket, level = "lt30", "error"
+        elif ok24 < 0.6 and okb is not None and attb >= _PLAYBACK_MIN_BASE_ATTEMPTS and okb - ok24 >= 0.25:
+            bucket, level = "lt60", "warning"
+        if not bucket:
+            continue
+        base_txt = f", tydzien wczesniej {okb*100:.0f}%" if okb is not None else ""
+        out.append(
+            {
+                "origin": r.origin,
+                "bucket": bucket,
+                "ok24": round(ok24, 3),
+                "att24": att24,
+                "level": level,
+                "message": f"playback-watchdog: {r.origin} {ok24*100:.0f}% udanych z {att24} prob w 24 h{base_txt}",
+            }
+        )
+    return out
+
+
+def _emit(items: list[dict[str, Any]], *, kind: str, key: str, sub: str) -> None:
+    try:
+        import sentry_sdk
+    except Exception:  # pragma: no cover
+        return
+    for x in items:
+        try:
+            with sentry_sdk.push_scope() as scope:
+                scope.level = x["level"]
+                scope.set_tag("watchdog_kind", kind)
+                scope.set_tag("watchdog_origin", str(x.get(key)))
+                for k, v in x.items():
+                    if k not in ("message", "level"):
+                        scope.set_extra(k, v)
+                scope.fingerprint = [kind, str(x.get(key)), str(x.get(sub))]
+                sentry_sdk.capture_message(x["message"])
+        except Exception:  # pragma: no cover
+            log.exception("quality-watchdog: Sentry capture failed")
+
+
+def run_quality_watchdogs(*, to_sentry: bool = True) -> dict[str, Any]:
+    with session_scope() as s:
+        listing = check_listing_health(s)
+        parser = check_parser_quality(s)
+        playback = check_playback_health(s)
+    for group, name in ((listing, "listing"), (parser, "parser"), (playback, "playback")):
+        for x in group:
+            log.warning("quality-watchdog [%s]: %s", name, x["message"])
+    if to_sentry:
+        _emit(listing, kind="listing-watchdog", key="sitetag", sub="problem")
+        _emit(parser, kind="parser-watchdog", key="origin", sub="metric")
+        _emit(playback, kind="playback-watchdog", key="origin", sub="bucket")
+    summary = {"listing": len(listing), "parser": len(parser), "playback": len(playback)}
+    if not any(summary.values()):
+        log.info("quality-watchdog: bez uwag")
+    return {"summary": summary, "listing": listing, "parser": parser, "playback": playback}