"""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 datetime, timedelta, timezone from sqlalchemy import or_, select from sqlalchemy.orm import Session from app import progress, quota from app.config import settings from app.models import Playlist, PlaylistItem, Video, VideoState 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.youtube.client import YouTubeClient log = logging.getLogger("subfeed.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 _now() -> datetime: return datetime.now(timezone.utc) 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 {} # Saved (VideoState.status == "saved"). for user_id, video_id in db.execute( select(VideoState.user_id, VideoState.video_id).where( VideoState.video_id.in_(ids), VideoState.status == "saved" ) ): impact.setdefault(user_id, set()).add(video_id) # In any of the user's playlists (covers watch-later, which is a playlist). 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) <= settings.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) > settings.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"], }