0df2236b37
Tabele per urzadzenie (migracja 0037): clip_seen (wykluczenie, wiersz przy
podaniu klipu), clip_feed_events (append-only: impression z dwell, skip,
complete, like, unlike, not_interested, autoswipe), clip_affinity (wagi
tag/nisza/tworca w JSONB), clip_favorites, clip_blacklist. GIN na clips.niches.
GET /clips/feed: kursor base64 {seed, pk, page} zamiast offsetu; 80% z puli
afinicznosci (skany po tagu/niszy/tworcy od losowego uuid, punktacja
sum wag + 0,3 ln(1+lajki) + swiezosc, max 2 klipy tworcy na strone) + 20%
explore (okno 2000 id po PK, losowanie wazone lajkami, zawijanie na koncu).
Bez wag 100% explore. POST /clips/events: zapis, podniesienie outcome,
ulubione, delty wag (like +5, complete +3, dwell>=10 s +1, skip<2 s -1,
not_interested -4 i -8 na wskazany cel). Onboarding 3 tapow, opcje,
blacklista tag/nisza/tworca, reset feedu, ulubione. Import miedzy
urzadzeniami kopiuje clip_seen, clip_favorites i clip_blacklist.
Job clips-maintenance co 24 h: decay wag x0,97, retencja zdarzen 90 dni,
reaper (HEAD na 2000 najdawniej sprawdzanych). Logika bez bazy w
app/clips_feed.py z testami (kursor, delty, cap/decay, punktacja, sklad).
260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""Ingest klipów (feed Swipe): upsert do `clips` + `clip_tags`, joby RedGifs.
|
||
|
||
Potok celowo prosty, bez `_ingest_iter_into_run`: klipy nie mają dedupu, matchera
|
||
performerów ani fingerprintów, a `(source, source_id)` jednoznacznie identyfikuje rekord.
|
||
Upsert `ON CONFLICT ... DO UPDATE` odświeża lajki/widoki/adresy i **zdejmuje `dead_at`**
|
||
(klip, który źródło znów podaje, żyje). Tagi tylko dokładamy (`DO NOTHING`), źródło ich
|
||
nie odbiera.
|
||
|
||
Dwa tryby:
|
||
- `run_redgifs_latest(pages)`: globalne „najnowsze", co godzinę z schedulera. 5 stron =
|
||
~500 klipów/run, RedGifs publikuje kilkaset na godzinę, więc to łapie prawie wszystko.
|
||
- `run_redgifs_niches(...)`: backfill po niszach (społecznościach), ręcznie ze skryptu
|
||
albo z `docker compose run` (patrz notatka o restarcie workera). Każde zapytanie widzi
|
||
maks. 10 000 wyników, dlatego katalog chodzi się per nisza, nie jednym listingiem.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections.abc import Iterable
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.connectors.clips.redgifs import RateLimited, RawClip, RedgifsClient, map_gif
|
||
from app.db import session_scope
|
||
from app.models.clip import Clip, ClipTag
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
def upsert_clips(session: Session, clips: Iterable[RawClip]) -> dict[str, int]:
|
||
seen = new = 0
|
||
for c in clips:
|
||
seen += 1
|
||
stmt = (
|
||
pg_insert(Clip)
|
||
.values(
|
||
source=c.source,
|
||
source_id=c.source_id,
|
||
title=c.title,
|
||
creator=c.creator,
|
||
creator_verified=c.creator_verified,
|
||
duration_sec=c.duration_sec,
|
||
width=c.width,
|
||
height=c.height,
|
||
has_audio=c.has_audio,
|
||
content_type=c.content_type,
|
||
niches=c.niches,
|
||
hd_url=c.hd_url,
|
||
sd_url=c.sd_url,
|
||
poster_url=c.poster_url,
|
||
thumb_url=c.thumb_url,
|
||
hls=c.hls,
|
||
likes=c.likes,
|
||
views=c.views,
|
||
published_at=c.published_at,
|
||
)
|
||
.on_conflict_do_update(
|
||
constraint="uq_clips_source_source_id",
|
||
set_={
|
||
"title": c.title,
|
||
"creator": c.creator,
|
||
"creator_verified": c.creator_verified,
|
||
"duration_sec": c.duration_sec,
|
||
"has_audio": c.has_audio,
|
||
"niches": c.niches,
|
||
"hd_url": c.hd_url,
|
||
"sd_url": c.sd_url,
|
||
"poster_url": c.poster_url,
|
||
"thumb_url": c.thumb_url,
|
||
"likes": c.likes,
|
||
"views": c.views,
|
||
"dead_at": None,
|
||
"dead_reason": None,
|
||
"updated_at": text("now()"),
|
||
},
|
||
)
|
||
# xmax = 0 tylko dla świeżo wstawionego wiersza (sztuczka Postgresa na
|
||
# rozróżnienie INSERT od UPDATE w jednym zapytaniu).
|
||
.returning(Clip.id, text("(xmax = 0) AS inserted"))
|
||
)
|
||
row = session.execute(stmt).one()
|
||
clip_id, inserted = row[0], bool(row[1])
|
||
if inserted:
|
||
new += 1
|
||
if c.tags:
|
||
session.execute(
|
||
pg_insert(ClipTag)
|
||
.values([{"clip_id": clip_id, "tag": t} for t in c.tags])
|
||
.on_conflict_do_nothing()
|
||
)
|
||
return {"seen": seen, "new": new, "updated": seen - new}
|
||
|
||
|
||
def _merge(total: dict[str, int], part: dict[str, int]) -> None:
|
||
for k, v in part.items():
|
||
total[k] = total.get(k, 0) + v
|
||
|
||
|
||
def run_redgifs_latest(*, pages: int = 5) -> dict[str, int]:
|
||
"""Globalne „najnowsze" RedGifs, `pages` stron po 100. Zatrzymuje się na pierwszej
|
||
stronie w całości już znanej (kolejne są starsze), żeby godzinny run nie mielił
|
||
w kółko tych samych 500 klipów."""
|
||
total: dict[str, int] = {"seen": 0, "new": 0, "updated": 0, "errors": 0}
|
||
client = RedgifsClient()
|
||
try:
|
||
for page in range(1, pages + 1):
|
||
try:
|
||
gifs = client.latest(page=page)
|
||
except Exception as e:
|
||
log.warning("clips/redgifs latest: strona %d padla: %s", page, e)
|
||
total["errors"] += 1
|
||
break
|
||
clips = [c for c in (map_gif(g) for g in gifs) if c is not None]
|
||
if not clips:
|
||
break
|
||
with session_scope() as session:
|
||
part = upsert_clips(session, clips)
|
||
_merge(total, part)
|
||
if part["new"] == 0 and page > 1:
|
||
break
|
||
finally:
|
||
client.close()
|
||
log.info("clips/redgifs latest: %s", total)
|
||
return total
|
||
|
||
|
||
def run_redgifs_niches(
|
||
*, niche_pages: int = 1, per_niche_pages: int = 1, order: str = "latest", skip: int = 0
|
||
) -> dict[str, int]:
|
||
"""Backfill po niszach: `niche_pages` stron listy nisz (100 na stronę, wg API
|
||
najpopularniejsze pierwsze), dla każdej niszy `per_niche_pages` stron gifów."""
|
||
total: dict[str, int] = {"seen": 0, "new": 0, "updated": 0, "errors": 0, "niches": 0}
|
||
client = RedgifsClient()
|
||
try:
|
||
niche_ids: list[str] = []
|
||
for p in range(1, niche_pages + 1):
|
||
try:
|
||
data = client.niches(page=p)
|
||
except Exception as e:
|
||
log.warning("clips/redgifs niches: lista strona %d padla: %s", p, e)
|
||
total["errors"] += 1
|
||
break
|
||
niche_ids.extend(str(n.get("id")) for n in data.get("niches") or [] if n.get("id"))
|
||
for nid in niche_ids[skip:]:
|
||
total["niches"] += 1
|
||
for page in range(1, per_niche_pages + 1):
|
||
try:
|
||
gifs = client.niche_gifs(nid, order=order, page=page)
|
||
except RateLimited as e:
|
||
# Limit trwaly: koniec przebiegu, wznowienie przez --skip <niches>.
|
||
# Pierwszy backfill (2026-09-14) mielil 40 nisz po kolei w 429.
|
||
log.warning("clips/redgifs niches: %s; przerywam po %d niszach", e, total["niches"])
|
||
total["errors"] += 1
|
||
return total
|
||
except Exception as e:
|
||
log.warning("clips/redgifs niche %s strona %d padla: %s", nid, page, e)
|
||
total["errors"] += 1
|
||
break
|
||
clips = [c for c in (map_gif(g) for g in gifs) if c is not None]
|
||
if not clips:
|
||
break
|
||
with session_scope() as session:
|
||
_merge(total, upsert_clips(session, clips))
|
||
if total["niches"] % 20 == 0:
|
||
log.info("clips/redgifs niches: %d nisz, %s", total["niches"], total)
|
||
finally:
|
||
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
|