"""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 #: `error`: odtwarzacz w apce sie wywrocil albo stal bez obrazu (reason 'stall'). Nie #: rusza wag, a API od razu sprawdza plik i oznacza klip martwy (check_clips_now). ALLOWED_KINDS = frozenset( {"impression", "skip", "complete", "like", "unlike", "not_interested", "autoswipe", "open", "error"} ) #: 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))