From 05df599f7a80ddd2c7a9900aff275d607628007f Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 18 Jun 2026 03:20:28 +0200 Subject: [PATCH] feat(maintenance): scheduled job to retire unplayable videos Add a daily maintenance/validation job that detects videos which can't be played anywhere and retires them safely. Two phases: re-check already-flagged videos (recover if available again, else hard-delete once the grace period elapses, cascading to states/playlist items), and a rolling re-validation of the least-recently-checked currently-available videos that flags newly-unplayable ones (hidden from the feed immediately via unavailable_since). Detection is ~free: a video missing from the videos.list response is deleted-or-private; an `upcoming` premiere >2 days past its scheduled start that never went live is abandoned. A still-live broadcast is kept (legit 24/7 stream). Enrichment now also fetches part=status to populate the status columns. Grace is 7 days for removed videos, none for abandoned. Before deleting, affected users get one batched notification (never per-video). Interval is admin-tunable via the Scheduler dashboard; batch size and grace are config. Quota-attributed to the system and bounded by the same backfill reserve as the other jobs. --- backend/app/config.py | 11 ++ backend/app/routes/feed.py | 5 + backend/app/scheduler.py | 11 ++ backend/app/sync/maintenance.py | 246 ++++++++++++++++++++++++++++++++ backend/app/sync/videos.py | 6 + backend/app/youtube/client.py | 2 +- 6 files changed, 280 insertions(+), 1 deletion(-) create mode 100644 backend/app/sync/maintenance.py diff --git a/backend/app/config.py b/backend/app/config.py index 52bbd8a..af60b2b 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -90,6 +90,17 @@ class Settings(BaseSettings): autotag_interval_minutes: int = 30 subscriptions_resync_minutes: int = 360 playlist_sync_minutes: int = 360 + # --- Maintenance / validation job --- + # Runs once a day by default (admin-tunable via the Scheduler dashboard). Grace periods + # are in days because that's the granularity that matters; the rolling re-validation + # re-checks the least-recently-checked N videos per run (1 unit / 50 videos), so on a + # ~233k catalog ~15000/day ≈ a 15-day full cycle for ~300 quota units. + maintenance_interval_minutes: int = 1440 + maintenance_revalidate_batch: int = 15000 + # Grace before a flagged-unavailable video is hard-deleted (cascades to states/playlist + # items). Deleted/private/paywalled get a recovery window; abandoned upcoming and + # confirmed-dead stuck-live have no age grace (deleted on the sweep that confirms them). + maintenance_grace_days: int = 7 # live_status values hidden from the feed by default. Completed-stream VODs # ("was_live") are real watchable content and stay visible. feed_default_hidden_live: str = "live,upcoming" diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 04ef3be..8afa915 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -154,6 +154,11 @@ def _filtered_query( state, and_(state.video_id == Video.id, state.user_id == user.id) ) + # Hide videos the maintenance job has flagged as currently unplayable (deleted/private/ + # abandoned). They're either deleted after a grace period or unhidden if they recover; + # either way they shouldn't show in any feed view meanwhile. + query = query.where(Video.unavailable_since.is_(None)) + if channel_id: query = query.where(Video.channel_id == channel_id) if min_duration is not None: diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 656db73..b6188ab 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -13,6 +13,7 @@ from app.db import SessionLocal from app.models import SchedulerSetting from app.state import is_sync_paused from app.sync.autotag import run_autotag_all +from app.sync.maintenance import run_maintenance from app.sync.playlists import sync_all_playlists from app.sync.runner import ( run_deep_backfill, @@ -44,6 +45,7 @@ JOB_INTERVALS: dict[str, int] = { "shorts": settings.shorts_probe_interval_minutes, "subscriptions": settings.subscriptions_resync_minutes, "playlist_sync": settings.playlist_sync_minutes, + "maintenance": settings.maintenance_interval_minutes, } # Sane bounds for an admin-set interval (minutes). @@ -201,6 +203,14 @@ def _playlist_sync_job() -> None: _job("playlist_sync", sync_all_playlists) +def _maintenance_job() -> None: + def work(db): + with quota.attribute(None, "maintenance"): + return run_maintenance(db) + + _job("maintenance", work) + + def start_scheduler() -> None: global _scheduler if not settings.scheduler_enabled or _scheduler is not None: @@ -219,6 +229,7 @@ def start_scheduler() -> None: "shorts": _shorts_job, "subscriptions": _subscriptions_job, "playlist_sync": _playlist_sync_job, + "maintenance": _maintenance_job, } scheduler = BackgroundScheduler(timezone="UTC") for job_id, fn in fns.items(): diff --git a/backend/app/sync/maintenance.py b/backend/app/sync/maintenance.py new file mode 100644 index 0000000..be9f4c9 --- /dev/null +++ b/backend/app/sync/maintenance.py @@ -0,0 +1,246 @@ +"""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 quota +from app.config import settings +from app.models import Playlist, PlaylistItem, Video, VideoState +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 = 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} + 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 = settings.maintenance_revalidate_batch + 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) + 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"], + } diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index c4f002d..eaeeff2 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -217,6 +217,7 @@ def apply_video_details(video: Video, item: dict) -> None: stats = item.get("statistics", {}) topics = item.get("topicDetails", {}) live = item.get("liveStreamingDetails") + status = item.get("status", {}) video.title = snippet.get("title") or video.title video.description = snippet.get("description") @@ -232,6 +233,11 @@ def apply_video_details(video: Video, item: dict) -> None: "defaultAudioLanguage" ) video.live_status = _live_status(snippet, live) + # status part: used by the maintenance/validation job to judge playability. + if status: + video.embeddable = status.get("embeddable") + video.privacy_status = status.get("privacyStatus") + video.upload_status = status.get("uploadStatus") # is_short is decided separately by the youtube.com/shorts probe (run_shorts_classification). diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index f0d0da4..c65c333 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -327,7 +327,7 @@ class YouTubeClient: data = self._get( "videos", { - "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails", + "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails,status", "id": ",".join(batch), "maxResults": 50, },