"""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 . # 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