goon/app/scheduler/duration_backfill.py

"""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, 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, 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, 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 = _take_batch(s, batch=batch)

    if not rows:
        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, _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:
            # 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 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(
                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),
        "gone": len(gone),
    }