goon/app/scheduler/deep_crawl.py

"""Deep-crawl pełnych katalogów browse-tube'ów (Faza 2a — "ingest-all").

Browse scrapery (ALL_BROWSE_SCRAPERS) mają pełne listingi (np. porndoe >62k scen),
a my mieliśmy ~3% katalogu (search-by-performer + top-N browse). Ten job paginuje
DEEP: per tube trzyma kursor `last_page`, co run crawluje kolejne N stron od kursora,
idempotentnie (resolver pomija znane po raw_hash). Po dojściu do końca katalogu
(pusty listing) tube jest `exhausted`; gdy wszystkie exhausted — reset kursorów i
re-sweep od page 1 (incremental: łapie nowe + potwierdza istniejące).

Pilot 2026-06-03 (porndoe ogon, strony 64-110): 1119 nowych scen, 100% grywalne +
100% otagowane, 0% canonical-overlap (czysto addytywny content, nie duplikuje TPDB/
StashDB). ~1.2s/scenę.

Stan w JSON (mounted `app/_state/deepcrawl_state.json`) — wznawia między runami bez
migracji DB. Round-robin po `updated_at` → wszystkie tube'y postępują równomiernie.
"""
from __future__ import annotations

import json
import logging
import time
from pathlib import Path

from app.config import get_settings
from app.connectors.direct_scrapers import ALL_BROWSE_SCRAPERS, SCRAPER_SOURCE_NAME
from app.db import session_scope
from app.ingest import _process_scene, get_or_create_source
from app.models.source import SourceKind

log = logging.getLogger(__name__)

_DEFAULT_STATE = Path(__file__).resolve().parent.parent / "_state" / "deepcrawl_state.json"

# Per-tube depth cap (stron). Mega-tube'y (xvideos ~13M scen) crawlowane do końca
# zmonopolizowałyby round-robin i zalały bazę — capujemy do ~najnowszych N stron, potem
# exhausted→reset (incremental re-sweep świeżych). Tube'y skończone (porndoe/eporner) bez
# capu (None) → naturalny koniec katalogu. xvideos /new/ ~27 scen/stronę → 1800 ≈ ~50k.
_PAGE_CAP: dict[str, int] = {
    "xvideoscom": 1800,
}

# Miękki budżet czasu na run (s). Detail-fetch scrapery (per-scena fetch strony, np.
# przez wolne proxy) potrafią przekroczyć hard-timeout 3600s z _job_deep_crawl → run
# ubijany w locie, kursor NIE zapisany (orphan thread), tube zero postępu + alert
# GOON-V. Budżet < hard-timeout: przerywamy PO skończonej stronie, zapisujemy kursor,
# wracamy czysto, następny run kontynuuje. Margines 600s na dokończenie strony w toku.
_RUN_BUDGET_SEC = 3000

# Strony <= tego progu = "latest" (świeże posty tube'a) → nowe sceny stąd to genuine
# nowości. Głębsze strony = backfill starego katalogu → nowe sceny stąd dostają
# Scene.backfill=True i NIE liczą się jako "nowe" w ulubionych (tube podaje datę importu
# jako release_date, więc stary backfill inaczej udaje świeżość). Reset kursora re-sweepuje
# od strony 1, więc genuinnie nowe zawsze łapią się na stronach 1-2.
_LATEST_PAGE_THRESHOLD = 2


def _state_path() -> Path:
    return Path(getattr(get_settings(), "deepcrawl_state_path", None) or _DEFAULT_STATE)


def _load_state() -> dict:
    p = _state_path()
    if p.exists():
        try:
            return json.loads(p.read_text(encoding="utf-8"))
        except Exception as e:  # pragma: no cover - obronnie
            log.warning("deep-crawl: bad state file %s: %s — starting fresh", p, e)
    return {}


def _save_state(state: dict) -> None:
    p = _state_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2), encoding="utf-8")
    tmp.replace(p)  # atomic


def _browse_scrapers() -> dict:
    """{sitetag: scraper_cls} dla zarejestrowanych browse-scraperów."""
    out: dict = {}
    for cls in ALL_BROWSE_SCRAPERS:
        try:
            out[cls().sitetag] = cls
        except Exception as e:  # pragma: no cover
            log.warning("deep-crawl: skip scraper %s: %s", cls.__name__, e)
    return out


def _pick_target(state: dict, targets: list[str]) -> str | None:
    """Wybierz tube do crawla: najmniej-ostatnio-crawlowany, pomijając exhausted.
    Gdy wszystkie exhausted → reset (incremental re-sweep od page 1)."""
    live = [t for t in targets if not state.get(t, {}).get("exhausted")]
    if not live:
        if not targets:
            return None
        log.info("deep-crawl: all tubes exhausted → reset cursors for incremental re-sweep")
        for t in targets:
            state.setdefault(t, {})
            state[t]["exhausted"] = False
            state[t]["last_page"] = 0
        live = targets
    live.sort(key=lambda t: state.get(t, {}).get("updated_at", 0))
    return live[0]


def run_deep_crawl(*, pages_per_run: int = 60, sitetags: list[str] | None = None) -> dict:
    """Jeden run: wybierz tube, crawl kolejne `pages_per_run` stron od kursora, ingest.
    Zwraca podsumowanie (sitetag, zakres stron, counters, exhausted)."""
    scrapers = _browse_scrapers()
    targets = [t for t in (sitetags or list(scrapers)) if t in scrapers]
    if not targets:
        log.warning("deep-crawl: no browse scrapers / matching sitetags")
        return {}

    state = _load_state()
    sitetag = _pick_target(state, targets)
    if sitetag is None:
        return {}

    scraper = scrapers[sitetag]()
    cap = _PAGE_CAP.get(sitetag)  # mega-tube depth cap (None = crawl do końca katalogu)
    start = int(state.get(sitetag, {}).get("last_page", 0)) + 1
    # swept_once = tube był już RAZ przecrawlowany do końca. Wtedy jesteśmy w re-sweepie
    # (reset kursora), a każda NOWA scena to realny przyrost katalogu, NIE backfill,
    # niezależnie od numeru strony (review 23). Backfill tagujemy tylko na PIERWSZYM
    # przejściu, gdy strony > progu to nurkowanie w stary katalog.
    swept_once = bool(state.get(sitetag, {}).get("swept_once", False))
    end = start + pages_per_run - 1
    if cap is not None:
        end = min(end, cap)

    with session_scope() as session:
        src = get_or_create_source(session, kind=SourceKind.scraper, name=SCRAPER_SOURCE_NAME)
        source_id = src.id

    counters = {"seen": 0, "new": 0, "updated": 0, "skipped": 0, "errors": 0}
    t0 = time.time()
    last_done = start - 1
    exhausted = False
    budget_hit = False

    st = state.setdefault(sitetag, {})

    def _persist() -> None:
        # Zapisz kursor po KAŻDEJ skończonej stronie. Bez tego stan leciał raz na końcu
        # runu, więc hard-timeout ubijający run mid-page (mimo miękkiego budżetu, gdy
        # pojedyncza strona przez wolne proxy przekroczy margines) gubił cały postęp
        # (orphan thread, GOON-V). Per-page persist = najgorszy przypadek to powtórka
        # jednej strony. Zapis jest atomowy (tmp+replace) i tani względem fetcha strony.
        st["last_page"] = last_done
        st["exhausted"] = exhausted
        st["swept_once"] = swept_once or exhausted
        st["updated_at"] = int(time.time())
        _save_state(state)

    if cap is not None and start > cap:
        # kursor osiągnął per-tube cap → traktuj jak koniec katalogu (reset re-sweepuje od 1)
        exhausted = True
    else:
        for page in range(start, end + 1):
            scenes = scraper.crawl_page(page)
            if scenes is None:
                # transient fetch-fail listingu — NIE awansuj kursora, następny run powtórzy
                break
            if not scenes:
                log.info("deep-crawl %s: empty page %d → catalog end (exhausted)", sitetag, page)
                exhausted = True
                last_done = page
                break
            page_is_backfill = (page > _LATEST_PAGE_THRESHOLD) and not swept_once
            for raw in scenes:
                counters["seen"] += 1
                try:
                    _process_scene(
                        source_id=source_id,
                        raw_scene=raw,
                        counters=counters,
                        backfill=page_is_backfill,
                    )
                except Exception:
                    counters["errors"] += 1
            last_done = page
            _persist()  # kursor po każdej stronie — mid-page hard-kill nie gubi postępu
            # Miękki budżet: stop po skończonej stronie (kursor już zapisany), zanim
            # hard-timeout ubije run mid-page (orphan thread, kursor zgubiony, GOON-V).
            if time.time() - t0 > _RUN_BUDGET_SEC:
                budget_hit = True
                log.warning(
                    "deep-crawl %s: run budget %ds hit at page %d (%d/%d stron), stop czysto, "
                    "kursor zapisany, kontynuacja w następnym runie",
                    sitetag, _RUN_BUDGET_SEC, page, page - start + 1, pages_per_run,
                )
                break
        if not budget_hit and cap is not None and last_done >= cap:
            log.info("deep-crawl %s: reached page cap %d (exhausted)", sitetag, cap)
            exhausted = True

    # Zapis terminalny: utrwala finalne flagi (exhausted z empty-page/cap, swept_once).
    # swept_once=True gdy tube dobił do końca katalogu — kolejne przejścia (po resecie
    # kursora) to re-sweep, gdzie nowe sceny są genuine, nie backfill.
    _persist()

    log.info(
        "deep-crawl %s pages %d-%d: %s exhausted=%s budget_hit=%s (%.0fs)",
        sitetag, start, last_done, counters, exhausted, budget_hit, time.time() - t0,
    )
    return {
        "sitetag": sitetag, "start": start, "end": last_done,
        "exhausted": exhausted, "budget_hit": budget_hit, **counters,
    }