peter
bd2340c204
The import asks YouTube how long the watched videos of the candidate channels are (five or more watched, none the user follows, blocked, put out or owns), keeps the length on the Takeout row and names the channels that make the list: five videos of five minutes or more. The shorts job asks what an import could not. A channel's counted videos come in as an exploration's when its list is opened, ready to play; a channel can be put out of the list and back.
475 lines
19 KiB
Python
475 lines
19 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 collections.abc import Callable
|
|
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 logs, progress, quota, sysconfig
|
|
from app.catalog.models import LIVE_OR_UPCOMING, Channel, Video
|
|
from app.titles import normalize_title
|
|
from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail, is_playlist_gone
|
|
from app.youtube.shorts import make_client, probe_is_short
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
_DURATION_RE = re.compile(r"P(?:(\d+)D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE)
|
|
|
|
|
|
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"))
|
|
# silent-ok: a pure parser with no video to name — the only thing it could put in a line is
|
|
# the upstream string itself. None is stored, and absent is how the UI renders it.
|
|
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:
|
|
# Normalize the title at first insert too (RSS/backfill stubs), so a brand-new video
|
|
# never flashes its raw title before enrichment refreshes it.
|
|
raw = row.get("title")
|
|
if raw:
|
|
row = {**row, "original_title": raw, "title": normalize_title(raw)}
|
|
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 = datetime.now(timezone.utc)
|
|
db.add(channel)
|
|
db.commit()
|
|
return inserted
|
|
|
|
|
|
# --- 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_at is NOT NULL (migration 0068); fall back to now() for the rare item that carries no
|
|
# date at all, so the bulk stub insert never writes a NULL. The enrich pass replaces it if the
|
|
# video's own snippet later yields a real publish date.
|
|
published = parse_dt(
|
|
content.get("videoPublishedAt") or snippet.get("publishedAt")
|
|
) or datetime.now(timezone.utc)
|
|
return {
|
|
"id": video_id,
|
|
"channel_id": channel_id,
|
|
"title": snippet.get("title"),
|
|
"published_at": published,
|
|
"thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
|
|
}
|
|
|
|
|
|
def _uploads_page(yt: YouTubeClient, channel_id: str, uploads: str, page_token: str | None) -> dict:
|
|
"""One page of the channel's uploads. YouTube answers the uploads list of a channel with no
|
|
uploads at all as not found: an empty back-catalogue, so an empty last page — not a failure
|
|
that left `recent_synced_at` unstamped and charged the channel again on every run.
|
|
Any other failure raises."""
|
|
try:
|
|
return yt.get_playlist_items(uploads, page_token)
|
|
except YouTubeError as exc:
|
|
if not is_playlist_gone(exc):
|
|
raise
|
|
log.info("backfill channel=%s has no uploads playlist", channel_id)
|
|
return {}
|
|
|
|
|
|
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 = datetime.now(timezone.utc)
|
|
db.add(channel)
|
|
db.commit()
|
|
return 0
|
|
|
|
cutoff = datetime.now(timezone.utc) - 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 = _uploads_page(yt, channel.id, 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 = datetime.now(timezone.utc)
|
|
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).
|
|
|
|
Commits the advanced cursor after every page (not just at the end) so progress
|
|
is durable: a restart resumes from the last stored page instead of re-fetching —
|
|
and re-charging quota for — pages already ingested. Re-checks the shared budget
|
|
before each page and yields once it hits the backfill reserve, so a single large
|
|
channel can't drain the day's quota in one run even with a high max_pages; the
|
|
paced scheduler finishes the tail on later ticks (deep_requested stays set)."""
|
|
if channel.backfill_done or not channel.uploads_playlist_id:
|
|
return 0
|
|
page_token = channel.backfill_cursor
|
|
total = 0
|
|
for _ in range(max_pages):
|
|
if quota.backfill_reserve_reached():
|
|
break
|
|
data = _uploads_page(yt, channel.id, 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
|
|
channel.backfill_done = not page_token
|
|
db.add(channel)
|
|
db.commit()
|
|
if not page_token:
|
|
break
|
|
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", {})
|
|
|
|
raw_title = snippet.get("title")
|
|
if raw_title:
|
|
# Store the raw title and a normalized one for display (feed/search/downloads).
|
|
video.original_title = raw_title
|
|
video.title = normalize_title(raw_title)
|
|
video.description = snippet.get("description")
|
|
# published_at is NOT NULL (0068). Prefer the enrich snippet's real date whenever it has one —
|
|
# this also corrects a now() placeholder written at stub time (e.g. a scrape-search stub, which
|
|
# carries no publish date). Only fall back to now() when the row still has nothing at all; never
|
|
# clobber an existing real date with now() (a re-enrich whose snippet lacks a date).
|
|
snippet_published = parse_dt(snippet.get("publishedAt"))
|
|
if snippet_published:
|
|
video.published_at = snippet_published
|
|
elif not video.published_at:
|
|
video.published_at = datetime.now(timezone.utc)
|
|
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
|
|
# Creator keyword tags (free with the snippet we already fetch) → search document.
|
|
tags = snippet.get("tags")
|
|
video.keywords = " ".join(tags) if tags 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 _apply_video_batch(
|
|
db: Session,
|
|
yt: YouTubeClient,
|
|
videos: list[Video],
|
|
on_missing: Callable[[Video], None],
|
|
) -> int:
|
|
"""Fetch videos.list details for the given rows, apply them (stamping enriched_at), and
|
|
return how many were applied. Rows YouTube omits (deleted/private) are handled by the
|
|
caller-supplied `on_missing` so each caller decides how to retire them."""
|
|
if not videos:
|
|
return 0
|
|
items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
|
|
now = datetime.now(timezone.utc)
|
|
applied = 0
|
|
for video in videos:
|
|
item = items.get(video.id)
|
|
if item is None:
|
|
on_missing(video)
|
|
continue
|
|
apply_video_details(video, item)
|
|
video.enriched_at = now
|
|
applied += 1
|
|
db.commit()
|
|
return applied
|
|
|
|
|
|
def _mark_enriched(video: Video) -> None:
|
|
# Unavailable (deleted/private): stamp so we don't keep retrying.
|
|
video.enriched_at = datetime.now(timezone.utc)
|
|
|
|
|
|
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()
|
|
)
|
|
return _apply_video_batch(db, yt, list(videos), _mark_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_OR_UPCOMING),
|
|
and_(
|
|
Video.live_status == "was_live",
|
|
Video.duration_seconds.is_(None),
|
|
),
|
|
)
|
|
)
|
|
.limit(limit)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
def _retire(video: Video) -> 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"
|
|
|
|
return _apply_video_batch(db, yt, list(videos), _retire)
|
|
|
|
|
|
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}
|
|
|
|
|
|
def ensure_videos(
|
|
db: Session, yt: YouTubeClient, video_ids: list[str], *, explore: bool = False
|
|
) -> None:
|
|
"""Make sure every given video id exists in the catalog, fetching + ingesting the
|
|
missing ones (and a stub channel for any unknown channel). Deleted/private videos that
|
|
the API won't return are simply left out. The catalog's write for a playlist sync that names
|
|
videos it has never seen; commits. `explore` adds the new ones as an exploration's
|
|
(`via_explore`): visible to who explores or follows their channel, and reclaimed by the cleanup
|
|
once nobody keeps them."""
|
|
if not video_ids:
|
|
return
|
|
have = set(db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all())
|
|
missing = [v for v in dict.fromkeys(video_ids) if v not in have]
|
|
if not missing:
|
|
return
|
|
items = yt.get_videos(missing)
|
|
chan_ids = {
|
|
it.get("snippet", {}).get("channelId")
|
|
for it in items
|
|
if it.get("snippet", {}).get("channelId")
|
|
}
|
|
existing_ch = set(
|
|
db.execute(select(Channel.id).where(Channel.id.in_(chan_ids))).scalars().all()
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
for it in items:
|
|
sn = it.get("snippet", {})
|
|
cid = sn.get("channelId")
|
|
if cid and cid not in existing_ch:
|
|
db.add(Channel(id=cid, title=sn.get("channelTitle")))
|
|
existing_ch.add(cid)
|
|
db.flush()
|
|
seen: set[str] = set()
|
|
for it in items:
|
|
vid = it.get("id")
|
|
sn = it.get("snippet", {})
|
|
cid = sn.get("channelId")
|
|
if not vid or not cid or vid in seen:
|
|
continue
|
|
seen.add(vid)
|
|
# published_at is NOT NULL (0068); apply_video_details re-fills a missing date, but seed a now()
|
|
# fallback here too so the constructor never carries an explicit NULL.
|
|
v = Video(
|
|
id=vid,
|
|
channel_id=cid,
|
|
published_at=parse_dt(sn.get("publishedAt")) or now,
|
|
via_explore=explore,
|
|
)
|
|
apply_video_details(v, it)
|
|
v.enriched_at = now
|
|
db.add(v)
|
|
db.commit()
|