goon/app/scheduler/taxonomy_counts.py

"""Refresh denormalizowanych `scene_count` na tags / performers / studios.

Liczniki są utrzymywane w tle (zamiast liczone per-request) bo agregacja po 6.3M
scene_tags / 3M scene_performers z EXISTS do 1.15M playback_sources zajmuje ~4.3s,
nie do zaakceptowania w hot-path UI (/tags, /performers, /studios, /favorites).

Definicja: scene_count = liczba scen z danym tagiem/performerem/studiem mających ≥1
  playback_source z dead_at IS NULL **ORAZ** duration_sec >= 60. Filtr duration mirror
  mobilnego `min_duration_sec=60` (api.ts default), listy scen w apce ZAWSZE filtrują
  ≥60s, więc badge/licznik musi liczyć to samo, inaczej "119 w ulubionych, 39 po wejściu"
  (bug-report 2026-06-01 Safira Yakkuza). Wymaga propagacji duration z playback_source na
  Scene.duration_sec (backfill + _effective_duration w resolverze, 2026-06-01).

Każdy UPDATE robi pełny LEFT JOIN (tag/performer/studio) ⨝ agregat → ustawia 0 dla
sierot. `IS DISTINCT FROM` pomija przepisywanie niezmienionych wierszy (mniej WAL/bloat).
Całość ~5-10s, leci co kilka godzin, counts do tego stale, co dla sortu "popular" i
badge "(N)" jest bez znaczenia.
"""
from __future__ import annotations

import logging

from sqlalchemy import text

from app.db import session_scope
from app.verticals import ALL_VERTICAL_ORIGINS, VERTICALS

log = logging.getLogger(__name__)

# Mirror mobilnego api.ts `min_duration_sec ?? 60`, listy scen filtrują ≥60s.
_MIN_DURATION_SEC = 60

# Wspólny predykat: scena ma ≥1 żywy playback_source.
_LIVE = (
    "EXISTS (SELECT 1 FROM playback_sources ps "
    "WHERE ps.scene_id = {scene_col} AND ps.dead_at IS NULL)"
)

# Sceny z wertykałów (JAV, animacja) NIE wchodzą do głównego feedu, więc nie mogą
# wchodzić do licznika, który ten feed opisuje. Bez tego lista tagów obiecywała treść,
# której filtr i tak nie pokaże: 2664 tagi wyłącznie animacyjne (nazwy postaci, serie)
# i cały słownik JAV. To druga strona bramy z `app/api/scenes.py`, musi się z nią zgadzać.
_NOT_VERTICAL = (
    "NOT EXISTS (SELECT 1 FROM playback_sources pv "
    "WHERE pv.scene_id = {scene_col} AND pv.dead_at IS NULL "
    "AND pv.origin = ANY(:vertical_origins))"
)

# Duration check po scene_id (dla tags/performers gdzie nie mamy scene-row w scope).
_DUR = "(SELECT s2.duration_sec FROM scenes s2 WHERE s2.id = {scene_col}) >= " + str(_MIN_DURATION_SEC)

_TAGS_SQL = text(
    f"""
    UPDATE tags t SET scene_count = COALESCE(a.c, 0)
    FROM tags base
    LEFT JOIN (
        SELECT st.tag_id, count(*) AS c
        FROM scene_tags st
        WHERE {_LIVE.format(scene_col="st.scene_id")}
          AND {_DUR.format(scene_col="st.scene_id")}
          AND {_NOT_VERTICAL.format(scene_col="st.scene_id")}
        GROUP BY st.tag_id
    ) a ON a.tag_id = base.id
    WHERE t.id = base.id AND t.scene_count IS DISTINCT FROM COALESCE(a.c, 0)
    """
)

_PERFORMERS_SQL = text(
    f"""
    UPDATE performers p SET scene_count = COALESCE(a.c, 0)
    FROM performers base
    LEFT JOIN (
        SELECT sp.performer_id, count(*) AS c
        FROM scene_performers sp
        WHERE {_LIVE.format(scene_col="sp.scene_id")}
          AND {_DUR.format(scene_col="sp.scene_id")}
          AND {_NOT_VERTICAL.format(scene_col="sp.scene_id")}
        GROUP BY sp.performer_id
    ) a ON a.performer_id = base.id
    WHERE p.id = base.id AND p.scene_count IS DISTINCT FROM COALESCE(a.c, 0)
    """
)

# scenes.max_quality_p: najwyzsza ZNANA jakosc sposrod zywych zrodel sceny. Odswiezamy
# tu, a nie przy ingescie, bo `quality` moze dojsc pozniej (re-scrape, nowe zrodlo po
# merge), a zapytanie jest tanie: dotyka tylko scen, ktore maja zrodlo z jakoscia.
# Sceny, ktore jakosc STRACILY (zrodlo umarlo), musza wrocic do NULL, stad druga czesc.
_MAX_QUALITY_SQL = text(
    r"""
    WITH q AS (
        SELECT ps.scene_id,
               max(CAST(COALESCE(SUBSTRING(ps.quality FROM '\d+'), '0') AS INTEGER)) AS maxq
        FROM playback_sources ps
        WHERE ps.dead_at IS NULL AND ps.quality IS NOT NULL
        GROUP BY ps.scene_id
    )
    UPDATE scenes s SET max_quality_p = q.maxq
    FROM q
    WHERE q.scene_id = s.id
      AND s.max_quality_p IS DISTINCT FROM q.maxq
    """
)

_MAX_QUALITY_CLEAR_SQL = text(
    """
    UPDATE scenes s SET max_quality_p = NULL
    WHERE s.max_quality_p IS NOT NULL
      AND NOT EXISTS (
          SELECT 1 FROM playback_sources ps
          WHERE ps.scene_id = s.id AND ps.dead_at IS NULL AND ps.quality IS NOT NULL
      )
    """
)

_STUDIOS_SQL = text(
    f"""
    UPDATE studios s SET scene_count = COALESCE(a.c, 0)
    FROM studios base
    LEFT JOIN (
        SELECT sc.studio_id, count(*) AS c
        FROM scenes sc
        WHERE sc.studio_id IS NOT NULL AND {_LIVE.format(scene_col="sc.id")}
          AND sc.duration_sec >= {_MIN_DURATION_SEC}
          AND {_NOT_VERTICAL.format(scene_col="sc.id")}
        GROUP BY sc.studio_id
    ) a ON a.studio_id = base.id
    WHERE s.id = base.id AND s.scene_count IS DISTINCT FROM COALESCE(a.c, 0)
    """
)


# Tagi wertykalu (lista w zakladce JAV/animacja): sceny z zywym zrodlem z jego originow,
# bez progu dlugosci, ta sama definicja co dawny agregat w /tags?vertical=. Podmiana
# calego kompletu w jednej transakcji, to kilka tysiecy wierszy (migracja 0047).
_VERTICAL_TAGS_DELETE_SQL = text("DELETE FROM tag_vertical_counts WHERE vertical = :vertical")
_VERTICAL_TAGS_INSERT_SQL = text(
    """
    INSERT INTO tag_vertical_counts (vertical, tag_id, scene_count)
    SELECT :vertical, st.tag_id, count(*)
    FROM scene_tags st
    JOIN (
        SELECT DISTINCT ps.scene_id FROM playback_sources ps
        WHERE ps.dead_at IS NULL AND ps.origin = ANY(:origins)
    ) v ON v.scene_id = st.scene_id
    GROUP BY st.tag_id
    """
)


def refresh_taxonomy_counts() -> dict[str, int]:
    """Przelicza scene_count dla tags/performers/studios. Zwraca rowcount per tabela
    (ile wierszy faktycznie się zmieniło)."""
    out: dict[str, int] = {}
    with session_scope() as session:
        for vertical, origins in VERTICALS.items():
            session.execute(_VERTICAL_TAGS_DELETE_SQL, {"vertical": vertical})
            out[f"tags_{vertical}"] = session.execute(
                _VERTICAL_TAGS_INSERT_SQL, {"vertical": vertical, "origins": list(origins)}
            ).rowcount or 0
            session.commit()
        for name, stmt in (
            ("tags", _TAGS_SQL),
            ("performers", _PERFORMERS_SQL),
            ("studios", _STUDIOS_SQL),
        ):
            res = session.execute(stmt, {"vertical_origins": list(ALL_VERTICAL_ORIGINS)})
            out[name] = res.rowcount or 0
            # Commit per-tabela, długie transakcje trzymałyby locki na hot tables.
            session.commit()
        # max_quality_p, bez parametru wertykałów: jakość opisuje samo źródło, więc
        # dotyczy tak samo głównego feedu jak wertykałów.
        for name, stmt in (
            ("max_quality", _MAX_QUALITY_SQL),
            ("max_quality_cleared", _MAX_QUALITY_CLEAR_SQL),
        ):
            out[name] = session.execute(stmt).rowcount or 0
            session.commit()
    return out