"""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. """ import logging from datetime import timedelta from sqlalchemy import select from sqlalchemy.orm import Session from app import progress, quota, sysconfig from app.config import settings from app.models import Playlist, PlaylistItem, Video from app.notifications import create_notification from app.state import get_maintenance_batch from app.sync.runner import get_service_user from app.sync.videos import apply_video_details, parse_dt from app.utils import now_utc as _now from app.youtube.client import YouTubeClient log = logging.getLogger("siftlode.maintenance") # 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 _now() - 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) -> dict: """Phase 1: recover or hard-delete videos already flagged unavailable.""" 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])} now = _now() 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) -> 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 = get_maintenance_batch(db) # admin override, else config default flagged_new = 0 checked = 0 now = _now() # Process in videos.list-sized pages so we commit incrementally and stop on low quota. while checked < batch: if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"): break page = ( db.execute( select(Video) .where(Video.unavailable_since.is_(None)) .order_by( Video.last_checked_at.is_(None).desc(), # never-checked first Video.last_checked_at.asc(), ) .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"} with YouTubeClient(db, user) as yt: phase1 = _recheck_flagged(db, yt) phase2 = ( _revalidate_rolling(db, yt) if quota.remaining_today(db) > sysconfig.get_int(db, "backfill_quota_reserve") 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"], }