goon/app/api/clips.py

"""Klipy (feed Swipe): feed z afinicznością, pamięć widzianych, telemetria, ulubione,
blacklista, onboarding. Logika bez bazy w `app/clips_feed.py`.

**Feed** `GET /clips/feed`: strona = 80 % z puli afiniczności (tagi, nisze, twórcy z wag
urządzenia) + 20 % explore (okno po kluczu głównym od kursora, losowanie ważone lajkami).
Bez wag (nowe urządzenie, brak onboardingu) 100 % explore. Każdy podany klip ląduje
w `clip_seen` od razu, więc nigdy nie wraca. Widoczność: żywy, do `clips_feed_max_duration_sec`,
z plikiem, nie widziany, nie na blackliście (tag / nisza / twórca).

**Zdarzenia** `POST /clips/events`: paczka z apki (co ~5 s). Zapis do `clip_feed_events`,
podniesienie `clip_seen.outcome`, ulubione przy like/unlike, delty na wagi afiniczności
(`clips_feed.apply_event`).

Adresy plików i posterów są publiczne (`media.redgifs.com`), więc jadą do apki wprost,
bez image proxy: RedGifs nie sprawdza Referera, a proxy tylko dodałoby VPS do toru.
"""
from __future__ import annotations

import logging
import math
import threading
import time
import uuid
from datetime import UTC, datetime

from fastapi import APIRouter, Depends, HTTPException, Query, Response
from pydantic import BaseModel, Field
from sqlalchemy import String, cast, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from app import clips_feed as cf
from app.api.device import get_device_id
from app.auth import require_api_key
from app.config import get_settings
from app.db import get_session
from app.models.clip import Clip, ClipTag
from app.models.clip_feed import ClipAffinity, ClipBlacklist, ClipFavorite, ClipFeedEvent, ClipSeen

log = logging.getLogger(__name__)
router = APIRouter(prefix="/clips", tags=["clips"], dependencies=[Depends(require_api_key)])

_EXPLORE_WINDOW = 2000
_AFFINITY_TAGS, _AFFINITY_NICHES, _AFFINITY_CREATORS = 6, 3, 3
_PER_TAG, _PER_NICHE, _PER_CREATOR = 60, 60, 40
_UUID_ZERO = uuid.UUID(int=0)
_BLACKLIST_KINDS = frozenset({"tag", "niche", "creator"})
#: Zdarzenia oznaczajace, ze klip zostal obejrzany (zejscie ze strony), do clips.feed_views.
_VIEW_KINDS = frozenset({"complete", "skip", "impression", "autoswipe"})


# ---- schematy ---------------------------------------------------------------------------
class ClipOut(BaseModel):
    id: uuid.UUID
    source: str
    source_id: str
    title: str | None
    creator: str | None
    creator_verified: bool
    duration_sec: float | None
    width: int | None
    height: int | None
    has_audio: bool
    content_type: str | None
    niches: list[str]
    tags: list[str]
    hd_url: str | None
    sd_url: str | None
    poster_url: str | None
    thumb_url: str | None
    likes: int
    views: int
    published_at: datetime | None
    is_favorite: bool = False


class ClipListOut(BaseModel):
    items: list[ClipOut]
    limit: int
    offset: int
    has_more: bool
    #: Liczba wszystkich klipow tworcy (tylko przy filtrze `creator`, indeks ix_clips_creator).
    #: Strona tworcy w apce pokazuje ja w naglowku; przy innych listach nie liczymy.
    total: int | None = None


class ClipStatsOut(BaseModel):
    total: int
    alive: int
    dead: int
    with_audio: int
    verified: int
    by_source: dict[str, int]
    newest_published_at: datetime | None
    tags: int


class FeedItemOut(BaseModel):
    clip: ClipOut
    #: `affinity` | `explore`, do telemetrii i do pomiaru, czy personalizacja coś daje.
    slot: str
    is_fresh: bool


class FeedPageOut(BaseModel):
    items: list[FeedItemOut]
    next_cursor: str
    seed: str
    #: True gdy okno explore doszło do końca katalogu i zawinęło się na początek.
    wrapped: bool
    #: Ile klipów to urządzenie już widziało (pigułka „N nowych").
    served_total: int
    personalized: bool


class FeedEventIn(BaseModel):
    clip_id: uuid.UUID
    kind: str = Field(min_length=1, max_length=16)
    dwell_ms: int | None = Field(default=None, ge=0, le=3_600_000)
    reason: str | None = Field(default=None, max_length=160)
    slot: str | None = Field(default=None, max_length=12)
    position: int | None = Field(default=None, ge=0, le=32767)


class FeedEventsIn(BaseModel):
    session_id: str | None = Field(default=None, max_length=32)
    events: list[FeedEventIn] = Field(max_length=100)


class OnboardingIn(BaseModel):
    tags: list[str] = Field(default_factory=list, max_length=20)
    niches: list[str] = Field(default_factory=list, max_length=20)


class OnboardingOptionOut(BaseModel):
    value: str
    count: int


class OnboardingOptionsOut(BaseModel):
    tags: list[OnboardingOptionOut]
    niches: list[OnboardingOptionOut]


class AffinityOut(BaseModel):
    tags: dict[str, float]
    niches: dict[str, float]
    creators: dict[str, float]
    events_seen: int


class BlacklistItemOut(BaseModel):
    kind: str
    value: str


# ---- pomocnicze -------------------------------------------------------------------------
def _tags_for(session: Session, clip_ids: list[uuid.UUID]) -> dict[uuid.UUID, list[str]]:
    if not clip_ids:
        return {}
    out: dict[uuid.UUID, list[str]] = {}
    for cid, tag in session.execute(
        select(ClipTag.clip_id, ClipTag.tag).where(ClipTag.clip_id.in_(clip_ids)).order_by(ClipTag.tag)
    ):
        out.setdefault(cid, []).append(tag)
    return out


def _favorites_for(session: Session, device_id: str, clip_ids: list[uuid.UUID]) -> set[uuid.UUID]:
    if not clip_ids:
        return set()
    return {
        r[0]
        for r in session.execute(
            select(ClipFavorite.clip_id).where(ClipFavorite.device_id == device_id, ClipFavorite.clip_id.in_(clip_ids))
        )
    }


def _clip_out(r: Clip, tags: list[str], fav: bool = False) -> ClipOut:
    return ClipOut(
        id=r.id,
        source=r.source,
        source_id=r.source_id,
        title=r.title,
        creator=r.creator,
        creator_verified=r.creator_verified,
        duration_sec=r.duration_sec,
        width=r.width,
        height=r.height,
        has_audio=r.has_audio,
        content_type=r.content_type,
        niches=list(r.niches or []),
        tags=tags,
        hd_url=r.hd_url,
        sd_url=r.sd_url,
        poster_url=r.poster_url,
        thumb_url=r.thumb_url,
        likes=r.likes,
        views=r.views,
        published_at=r.published_at,
        is_favorite=fav,
    )


def _load_affinity(session: Session, device_id: str) -> tuple[ClipAffinity | None, cf.Affinity]:
    row = session.get(ClipAffinity, device_id)
    if row is None:
        return None, cf.Affinity()
    return row, cf.Affinity(
        tags=dict(row.tag_weights or {}), niches=dict(row.niche_weights or {}), creators=dict(row.creator_weights or {})
    )


def _save_affinity(session: Session, device_id: str, row: ClipAffinity | None, aff: cf.Affinity, events: int) -> None:
    if row is None:
        # Pierwszy zapis dla urzadzenia. Onboarding i zdarzenia z pierwszych klipow
        # przychodza naraz, oba widzialy brak wiersza i oba robily INSERT, drugi konczyl
        # sie UniqueViolation i bledem 500 (GOON-7X). Wstawiamy pusty wiersz z ON CONFLICT
        # i bierzemy go z blokada. Nasze wagi liczylismy od zera, wiec to same przyrosty:
        # dodajemy je do tego, co zdazylo zapisac rownolegle zadanie.
        session.execute(
            pg_insert(ClipAffinity)
            .values(device_id=device_id, tag_weights={}, niche_weights={}, creator_weights={}, events_seen=0)
            .on_conflict_do_nothing(index_elements=["device_id"])
        )
        row = session.get(ClipAffinity, device_id, with_for_update=True, populate_existing=True)
        for mine, theirs in ((aff.tags, row.tag_weights), (aff.niches, row.niche_weights), (aff.creators, row.creator_weights)):
            for k, v in (theirs or {}).items():
                mine[k] = round(mine.get(k, 0.0) + v, 3)
    row.tag_weights = aff.tags
    row.niche_weights = aff.niches
    row.creator_weights = aff.creators
    row.events_seen = (row.events_seen or 0) + events


def _blacklist(session: Session, device_id: str) -> dict[str, set[str]]:
    out: dict[str, set[str]] = {"tag": set(), "niche": set(), "creator": set()}
    for kind, value in session.execute(
        select(ClipBlacklist.kind, ClipBlacklist.value).where(ClipBlacklist.device_id == device_id)
    ):
        out.setdefault(kind, set()).add(value)
    return out


def _visible_clauses(device_id: str, bl: dict[str, set[str]]) -> list:
    clauses = [
        Clip.dead_at.is_(None),
        Clip.duration_sec.is_not(None),
        Clip.duration_sec <= get_settings().clips_feed_max_duration_sec,
        (Clip.hd_url.is_not(None)) | (Clip.sd_url.is_not(None)),
        ~select(ClipSeen.clip_id)
        .where(ClipSeen.device_id == device_id, ClipSeen.clip_id == Clip.id)
        .exists(),
    ]
    if bl["creator"]:
        clauses.append((Clip.creator.is_(None)) | (~Clip.creator.in_(list(bl["creator"]))))
    if bl["niche"]:
        clauses.append(~Clip.niches.overlap(list(bl["niche"])))
    if bl["tag"]:
        clauses.append(
            ~select(ClipTag.clip_id)
            .where(ClipTag.clip_id == Clip.id, ClipTag.tag.in_(list(bl["tag"])))
            .exists()
        )
    return clauses


_CAND_COLS = (
    Clip.id, Clip.creator, Clip.likes, Clip.published_at, Clip.niches,
    Clip.feed_views, Clip.feed_skips, Clip.feed_likes,
)


def _rows_to_candidates(session: Session, rows, *, with_tags: bool) -> list[cf.Candidate]:
    """Wiersze `_CAND_COLS` → Candidate. Tagi tylko na życzenie: pula explore punktuje
    wyłącznie lajkami, więc ciągnięcie tagów dla 300 kandydatów było czystym kosztem."""
    tags = _tags_for(session, [r[0] for r in rows]) if with_tags else {}
    return [
        cf.Candidate(
            id=r[0], creator=r[1], likes=r[2], published_at=r[3], tags=tags.get(r[0], []), niches=list(r[4] or []),
            feed_views=r[5] or 0, feed_skips=r[6] or 0, feed_likes=r[7] or 0,
        )
        for r in rows
    ]


def _candidates(session: Session, ids: list[uuid.UUID], clauses: list) -> list[cf.Candidate]:
    if not ids:
        return []
    rows = session.execute(select(*_CAND_COLS).where(Clip.id.in_(ids), *clauses)).all()
    return _rows_to_candidates(session, rows, with_tags=True)


def _explore_candidates(
    session: Session, pk: uuid.UUID, clauses: list, limit: int
) -> tuple[list[cf.Candidate], uuid.UUID, bool]:
    """Widoczni kandydaci z okna `_EXPLORE_WINDOW` id po `pk` (podzapytanie, bez
    przenoszenia 2000 uuid przez SQL). Zwraca (kandydaci, koniec okna, czy okno było
    ostatnie). Ostatnie okno = katalog wyczerpany od `pk`, następna strona zawija do zera."""
    window = (
        select(Clip.id).where(Clip.id > pk, Clip.dead_at.is_(None)).order_by(Clip.id).limit(_EXPLORE_WINDOW)
    ).subquery()
    # Postgres nie ma max(uuid); tekstowo porządek jest ten sam (małe hex, stały format).
    end_txt, n = session.execute(
        select(func.max(cast(window.c.id, String)), func.count()).select_from(window)
    ).one()
    if not n:
        return [], _UUID_ZERO, True
    end_id = uuid.UUID(end_txt)
    rows = session.execute(
        select(*_CAND_COLS).where(Clip.id.in_(select(window.c.id)), *clauses).order_by(Clip.id).limit(limit)
    ).all()
    last = n < _EXPLORE_WINDOW
    return _rows_to_candidates(session, rows, with_tags=False), (end_id if not last else _UUID_ZERO), last


# Rzadkosc tagow (IDF) do punktacji: ln(liczba klipow / klipy z tagiem). Liczenie to jedno
# GROUP BY po ~1,7 mln clip_tags (~1 s), slownik zmienia sie wolno, wiec cache 6 h w procesie.
_IDF_TTL_SEC = 6 * 3600
_idf_cache: tuple[float, dict[str, float]] | None = None
_idf_lock = threading.Lock()


def _tag_idf(session: Session) -> dict[str, float]:
    global _idf_cache
    hit = _idf_cache
    if hit is not None and time.monotonic() - hit[0] < _IDF_TTL_SEC:
        return hit[1]
    with _idf_lock:
        hit = _idf_cache
        if hit is not None and time.monotonic() - hit[0] < _IDF_TTL_SEC:
            return hit[1]
        n_all = session.execute(select(func.count()).select_from(Clip)).scalar_one()
        rows = session.execute(text("SELECT tag, count(*) FROM clip_tags GROUP BY tag")).all()
        idf = {t: math.log((n_all + 1) / (n + 1)) for t, n in rows}
        _idf_cache = (time.monotonic(), idf)
        return idf


def _affinity_candidate_ids(session: Session, aff: cf.Affinity, rng) -> set[uuid.UUID]:
    ids: set[uuid.UUID] = set()
    for tag in aff.top("tags", _AFFINITY_TAGS):
        start = uuid.UUID(int=rng.getrandbits(128))
        rows = session.execute(
            select(ClipTag.clip_id).where(ClipTag.tag == tag, ClipTag.clip_id > start).order_by(ClipTag.clip_id).limit(_PER_TAG)
        ).all()
        if len(rows) < _PER_TAG:
            rows += session.execute(
                select(ClipTag.clip_id).where(ClipTag.tag == tag).order_by(ClipTag.clip_id).limit(_PER_TAG - len(rows))
            ).all()
        ids.update(r[0] for r in rows)
    for niche in aff.top("niches", _AFFINITY_NICHES):
        start = uuid.UUID(int=rng.getrandbits(128))
        rows = session.execute(
            select(Clip.id).where(Clip.niches.overlap([niche]), Clip.id > start).order_by(Clip.id).limit(_PER_NICHE)
        ).all()
        ids.update(r[0] for r in rows)
    for creator in aff.top("creators", _AFFINITY_CREATORS):
        start = uuid.UUID(int=rng.getrandbits(128))
        rows = session.execute(
            select(Clip.id).where(Clip.creator == creator, Clip.id > start).order_by(Clip.id).limit(_PER_CREATOR)
        ).all()
        if len(rows) < _PER_CREATOR:
            rows += session.execute(
                select(Clip.id).where(Clip.creator == creator).order_by(Clip.id).limit(_PER_CREATOR - len(rows))
            ).all()
        ids.update(r[0] for r in rows)
    return ids


# ---- feed -------------------------------------------------------------------------------
@router.get("/feed", response_model=FeedPageOut)
def clip_feed(
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
    seed: str | None = Query(default=None, max_length=16),
    cursor: str | None = Query(default=None, max_length=200),
    limit: int = Query(default=10, ge=1, le=30),
) -> FeedPageOut:
    t0 = time.monotonic()
    decoded = cf.decode_cursor(cursor) if cursor else None
    if decoded:
        seed, pk, page = decoded
    else:
        seed = (seed or cf.new_seed())[:16]
        pk, page = cf.seed_start_uuid(seed), 0
    rng = cf.page_rng(seed, page)

    _row, aff = _load_affinity(session, device_id)
    personalized = not aff.is_empty()
    bl = _blacklist(session, device_id)
    clauses = _visible_clauses(device_id, bl)
    n_aff, n_explore = cf.split_counts(limit, personalized)

    picked_aff: list[cf.Candidate] = []
    if n_aff:
        cand_ids = _affinity_candidate_ids(session, aff, rng)
        picked_aff = cf.pick_affinity(
            _candidates(session, list(cand_ids), clauses),
            aff,
            n_aff,
            idf=_tag_idf(session),
            v1=cf.uses_v1_scoring(device_id),
        )
    taken = {c.id for c in picked_aff}

    # Explore: okno po PK; gdy w oknie brak widocznych, przesuwamy sie dalej (max 3 okna).
    # Braki z afinicznosci (za malo kandydatow po filtrach) dopelnia explore.
    picked_explore: list[cf.Candidate] = []
    wrapped = False
    next_pk = pk
    want = n_explore + max(0, n_aff - len(picked_aff))
    for _ in range(3):
        pool, next_pk, last = _explore_candidates(session, next_pk, clauses, limit=max(60, want * 15))
        pool = [c for c in pool if c.id not in taken]
        picked_explore += cf.pick_explore(pool, want - len(picked_explore), rng)
        taken.update(c.id for c in picked_explore)
        if last:
            wrapped = True
        if len(picked_explore) >= want or last:
            break

    page_items = cf.compose_page(picked_aff, picked_explore, rng)
    clip_ids = [c.id for c, _ in page_items]
    if clip_ids:
        session.execute(
            pg_insert(ClipSeen)
            .values([{"device_id": device_id, "clip_id": cid, "outcome": "served"} for cid in clip_ids])
            .on_conflict_do_nothing()
        )
    rows = {r.id: r for r in session.execute(select(Clip).where(Clip.id.in_(clip_ids))).scalars()} if clip_ids else {}
    tags = _tags_for(session, clip_ids)
    favs = _favorites_for(session, device_id, clip_ids)
    served_total = session.execute(
        select(func.count()).select_from(ClipSeen).where(ClipSeen.device_id == device_id)
    ).scalar_one()
    session.commit()

    items = [
        FeedItemOut(
            clip=_clip_out(rows[c.id], tags.get(c.id, []), c.id in favs),
            slot=slot,
            is_fresh=cf.is_fresh(rows[c.id].published_at),
        )
        for c, slot in page_items
        if c.id in rows
    ]
    log.info(
        "clips/feed: device=%s page=%d served=%d aff=%d explore=%d wrapped=%s ms=%d",
        device_id[:8], page, len(items), len(picked_aff), len(picked_explore), wrapped,
        int((time.monotonic() - t0) * 1000),
    )
    return FeedPageOut(
        items=items,
        next_cursor=cf.encode_cursor(seed, next_pk, page + 1),
        seed=seed,
        wrapped=wrapped,
        served_total=int(served_total),
        personalized=personalized,
    )


# ---- zdarzenia --------------------------------------------------------------------------
@router.post("/events", status_code=204)
def clip_events(
    body: FeedEventsIn,
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
) -> Response:
    events = [e for e in body.events if e.kind in cf.ALLOWED_KINDS]
    if not events:
        return Response(status_code=204)
    clip_ids = list({e.clip_id for e in events})
    clips = {r.id: r for r in session.execute(select(Clip).where(Clip.id.in_(clip_ids))).scalars()}
    tags = _tags_for(session, clip_ids)
    seen = {
        r.clip_id: r
        for r in session.execute(
            select(ClipSeen).where(ClipSeen.device_id == device_id, ClipSeen.clip_id.in_(clip_ids))
        ).scalars()
    }
    row, aff = _load_affinity(session, device_id)
    changed = 0
    # Liczniki przyjecia klipu u nas (clips.feed_*), z nich korzysta punktacja feedu.
    counters: dict[uuid.UUID, list[int]] = {}
    for e in events:
        d = counters.setdefault(e.clip_id, [0, 0, 0])
        if e.kind in _VIEW_KINDS:
            d[0] += 1
        if e.kind == "skip":
            d[1] += 1
        if e.kind == "like":
            d[2] += 1
        elif e.kind == "unlike":
            d[2] -= 1
    for cid, (dv, ds, dl) in counters.items():
        if dv or ds or dl:
            session.execute(
                text(
                    "UPDATE clips SET feed_views = feed_views + :v, feed_skips = feed_skips + :s, "
                    "feed_likes = greatest(feed_likes + :l, 0) WHERE id = :id"
                ),
                {"v": dv, "s": ds, "l": dl, "id": cid},
            )
    for e in events:
        session.add(
            ClipFeedEvent(
                device_id=device_id,
                session_id=body.session_id,
                clip_id=e.clip_id,
                kind=e.kind,
                dwell_ms=e.dwell_ms,
                reason=e.reason,
                slot=e.slot,
                position=e.position,
            )
        )
        clip = clips.get(e.clip_id)
        if clip is None:
            continue
        # clip_seen: podnies wynik, nigdy nie cofaj.
        outcome = e.kind if e.kind in cf.OUTCOME_RANK else None
        if outcome:
            s = seen.get(e.clip_id)
            if s is None:
                s = ClipSeen(device_id=device_id, clip_id=e.clip_id, outcome=outcome)
                session.add(s)
                seen[e.clip_id] = s
            elif cf.OUTCOME_RANK[outcome] > cf.OUTCOME_RANK.get(s.outcome, 0):
                s.outcome = outcome
        if e.kind == "like":
            session.execute(
                pg_insert(ClipFavorite).values(device_id=device_id, clip_id=e.clip_id).on_conflict_do_nothing()
            )
        elif e.kind == "unlike":
            session.execute(
                text("DELETE FROM clip_favorites WHERE device_id = :d AND clip_id = :c"),
                {"d": device_id, "c": e.clip_id},
            )
        if cf.apply_event(
            aff,
            kind=e.kind,
            dwell_ms=e.dwell_ms,
            reason=e.reason,
            tags=tags.get(e.clip_id, []),
            niches=list(clip.niches or []),
            creator=clip.creator,
        ):
            changed += 1
    if changed:
        _save_affinity(session, device_id, row, aff, changed)
    session.commit()
    broken = list({e.clip_id for e in events if e.kind == "error"})
    if broken:
        # W tle, zeby apka nie czekala na HEAD do RedGifs.
        threading.Thread(target=_check_broken, args=(broken,), name="clip-check", daemon=True).start()
    return Response(status_code=204)


def _check_broken(clip_ids: list) -> None:
    from app.scheduler.clips_ingest import check_clips_now

    try:
        check_clips_now(clip_ids)
    except Exception:  # noqa: BLE001
        log.exception("clips: sprawdzenie zgloszonych klipow nie wyszlo")


# ---- onboarding, afinicznosc ------------------------------------------------------------
_OPTIONS_CACHE: dict[str, tuple[float, OnboardingOptionsOut]] = {}


@router.get("/onboarding-options", response_model=OnboardingOptionsOut)
def onboarding_options(
    session: Session = Depends(get_session),
    q: str | None = Query(default=None, max_length=64, description="Szukaj w slowniku tagow"),
) -> OnboardingOptionsOut:
    """Najpopularniejsze tagi i nisze do wyboru przy pierwszym wejsciu w Edge.
    Liczone z bazy, cache 10 min w procesie (szukanie z pominieciem cache).

    `q` szuka po calym slowniku (1,7 tys. tagow), bo sama czolowka nie wystarcza: „anal"
    to 27. tag co do liczby klipow i wypadal poza liste (uwaga Jana 2026-09-16)."""
    if q and q.strip():
        needle = f"%{q.strip().lower()}%"
        found = [
            OnboardingOptionOut(value=t, count=n)
            for t, n in session.execute(
                select(ClipTag.tag, func.count())
                .where(ClipTag.tag.like(needle))
                .group_by(ClipTag.tag)
                .order_by(func.count().desc())
                .limit(30)
            )
        ]
        return OnboardingOptionsOut(tags=found, niches=[])
    hit = _OPTIONS_CACHE.get("v1")
    if hit and time.monotonic() - hit[0] < 600:
        return hit[1]
    tags = [
        OnboardingOptionOut(value=t, count=n)
        for t, n in session.execute(
            select(ClipTag.tag, func.count()).group_by(ClipTag.tag).order_by(func.count().desc()).limit(60)
        )
    ]
    niches = [
        OnboardingOptionOut(value=v, count=n)
        for v, n in session.execute(
            text("SELECT n, count(*) FROM clips, unnest(niches) n WHERE dead_at IS NULL GROUP BY n ORDER BY 2 DESC LIMIT 30")
        )
    ]
    out = OnboardingOptionsOut(tags=tags, niches=niches)
    _OPTIONS_CACHE["v1"] = (time.monotonic(), out)
    return out


@router.post("/onboarding", response_model=AffinityOut)
def onboarding(
    body: OnboardingIn,
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
) -> AffinityOut:
    # +8 za wskazany tag, wiecej niz polubienie klipu (+5): to jawna deklaracja, a feed
    # inaczej zbyt dlugo dochodzil do tego, co user i tak powiedzial na starcie (uwaga
    # Jana 2026-09-16). Ten sam endpoint sluzy pozniejszemu dostrajaniu z zakladki Edge,
    # wiec kolejne wskazania dokladaja sie do wag.
    row, aff = _load_affinity(session, device_id)
    for t in body.tags:
        t = t.strip().lower()[:64]
        if t:
            aff.tags[t] = round(aff.tags.get(t, 0.0) + 8.0, 3)
    for n in body.niches:
        n = n.strip()[:64]
        if n:
            aff.niches[n] = round(aff.niches.get(n, 0.0) + 8.0, 3)
    _save_affinity(session, device_id, row, aff, 0)
    session.commit()
    row2 = session.get(ClipAffinity, device_id)
    return AffinityOut(tags=aff.tags, niches=aff.niches, creators=aff.creators, events_seen=row2.events_seen if row2 else 0)


@router.get("/affinity", response_model=AffinityOut)
def get_affinity(session: Session = Depends(get_session), device_id: str = Depends(get_device_id)) -> AffinityOut:
    row, aff = _load_affinity(session, device_id)
    return AffinityOut(tags=aff.tags, niches=aff.niches, creators=aff.creators, events_seen=row.events_seen if row else 0)


@router.delete("/affinity", status_code=204)
def reset_affinity(
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
    seen: bool = Query(default=False, description="True = zapomnij tez widziane klipy (feed od zera)"),
) -> Response:
    """„Resetuj moj feed": kasuje wagi; z `seen=true` takze pamiec widzianych."""
    session.execute(text("DELETE FROM clip_affinity WHERE device_id = :d"), {"d": device_id})
    if seen:
        session.execute(text("DELETE FROM clip_seen WHERE device_id = :d"), {"d": device_id})
    session.commit()
    return Response(status_code=204)


# ---- ulubione, blacklista ----------------------------------------------------------------
@router.get("/favorites", response_model=ClipListOut)
def clip_favorites(
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
    limit: int = Query(default=50, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
) -> ClipListOut:
    rows = session.execute(
        select(Clip)
        .join(ClipFavorite, ClipFavorite.clip_id == Clip.id)
        .where(ClipFavorite.device_id == device_id)
        .order_by(ClipFavorite.created_at.desc())
        .offset(offset)
        .limit(limit + 1)
    ).scalars().all()
    has_more = len(rows) > limit
    rows = rows[:limit]
    tags = _tags_for(session, [r.id for r in rows])
    return ClipListOut(
        items=[_clip_out(r, tags.get(r.id, []), True) for r in rows], limit=limit, offset=offset, has_more=has_more
    )


@router.get("/blacklist", response_model=list[BlacklistItemOut])
def clip_blacklist(session: Session = Depends(get_session), device_id: str = Depends(get_device_id)) -> list[BlacklistItemOut]:
    return [
        BlacklistItemOut(kind=k, value=v)
        for k, v in session.execute(
            select(ClipBlacklist.kind, ClipBlacklist.value).where(ClipBlacklist.device_id == device_id).order_by(ClipBlacklist.kind, ClipBlacklist.value)
        )
    ]


@router.post("/blacklist/{kind}/{value}", status_code=204)
def clip_blacklist_add(
    kind: str, value: str, session: Session = Depends(get_session), device_id: str = Depends(get_device_id)
) -> Response:
    if kind not in _BLACKLIST_KINDS or not value.strip():
        raise HTTPException(status_code=400, detail="kind must be tag|niche|creator")
    session.execute(
        pg_insert(ClipBlacklist).values(device_id=device_id, kind=kind, value=value.strip()[:128]).on_conflict_do_nothing()
    )
    session.commit()
    return Response(status_code=204)


@router.delete("/blacklist/{kind}/{value}", status_code=204)
def clip_blacklist_remove(
    kind: str, value: str, session: Session = Depends(get_session), device_id: str = Depends(get_device_id)
) -> Response:
    session.execute(
        text("DELETE FROM clip_blacklist WHERE device_id = :d AND kind = :k AND value = :v"),
        {"d": device_id, "k": kind, "v": value.strip()[:128]},
    )
    session.commit()
    return Response(status_code=204)


# ---- listing i statystyki (pilot) --------------------------------------------------------
@router.get("", response_model=ClipListOut)
def list_clips(
    session: Session = Depends(get_session),
    device_id: str = Depends(get_device_id),
    tag: str | None = Query(default=None, max_length=64),
    source: str | None = Query(default=None, max_length=16),
    creator: str | None = Query(default=None, max_length=128),
    has_audio: bool | None = None,
    max_duration_sec: float | None = Query(default=None, gt=0),
    limit: int = Query(default=50, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
) -> ClipListOut:
    q = select(Clip).where(Clip.dead_at.is_(None))
    if tag:
        q = q.where(Clip.id.in_(select(ClipTag.clip_id).where(ClipTag.tag == tag)))
    if source:
        q = q.where(Clip.source == source)
    if creator:
        q = q.where(Clip.creator == creator)
    if has_audio is not None:
        q = q.where(Clip.has_audio.is_(has_audio))
    if max_duration_sec is not None:
        q = q.where(Clip.duration_sec <= max_duration_sec)
    rows = session.execute(
        q.order_by(Clip.published_at.desc().nullslast(), Clip.id).offset(offset).limit(limit + 1)
    ).scalars().all()
    has_more = len(rows) > limit
    rows = rows[:limit]
    ids = [r.id for r in rows]
    tags = _tags_for(session, ids)
    favs = _favorites_for(session, device_id, ids)
    total = None
    if creator:
        total = session.execute(
            select(func.count()).select_from(Clip).where(Clip.dead_at.is_(None), Clip.creator == creator)
        ).scalar_one()
    return ClipListOut(
        items=[_clip_out(r, tags.get(r.id, []), r.id in favs) for r in rows],
        limit=limit,
        offset=offset,
        has_more=has_more,
        total=total,
    )


@router.get("/stats", response_model=ClipStatsOut)
def clip_stats(session: Session = Depends(get_session)) -> ClipStatsOut:
    row = session.execute(
        select(
            func.count(),
            func.count().filter(Clip.dead_at.is_(None)),
            func.count().filter(Clip.dead_at.is_not(None)),
            func.count().filter(Clip.has_audio.is_(True), Clip.dead_at.is_(None)),
            func.count().filter(Clip.creator_verified.is_(True), Clip.dead_at.is_(None)),
            func.max(Clip.published_at),
        )
    ).one()
    by_source = {s: n for s, n in session.execute(select(Clip.source, func.count()).group_by(Clip.source))}
    tags = session.execute(select(func.count(func.distinct(ClipTag.tag)))).scalar_one()
    return ClipStatsOut(
        total=row[0], alive=row[1], dead=row[2], with_audio=row[3], verified=row[4],
        by_source=by_source, newest_published_at=row[5], tags=tags,
    )