e6bcf5ba1e
Register 8 more keys in the sysconfig registry (quota daily budget + backfill reserve, recent-backfill window, shorts-probe params, enrich/autotag batch sizes) and route their reads through the DB-override-or-env resolver at every call site (quota.py, sync/runner.py, sync/videos.py, sync/autotag.py, sync/maintenance.py, routes/scheduler.py). All read sites already had a db session in scope. Reads are read-through (no cache), matching app.state; admin can tune them live, no redeploy.
376 lines
14 KiB
Python
376 lines
14 KiB
Python
"""Video ingestion: free RSS detection, recent-first (and deep) backfill from the
|
|
uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts
|
|
and livestream classification)."""
|
|
import re
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import and_, func, or_, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import progress, sysconfig
|
|
from app.models import Channel, Video
|
|
from app.youtube.client import YouTubeClient, best_thumbnail
|
|
from app.youtube.rss import fetch_channel_feed
|
|
from app.youtube.shorts import make_client, probe_is_short
|
|
|
|
_DURATION_RE = re.compile(
|
|
r"P(?:(\d+)D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE
|
|
)
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def parse_iso8601_duration(value: str | None) -> int | None:
|
|
if not value:
|
|
return None
|
|
match = _DURATION_RE.fullmatch(value)
|
|
if not match:
|
|
return None
|
|
days, hours, minutes, seconds = (int(g) if g else 0 for g in match.groups())
|
|
return days * 86400 + hours * 3600 + minutes * 60 + seconds
|
|
|
|
|
|
def parse_dt(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _insert_stubs(db: Session, rows: list[dict]) -> int:
|
|
"""Idempotently insert video stubs; returns the number of newly inserted rows.
|
|
|
|
Uses RETURNING so the count is exact (ON CONFLICT DO NOTHING leaves rowcount
|
|
unreliable across drivers). Duplicates within the batch are dropped first so a
|
|
single INSERT can't raise CardinalityViolation on the conflict target."""
|
|
if not rows:
|
|
return 0
|
|
seen: dict[str, dict] = {}
|
|
for row in rows:
|
|
seen[row["id"]] = row
|
|
stmt = (
|
|
pg_insert(Video)
|
|
.values(list(seen.values()))
|
|
.on_conflict_do_nothing(index_elements=[Video.id])
|
|
.returning(Video.id)
|
|
)
|
|
inserted = len(db.execute(stmt).fetchall())
|
|
db.commit()
|
|
return inserted
|
|
|
|
|
|
# --- RSS (free) ---
|
|
def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int:
|
|
"""Insert new video stubs from an already-fetched feed and stamp the channel's last RSS
|
|
time. DB-only — must run on the session's own thread; the network fetch
|
|
(fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it."""
|
|
inserted = _insert_stubs(db, entries)
|
|
channel.last_rss_at = _now()
|
|
db.add(channel)
|
|
db.commit()
|
|
return inserted
|
|
|
|
|
|
def poll_rss_channel(db: Session, channel: Channel) -> int:
|
|
return apply_rss_feed(db, channel, fetch_channel_feed(channel.id))
|
|
|
|
|
|
# --- Backfill (uploads playlist; costs quota) ---
|
|
def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
|
|
content = item.get("contentDetails", {})
|
|
snippet = item.get("snippet", {})
|
|
video_id = content.get("videoId")
|
|
if not video_id:
|
|
return None
|
|
published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt"))
|
|
return {
|
|
"id": video_id,
|
|
"channel_id": channel_id,
|
|
"title": snippet.get("title"),
|
|
"published_at": published,
|
|
"thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
|
|
}
|
|
|
|
|
|
def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> int:
|
|
"""First pass: newest videos up to the configured count or age cutoff. Records a
|
|
resume cursor for the later deep backfill."""
|
|
if not channel.uploads_playlist_id:
|
|
channel.recent_synced_at = _now()
|
|
db.add(channel)
|
|
db.commit()
|
|
return 0
|
|
|
|
cutoff = _now() - timedelta(days=sysconfig.get_int(db, "backfill_recent_max_days"))
|
|
collected: list[dict] = []
|
|
page_token: str | None = None
|
|
next_cursor: str | None = None
|
|
done = False
|
|
|
|
while True:
|
|
page_start = page_token # token that fetched THIS page (None = first page)
|
|
data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
|
|
stopped_mid_page = False
|
|
for item in data.get("items", []):
|
|
stub = _stub_from_playlist_item(item, channel.id)
|
|
if stub is None:
|
|
continue
|
|
if stub["published_at"] and stub["published_at"] < cutoff:
|
|
stopped_mid_page = True
|
|
break
|
|
collected.append(stub)
|
|
if len(collected) >= sysconfig.get_int(db, "backfill_recent_max_videos"):
|
|
stopped_mid_page = True
|
|
break
|
|
page_token = data.get("nextPageToken")
|
|
if stopped_mid_page:
|
|
# We stopped partway through this page (age cutoff or count cap), so the older
|
|
# items still on it weren't collected. Resume the deep backfill FROM THIS page
|
|
# (re-fetch it; inserts are idempotent) so that tail isn't silently skipped.
|
|
next_cursor = page_start
|
|
done = False
|
|
break
|
|
if not page_token:
|
|
next_cursor = None
|
|
done = True
|
|
break
|
|
next_cursor = page_token # whole page consumed, more pages remain
|
|
|
|
channel.recent_synced_at = _now()
|
|
channel.backfill_cursor = next_cursor
|
|
channel.backfill_done = done
|
|
inserted = _insert_stubs(db, collected)
|
|
db.add(channel)
|
|
db.commit()
|
|
return inserted
|
|
|
|
|
|
def backfill_channel_deep(
|
|
db: Session, yt: YouTubeClient, channel: Channel, max_pages: int = 5
|
|
) -> int:
|
|
"""Continue paging older uploads from the stored cursor (bounded per run)."""
|
|
if channel.backfill_done or not channel.uploads_playlist_id:
|
|
return 0
|
|
page_token = channel.backfill_cursor
|
|
total = 0
|
|
for _ in range(max_pages):
|
|
data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
|
|
rows = [
|
|
s
|
|
for s in (
|
|
_stub_from_playlist_item(it, channel.id) for it in data.get("items", [])
|
|
)
|
|
if s is not None
|
|
]
|
|
total += _insert_stubs(db, rows)
|
|
page_token = data.get("nextPageToken")
|
|
channel.backfill_cursor = page_token
|
|
if not page_token:
|
|
channel.backfill_done = True
|
|
break
|
|
db.add(channel)
|
|
db.commit()
|
|
return total
|
|
|
|
|
|
def reconcile_full_history(db: Session) -> int:
|
|
"""Mark channels fully backfilled once their stored uploads meet or exceed YouTube's
|
|
advertised video_count. The uploads playlist length is normally <= the advertised
|
|
count, so 'stored >= video_count' means we already hold the whole history — even if
|
|
the deep cursor never reached the end (e.g. a small channel that was never
|
|
deep-requested, so the demand-driven deep job never ran). Without this such a channel
|
|
sits in 'needs full history' forever despite having every video. Idempotent, no API
|
|
quota; returns the number of channels flipped."""
|
|
stored = (
|
|
select(func.count(Video.id))
|
|
.where(Video.channel_id == Channel.id)
|
|
.scalar_subquery()
|
|
)
|
|
result = db.execute(
|
|
update(Channel)
|
|
.where(
|
|
Channel.backfill_done.is_(False),
|
|
Channel.video_count.is_not(None),
|
|
stored >= Channel.video_count,
|
|
)
|
|
.values(backfill_done=True, backfill_cursor=None)
|
|
)
|
|
db.commit()
|
|
return result.rowcount
|
|
|
|
|
|
# --- Enrichment (videos.list) ---
|
|
def _live_status(snippet: dict, live_details: dict | None) -> str:
|
|
broadcast = snippet.get("liveBroadcastContent")
|
|
if broadcast == "live":
|
|
return "live"
|
|
if broadcast == "upcoming":
|
|
return "upcoming"
|
|
if live_details and live_details.get("actualEndTime"):
|
|
return "was_live"
|
|
return "none"
|
|
|
|
|
|
def apply_video_details(video: Video, item: dict) -> None:
|
|
snippet = item.get("snippet", {})
|
|
content = item.get("contentDetails", {})
|
|
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")
|
|
if not video.published_at:
|
|
video.published_at = parse_dt(snippet.get("publishedAt"))
|
|
video.thumbnail_url = best_thumbnail(snippet.get("thumbnails")) or video.thumbnail_url
|
|
video.duration_seconds = parse_iso8601_duration(content.get("duration"))
|
|
video.view_count = int(stats["viewCount"]) if stats.get("viewCount") else None
|
|
video.like_count = int(stats["likeCount"]) if stats.get("likeCount") else None
|
|
video.category_id = int(snippet["categoryId"]) if snippet.get("categoryId") else None
|
|
video.topic_categories = topics.get("topicCategories")
|
|
video.default_language = snippet.get("defaultLanguage") or snippet.get(
|
|
"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).
|
|
|
|
|
|
def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
|
|
limit = limit or sysconfig.get_int(db, "enrich_batch_size")
|
|
videos = (
|
|
db.execute(select(Video).where(Video.enriched_at.is_(None)).limit(limit))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
if not videos:
|
|
return 0
|
|
items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
|
|
now = _now()
|
|
enriched = 0
|
|
for video in videos:
|
|
item = items.get(video.id)
|
|
if item is None:
|
|
# Unavailable (deleted/private): stamp so we don't keep retrying.
|
|
video.enriched_at = now
|
|
continue
|
|
apply_video_details(video, item)
|
|
video.enriched_at = now
|
|
enriched += 1
|
|
db.commit()
|
|
return enriched
|
|
|
|
|
|
def refresh_live(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
|
|
"""Re-fetch videos in a transient live state so they pick up their final shape.
|
|
|
|
`enrich_pending` is one-shot (it only touches enriched_at IS NULL), but a live broadcast
|
|
is not a final state: a stream ends and becomes a normal VOD — gaining a real duration
|
|
and flipping live_status to was_live — and an upcoming premiere eventually goes live then
|
|
ends. So we revisit anything still live/upcoming, plus a just-ended stream that hasn't got
|
|
its duration yet (YouTube can lag a bit after the broadcast), until it settles. Cheap:
|
|
videos.list is 1 unit per 50 ids, and the live/upcoming set is normally tiny."""
|
|
limit = limit or sysconfig.get_int(db, "enrich_batch_size")
|
|
videos = (
|
|
db.execute(
|
|
select(Video)
|
|
.where(
|
|
or_(
|
|
Video.live_status.in_(("live", "upcoming")),
|
|
and_(
|
|
Video.live_status == "was_live",
|
|
Video.duration_seconds.is_(None),
|
|
),
|
|
)
|
|
)
|
|
.limit(limit)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
if not videos:
|
|
return 0
|
|
items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
|
|
now = _now()
|
|
updated = 0
|
|
for video in videos:
|
|
item = items.get(video.id)
|
|
if item is None:
|
|
# The broadcast vanished (deleted/private). Stop treating it as live so it
|
|
# doesn't get refreshed forever; it just becomes an ordinary (dead) row.
|
|
video.live_status = "none"
|
|
continue
|
|
apply_video_details(video, item)
|
|
video.enriched_at = now
|
|
updated += 1
|
|
db.commit()
|
|
return updated
|
|
|
|
|
|
def run_shorts_classification(db: Session, limit: int | None = None) -> dict:
|
|
"""Finalize Shorts classification: cheaply mark non-candidates, then probe the
|
|
youtube.com/shorts URL for short-enough, non-live, enriched videos."""
|
|
limit = limit or sysconfig.get_int(db, "shorts_probe_batch")
|
|
probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds")
|
|
|
|
# 1) Anything enriched that can't be a Short -> finalize without a probe.
|
|
db.execute(
|
|
update(Video)
|
|
.where(
|
|
Video.shorts_probed.is_(False),
|
|
Video.enriched_at.is_not(None),
|
|
or_(
|
|
Video.duration_seconds.is_(None),
|
|
Video.duration_seconds > probe_max,
|
|
Video.live_status != "none",
|
|
),
|
|
)
|
|
.values(is_short=False, shorts_probed=True)
|
|
)
|
|
db.commit()
|
|
|
|
# 2) Probe the remaining candidates.
|
|
candidates = (
|
|
db.execute(
|
|
select(Video).where(
|
|
Video.shorts_probed.is_(False),
|
|
Video.enriched_at.is_not(None),
|
|
Video.duration_seconds <= probe_max,
|
|
Video.live_status == "none",
|
|
).limit(limit)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
if not candidates:
|
|
return {"probed": 0, "shorts": 0}
|
|
|
|
probed = 0
|
|
shorts = 0
|
|
with make_client() as client:
|
|
def work(video: Video):
|
|
return video, probe_is_short(client, video.id)
|
|
|
|
with ThreadPoolExecutor(max_workers=16) as pool:
|
|
for video, result in pool.map(work, candidates):
|
|
if result is None:
|
|
continue # leave unprobed; retry on a later run
|
|
video.is_short = result
|
|
video.shorts_probed = True
|
|
probed += 1
|
|
if result:
|
|
shorts += 1
|
|
progress.report(probed, len(candidates), "probe")
|
|
db.commit()
|
|
return {"probed": probed, "shorts": shorts}
|