"""Maintenance / validation job: detect videos that can't be played anywhere and retire them safely. Lifecycle (no permanent soft-delete state — just a grace timer before a hard delete): 1. Re-check flagged videos (those with unavailable_since set). If a video has recovered (e.g. a private video made public again), clear the flag. Otherwise, if its grace period has elapsed, HARD DELETE it (cascades to video_states / playlist_items). Before deleting, gather which users had it saved or in a playlist so we can tell them. 2. Rolling re-validation: re-fetch the least-recently-checked N currently-available videos. Newly-unplayable ones get flagged (hidden from the feed immediately via unavailable_since); the grace timer starts now and they're deleted in a later run if they don't recover. Detection is ~free: a video missing from the videos.list response is deleted-or-private. We never delete a live broadcast that still reports viewers (a legit 24/7 stream); we only flag an `upcoming` premiere that blew past its scheduled start and never went live. Quota: phase 1 costs ~(flagged / 50) units, phase 2 ~(batch / 50). Both respect the same backfill reserve as the other jobs so interactive syncs never starve. """ from datetime import datetime, timedelta, timezone from sqlalchemy import or_, select from sqlalchemy.orm import Session from app import logs, progress, quota, sysconfig from app.config import settings from app.models import Video from app.reader.models import Playlist, PlaylistItem from app.notifications import create_notification from app.sync.runner import get_service_user from app.sync.videos import apply_video_details, parse_dt from app.youtube.client import YouTubeClient log = logs.get_logger(__name__) # Reasons recorded on Video.unavailable_reason. "removed" = absent from videos.list # (deleted or made private); "abandoned" = an upcoming premiere that never went live. GRACE_DAYS_BY_REASON = { "removed": settings.maintenance_grace_days, "abandoned": 0, # already past-due when detected; deleted on the confirming sweep } DEFAULT_GRACE_DAYS = settings.maintenance_grace_days def _grace_days(reason: str | None) -> int: return GRACE_DAYS_BY_REASON.get(reason or "", DEFAULT_GRACE_DAYS) def _is_abandoned_upcoming(item: dict) -> bool: """True for an `upcoming` broadcast whose scheduled start is >2 days past and which never actually went live (no actualStartTime).""" snippet = item.get("snippet", {}) if snippet.get("liveBroadcastContent") != "upcoming": return False live = item.get("liveStreamingDetails") or {} if live.get("actualStartTime"): return False # it did go live at some point — not abandoned scheduled = parse_dt(live.get("scheduledStartTime")) if scheduled is None: return False return datetime.now(timezone.utc) - scheduled > timedelta(days=2) def _evaluate(video: Video, item: dict | None) -> str | None: """Given a fresh videos.list item (or None if absent), return an unavailable reason, or None if the video is fine. Applies fresh metadata as a side effect when present.""" if item is None: return "removed" apply_video_details(video, item) if _is_abandoned_upcoming(item): return "abandoned" # A still-"live" broadcast is kept (a legit 24/7 stream); ended streams have already # settled to was_live via apply_video_details. Nothing else here is a delete trigger — # embeddable=false only blocks our in-app embed, not playback on youtube.com. return None def _collect_user_impact(db: Session, videos: list[Video]) -> dict[int, list[dict]]: """For videos about to be deleted, map user_id -> list of {id, title, reason} for the videos that user had saved or in one of their playlists. Must run BEFORE the delete, while the video_states / playlist_items rows still exist.""" by_id = {v.id: v for v in videos} ids = list(by_id) impact: dict[int, set[str]] = {} if not ids: return {} # Anyone who had it in a playlist — which now also covers "saved": Watch later became a # built-in playlist, so "saved" is no longer a VideoState status (the old status=="saved" # scan matched nothing and under-counted impacted users). for user_id, video_id in db.execute( select(Playlist.user_id, PlaylistItem.video_id) .join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id) .where(PlaylistItem.video_id.in_(ids)) ): impact.setdefault(user_id, set()).add(video_id) return { user_id: [ { "id": vid, "title": by_id[vid].title, "reason": by_id[vid].unavailable_reason, } for vid in vids ] for user_id, vids in impact.items() } def _notify_removed(db: Session, impact: dict[int, list[dict]]) -> int: """One batched notification per affected user (never per-video).""" for user_id, videos in impact.items(): n = len(videos) create_notification( db, user_id=user_id, type="maintenance", title="Videos removed", body=( f"{n} saved or playlisted " f"{'video was' if n == 1 else 'videos were'} removed because " "they are no longer available on YouTube." ), data={"count": n, "videos": videos}, commit=False, ) db.commit() return len(impact) def _recheck_flagged(db: Session, yt: YouTubeClient, now: datetime) -> dict: """Phase 1: recover or hard-delete videos already flagged unavailable. `now` is the run's: phase 2 skips a video stamped with it, so one recovered here is not fetched again.""" flagged = db.execute(select(Video).where(Video.unavailable_since.is_not(None))).scalars().all() if not flagged: return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0} progress.report(0, len(flagged), "recheck") items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])} recovered = 0 to_delete: list[Video] = [] for video in flagged: reason = _evaluate(video, items.get(video.id)) video.last_checked_at = now if reason is None: # Recovered: unhide. video.unavailable_since = None video.unavailable_reason = None recovered += 1 continue # Still unavailable; keep the original reason/timer unless it changed. if reason != video.unavailable_reason: video.unavailable_reason = reason deadline = video.unavailable_since + timedelta(days=_grace_days(reason)) if now >= deadline: to_delete.append(video) db.commit() # persist recoveries / reason updates / last_checked_at notified = 0 deleted = 0 if to_delete: impact = _collect_user_impact(db, to_delete) for video in to_delete: db.delete(video) db.commit() # cascades to video_states / playlist_items deleted = len(to_delete) notified = _notify_removed(db, impact) return { "checked": len(flagged), "recovered": recovered, "deleted": deleted, "notified": notified, } def _revalidate_rolling(db: Session, yt: YouTubeClient, now: datetime) -> dict: """Phase 2: re-check the least-recently-checked currently-available videos and flag any that have become unplayable (hidden immediately; deleted later if they don't recover).""" batch = sysconfig.get_int( db, "maintenance_revalidate_batch" ) # admin override, else config default flagged_new = 0 checked = 0 # Process in videos.list-sized pages so we commit incrementally and stop on low quota. while checked < batch: if quota.backfill_reserve_reached(): break page = ( db.execute( select(Video) .where(Video.unavailable_since.is_(None)) # Not one this run already checked: with fewer available videos than the batch, the # loop used to wrap around and re-fetch the same ones until `checked` reached the # batch, spending quota on answers it already had. .where(or_(Video.last_checked_at.is_(None), Video.last_checked_at < now)) # NULLS FIRST = never-checked first, then least-recently-checked; equivalent to the old # `(last_checked_at IS NULL) DESC, last_checked_at ASC` but in the exact shape of the # partial ix_videos_recheck_order (migration 0069, PERF-18) so it rides that index. .order_by(Video.last_checked_at.asc().nulls_first()) .limit(min(50, batch - checked)) ) .scalars() .all() ) if not page: break items = {it["id"]: it for it in yt.get_videos([v.id for v in page])} for video in page: reason = _evaluate(video, items.get(video.id)) video.last_checked_at = now if reason is not None: video.unavailable_since = now video.unavailable_reason = reason flagged_new += 1 db.commit() checked += len(page) progress.report(checked, batch, "revalidate") return {"checked": checked, "flagged": flagged_new} def run_maintenance(db: Session) -> dict: """Entry point invoked by the scheduler. Quota-attributed to the system (no actor).""" user = get_service_user(db) if user is None: return {"reason": "no service user"} # One `now` for the whole run: each phase taking its own let a video phase 1 had just recovered # (stamped with phase 1's) pass phase 2's "not checked this run" test and be fetched twice. now = datetime.now(timezone.utc) with YouTubeClient(db, user) as yt: phase1 = _recheck_flagged(db, yt, now) phase2 = ( _revalidate_rolling(db, yt, now) if not quota.backfill_reserve_reached() else {"checked": 0, "flagged": 0, "skipped": "low quota"} ) return { "rechecked": phase1["checked"], "recovered": phase1["recovered"], "deleted": phase1["deleted"], "notified": phase1["notified"], "revalidated": phase2["checked"], "flagged": phase2["flagged"], }