ed4194a8d3
Add per-job "Run now" buttons and a "Start all now" button to the admin Scheduler dashboard (admin-gated endpoints). Triggers run the job in a background thread independent of its interval, refusing a concurrent run (409). While running, the long jobs (maintenance, enrich, backfill, shorts) report live progress through a decoupled contextvar sink, shown as a progress bar on the job row via the existing 4s poll. A manually-triggered run posts a completion notification to the triggering admin's inbox (scheduled runs stay silent to avoid spam); the inbox renders the "scheduler" type trilingually from type+data. While here, give the maintenance job its missing dashboard label/description in all three languages.
248 lines
9.5 KiB
Python
248 lines
9.5 KiB
Python
"""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.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 = 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)
|
|
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"],
|
|
}
|