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). Sieć i parsowanie, bez dotykania bazy."""
    scene_id, page_url, origin = row
    from app.extractors._fetch import browser_get

    try:
        r = browser_get(page_url, timeout=_FETCH_TIMEOUT, follow_redirects=True)
        if r.status_code >= 400:
            return (scene_id, None)
        return (scene_id, extract_duration_for_origin(origin, r.text, page_url))
    except Exception as e:  # pojedyncza strona nie może wywrócić paczki
        log.debug("duration-backfill: %s padlo (%s)", page_url, e)
        return (scene_id, None)


def run_duration_backfill(*, batch: int = _DEFAULT_BATCH) -> dict[str, int]:
    """Bierze paczkę z kolejki, dolicza długość, zapisuje. Pusta kolejka = no-op."""
    with session_scope() as s:
        rows = s.execute(
            text(
                """
                SELECT scene_id, page_url, origin
                  FROM duration_backfill_queue
                 WHERE done_at IS NULL AND attempts < :max_attempts
                 ORDER BY attempts, created_at
                 LIMIT :batch
                """
            ),
            {"max_attempts": _MAX_ATTEMPTS, "batch": batch},
        ).fetchall()

    if not rows:
        return {"taken": 0, "filled": 0, "missed": 0}

    with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
        results = list(pool.map(_fetch_one, [tuple(r) for r in rows]))

    filled = [(sid, dur) for sid, dur in results if dur]
    missed = [sid for sid, dur in results if not dur]

    with session_scope() as s:
        for scene_id, dur in filled:
            # Warunek na NULL, bo między planem a zapisem scena mogła zostać scalona
            # albo wzbogacona z kanonu, a ten jest wiarygodniejszy niż tube.
            s.execute(
                text(
                    "UPDATE scenes SET duration_sec = :d "
                    " WHERE id = :id AND duration_sec IS NULL"
                ),
                {"d": dur, "id": scene_id},
            )
            s.execute(
                text(
                    "UPDATE duration_backfill_queue SET done_at = now() WHERE scene_id = :id"
                ),
                {"id": scene_id},
            )
        if missed:
            # Wyczerpane próby zamykamy, żeby kolejka nie kręciła się w kółko na martwych.
            s.execute(
                text(
                    """
                    UPDATE duration_backfill_queue
                       SET attempts = attempts + 1,
                           done_at = CASE WHEN attempts + 1 >= :max_attempts
                                          THEN now() ELSE NULL END
                     WHERE scene_id = ANY(:ids)
                    """
                ),
                {"ids": missed, "max_attempts": _MAX_ATTEMPTS},
            )

    return {"taken": len(rows), "filled": len(filled), "missed": len(missed)}