peter/siftlode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(maintenance): scheduled job to retire unplayable videos

Browse source 
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.
This commit is contained in:
npeter83 2026-06-18 03:20:28 +02:00
parent b9a3a9012d
commit c4aecf9f77
6 changed files with 280 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

246
backend/app/sync/maintenance.py Normal file
View file

@ -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"],
}