siftlode/backend/app/sync/explore.py

181 lines
7.2 KiB
Python
Raw Permalink Normal View History

"""Channel exploration: ingest an un-subscribed channel's uploads on demand (flagged
via_explore, kept per-explorer) and purge channels nobody ends up keeping.
A user opens a channel's page without subscribing → we pull a page of its recent uploads,
mark the new rows via_explore, and enrich them so they're immediately browsable. The videos
stay private to users who explored (or subscribe to) the channel the feed query gates them
so one user's curiosity never floods everyone's catalog. Subscribing is the "keep" signal;
otherwise the cleanup job hard-deletes the channel and its untouched ephemeral videos after a
grace period (mirrors the maintenance job's grace-then-delete safety)."""
import logging
from datetime import timedelta
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from app import sysconfig
from app.models import (
Channel,
ExploredChannel,
PlaylistItem,
SearchFind,
Subscription,
Video,
VideoState,
)
from app.sync.videos import _insert_stubs, _stub_from_playlist_item, apply_video_details
from app.utils import now_utc as _now
from app.youtube.client import YouTubeClient
log = logging.getLogger("siftlode.explore")
def _enrich_new(db: Session, yt: YouTubeClient, video_ids: list[str]) -> int:
"""Enrich just the not-yet-enriched videos among `video_ids` (videos.list, 1 unit/50)."""
if not video_ids:
return 0
videos = (
db.execute(
select(Video).where(
Video.id.in_(video_ids), Video.enriched_at.is_(None)
)
)
.scalars()
.all()
)
if not videos:
return 0
items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
now = _now()
for video in videos:
item = items.get(video.id)
if item is None:
video.enriched_at = now # unavailable; stop retrying
continue
apply_video_details(video, item)
video.enriched_at = now
db.commit()
return len(videos)
def explore_ingest_page(
db: Session, yt: YouTubeClient, channel: Channel, page_token: str | None = None
) -> dict:
"""Ingest one page (up to 50, newest-first) of the channel's uploads, marking the new
videos via_explore, then enrich them. Returns the count ingested and the next page token
(None when the back-catalogue is exhausted) so the caller can offer "load more". Stateless
pagination: the caller round-trips the token, so the channel's own backfill cursor (used by
the organic deep backfill) is never touched."""
if not channel.uploads_playlist_id:
return {"ingested": 0, "next_page_token": None}
data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
stubs: list[dict] = []
for item in data.get("items", []):
stub = _stub_from_playlist_item(item, channel.id)
if stub is not None:
stub["via_explore"] = True
stubs.append(stub)
inserted = _insert_stubs(db, stubs)
self_enriched = _enrich_new(db, yt, [s["id"] for s in stubs])
log.info(
"explore ingest channel=%s ingested=%s enriched=%s", channel.id, inserted, self_enriched
)
return {"ingested": inserted, "next_page_token": data.get("nextPageToken")}
def purge_unkept_explored(db: Session) -> dict:
"""Cleanup job: hard-delete explored channels nobody kept and their untouched ephemeral
videos, after a grace period (mirrors the maintenance grace-then-delete safety).
Keep signals that spare a channel/video:
- any user SUBSCRIBES to the channel (subscribe promotes it to organic channel-level), or
- a live ExploredChannel row (someone explored it within the grace window channel-level), or
- a video has per-user state: watched/in-progress (VideoState), in a playlist (PlaylistItem),
or surfaced by the user's own search (SearchFind) — these survive even if the channel goes.
A channel row is dropped only once it holds no videos at all. 0 quota."""
grace_days = sysconfig.get_int(db, "explore_grace_days")
cutoff = _now() - timedelta(days=grace_days)
expired = db.execute(
delete(ExploredChannel).where(ExploredChannel.created_at < cutoff)
).rowcount
db.commit()
sub_exists = (
select(Subscription.id).where(Subscription.channel_id == Channel.id).exists()
)
exp_exists = (
select(ExploredChannel.id).where(ExploredChannel.channel_id == Channel.id).exists()
)
orphan_ids = (
db.execute(
select(Channel.id).where(
Channel.from_explore.is_(True), ~sub_exists, ~exp_exists
)
)
.scalars()
.all()
)
if not orphan_ids:
return {"expired": expired, "videos_deleted": 0, "channels_deleted": 0}
state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
find_exists = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
videos_deleted = db.execute(
delete(Video).where(
Video.channel_id.in_(orphan_ids),
Video.via_explore.is_(True),
~state_exists,
~pl_exists,
~find_exists,
)
).rowcount
db.commit()
has_videos = select(Video.id).where(Video.channel_id == Channel.id).exists()
channels_deleted = db.execute(
delete(Channel).where(Channel.id.in_(orphan_ids), ~has_videos)
).rowcount
db.commit()
return {
"expired": expired,
"videos_deleted": videos_deleted,
"channels_deleted": channels_deleted,
}
def purge_unkept_search(db: Session, grace_days: int | None = None) -> dict:
"""Remove live-search result videos nobody kept, after a grace period — "as if never added".
A `via_search` video is KEPT (and spared) when anyone has watched / in-progress / hidden it
(a VideoState), saved/playlisted it (a PlaylistItem), or subscribes to its channel (it's then
organically owned). A bare SearchFind does NOT count it's the ephemeral "shown in search"
marker itself so a result you only glanced at is reclaimed. Deleting the video cascades its
SearchFinds. `grace_days=0` purges immediately (the admin "purge now" path); None uses config.
0 quota."""
grace = sysconfig.get_int(db, "search_grace_days") if grace_days is None else grace_days
cutoff = _now() - timedelta(days=grace)
state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
sub_exists = (
select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
)
deleted = db.execute(
delete(Video).where(
Video.via_search.is_(True),
Video.created_at < cutoff,
~state_exists,
~pl_exists,
~sub_exists,
)
).rowcount
db.commit()
return {"search_videos_deleted": deleted}
def purge_ephemeral(db: Session) -> dict:
"""The scheduler's discovery-cleanup pass: reclaim un-kept explored channels AND un-kept
live-search result videos in one run."""
return {**purge_unkept_explored(db), **purge_unkept_search(db)}