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** (od 2026-10-02, uzasadnienie przy `score_candidate`) = log(Σ waga
tagu × rzadkość tagu) + 0,25·log(nisze) + 2,5·log(twórca) + 0,8·ln(1+lajki RedGifs)
+ 120·(odsetek lajków u nas − 3 %) − 15·(odsetek ucieczek u nas − 50 %) + 0,3 za świeżość.
25 % urządzeń zostaje na starej punktacji jako grupa kontrolna. 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]
    #: Przyjecie klipu u nas (kolumny clips.feed_*, migracja 0052).
    feed_views: int = 0
    feed_skips: int = 0
    feed_likes: int = 0


# Punktacja od 2026-10-02, dobrana na odtworzeniu 189 tys. zdarzen (wagi urzadzenia
# liczone zdarzenie po zdarzeniu, bez zagladania w przyszlosc; trening na 70 % najstarszych,
# test na reszcie). Miara: czy wyzszy wynik u TEGO urzadzenia znaczy lajk albo >= 10 s
# ogladania, a nie ucieczke < 2 s (AUC w obrebie urzadzenia). Stara punktacja: 0,576,
# ta: 0,652, regresja logistyczna na tych samych cechach: 0,654. Wnioski:
#   * suma wag tagow niewiele mowi, bo tagi RedGifs sa liczne i ogolne, a lajk dokladal
#     +5 kazdemu z ~10 tagow klipu; waga razy rzadkosc tagu (IDF) i log daje wiecej,
#   * nisze prawie nic (0,54), stad 0,25 zamiast 2,
#   * tworca to najlepszy sygnal z wag (0,58),
#   * najmocniejszy jest odsetek lajkow klipu u nas (0,60), wygladzony, zeby klip bez
#     historii nie dostawal skrajnych ocen; popularnosc z RedGifs slaba, ale nie zerowa.
W_NICHE = 0.25
W_CREATOR = 2.5
W_POP = 0.8
W_FEED_LIKE = 120.0
W_FEED_SKIP = 15.0
W_FRESH = 0.3
#: Wygladzanie odsetkow: klip zaczyna od 10 „wirtualnych" wyswietlen ze srednimi wartosciami.
PRIOR_VIEWS = 10.0
PRIOR_LIKE_RATE = 0.03
PRIOR_SKIP_RATE = 0.5


def _slog(x: float) -> float:
    return math.copysign(math.log1p(abs(x)), x)


def score_candidate(
    c: Candidate, aff: Affinity, now: datetime | None = None, idf: dict[str, float] | None = None
) -> float:
    """`idf`: tag -> ln(N/df) po calym katalogu klipow; bez niego kazdy tag waży 1."""
    now = now or datetime.now(UTC)
    default_idf = max(idf.values()) if idf else 1.0
    tag = sum(aff.tags.get(t, 0.0) * (idf.get(t, default_idf) if idf else 1.0) for t in c.tags)
    s = _slog(tag)
    s += W_NICHE * _slog(sum(aff.niches.get(n, 0.0) for n in c.niches))
    s += W_CREATOR * _slog(aff.creators.get(c.creator or "", 0.0))
    s += W_POP * math.log1p(max(c.likes, 0))
    like_rate = (c.feed_likes + PRIOR_VIEWS * PRIOR_LIKE_RATE) / (c.feed_views + PRIOR_VIEWS)
    skip_rate = (c.feed_skips + PRIOR_VIEWS * PRIOR_SKIP_RATE) / (c.feed_views + PRIOR_VIEWS)
    s += W_FEED_LIKE * (like_rate - PRIOR_LIKE_RATE) - W_FEED_SKIP * (skip_rate - PRIOR_SKIP_RATE)
    if c.published_at and c.published_at >= now - timedelta(days=FRESH_DAYS):
        s += W_FRESH
    return s


def score_candidate_v1(c: Candidate, aff: Affinity, now: datetime | None = None) -> float:
    """Punktacja sprzed 2026-10-02, zostaje dla grupy kontrolnej (patrz `uses_v1_scoring`)."""
    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


#: Odsetek urzadzen na starej punktacji (grupa kontrolna). Przydzial staly po hashu id
#: urzadzenia, wiec do porownania wystarczy ten sam hash przy analizie zdarzen.
V1_CONTROL_PERCENT = 25


def uses_v1_scoring(device_id: str) -> bool:
    h = int(hashlib.sha256(device_id.encode("utf-8")).hexdigest()[:8], 16)
    return h % 100 < V1_CONTROL_PERCENT


def pick_affinity(
    cands: list[Candidate],
    aff: Affinity,
    n: int,
    *,
    max_per_creator: int = MAX_PER_CREATOR,
    idf: dict[str, float] | None = None,
    v1: bool = False,
) -> list[Candidate]:
    """Najlepsze `n` po punktach, max `max_per_creator` klipów jednego twórcy na stronę."""
    if v1:
        ranked = sorted(cands, key=lambda c: -score_candidate_v1(c, aff))
    else:
        ranked = sorted(cands, key=lambda c: -score_candidate(c, aff, idf=idf))
    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))