From 09f389100a9b7c16c01bcd7255ee3e5c94191781 Mon Sep 17 00:00:00 2001 From: goon-foss Date: Wed, 14 Oct 2026 12:13:13 +0200 Subject: [PATCH] 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 | 41 +++ app/connectors/direct_scrapers/fullmovies.py | 11 + app/connectors/direct_scrapers/galaxyporn.py | 8 + app/connectors/direct_scrapers/hdporngg.py | 11 + app/connectors/direct_scrapers/hentaimama.py | 8 + app/connectors/direct_scrapers/javflix.py | 19 ++ app/extractors/duration_extract.py | 19 +- app/models/__init__.py | 2 + app/models/browse_run.py | 34 +++ app/scheduler/browse_latest.py | 27 ++ app/scheduler/duration_backfill.py | 101 +++++-- app/scheduler/ingest_watchdog.py | 45 +++- app/scheduler/jobs.py | 9 + app/scheduler/quality_watchdog.py | 252 ++++++++++++++++++ 14 files changed, 565 insertions(+), 22 deletions(-) create mode 100644 alembic/versions/20261014_0056_browse_runs.py create mode 100644 app/models/browse_run.py create mode 100644 app/scheduler/quality_watchdog.py diff --git a/alembic/versions/20261014_0056_browse_runs.py b/alembic/versions/20261014_0056_browse_runs.py new file mode 100644 index 0000000..afd1775 --- /dev/null +++ b/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") diff --git a/app/connectors/direct_scrapers/fullmovies.py b/app/connectors/direct_scrapers/fullmovies.py index 56e5c4a..93fa16f 100644 --- a/app/connectors/direct_scrapers/fullmovies.py +++ b/app/connectors/direct_scrapers/fullmovies.py @@ -7,6 +7,7 @@ Identyczny engine co hdporn.gg (KVS sponsor_groups stack): `/videos//`, 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, diff --git a/app/connectors/direct_scrapers/galaxyporn.py b/app/connectors/direct_scrapers/galaxyporn.py index 483360c..20219ef 100644 --- a/app/connectors/direct_scrapers/galaxyporn.py +++ b/app/connectors/direct_scrapers/galaxyporn.py @@ -62,6 +62,9 @@ _IFRAME_RE = re.compile(r']+src="([^"]+)"', re.IGNORECASE) _TITLE_STUDIO_DATE_RE = re.compile(r"^(?P[A-Za-z0-9][A-Za-z0-9._-]{1,30})\s+(?P\d{2})\s+(?P\d{2})\s+(?P\d{2})\s+(?P.+)$") # `[Onlyfans] Some Title` / `[Darkkotv] …` _TITLE_BRACKET_RE = re.compile(r"^\[(?P[^\]]{2,30})\]\s*(?P.+)$") +# 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( diff --git a/app/connectors/direct_scrapers/hdporngg.py b/app/connectors/direct_scrapers/hdporngg.py index 37282f7..f7e746f 100644 --- a/app/connectors/direct_scrapers/hdporngg.py +++ b/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, diff --git a/app/connectors/direct_scrapers/hentaimama.py b/app/connectors/direct_scrapers/hentaimama.py index 0034e30..16b9488 100644 --- a/app/connectors/direct_scrapers/hentaimama.py +++ b/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'
(.*?)
', re.DOTALL) _ANCHOR_TEXT_RE = re.compile(r">\s*([^<]{2,40}?)\s*") _DATE_RE = re.compile(r'\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ż ``, # 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, ) ], diff --git a/app/connectors/direct_scrapers/javflix.py b/app/connectors/direct_scrapers/javflix.py index 15388cd..247d26d 100644 --- a/app/connectors/direct_scrapers/javflix.py +++ b/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 `
Actress: `, +# 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'
(.*?)
', re.IGNORECASE | re.DOTALL) +_ACTOR_RE = re.compile(r'href="https://javflix\.cc/actor/([a-z0-9\-]+)/"[^>]*>([^<]+)', 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( diff --git a/app/extractors/duration_extract.py b/app/extractors/duration_extract.py index 6fe8ded..805f4ad 100644 --- a/app/extractors/duration_extract.py +++ b/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: `` w to + dlugosc TEJ sceny; karty powiazane maja tylko `40:33`, + 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, } diff --git a/app/models/__init__.py b/app/models/__init__.py index c4c38d3..9704a4d 100644 --- a/app/models/__init__.py +++ b/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", diff --git a/app/models/browse_run.py b/app/models/browse_run.py new file mode 100644 index 0000000..9c10035 --- /dev/null +++ b/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) diff --git a/app/scheduler/browse_latest.py b/app/scheduler/browse_latest.py index b366139..b85dc2a 100644 --- a/app/scheduler/browse_latest.py +++ b/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 diff --git a/app/scheduler/duration_backfill.py b/app/scheduler/duration_backfill.py index b5f2953..f2153fa 100644 --- a/app/scheduler/duration_backfill.py +++ b/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), + } diff --git a/app/scheduler/ingest_watchdog.py b/app/scheduler/ingest_watchdog.py index 5e22514..e513c68 100644 --- a/app/scheduler/ingest_watchdog.py +++ b/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( diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 641c83e..d8df5b2 100644 --- a/app/scheduler/jobs.py +++ b/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: diff --git a/app/scheduler/quality_watchdog.py b/app/scheduler/quality_watchdog.py new file mode 100644 index 0000000..e913f3a --- /dev/null +++ b/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}