diff --git a/alembic/versions/20260914_0037_clip_feed.py b/alembic/versions/20260914_0037_clip_feed.py new file mode 100644 index 0000000..96c70ca --- /dev/null +++ b/alembic/versions/20260914_0037_clip_feed.py @@ -0,0 +1,86 @@ +"""Feed Swipe, stan per urządzenie: clip_seen, clip_feed_events, clip_affinity, +clip_favorites, clip_blacklist. Plus GIN na clips.niches pod skan po niszy. + +Uzasadnienie w docstringu `app/models/clip_feed.py`. + +Revision ID: 0037_clip_feed +Revises: 0036_clips +Create Date: 2026-09-14 +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "0037_clip_feed" +down_revision: str | None = "0036_clips" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "clip_seen", + sa.Column("device_id", sa.String(64), nullable=False), + sa.Column("clip_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("outcome", sa.String(16), nullable=False, server_default="served"), + sa.Column("seen_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.ForeignKeyConstraint(["clip_id"], ["clips.id"], name="fk_clip_seen_clip_id_clips", ondelete="CASCADE"), + sa.PrimaryKeyConstraint("device_id", "clip_id", name="pk_clip_seen"), + ) + op.create_table( + "clip_feed_events", + sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("device_id", sa.String(64), nullable=False), + sa.Column("session_id", sa.String(32), nullable=True), + sa.Column("clip_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("kind", sa.String(16), nullable=False), + sa.Column("dwell_ms", sa.Integer(), nullable=True), + sa.Column("reason", sa.String(160), nullable=True), + sa.Column("slot", sa.String(12), nullable=True), + sa.Column("position", sa.SmallInteger(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("id", name="pk_clip_feed_events"), + ) + op.create_index("ix_clip_feed_events_device_created", "clip_feed_events", ["device_id", "created_at"]) + op.create_index("ix_clip_feed_events_created", "clip_feed_events", ["created_at"]) + op.create_table( + "clip_affinity", + sa.Column("device_id", sa.String(64), nullable=False), + sa.Column("tag_weights", postgresql.JSONB(), nullable=False, server_default="{}"), + sa.Column("niche_weights", postgresql.JSONB(), nullable=False, server_default="{}"), + sa.Column("creator_weights", postgresql.JSONB(), nullable=False, server_default="{}"), + sa.Column("events_seen", sa.Integer(), nullable=False, server_default="0"), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("device_id", name="pk_clip_affinity"), + ) + op.create_table( + "clip_favorites", + sa.Column("device_id", sa.String(64), nullable=False), + sa.Column("clip_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.ForeignKeyConstraint(["clip_id"], ["clips.id"], name="fk_clip_favorites_clip_id_clips", ondelete="CASCADE"), + sa.PrimaryKeyConstraint("device_id", "clip_id", name="pk_clip_favorites"), + ) + op.create_table( + "clip_blacklist", + sa.Column("device_id", sa.String(64), nullable=False), + sa.Column("kind", sa.String(8), nullable=False), + sa.Column("value", sa.String(128), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("device_id", "kind", "value", name="pk_clip_blacklist"), + ) + # Skan kandydatów po niszy (`niches && ARRAY[...]`) bez pełnego przebiegu tabeli. + op.execute("CREATE INDEX IF NOT EXISTS ix_clips_niches_gin ON clips USING gin (niches)") + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS ix_clips_niches_gin") + op.drop_table("clip_blacklist") + op.drop_table("clip_favorites") + op.drop_table("clip_affinity") + op.drop_index("ix_clip_feed_events_created", table_name="clip_feed_events") + op.drop_index("ix_clip_feed_events_device_created", table_name="clip_feed_events") + op.drop_table("clip_feed_events") + op.drop_table("clip_seen") diff --git a/app/api/clips.py b/app/api/clips.py index dc682b0..a5f17f3 100644 --- a/app/api/clips.py +++ b/app/api/clips.py @@ -1,27 +1,51 @@ -"""Klipy (feed Swipe), etap 2: podstawowy listing i statystyki do pilota. +"""Klipy (feed Swipe): feed z afinicznością, pamięć widzianych, telemetria, ulubione, +blacklista, onboarding. Logika bez bazy w `app/clips_feed.py`. -Właściwy feed z afinicznością, pamięcią widzianych i telemetrią to etap 3 -(`/clips/feed`, `/clips/events`). Tu tylko to, czego trzeba, żeby zweryfikować ingest -i odtwarzanie z apki/emulatora: lista najnowszych żywych klipów z filtrem po tagu, -źródle i audio oraz zbiorcze liczniki. +**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 time import uuid -from datetime import datetime +from datetime import UTC, datetime -from fastapi import APIRouter, Depends, Query -from pydantic import BaseModel -from sqlalchemy import func, select +from fastapi import APIRouter, Depends, HTTPException, Query, Response +from pydantic import BaseModel, Field +from sqlalchemy import 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"}) + +# ---- schematy --------------------------------------------------------------------------- class ClipOut(BaseModel): id: uuid.UUID source: str @@ -43,6 +67,7 @@ class ClipOut(BaseModel): likes: int views: int published_at: datetime | None + is_favorite: bool = False class ClipListOut(BaseModel): @@ -63,6 +88,66 @@ class ClipStatsOut(BaseModel): 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 {} @@ -74,9 +159,451 @@ def _tags_for(session: Session, clip_ids: list[uuid.UUID]) -> dict[uuid.UUID, li 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: + row = ClipAffinity(device_id=device_id) + session.add(row) + 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 + + +def _candidates(session: Session, ids: list[uuid.UUID], clauses: list) -> list[cf.Candidate]: + if not ids: + return [] + rows = session.execute(select(Clip).where(Clip.id.in_(ids), *clauses)).scalars().all() + tags = _tags_for(session, [r.id for r in rows]) + return [ + cf.Candidate( + id=r.id, creator=r.creator, likes=r.likes, published_at=r.published_at, + tags=tags.get(r.id, []), niches=list(r.niches or []), + ) + for r in rows + ] + + +def _explore_window(session: Session, pk: uuid.UUID) -> tuple[list[uuid.UUID], bool]: + """Kolejne `_EXPLORE_WINDOW` id po `pk`; gdy okno się kończy, zawija do zera.""" + ids = [ + r[0] + for r in session.execute( + select(Clip.id).where(Clip.id > pk, Clip.dead_at.is_(None)).order_by(Clip.id).limit(_EXPLORE_WINDOW) + ) + ] + wrapped = False + if len(ids) < _EXPLORE_WINDOW and pk != _UUID_ZERO: + wrapped = True + more = [ + r[0] + for r in session.execute( + select(Clip.id).where(Clip.dead_at.is_(None)).order_by(Clip.id).limit(_EXPLORE_WINDOW - len(ids)) + ) + ] + ids.extend(more) + return ids, wrapped + + +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) + taken = {c.id for c in picked_aff} + + # Explore: okno po PK; gdy w oknie brak widocznych, przesuwamy sie dalej (max 3 okna). + picked_explore: list[cf.Candidate] = [] + wrapped = False + next_pk = pk + want = n_explore + max(0, n_aff - len(picked_aff)) # braki z afinicznosci dopelnia explore + for _ in range(3): + ids, w = _explore_window(session, next_pk) + wrapped = wrapped or w + if not ids: + break + next_pk = ids[-1] if not w else _UUID_ZERO + pool = [c for c in _candidates(session, ids, clauses) if c.id not in taken] + picked_explore += cf.pick_explore(pool, want - len(picked_explore), rng) + if len(picked_explore) >= want or len(ids) < _EXPLORE_WINDOW: + 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 + 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() + return Response(status_code=204) + + +# ---- onboarding, afinicznosc ------------------------------------------------------------ +_OPTIONS_CACHE: dict[str, tuple[float, OnboardingOptionsOut]] = {} + + +@router.get("/onboarding-options", response_model=OnboardingOptionsOut) +def onboarding_options(session: Session = Depends(get_session)) -> OnboardingOptionsOut: + """Najpopularniejsze tagi i nisze do wyboru „3 tapów" przy pierwszym wejsciu. + Liczone z bazy, cache 10 min w procesie.""" + 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(40) + ) + ] + 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: + 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) + 5.0, 3) + for n in body.niches: + n = n.strip()[:64] + if n: + aff.niches[n] = round(aff.niches.get(n, 0.0) + 5.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), @@ -101,33 +628,12 @@ def list_clips( ).scalars().all() has_more = len(rows) > limit rows = rows[:limit] - tags = _tags_for(session, [r.id for r in rows]) - items = [ - 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.get(r.id, []), - 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, - ) - for r in rows - ] - return ClipListOut(items=items, limit=limit, offset=offset, has_more=has_more) + ids = [r.id for r in rows] + tags = _tags_for(session, ids) + favs = _favorites_for(session, device_id, ids) + 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 + ) @router.get("/stats", response_model=ClipStatsOut) @@ -142,17 +648,9 @@ def clip_stats(session: Session = Depends(get_session)) -> ClipStatsOut: 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)) - } + 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, + 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, ) diff --git a/app/api/me.py b/app/api/me.py index bcb9bfd..4048e92 100644 --- a/app/api/me.py +++ b/app/api/me.py @@ -33,6 +33,11 @@ _TABLES: list[tuple[str, str]] = [ ("blacklisted_performers", "performer_id"), ("blacklisted_studios", "studio_id"), ("blacklisted_tags", "tag_id"), + # Feed Swipe: widziane klipy i ulubione klipy jadą z użytkownikiem na nowy telefon. + # Afiniczność (clip_affinity) celowo NIE: ma jeden wiersz na urządzenie bez kolumny + # encji, a i tak odbuduje się z pierwszych zdarzeń. + ("clip_seen", "clip_id"), + ("clip_favorites", "clip_id"), ] @@ -48,7 +53,10 @@ class ImportOut(BaseModel): # Tabele do przeniesienia przy transferze między urządzeniami + kolumna-encji do dedupu. # saved_searches ma własne `id` PK (nie (device_id,encja)), więc dedup po `query`. -_COPY_TABLES: list[tuple[str, str]] = _TABLES + [("saved_searches", "query")] +_COPY_TABLES: list[tuple[str, str]] = _TABLES + [ + ("saved_searches", "query"), + ("clip_blacklist", "value"), +] @router.post("/import/{source_device}", response_model=ImportOut) diff --git a/app/clips_feed.py b/app/clips_feed.py new file mode 100644 index 0000000..168a531 --- /dev/null +++ b/app/clips_feed.py @@ -0,0 +1,243 @@ +"""Logika feedu Swipe bez bazy: kursor, wagi afiniczności, punktacja, składanie strony. + +Wyniesione z `app/api/clips.py`, żeby dało się testować bez Postgresa (`tests/test_clips_feed.py`). + +**Kursor** = base64 JSON `{"s": seed, "pk": , "p": }`. Nie offset: pula +„explore" idzie oknem po kluczu głównym od `pk` (ten sam trik co `sort=random` w +`list_scenes`), pula afiniczności losuje punkty startu z `Random(seed, p)`. Dzięki temu +dwa telefony z tym samym seedem dostają inne strony tylko przez różne `clip_seen`, +a pull-to-refresh to po prostu nowy seed. + +**Wagi**: trzy słowniki (tag, nisza, twórca) → liczba. Zdarzenia dodają delty +(`EVENT_DELTAS`), powód „nie interesuje" dokłada karę na wskazany cel, nocny decay ×0,97 +zapomina stare gusta, cap trzyma słowniki małe (60/40/40). Wszystko celowo proste i +przejrzyste: da się wytłumaczyć użytkownikowi, dlaczego coś widzi, i skasować jednym +wywołaniem. + +**Punktacja kandydata** = Σ wag tagów + 2·waga niszy + 3·waga twórcy + 0,3·ln(1+lajki) ++ 0,5 gdy klip z ostatnich 7 dni. Strona: 80 % z afiniczności (max 2 klipy jednego twórcy), +20 % explore, przemieszane deterministycznie z seeda. +""" +from __future__ import annotations + +import base64 +import hashlib +import json +import math +import random +import uuid +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta + +FRESH_DAYS = 7 +EXPLORE_SHARE = 0.2 +MAX_PER_CREATOR = 2 +CAP_TAGS, CAP_NICHES, CAP_CREATORS = 60, 40, 40 +DECAY_FACTOR = 0.97 +DECAY_FLOOR = 0.2 +#: Poniżej tego dwell „impression" nic nie znaczy, powyżej `DWELL_POSITIVE_MS` to plus. +SKIP_MS = 2000 +DWELL_POSITIVE_MS = 10000 + +#: kind → (delta tag, delta nisza, delta twórca) +EVENT_DELTAS: dict[str, tuple[float, float, float]] = { + "complete": (3.0, 3.0, 2.0), + "like": (5.0, 5.0, 4.0), + "unlike": (-5.0, -5.0, -4.0), + "dwell": (1.0, 1.0, 0.5), + "skip": (-1.0, -1.0, -0.5), + "not_interested": (-4.0, -4.0, -3.0), +} +REASON_PENALTY = -8.0 +ALLOWED_KINDS = frozenset( + {"impression", "skip", "complete", "like", "unlike", "not_interested", "autoswipe", "open"} +) +#: Kolejność „siły" wyniku w `clip_seen`: nie cofamy do słabszego. +OUTCOME_RANK = {"served": 0, "skip": 1, "complete": 2, "like": 3, "not_interested": 4} + + +# ---- kursor ----------------------------------------------------------------------------- +def new_seed() -> str: + return uuid.uuid4().hex[:12] + + +def seed_start_uuid(seed: str, salt: str = "") -> uuid.UUID: + return uuid.UUID(hashlib.md5(f"{seed}:{salt}".encode()).hexdigest()) + + +def encode_cursor(seed: str, pk: uuid.UUID, page: int) -> str: + raw = json.dumps({"s": seed, "pk": str(pk), "p": page}, separators=(",", ":")) + return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=") + + +def decode_cursor(cursor: str) -> tuple[str, uuid.UUID, int] | None: + """None gdy kursor jest zepsuty (klient zaczyna wtedy od nowa, bez 400).""" + try: + pad = "=" * (-len(cursor) % 4) + d = json.loads(base64.urlsafe_b64decode(cursor + pad).decode()) + seed = str(d["s"])[:16] + pk = uuid.UUID(str(d["pk"])) + page = int(d["p"]) + if not seed or page < 0 or page > 100000: + return None + return seed, pk, page + except Exception: + return None + + +def page_rng(seed: str, page: int) -> random.Random: + return random.Random(f"{seed}:{page}") + + +# ---- wagi ------------------------------------------------------------------------------- +@dataclass +class Affinity: + tags: dict[str, float] = field(default_factory=dict) + niches: dict[str, float] = field(default_factory=dict) + creators: dict[str, float] = field(default_factory=dict) + + def is_empty(self) -> bool: + return not ( + any(w > 0 for w in self.tags.values()) + or any(w > 0 for w in self.niches.values()) + or any(w > 0 for w in self.creators.values()) + ) + + def top(self, which: str, n: int) -> list[str]: + d = getattr(self, which) + return [k for k, w in sorted(d.items(), key=lambda kv: -kv[1]) if w > 0][:n] + + +def _bump(d: dict[str, float], key: str | None, delta: float) -> None: + if not key or not delta: + return + d[key] = round(d.get(key, 0.0) + delta, 3) + + +def cap_weights(d: dict[str, float], n: int) -> dict[str, float]: + """Zostaw `n` pozycji o największej wartości bezwzględnej (kary też są informacją).""" + if len(d) <= n: + return d + keep = sorted(d.items(), key=lambda kv: -abs(kv[1]))[:n] + return dict(keep) + + +def decay_weights(d: dict[str, float], factor: float = DECAY_FACTOR, floor: float = DECAY_FLOOR) -> dict[str, float]: + out = {k: round(v * factor, 3) for k, v in d.items()} + return {k: v for k, v in out.items() if abs(v) >= floor} + + +def event_kind_for_weights(kind: str, dwell_ms: int | None) -> str | None: + """Zdarzenie z apki → klucz w `EVENT_DELTAS`. `impression` liczy się tylko przy długim + dwell, `skip` tylko przy krótkim; `autoswipe`/`open` nie ruszają wag.""" + if kind == "impression": + return "dwell" if (dwell_ms or 0) >= DWELL_POSITIVE_MS else None + if kind == "skip": + return "skip" if (dwell_ms or 0) < SKIP_MS else None + return kind if kind in EVENT_DELTAS else None + + +def apply_event( + aff: Affinity, + *, + kind: str, + dwell_ms: int | None, + reason: str | None, + tags: list[str], + niches: list[str], + creator: str | None, +) -> bool: + """Nakłada delty zdarzenia na wagi. True gdy coś się zmieniło.""" + wkind = event_kind_for_weights(kind, dwell_ms) + changed = False + if wkind: + dt, dn, dc = EVENT_DELTAS[wkind] + for t in tags: + _bump(aff.tags, t, dt) + for n in niches: + _bump(aff.niches, n, dn) + _bump(aff.creators, creator, dc) + changed = True + if kind == "not_interested" and reason and ":" in reason: + target_kind, _, target = reason.partition(":") + target = target.strip() + if target_kind == "tag": + _bump(aff.tags, target, REASON_PENALTY) + elif target_kind == "niche": + _bump(aff.niches, target, REASON_PENALTY) + elif target_kind == "creator": + _bump(aff.creators, target, REASON_PENALTY) + changed = True + if changed: + aff.tags = cap_weights(aff.tags, CAP_TAGS) + aff.niches = cap_weights(aff.niches, CAP_NICHES) + aff.creators = cap_weights(aff.creators, CAP_CREATORS) + return changed + + +# ---- punktacja i składanie strony ----------------------------------------------------- +@dataclass +class Candidate: + id: uuid.UUID + creator: str | None + likes: int + published_at: datetime | None + tags: list[str] + niches: list[str] + + +def score_candidate(c: Candidate, aff: Affinity, now: datetime | None = None) -> float: + now = now or datetime.now(UTC) + s = sum(aff.tags.get(t, 0.0) for t in c.tags) + s += 2.0 * sum(aff.niches.get(n, 0.0) for n in c.niches) + s += 3.0 * aff.creators.get(c.creator or "", 0.0) + s += 0.3 * math.log1p(max(c.likes, 0)) + if c.published_at and c.published_at >= now - timedelta(days=FRESH_DAYS): + s += 0.5 + return s + + +def pick_affinity(cands: list[Candidate], aff: Affinity, n: int, *, max_per_creator: int = MAX_PER_CREATOR) -> list[Candidate]: + """Najlepsze `n` po punktach, max `max_per_creator` klipów jednego twórcy na stronę.""" + ranked = sorted(cands, key=lambda c: -score_candidate(c, aff)) + out: list[Candidate] = [] + per: dict[str, int] = {} + for c in ranked: + key = c.creator or "" + if per.get(key, 0) >= max_per_creator: + continue + per[key] = per.get(key, 0) + 1 + out.append(c) + if len(out) >= n: + break + return out + + +def pick_explore(cands: list[Candidate], n: int, rng: random.Random) -> list[Candidate]: + """Losowanie ważone popularnością: ln(1+lajki)+1, żeby zero lajków też miało szansę.""" + pool = list(cands) + out: list[Candidate] = [] + while pool and len(out) < n: + weights = [math.log1p(max(c.likes, 0)) + 1.0 for c in pool] + pick = rng.choices(range(len(pool)), weights=weights, k=1)[0] + out.append(pool.pop(pick)) + return out + + +def split_counts(limit: int, has_affinity: bool) -> tuple[int, int]: + """(z afiniczności, explore). Bez wag wszystko idzie z explore.""" + if not has_affinity: + return 0, limit + n_explore = max(1, round(limit * EXPLORE_SHARE)) + return limit - n_explore, n_explore + + +def compose_page(aff_items: list[Candidate], explore_items: list[Candidate], rng: random.Random) -> list[tuple[Candidate, str]]: + page = [(c, "affinity") for c in aff_items] + [(c, "explore") for c in explore_items] + rng.shuffle(page) + return page + + +def is_fresh(published_at: datetime | None, now: datetime | None = None) -> bool: + now = now or datetime.now(UTC) + return bool(published_at and published_at >= now - timedelta(days=FRESH_DAYS)) diff --git a/app/config.py b/app/config.py index 6207abc..329c6a3 100644 --- a/app/config.py +++ b/app/config.py @@ -230,6 +230,13 @@ class Settings(BaseSettings): # niszach ręcznie: `python -m scripts.clips_backfill_redgifs`. sched_clips_hours: int = Field(default=1, validation_alias="GOON_SCHED_CLIPS_HOURS") clips_latest_pages: int = Field(default=5, validation_alias="GOON_CLIPS_LATEST_PAGES") + # Konserwacja feedu Swipe raz na dobę: decay wag afiniczności (×0,97), retencja + # zdarzeń (90 dni), reaper martwych plików (HEAD na 2000 najdawniej sprawdzanych). + sched_clips_maintenance_hours: int = Field( + default=24, validation_alias="GOON_SCHED_CLIPS_MAINTENANCE_HOURS" + ) + # Feed podaje tylko klipy do tej długości (RedGifs p90 = 60 s; dłuższe to nie „swipe"). + clips_feed_max_duration_sec: int = Field(default=120, validation_alias="GOON_CLIPS_FEED_MAX_DURATION_SEC") # Bright Data ISP proxy (stałe IP od ISP, rozliczane ryczałtem NIE per-GB), # używany do ingestu HTML (scrape) tubów które blokują VPS IP twardym Cloudflare diff --git a/app/models/__init__.py b/app/models/__init__.py index 600348d..89a7c8d 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -2,6 +2,7 @@ from app.models.base import Base from app.models.blacklist import BlacklistedPerformer, BlacklistedStudio, BlacklistedTag from app.models.bug_report import BugReport from app.models.clip import Clip, ClipTag +from app.models.clip_feed import ClipAffinity, ClipBlacklist, ClipFavorite, ClipFeedEvent, ClipSeen from app.models.external_record import ExternalRecord from app.models.favorite_movie import FavoriteMovie from app.models.favorite_performer import FavoritePerformer @@ -45,6 +46,11 @@ __all__ = [ "BlacklistedTag", "BugReport", "Clip", + "ClipAffinity", + "ClipBlacklist", + "ClipFavorite", + "ClipFeedEvent", + "ClipSeen", "ClipTag", "ExternalRecord", "FavoriteMovie", diff --git a/app/models/clip_feed.py b/app/models/clip_feed.py new file mode 100644 index 0000000..33a01fb --- /dev/null +++ b/app/models/clip_feed.py @@ -0,0 +1,104 @@ +"""Stan feedu Swipe per urządzenie: widziane, zdarzenia, afiniczność, ulubione, blacklista. + +Tożsamość = `X-Device-Id` (bez kont), jak w reszcie aplikacji. Wszystko kluczowane po +`device_id`, więc przenosi się kodem synchronizacji (`/me/import`). + +**`clip_seen`**: zbiór wykluczeń feedu. Wiersz powstaje **w chwili podania** klipu +(`outcome='served'`), zdarzenia z apki podnoszą `outcome`. Złożony PK to jedyny indeks, +jakiego potrzebuje `NOT EXISTS` w zapytaniu feedu (jedno trafienie na kandydata). Koszt +zapisu przy podaniu: klip z prefetchu, którego użytkownik nie doszedł, znika z feedu +(kilka na sesję); zysk: kursor bez listy widzianych i zero powtórek między stronami. + +**`clip_feed_events`**: append-only telemetria (impression z dwell, skip, complete, like, +unlike, not_interested, autoswipe). Bez FK na klip (klip może zostać skasowany). Karmi +afiniczność i mierzy, czy personalizacja coś daje (`slot` affinity vs explore). + +**`clip_affinity`**: wagi per urządzenie w JSONB (tag → waga, nisza → waga, twórca → waga), +aktualizowane przy zdarzeniu, nocny decay ×0,97, cap 60/40/40 pozycji. To najbardziej +intymne dane w aplikacji: kasowalne jednym wywołaniem (`DELETE /clips/affinity`), nie +trafiają do logów. +""" +from __future__ import annotations + +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Index, Integer, SmallInteger, String, func +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.models.base import Base + + +class ClipSeen(Base): + __tablename__ = "clip_seen" + + device_id: Mapped[str] = mapped_column(String(64), primary_key=True) + clip_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), ForeignKey("clips.id", ondelete="CASCADE"), primary_key=True + ) + #: served | skip | complete | like | not_interested (rosnąca „siła", nie cofamy). + outcome: Mapped[str] = mapped_column(String(16), nullable=False, default="served") + seen_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + +class ClipFeedEvent(Base): + __tablename__ = "clip_feed_events" + __table_args__ = ( + Index("ix_clip_feed_events_device_created", "device_id", "created_at"), + Index("ix_clip_feed_events_created", "created_at"), + ) + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + device_id: Mapped[str] = mapped_column(String(64), nullable=False) + session_id: Mapped[str | None] = mapped_column(String(32), nullable=True) + clip_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True) + kind: Mapped[str] = mapped_column(String(16), nullable=False) + dwell_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + #: Powód „nie interesuje": `tag:` | `niche:` | `creator:` | `seen_it`. + reason: Mapped[str | None] = mapped_column(String(160), nullable=True) + #: `affinity` | `explore` (z którego koszyka klip przyszedł). + slot: Mapped[str | None] = mapped_column(String(12), nullable=True) + position: Mapped[int | None] = mapped_column(SmallInteger, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + +class ClipAffinity(Base): + __tablename__ = "clip_affinity" + + device_id: Mapped[str] = mapped_column(String(64), primary_key=True) + tag_weights: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + niche_weights: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + creator_weights: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + events_seen: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False + ) + + +class ClipFavorite(Base): + __tablename__ = "clip_favorites" + + device_id: Mapped[str] = mapped_column(String(64), primary_key=True) + clip_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), ForeignKey("clips.id", ondelete="CASCADE"), primary_key=True + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + +class ClipBlacklist(Base): + __tablename__ = "clip_blacklist" + + device_id: Mapped[str] = mapped_column(String(64), primary_key=True) + #: tag | niche | creator + kind: Mapped[str] = mapped_column(String(8), primary_key=True) + value: Mapped[str] = mapped_column(String(128), primary_key=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) diff --git a/app/scheduler/clips_ingest.py b/app/scheduler/clips_ingest.py index 19df4e5..8c279f8 100644 --- a/app/scheduler/clips_ingest.py +++ b/app/scheduler/clips_ingest.py @@ -169,3 +169,92 @@ def run_redgifs_niches( client.close() log.info("clips/redgifs niches: %s", total) return total + + +# ---- konserwacja (raz na dobe) --------------------------------------------------------- +_EVENTS_RETENTION_DAYS = 90 +_DEAD_CODES = {403, 404, 410} +_REAP_UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/152.0.0.0 Safari/537.36"} + + +def reap_dead(*, limit: int = 2000, apply: bool = True) -> dict[str, int]: + """Reaper martwych klipow: tworcy kasuja klipy, RedGifs oddaje wtedy 404/410 na pliku. + + HEAD na `hd_url` (albo `sd_url`) dla najdawniej sprawdzanych zywych klipow + (`updated_at`); 403/404/410 → `dead_at`. Bledy sieci NIE oznaczaja smierci (patrz + notatka o dwoch kontrolach przy przemiale zrodel). Klip, ktory zrodlo znow poda + w listingu, upsert odzywia. Sprawdzone i zywe dostaja swiezy `updated_at`, zeby + kolejny przebieg bral inne. + """ + import time as _time + + import httpx + from sqlalchemy import select + + checked = dead = errors = 0 + dead_ids: list[tuple] = [] + with session_scope() as session: + rows = session.execute( + select(Clip.id, Clip.hd_url, Clip.sd_url) + .where(Clip.dead_at.is_(None)) + .order_by(Clip.updated_at.asc()) + .limit(limit) + ).all() + with httpx.Client(headers=_REAP_UA, timeout=20, follow_redirects=True) as http: + for cid, hd, sd in rows: + url = hd or sd + if not url: + continue + checked += 1 + try: + r = http.head(url) + except httpx.HTTPError: + errors += 1 + continue + if r.status_code in _DEAD_CODES: + dead += 1 + dead_ids.append((cid, f"http {r.status_code}")) + _time.sleep(0.2) + if apply and rows: + dead_set = {d[0] for d in dead_ids} + with session_scope() as session: + for cid, reason in dead_ids: + session.execute( + text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"), + {"r": reason, "id": cid}, + ) + alive = [cid for cid, _hd, _sd in rows if cid not in dead_set] + if alive: + session.execute(text("UPDATE clips SET updated_at = now() WHERE id = ANY(:ids)"), {"ids": alive}) + out = {"checked": checked, "dead": dead, "errors": errors, "applied": int(apply)} + log.info("clips reaper: %s", out) + return out + + +def decay_affinity() -> int: + """Nocne zapominanie: wszystkie wagi ×0,97, |w| < 0,2 znika. Zwraca liczbe urzadzen.""" + from app.clips_feed import decay_weights + from app.models.clip_feed import ClipAffinity + + n = 0 + with session_scope() as session: + for row in session.query(ClipAffinity).all(): + row.tag_weights = decay_weights(dict(row.tag_weights or {})) + row.niche_weights = decay_weights(dict(row.niche_weights or {})) + row.creator_weights = decay_weights(dict(row.creator_weights or {})) + n += 1 + return n + + +def run_clips_maintenance() -> dict[str, int]: + out: dict[str, int] = {} + out["affinity_rows"] = decay_affinity() + with session_scope() as session: + res = session.execute( + text("DELETE FROM clip_feed_events WHERE created_at < now() - make_interval(days => :d)"), + {"d": _EVENTS_RETENTION_DAYS}, + ) + out["events_pruned"] = int(res.rowcount or 0) + out.update({f"reap_{k}": v for k, v in reap_dead(limit=2000, apply=True).items()}) + log.info("clips maintenance: %s", out) + return out diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 914574b..d628484 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -487,6 +487,16 @@ def _job_clips_latest(pages: int) -> None: log.exception("[scheduler] clips-latest failed") +def _job_clips_maintenance() -> None: + """Feed Swipe: decay afiniczności, retencja zdarzeń, reaper martwych plików.""" + from app.scheduler.clips_ingest import run_clips_maintenance + + try: + _run_with_timeout(run_clips_maintenance, label="clips-maintenance") + except Exception: + log.exception("[scheduler] clips-maintenance failed") + + def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: """Buduje scheduler na podstawie cfg dictu. @@ -574,6 +584,17 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: ) log.info("scheduler: clips-latest every %dh (pages=%d)", cfg["clips_hours"], clips_pages) + if cfg.get("clips_maintenance_hours"): + sched.add_job( + _job_clips_maintenance, + IntervalTrigger(hours=cfg["clips_maintenance_hours"], start_date=INTERVAL_ANCHOR), + id="clips_maintenance", + replace_existing=True, + max_instances=1, + coalesce=True, + ) + log.info("scheduler: clips-maintenance every %dh", cfg["clips_maintenance_hours"]) + if cfg.get("bulk_dedup_hours"): sched.add_job( _job_bulk_dedup_performers, diff --git a/app/scheduler/worker.py b/app/scheduler/worker.py index 3727e8a..f9bc513 100644 --- a/app/scheduler/worker.py +++ b/app/scheduler/worker.py @@ -252,6 +252,7 @@ def run_forever() -> int: # Klipy do feedu Swipe (RedGifs „najnowsze"), co godzinę, N stron po 100. "clips_hours": getattr(settings, "sched_clips_hours", 1) or None, "clips_latest_pages": getattr(settings, "clips_latest_pages", 5), + "clips_maintenance_hours": getattr(settings, "sched_clips_maintenance_hours", 24) or None, } sched = build_scheduler(cfg) log.info("worker scheduled mode starting (jobs=%d)", len(sched.get_jobs())) diff --git a/scripts/clips_reap_dead.py b/scripts/clips_reap_dead.py index 0ba6e50..5dcd207 100644 --- a/scripts/clips_reap_dead.py +++ b/scripts/clips_reap_dead.py @@ -1,73 +1,15 @@ -"""Reaper martwych klipów: twórcy kasują klipy, RedGifs oddaje wtedy 404/410 na pliku. +"""Reaper martwych klipów, ręczne uruchomienie. Logika w `app.scheduler.clips_ingest.reap_dead` +(ta sama, którą raz na dobę odpala job `clips-maintenance`). -Sprawdza HEAD `hd_url` (albo `sd_url`) dla próbki żywych klipów: najpierw te najdawniej -sprawdzane (`updated_at`), bo ingest `latest` i tak odświeża świeże. 404/410/403 → -`dead_at` + `dead_reason`; błędy sieci nie oznaczają śmierci (patrz notatka o dwóch -kontrolach przy przemiale źródeł). Klip, który źródło znów poda w listingu, upsert -odżywia (zdejmuje `dead_at`). - -Uruchomienie (kontener worker): python -m scripts.clips_reap_dead --limit 500 # dry-run - python -m scripts.clips_reap_dead --limit 500 --yes # zapisz + python -m scripts.clips_reap_dead --limit 500 --yes # zapisz dead_at """ from __future__ import annotations import argparse import logging -import time -import httpx -from sqlalchemy import select, text - -from app.db import session_scope -from app.models.clip import Clip - -log = logging.getLogger(__name__) -_UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/152.0.0.0 Safari/537.36"} -_DEAD_CODES = {403, 404, 410} - - -def run(*, limit: int, apply: bool) -> dict[str, int]: - checked = dead = errors = 0 - dead_ids: list = [] - with session_scope() as session: - rows = session.execute( - select(Clip.id, Clip.hd_url, Clip.sd_url) - .where(Clip.dead_at.is_(None)) - .order_by(Clip.updated_at.asc()) - .limit(limit) - ).all() - with httpx.Client(headers=_UA, timeout=20, follow_redirects=True) as http: - for cid, hd, sd in rows: - url = hd or sd - if not url: - continue - checked += 1 - try: - r = http.head(url) - except httpx.HTTPError: - errors += 1 - continue - if r.status_code in _DEAD_CODES: - dead += 1 - dead_ids.append((cid, f"http {r.status_code}")) - time.sleep(0.2) - if apply: - with session_scope() as session: - for cid, reason in dead_ids: - session.execute( - text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"), - {"r": reason, "id": cid}, - ) - # Sprawdzone i zywe: odswiez updated_at, zeby kolejny run bral inne. - alive = [cid for cid, _hd, _sd in rows if cid not in {d[0] for d in dead_ids}] - if alive: - session.execute( - text("UPDATE clips SET updated_at = now() WHERE id = ANY(:ids)"), {"ids": alive} - ) - out = {"checked": checked, "dead": dead, "errors": errors, "applied": int(apply)} - print(out) - return out +from app.scheduler.clips_ingest import reap_dead def main() -> None: @@ -76,7 +18,7 @@ def main() -> None: ap.add_argument("--limit", type=int, default=500) ap.add_argument("--yes", action="store_true", help="zapisz dead_at (bez tego dry-run)") a = ap.parse_args() - run(limit=a.limit, apply=a.yes) + print(reap_dead(limit=a.limit, apply=a.yes)) if __name__ == "__main__": diff --git a/tests/test_clips_feed.py b/tests/test_clips_feed.py new file mode 100644 index 0000000..b3bbb41 --- /dev/null +++ b/tests/test_clips_feed.py @@ -0,0 +1,97 @@ +import random +import uuid +from datetime import UTC, datetime, timedelta + +from app.clips_feed import ( + Affinity, + Candidate, + apply_event, + cap_weights, + compose_page, + decay_weights, + decode_cursor, + encode_cursor, + event_kind_for_weights, + pick_affinity, + pick_explore, + score_candidate, + split_counts, +) + + +def test_cursor_roundtrip_and_garbage(): + pk = uuid.uuid4() + c = encode_cursor("abc123", pk, 7) + assert decode_cursor(c) == ("abc123", pk, 7) + assert decode_cursor("not-a-cursor") is None + assert decode_cursor("") is None + + +def test_event_kind_mapping(): + assert event_kind_for_weights("impression", 12000) == "dwell" + assert event_kind_for_weights("impression", 3000) is None + assert event_kind_for_weights("skip", 500) == "skip" + assert event_kind_for_weights("skip", 5000) is None + assert event_kind_for_weights("like", None) == "like" + assert event_kind_for_weights("autoswipe", None) is None + + +def test_apply_event_like_and_not_interested_reason(): + aff = Affinity() + changed = apply_event(aff, kind="like", dwell_ms=None, reason=None, tags=["big-tits", "amateur"], niches=["just-boobs"], creator="kitsune") + assert changed + assert aff.tags["big-tits"] == 5.0 and aff.niches["just-boobs"] == 5.0 and aff.creators["kitsune"] == 4.0 + apply_event(aff, kind="not_interested", dwell_ms=None, reason="tag:amateur", tags=["amateur"], niches=[], creator=None) + # -4 za zdarzenie i -8 za wskazany tag + assert aff.tags["amateur"] == 5.0 - 4.0 - 8.0 + assert not aff.is_empty() + assert aff.top("tags", 1) == ["big-tits"] + + +def test_cap_and_decay(): + d = {f"t{i}": float(i) for i in range(100)} + capped = cap_weights(d, 10) + assert len(capped) == 10 and "t99" in capped and "t0" not in capped + decayed = decay_weights({"a": 1.0, "b": 0.2}, factor=0.5, floor=0.2) + assert decayed == {"a": 0.5} # b spada pod prog i znika + + +def _cand(creator, likes=0, tags=(), niches=(), days_old=30): + return Candidate( + id=uuid.uuid4(), creator=creator, likes=likes, + published_at=datetime.now(UTC) - timedelta(days=days_old), tags=list(tags), niches=list(niches), + ) + + +def test_pick_affinity_respects_creator_cap_and_order(): + aff = Affinity(tags={"x": 10.0}) + cands = [_cand("a", tags=["x"]) for _ in range(5)] + [_cand("b", tags=["x"], likes=1)] + picked = pick_affinity(cands, aff, 4, max_per_creator=2) + creators = [c.creator for c in picked] + assert creators.count("a") == 2 and "b" in creators and len(picked) == 3 + + +def test_score_prefers_fresh_and_liked(): + aff = Affinity() + old = _cand("a", likes=0, days_old=100) + fresh = _cand("a", likes=0, days_old=1) + liked = _cand("a", likes=1000, days_old=100) + assert score_candidate(fresh, aff) > score_candidate(old, aff) + assert score_candidate(liked, aff) > score_candidate(old, aff) + + +def test_split_and_compose_deterministic(): + assert split_counts(10, False) == (0, 10) + assert split_counts(10, True) == (8, 2) + a = [_cand("a") for _ in range(8)] + e = [_cand("e") for _ in range(2)] + p1 = compose_page(list(a), list(e), random.Random("s:1")) + p2 = compose_page(list(a), list(e), random.Random("s:1")) + assert [c.id for c, _ in p1] == [c.id for c, _ in p2] + assert sum(1 for _, slot in p1 if slot == "explore") == 2 + + +def test_pick_explore_weighted_but_complete(): + cands = [_cand("a", likes=0), _cand("b", likes=10**6)] + out = pick_explore(cands, 2, random.Random(1)) + assert {c.creator for c in out} == {"a", "b"}