8971d81ef9
Tworcy kasuja zwykle cale konto naraz, a przemial przy 1000 klipow na godzine robi okrazenie katalogu w okolo tydzien. Zgloszenie d343a9b6: strona tworcy z pustymi kafelkami, 120 "zywych" klipow, kazdy 403. Po oznaczeniu martwego klipu (przemial albo zgloszenie z apki) od razu HEAD na pozostalych zywych klipach tego tworcy niesprawdzanych od 6 h, do 600 na przebieg. Sprawdzone na produkcji: 120 z 120 oznaczone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
372 lines
15 KiB
Python
372 lines
15 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 _head_dead(http, url: str) -> str | None:
|
||
"""'http 403' itp., gdy plik klipu zniknal; None, gdy zyje albo nie wiadomo (blad sieci)."""
|
||
import httpx
|
||
|
||
try:
|
||
r = http.head(url)
|
||
except httpx.HTTPError:
|
||
return None
|
||
return f"http {r.status_code}" if r.status_code in _DEAD_CODES else None
|
||
|
||
|
||
def check_clips_now(clip_ids: list) -> int:
|
||
"""Natychmiastowe sprawdzenie klipow zgloszonych przez apke jako zepsute (zdarzenie
|
||
`error`). Ten sam test co reaper, tylko bez czekania na kolejke. Zwraca ile oznaczono."""
|
||
import httpx
|
||
from sqlalchemy import select
|
||
|
||
if not clip_ids:
|
||
return 0
|
||
with session_scope() as session:
|
||
rows = session.execute(
|
||
select(Clip.id, Clip.hd_url, Clip.sd_url).where(Clip.id.in_(clip_ids), Clip.dead_at.is_(None))
|
||
).all()
|
||
marked_ids: list = []
|
||
with httpx.Client(headers=_REAP_UA, timeout=15, follow_redirects=True) as http:
|
||
for cid, hd, sd in rows:
|
||
url = hd or sd
|
||
reason = _head_dead(http, url) if url else None
|
||
if reason is None:
|
||
continue
|
||
with session_scope() as session:
|
||
session.execute(
|
||
text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"),
|
||
{"r": f"{reason} (apka)", "id": cid},
|
||
)
|
||
marked_ids.append(cid)
|
||
marked = len(marked_ids)
|
||
if marked:
|
||
log.info("clips: %d zgloszonych z apki oznaczone jako martwe", marked)
|
||
sweep_creators(marked_ids)
|
||
return marked
|
||
|
||
|
||
#: Ile klipow jednego przebiegu zamiatania i jak dawno sprawdzony klip bierzemy ponownie.
|
||
_SWEEP_CAP = 600
|
||
_SWEEP_RECHECK_HOURS = 6
|
||
|
||
|
||
def sweep_creators(dead_clip_ids: list, *, cap: int = _SWEEP_CAP, sleep: float = 0.1) -> int:
|
||
"""Po znalezieniu skasowanego klipu sprawdza od razu pozostale zywe klipy jego tworcy.
|
||
|
||
Tworcy kasuja zwykle cale konto naraz, a przemial przy 1000 klipow na godzine robi
|
||
okrazenie katalogu w okolo tydzien. 2026-10-06 @thenaughtywoman mial 115 martwych
|
||
i 120 "zywych", z ktorych kazdy oddawal 403, wiec strona tworcy pokazywala puste
|
||
kafelki, a feed podsuwal czarne klipy. Kazdy klip sprawdzamy osobno (HEAD), nic nie
|
||
oznaczamy w ciemno. Sprawdzone zywe dostaja swiezy updated_at, wiec ten sam tworca
|
||
zgloszony z kilku telefonow naraz nie jest mielony w kolko. Zwraca ile oznaczono.
|
||
"""
|
||
import time as _time
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
import httpx
|
||
from sqlalchemy import select
|
||
|
||
if not dead_clip_ids:
|
||
return 0
|
||
recheck_before = datetime.now(UTC) - timedelta(hours=_SWEEP_RECHECK_HOURS)
|
||
with session_scope() as session:
|
||
creators = [
|
||
c for (c,) in session.execute(
|
||
select(Clip.creator).where(Clip.id.in_(dead_clip_ids), Clip.creator.is_not(None)).distinct()
|
||
).all()
|
||
]
|
||
if not creators:
|
||
return 0
|
||
rows = session.execute(
|
||
select(Clip.id, Clip.hd_url, Clip.sd_url)
|
||
.where(
|
||
Clip.creator.in_(creators),
|
||
Clip.dead_at.is_(None),
|
||
Clip.updated_at < recheck_before,
|
||
)
|
||
.order_by(Clip.updated_at.asc())
|
||
.limit(cap)
|
||
).all()
|
||
dead: list[tuple] = []
|
||
alive: list = []
|
||
with httpx.Client(headers=_REAP_UA, timeout=15, follow_redirects=True) as http:
|
||
for cid, hd, sd in rows:
|
||
url = hd or sd
|
||
if not url:
|
||
continue
|
||
reason = _head_dead(http, url)
|
||
if reason:
|
||
dead.append((cid, reason))
|
||
else:
|
||
alive.append(cid)
|
||
_time.sleep(sleep)
|
||
with session_scope() as session:
|
||
for cid, reason in dead:
|
||
session.execute(
|
||
text("UPDATE clips SET dead_at = now(), dead_reason = :r WHERE id = :id AND dead_at IS NULL"),
|
||
{"r": f"{reason} (tworca)", "id": cid},
|
||
)
|
||
if alive:
|
||
session.execute(text("UPDATE clips SET updated_at = now() WHERE id = ANY(:ids)"), {"ids": alive})
|
||
if rows:
|
||
log.info("clips: zamiatanie %d tworcow, sprawdzone %d, martwe %d", len(creators), len(rows), len(dead))
|
||
return len(dead)
|
||
|
||
|
||
def reap_dead(*, limit: int = 2000, apply: bool = True, sleep: float = 0.2) -> 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(sleep)
|
||
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})
|
||
dead += sweep_creators(list(dead_set))
|
||
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
|