diff --git a/alembic/versions/20260915_0038_scenes_studio_effective_date_idx.py b/alembic/versions/20260915_0038_scenes_studio_effective_date_idx.py new file mode 100644 index 0000000..b3f71b3 --- /dev/null +++ b/alembic/versions/20260915_0038_scenes_studio_effective_date_idx.py @@ -0,0 +1,37 @@ +"""Indeks pod liste scen studia i licznik "+N" ulubionych studiow. + +Oba sortuja po dacie premiery albo, gdy jej nie ma, dniu trafienia do katalogu (to samo +wyrazenie co ix_scenes_effective_date_desc z 0033), w obrebie jednego studia. Bez tego +Postgres czytal WSZYSTKIE sceny studia (Brazzers ma ich dziesiatki tysiecy), sprawdzal +kazdej playback i sortowal, a licznik dla 209 ulubionych studiow trwal 10-25 s. Z tym +indeksem idzie po kolei i konczy po 200 widocznych scenach na studio. + +CONCURRENTLY w autocommit_block, jak w 0035. Nieudany CONCURRENTLY zostawia indeks +INVALID; wtedy DROP INDEX i ponownie. + +Revision ID: 0038_scenes_studio_effdate_idx +Revises: 0037_clip_feed +Create Date: 2026-09-15 +""" +from __future__ import annotations + +from alembic import op + +revision = "0038_scenes_studio_effdate_idx" +down_revision: str | None = "0037_clip_feed" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_scenes_studio_effective_date_desc " + "ON scenes (studio_id, COALESCE(release_date, (created_at AT TIME ZONE 'UTC')::date) DESC, " + "created_at DESC, id DESC) WHERE studio_id IS NOT NULL" + ) + + +def downgrade() -> None: + with op.get_context().autocommit_block(): + op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_scenes_studio_effective_date_desc") diff --git a/app/api/favorites.py b/app/api/favorites.py index d6221b2..cbbebd5 100644 --- a/app/api/favorites.py +++ b/app/api/favorites.py @@ -21,18 +21,21 @@ Endpointy (studios): """ from __future__ import annotations +import logging +import threading +import time import uuid from datetime import UTC, datetime from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, Path, status +from fastapi import APIRouter, Depends, HTTPException, Path, Query, status from pydantic import BaseModel from sqlalchemy import exists, func, select from sqlalchemy.orm import Session from app.api.device import get_device_id from app.auth import require_api_key -from app.db import get_session +from app.db import get_session, session_scope from app.models.favorite_movie import FavoriteMovie from app.models.favorite_performer import FavoritePerformer from app.models.favorite_site import FavoriteSite @@ -42,6 +45,8 @@ from app.models.performer import Performer from app.models.scene import Scene, ScenePerformer from app.models.studio import Studio +log = logging.getLogger(__name__) + router = APIRouter( prefix="/favorites", tags=["favorites"], dependencies=[Depends(require_api_key)] ) @@ -61,55 +66,71 @@ def _new_counts(session: Session, device_id: str, *, kind: str) -> dict: """Policz per-favorite "+N nowych" = sceny created_at > last_seen_at, ale TYLKO wśród tych które user zobaczy na liście: te same filtry widoczności co list_scenes (żywy playback + blacklist + stub) i to samo okno (top-_FAVORITES_PAGE_CAP pod sortem - release_date desc), z pominięciem backfillu. Liczy w SQL (count(*) FILTER), zwraca - {group_id: n}, bez streamowania N*200 wierszy do Pythona. + `release_date` z list_scenes, czyli data premiery albo dzień dodania), z pominięciem + backfillu. Liczy w SQL (count(*) FILTER), zwraca {group_id: n}. kind="performer": grupuje po ScenePerformer.performer_id, join favorite_performers. kind="studio": grupuje po Scene.studio_id, join favorite_studios. Studia potrzebują odsiewu stub (mogą nie mieć performera); performerki nie (mają). """ - from sqlalchemy import and_, func + from sqlalchemy import and_, func, true - from app.api.scenes import blacklist_clauses, live_playback_exists, stub_exclusion_clause + from app.api.scenes import ( + _effective_date_desc, + blacklist_clauses, + live_playback_exists, + stub_exclusion_clause, + ) clauses = [live_playback_exists(), *blacklist_clauses(session, device_id)] + list_order = (_effective_date_desc(), Scene.created_at.desc(), Scene.id.desc()) - if kind == "performer": - gid = ScenePerformer.performer_id - last_seen = FavoritePerformer.last_seen_at - base = ( - select( - gid.label("gid"), - Scene.created_at.label("created_at"), - Scene.backfill.label("backfill"), - last_seen.label("last_seen"), - ) - .select_from(FavoritePerformer) - .join(ScenePerformer, ScenePerformer.performer_id == FavoritePerformer.performer_id) - .join(Scene, Scene.id == ScenePerformer.scene_id) - .where(FavoritePerformer.device_id == device_id) - ) - partition = ScenePerformer.performer_id - else: + if kind == "studio": + # Po kolei indeksem ix_scenes_studio_effective_date_desc (migracja 0038) i stop po + # 200 widocznych scenach na studio. Okno row_number() czytało wszystkie sceny + # wszystkich ulubionych studiów (324 tys. u najcięższego urządzenia, 10-25 s). clauses.append(stub_exclusion_clause()) - gid = Scene.studio_id - last_seen = FavoriteStudio.last_seen_at - base = ( + top = ( + select(Scene.created_at.label("created_at"), Scene.backfill.label("backfill")) + .where(Scene.studio_id == FavoriteStudio.studio_id, *clauses) + .order_by(*list_order) + .limit(_FAVORITES_PAGE_CAP) + .lateral("top") + ) + rows = session.execute( select( - gid.label("gid"), - Scene.created_at.label("created_at"), - Scene.backfill.label("backfill"), - last_seen.label("last_seen"), + FavoriteStudio.studio_id, + func.count().filter( + and_( + top.c.backfill.is_(False), + top.c.created_at > FavoriteStudio.last_seen_at, + ) + ), ) .select_from(FavoriteStudio) - .join(Scene, Scene.studio_id == FavoriteStudio.studio_id) + .join(top, true()) .where(FavoriteStudio.device_id == device_id) - ) - partition = Scene.studio_id + .group_by(FavoriteStudio.studio_id) + ).all() + return {gid_val: int(n) for gid_val, n in rows} + # Performerki: sceny idą przez scene_performers, więc indeks po dacie nie pomoże i + # trzeba przeczytać całą filmografię. To jest drogie (6-15 s przy ~1000 ulubionych), + # dlatego wynik siedzi w cache, patrz _cached_new_counts. + base = ( + select( + ScenePerformer.performer_id.label("gid"), + Scene.created_at.label("created_at"), + Scene.backfill.label("backfill"), + FavoritePerformer.last_seen_at.label("last_seen"), + ) + .select_from(FavoritePerformer) + .join(ScenePerformer, ScenePerformer.performer_id == FavoritePerformer.performer_id) + .join(Scene, Scene.id == ScenePerformer.scene_id) + .where(FavoritePerformer.device_id == device_id) + ) rn = func.row_number().over( - partition_by=partition, - order_by=(Scene.release_date.desc().nullslast(), Scene.created_at.desc()), + partition_by=ScenePerformer.performer_id, order_by=list_order ).label("rn") inner = base.add_columns(rn).where(*clauses).subquery() rows = session.execute( @@ -186,6 +207,94 @@ def _merged_new_counts(session: Session, device_id: str, *, kind: str) -> dict: return counts +# Cache liczników "+N" w pamięci procesu API (uvicorn chodzi jednym procesem). Liczenie +# dla urządzeń z setkami ulubionych trwa sekundy, a apka woła /favorites przy wejściu na +# listę scen, ulubione i każdą performerkę. Liczniki zmieniają się wolno (ingest), więc: +# świeży wynik oddajemy od razu, starszy niż _COUNTS_FRESH_SEC też od razu i liczymy +# nowy w tle, a dopiero starszy niż _COUNTS_MAX_AGE_SEC liczymy na miejscu. +# mark_seen zeruje licznik w cache, żeby badge zgasł natychmiast. +_COUNTS_FRESH_SEC = 300.0 +_COUNTS_MAX_AGE_SEC = 6 * 3600.0 +_counts_cache: dict[tuple[str, str], tuple[float, dict]] = {} +_counts_seen: dict[tuple[str, str], dict] = {} # gid -> kiedy oznaczono jako obejrzane +_counts_locks: dict[tuple[str, str], threading.Lock] = {} +_counts_guard = threading.Lock() + + +def _counts_lock(key: tuple[str, str]) -> threading.Lock: + with _counts_guard: + lock = _counts_locks.get(key) + if lock is None: + lock = _counts_locks[key] = threading.Lock() + return lock + + +def _store_counts(key: tuple[str, str], started: float, counts: dict) -> None: + with _counts_guard: + # mark_seen w trakcie liczenia: wynik policzono ze starym last_seen_at + for gid, at in _counts_seen.get(key, {}).items(): + if at >= started: + counts[gid] = 0 + _counts_seen.pop(key, None) + _counts_cache[key] = (time.monotonic(), counts) + stale = [k for k, (at, _) in _counts_cache.items() if time.monotonic() - at > _COUNTS_MAX_AGE_SEC] + for k in stale: + _counts_cache.pop(k, None) + _counts_locks.pop(k, None) + + +def _compute_counts(key: tuple[str, str]) -> None: + device_id, kind = key + started = time.monotonic() + with session_scope() as session: + counts = _merged_new_counts(session, device_id, kind=kind) + _store_counts(key, started, counts) + + +def _refresh_in_background(key: tuple[str, str]) -> None: + lock = _counts_lock(key) + if not lock.acquire(blocking=False): + return # już ktoś liczy + + def run() -> None: + try: + _compute_counts(key) + except Exception: # noqa: BLE001 + log.exception("favorites: odświeżenie liczników w tle nie wyszło %s", key) + finally: + lock.release() + + threading.Thread(target=run, name="fav-counts", daemon=True).start() + + +def _cached_new_counts(device_id: str, *, kind: str) -> dict: + key = (device_id, kind) + hit = _counts_cache.get(key) + if hit is not None: + age = time.monotonic() - hit[0] + if age < _COUNTS_MAX_AGE_SEC: + if age > _COUNTS_FRESH_SEC: + _refresh_in_background(key) + return dict(hit[1]) + # Brak w cache: liczymy teraz. Lock, żeby równoległe zapytania z kilku ekranów + # nie liczyły tego samego naraz, drugie poczeka i weźmie gotowy wynik. + with _counts_lock(key): + hit = _counts_cache.get(key) + if hit is not None and time.monotonic() - hit[0] < _COUNTS_FRESH_SEC: + return dict(hit[1]) + _compute_counts(key) + return dict(_counts_cache[key][1]) + + +def _mark_counts_seen(device_id: str, kind: str, gid: uuid.UUID) -> None: + key = (device_id, kind) + with _counts_guard: + _counts_seen.setdefault(key, {})[gid] = time.monotonic() + hit = _counts_cache.get(key) + if hit is not None and hit[1].get(gid): + hit[1][gid] = 0 + + class FavoriteOut(BaseModel): performer_id: uuid.UUID canonical_name: str @@ -206,6 +315,10 @@ class FavoriteListOut(BaseModel): def list_favorites( session: Annotated[Session, Depends(get_session)], device_id: Annotated[str, Depends(get_device_id)], + counts: bool = Query( + default=True, + description="false: bez liczników +N (ekrany, które chcą tylko wiedzieć, czy zapalić gwiazdkę)", + ), ) -> FavoriteListOut: rows = session.execute( select(FavoritePerformer, Performer) @@ -220,7 +333,7 @@ def list_favorites( # _job_refresh_taxonomy_counts), ta sama definicja co przed (sceny z żywym # playback). Wcześniej grouped count z EXISTS playback per-request. Migracja 0019. scene_counts: dict = {perf.id: perf.scene_count for _, perf in rows} - new_counts = _merged_new_counts(session, device_id, kind="performer") + new_counts = _cached_new_counts(device_id, kind="performer") if counts else {} items: list[FavoriteOut] = [] new_total = 0 @@ -297,6 +410,7 @@ def mark_seen( raise HTTPException(status_code=404, detail="not in favorites") fav.last_seen_at = datetime.now(UTC) session.commit() + _mark_counts_seen(device_id, "performer", performer_id) return SeenOut(performer_id=performer_id, last_seen_at=fav.last_seen_at) @@ -327,6 +441,10 @@ class FavoriteStudioListOut(BaseModel): def list_favorite_studios( session: Annotated[Session, Depends(get_session)], device_id: Annotated[str, Depends(get_device_id)], + counts: bool = Query( + default=True, + description="false: bez liczników +N i filmów (sam stan gwiazdki)", + ), ) -> FavoriteStudioListOut: rows = session.execute( select(FavoriteStudio, Studio) @@ -336,6 +454,24 @@ def list_favorite_studios( ).all() if not rows: return FavoriteStudioListOut(items=[], total=0, new_total=0) + if not counts: + return FavoriteStudioListOut( + items=[ + FavoriteStudioOut( + studio_id=st.id, + name=st.name, + slug=st.slug, + network=st.network, + scene_count=st.scene_count, + new_count=0, + last_seen_at=fav.last_seen_at, + created_at=fav.created_at, + ) + for fav, st in rows + ], + total=len(rows), + new_total=0, + ) # scene_count: zdenormalizowany Studio.scene_count (refresh w tle, migracja 0019). scene_counts: dict = {st.id: st.scene_count for _, st in rows} @@ -362,7 +498,7 @@ def list_favorite_studios( .group_by(Movie.studio_id) ).all() } - new_counts = _merged_new_counts(session, device_id, kind="studio") + new_counts = _cached_new_counts(device_id, kind="studio") items: list[FavoriteStudioOut] = [] new_total = 0 @@ -440,6 +576,7 @@ def mark_studio_seen( raise HTTPException(status_code=404, detail="not in favorites") fav.last_seen_at = datetime.now(UTC) session.commit() + _mark_counts_seen(device_id, "studio", studio_id) return SeenStudioOut(studio_id=studio_id, last_seen_at=fav.last_seen_at) diff --git a/app/api/scene_favorites.py b/app/api/scene_favorites.py index d99a1f8..09ad7cc 100644 --- a/app/api/scene_favorites.py +++ b/app/api/scene_favorites.py @@ -16,7 +16,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app.api.device import get_device_id -from app.api.scenes import _build_scene_out +from app.api.scenes import _build_scenes_out_batch from app.api.schemas import SceneOut from app.auth import require_api_key from app.db import get_session @@ -54,7 +54,11 @@ def list_scene_favorites( ) .all() ) - items = [_build_scene_out(session, scene, device_id=device_id) for scene, _ in rows] + # Wsadowo jak listy scen: ekran ulubionych rysuje ten sam kafelek, a pojedyncze + # _build_scene_out robilo 7 zapytan na scene (setki ulubionych = sekundy). + items = _build_scenes_out_batch( + session, [scene for scene, _ in rows], light=True, device_id=device_id + ) return SceneFavoriteListOut(items=items, total=len(items)) diff --git a/app/api/scenes.py b/app/api/scenes.py index d5bbde5..3b12823 100644 --- a/app/api/scenes.py +++ b/app/api/scenes.py @@ -9,7 +9,7 @@ from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel -from sqlalchemy import Date, distinct, exists, false, func, literal_column, select +from sqlalchemy import Date, and_, distinct, exists, false, func, literal_column, select, union from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -681,12 +681,35 @@ def similar_scenes( # Fill z tagów gdy performerów mało/brak (anonimowe tube sceny), inaczej pomijamy, # bo szeroki tag skanuje setki tysięcy scen. if tag_ids and len(cand_ids) < limit * 4: + # Pula z NAJRZADSZYCH tagów (najwięcej mówią o scenie), po _TAG_SEED_ROWS scen na + # tag, a wspólne tagi liczymy tylko dla tej puli. Grupowanie po wszystkich tagach + # sceny naraz czytało całe 'amateur' + 'teen-18' + 'handjob' (600 tys. wierszy, + # do 5 s). Sceny, które łączą z celem wyłącznie szerokie tagi, i tak są słabo + # podobne, więc ich pominięcie nic nie psuje. + _TAG_SEED_TAGS, _TAG_SEED_ROWS = 4, 1500 + tag_sizes = dict( + session.execute(select(Tag.id, Tag.scene_count).where(Tag.id.in_(tag_ids))).all() + ) + live_tags = [t for t in tag_ids if (tag_sizes.get(t) or 0) > 0] or list(tag_ids) + seed_tags = sorted(live_tags, key=lambda t: tag_sizes.get(t) or 0)[:_TAG_SEED_TAGS] + pool = union( + *[ + select(SceneTag.scene_id) + .where(SceneTag.tag_id == t, SceneTag.scene_id != scene_id) + .limit(_TAG_SEED_ROWS) + for t in seed_tags + ] + ).subquery("pool") + shared = func.count(distinct(SceneTag.tag_id)) rows = session.execute( - select(SceneTag.scene_id, func.count(distinct(SceneTag.tag_id)).label("c")) - .where(SceneTag.tag_id.in_(tag_ids), SceneTag.scene_id != scene_id) - .group_by(SceneTag.scene_id) - .having(func.count(distinct(SceneTag.tag_id)) >= 2) - .order_by(func.count(distinct(SceneTag.tag_id)).desc()) + select(pool.c.scene_id, shared.label("c")) + .join( + SceneTag, + and_(SceneTag.scene_id == pool.c.scene_id, SceneTag.tag_id.in_(tag_ids)), + ) + .group_by(pool.c.scene_id) + .having(shared >= 2) + .order_by(shared.desc()) .limit(_TAG_FILL_CAP) ).all() for sid, c in rows: @@ -1462,23 +1485,41 @@ def enrich_tags_from_tube( # (scene_id, tag_id), drugi flush dostaje UniqueViolation (GOON-H, 4 events # w 10h mimo wcześniejszego seen_tag_ids fix). ON CONFLICT skip'uje silently. from sqlalchemy.dialects.postgresql import insert as pg_insert - added = 0 - seen_tag_ids: set = set() + + from app.normalize.tag_stoplist import is_junk_tag + + # Istniejące tagi jednym zapytaniem po slugach. resolve_tag robi 1-4 zapytania na + # tag (Sentry N+1 przy 20-40 tagach ze strony), więc woła się go już tylko dla + # nowych nazw i dla slugów z końcówką litera+"2", które kanonizuje po swojemu. + slugs = {slugify(n) for n in tag_names} + direct = [ + s for s in slugs + if s and len(s) <= 120 and not is_junk_tag(s) + and not (len(s) > 1 and s[-1] == "2" and s[-2].isalpha()) + ] + by_slug = ( + {t.slug: t for t in session.execute(select(Tag).where(Tag.slug.in_(direct))).scalars()} + if direct else {} + ) + tag_ids: list = [] for name in tag_names: - norm = NormalizedTag(name=name, slug=slugify(name), external_id=None) - tag = resolve_tag(session, norm=norm) - if tag is None or tag.id in seen_tag_ids: - continue - seen_tag_ids.add(tag.id) - stmt = ( + slug = slugify(name) + tag = by_slug.get(slug) + if tag is None: + tag = resolve_tag(session, norm=NormalizedTag(name=name, slug=slug, external_id=None)) + if tag is not None and tag.id not in tag_ids: + tag_ids.append(tag.id) + added = 0 + if tag_ids: + # Jeden INSERT; sortowane id, żeby dwa równoległe wywołania dla tej samej sceny + # zakładały blokady w tej samej kolejności. + result = session.execute( pg_insert(SceneTag.__table__) - .values(scene_id=scene_id, tag_id=tag.id, source_id=None) + .values([{"scene_id": scene_id, "tag_id": t, "source_id": None} for t in sorted(tag_ids)]) .on_conflict_do_nothing(index_elements=["scene_id", "tag_id"]) + .returning(SceneTag.__table__.c.tag_id) ) - result = session.execute(stmt) - # rowcount == 1 gdy faktycznie wstawiony, 0 gdy ON CONFLICT skip - if result.rowcount and result.rowcount > 0: - added += 1 + added = len(result.fetchall()) session.commit() return EnrichTagsOut(scene_id=scene_id, added=added, tube_used=sitetag, tags=tag_names)