goon/app/clips_feed.py

"""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": <uuid>, "p": <nr strony>}`. 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))