peter
e3ee1baa94
The user-triggered "get full history" (backfill_channel_now) capped at max_pages=60 = 3000 videos. A channel just over that (BBC Earth: 3403 uploads = 69 pages) stopped one page short, so its tail dribbled in over the scheduler's paced 10-minute ticks (250 videos/tick) and the button spinner looked stuck for ~18 minutes with no visible progress. - Raise the immediate cap to max_pages=400 (20 000 videos) so any realistic channel completes in the single background run. - Make a high cap safe: backfill_channel_deep now re-checks the shared quota before each page and yields at the backfill reserve (previously quota was checked once by the caller before the whole run), and commits the advanced cursor after every page so a restart resumes instead of re-fetching — and re-charging quota for — pages already stored. - Channel page: while loading, the button shows the live stored/total count and a tooltip explaining the tail finishes in the background, so a very large channel's paced remainder reads as progress, not a hang. Adds deep-backfill paging tests (per-page cursor commit + resume, and the per-page quota yield).
338 lines
15 KiB
Python
338 lines
15 KiB
Python
"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints.
|
|
|
|
Channel/video data is shared across users, so background work acts through a "service
|
|
user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for
|
|
public reads.
|
|
"""
|
|
import logging
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import progress, quota, sysconfig
|
|
|
|
log = logging.getLogger("siftlode.sync")
|
|
from app.db import SessionLocal
|
|
from app.models import Channel, OAuthToken, Subscription, User
|
|
from app.sync.subscriptions import import_subscriptions
|
|
from app.sync.videos import (
|
|
apply_rss_feed,
|
|
backfill_channel_deep,
|
|
backfill_channel_recent,
|
|
enrich_pending,
|
|
reconcile_full_history,
|
|
refresh_live,
|
|
run_shorts_classification,
|
|
)
|
|
from app.youtube.client import YouTubeClient
|
|
from app.youtube.rss import RssError, fetch_channel_feed
|
|
|
|
# RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded.
|
|
RSS_POLL_WORKERS = 16
|
|
|
|
|
|
def token_users(db: Session) -> list[User]:
|
|
"""Every user with a stored YouTube refresh token (the read-scope service accounts), by id."""
|
|
return list(
|
|
db.execute(
|
|
select(User)
|
|
.join(OAuthToken)
|
|
.where(OAuthToken.refresh_token_enc.is_not(None))
|
|
.order_by(User.id)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def get_service_user(db: Session) -> User | None:
|
|
users = token_users(db)
|
|
return users[0] if users else None
|
|
|
|
|
|
def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str, int]:
|
|
"""Poll every channel's free RSS feed. Returns `{"new": inserted, "failed": channels_that_did
|
|
_not_complete}` — the failure count is part of the RESULT, not just a log line, so the scheduler
|
|
card and the audit summary can tell "no new uploads" apart from "nothing worked" (an egress
|
|
outage, or a database that can't accept writes, otherwise renders identically to a quiet hour).
|
|
|
|
`failed` counts a channel whose feed could not be READ *or* whose rows could not be APPLIED —
|
|
both leave the channel un-polled, and counting only the first would move the blind spot rather
|
|
than remove it.
|
|
|
|
Raises RuntimeError when NOTHING could be polled (`failed == total`, total > 0): a run that
|
|
achieved literally nothing is a failed run, and only an exception turns the scheduler card red
|
|
— a returned count leaves it green with the outage buried in the summary line. The message
|
|
carries the FIRST underlying error, because on a small instance "all channels failed" is also
|
|
what a single dead feed id looks like."""
|
|
if channels is None:
|
|
channels = db.execute(select(Channel)).scalars().all()
|
|
total = len(channels)
|
|
new = 0
|
|
done = 0
|
|
failed = 0
|
|
# Kept for the total-outage message: "failed for all N channels" alone can't tell an egress
|
|
# outage from one permanently dead feed id — which is the whole difference on an instance with
|
|
# a single subscription, where BOTH render as 100% failed.
|
|
first_error: str | None = None
|
|
# Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
|
|
# each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
|
|
# plain str) to the workers; the ORM instance is touched only here on the main thread.
|
|
futures = {}
|
|
with ThreadPoolExecutor(max_workers=RSS_POLL_WORKERS) as pool:
|
|
for channel in channels:
|
|
futures[pool.submit(fetch_channel_feed, channel.id)] = channel
|
|
for fut in as_completed(futures):
|
|
channel = futures[fut]
|
|
try:
|
|
entries = fut.result()
|
|
# A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where it
|
|
# was (stamping it made a permanently-broken feed look freshly polled). The shared
|
|
# counter/progress tail lives after the statement, so the two outcomes can stay two
|
|
# clauses: an RssError is EXPECTED (404, timeout) and gets one warning line; anything
|
|
# else is a bug on our side and earns a traceback.
|
|
except RssError as exc:
|
|
log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id}: {exc}"
|
|
except Exception as exc:
|
|
log.exception("RSS fetch failed for channel %s", channel.id)
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id}: {exc}"
|
|
else:
|
|
try:
|
|
new += apply_rss_feed(db, channel, entries)
|
|
except Exception as exc:
|
|
# A write failure (read-only DB, failover) leaves the channel just as un-polled
|
|
# as an unreachable feed — count it, or a wholly broken run reports "failed=0".
|
|
db.rollback()
|
|
failed += 1
|
|
first_error = first_error or f"{channel.id} (apply): {exc}"
|
|
log.exception("RSS apply failed for channel %s", channel.id)
|
|
# One place that advances the counter, whatever happened above.
|
|
done += 1
|
|
progress.report(done, total, "rss_poll")
|
|
if failed:
|
|
log.warning("RSS poll: %d/%d channels could not be polled", failed, total)
|
|
if total and failed == total:
|
|
# Nothing was polled at all — an egress outage (the tinyproxy-after-reboot incident), a
|
|
# dead DNS, a read-only database. Returning normally would record the run as `ok` with the
|
|
# failure hidden inside the result string, i.e. a green pill for a total outage.
|
|
# The first error TRAVELS WITH the message on purpose: on an instance with one or two
|
|
# subscriptions "all channels failed" is also what ONE dead feed id looks like, and a red
|
|
# card the user can't act on is its own kind of dishonest. `channel: HTTP 404` says
|
|
# "fix this subscription"; `Connection refused` says "the egress is down".
|
|
raise RuntimeError(f"RSS poll failed for all {total} channels — first: {first_error}")
|
|
return {"new": new, "failed": failed}
|
|
|
|
|
|
def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return 0
|
|
total = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for _ in range(max_batches):
|
|
if quota.remaining_today(db) <= floor:
|
|
break
|
|
n = enrich_pending(db, yt)
|
|
total += n
|
|
progress.report(total, None, "enrich") # total unknown (drains until empty)
|
|
if n == 0:
|
|
break
|
|
# Revisit transient live/upcoming videos (enrich_pending is one-shot) so an ended
|
|
# stream gains its final duration + was_live status instead of staying "live" forever.
|
|
if quota.remaining_today(db) > floor:
|
|
total += refresh_live(db, yt)
|
|
return total
|
|
|
|
|
|
def run_shorts(db: Session) -> dict:
|
|
return run_shorts_classification(db)
|
|
|
|
|
|
def run_subscription_resync(db: Session) -> dict:
|
|
"""Re-import every user's subscriptions so unsubscribes and new subscriptions on
|
|
YouTube are reflected automatically."""
|
|
users = token_users(db)
|
|
for i, user in enumerate(users, 1):
|
|
try:
|
|
import_subscriptions(db, user)
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Subscription resync failed for user %s", user.id)
|
|
progress.report(i, len(users), "subscriptions")
|
|
return {"users": len(users)}
|
|
|
|
|
|
def run_recent_backfill(
|
|
db: Session, channels: list[Channel] | None = None, max_channels: int | None = None
|
|
) -> dict:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
|
|
if channels is None:
|
|
channels = (
|
|
db.execute(select(Channel).where(Channel.recent_synced_at.is_(None)))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
if max_channels:
|
|
channels = channels[:max_channels]
|
|
processed = 0
|
|
videos_added = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for channel in channels:
|
|
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
|
break
|
|
try:
|
|
videos_added += backfill_channel_recent(db, yt, channel)
|
|
processed += 1
|
|
progress.report(processed, len(channels), "backfill_recent")
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Recent backfill failed for channel %s", channel.id)
|
|
reconciled = reconcile_full_history(db)
|
|
return {
|
|
"channels_processed": processed,
|
|
"videos_added": videos_added,
|
|
"reconciled_full": reconciled,
|
|
}
|
|
|
|
|
|
def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
|
|
# Demand-driven: only deep-backfill channels at least one user has opted into
|
|
# (Subscription.deep_requested). Recent backfill stays unconditional (cheap).
|
|
deep_requested = (
|
|
select(Subscription.id)
|
|
.where(
|
|
Subscription.channel_id == Channel.id,
|
|
Subscription.deep_requested.is_(True),
|
|
)
|
|
.exists()
|
|
)
|
|
channels = (
|
|
db.execute(
|
|
select(Channel).where(
|
|
Channel.backfill_done.is_(False),
|
|
Channel.recent_synced_at.is_not(None),
|
|
deep_requested,
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)[:max_channels]
|
|
processed = 0
|
|
videos_added = 0
|
|
with YouTubeClient(db, user) as yt:
|
|
for channel in channels:
|
|
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
|
break
|
|
# Claim the channel (or skip it if an interactive backfill already owns it) so the two
|
|
# never advance the same cursor concurrently — symmetric with backfill_channel_now.
|
|
with _backfill_lock:
|
|
if channel.id in _backfill_in_progress:
|
|
continue
|
|
_backfill_in_progress.add(channel.id)
|
|
try:
|
|
videos_added += backfill_channel_deep(db, yt, channel, max_pages)
|
|
processed += 1
|
|
progress.report(processed, len(channels), "backfill_deep")
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Deep backfill failed for channel %s", channel.id)
|
|
finally:
|
|
with _backfill_lock:
|
|
_backfill_in_progress.discard(channel.id)
|
|
reconciled = reconcile_full_history(db)
|
|
return {
|
|
"channels_processed": processed,
|
|
"videos_added": videos_added,
|
|
"reconciled_full": reconciled,
|
|
}
|
|
|
|
|
|
# Channels being deep-backfilled RIGHT NOW by EITHER path — the interactive backfill_channel_now (a
|
|
# FastAPI BackgroundTasks threadpool thread) OR the scheduler's run_deep_backfill (its own thread).
|
|
# BOTH register the channel they're paging (under the lock) so the other skips it: no two runs page
|
|
# the same channel's `backfill_cursor` concurrently, whichever started first. `_immediate_running`
|
|
# counts only the interactive path, capped so one user's burst can't drain the shared budget (the
|
|
# scheduler is already quota-paced). Single uvicorn process → one shared module state, lock suffices.
|
|
_backfill_in_progress: set[str] = set()
|
|
_immediate_running = 0
|
|
_backfill_lock = threading.Lock()
|
|
_MAX_CONCURRENT_IMMEDIATE = 2
|
|
|
|
|
|
def backfill_channel_now(channel_id: str, user_id: int, max_pages: int = 400) -> None:
|
|
"""User-initiated "get full history": deep-backfill ONE channel to completion right now, in a
|
|
background task with its OWN session, so the full catalogue lands in seconds instead of over the
|
|
paced scheduler's 10-minute ticks (max_pages=5/channel/tick). max_pages is high enough to finish
|
|
any realistic channel in a single run (400 pages = 20 000 videos; e.g. BBC Earth's ~3400 = 69
|
|
pages — the old 60-page cap stopped one page short of large channels, so their tail dribbled in
|
|
over the scheduler's 10-minute ticks and looked stuck). backfill_channel_deep re-checks the shared
|
|
budget before each page and yields at the reserve, so a high cap still can't drain the day's quota.
|
|
Quota-guarded, capped to _MAX_CONCURRENT_IMMEDIATE interactive runs so one user can't drain the
|
|
shared backfill budget by opting many channels in at once (the overflow, and any channel too large
|
|
to finish under the reserve, is finished by the scheduler since deep_requested stays set), and
|
|
mutually excluded per-channel with the scheduler via _backfill_in_progress. Best-effort — logs and
|
|
swallows failures."""
|
|
global _immediate_running
|
|
with _backfill_lock:
|
|
if channel_id in _backfill_in_progress or _immediate_running >= _MAX_CONCURRENT_IMMEDIATE:
|
|
return # already backfilling this channel, or too many interactive runs — leave it to the scheduler
|
|
_backfill_in_progress.add(channel_id)
|
|
_immediate_running += 1
|
|
try:
|
|
with SessionLocal() as db:
|
|
user = get_service_user(db)
|
|
if user is None:
|
|
return
|
|
channel = db.get(Channel, channel_id)
|
|
if channel is None or channel.backfill_done or not channel.uploads_playlist_id:
|
|
return
|
|
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
|
return
|
|
try:
|
|
# Attribute the spend to the requesting user under the full-backfill action, so it
|
|
# shows up correctly in the usage breakdown (not the default OTHER/system bucket).
|
|
with quota.attribute(user_id, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
|
|
with YouTubeClient(db, user) as yt:
|
|
backfill_channel_deep(db, yt, channel, max_pages)
|
|
reconcile_full_history(db)
|
|
except Exception:
|
|
db.rollback()
|
|
log.exception("Immediate backfill failed for channel %s", channel_id)
|
|
finally:
|
|
with _backfill_lock:
|
|
_backfill_in_progress.discard(channel_id)
|
|
_immediate_running -= 1
|
|
|
|
|
|
def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict:
|
|
"""Rough ETA to finish full-history backfill of `channels`. Quota-bound: total
|
|
estimated API units / the daily budget left for backfill. Deliberately approximate —
|
|
a channel's cost ~= playlistItems pages (1 unit / 50 videos) + enrichment (likewise)."""
|
|
pending = [c for c in channels if not c.backfill_done]
|
|
|
|
def units(c: Channel) -> int:
|
|
videos = c.video_count or 0
|
|
pages = max(1, (videos + 49) // 50)
|
|
return 2 * pages
|
|
|
|
total_units = sum(units(c) for c in pending)
|
|
daily_budget = max(
|
|
1, sysconfig.get_int(db, "quota_daily_budget") - sysconfig.get_int(db, "backfill_quota_reserve")
|
|
)
|
|
eta_seconds = int(total_units / daily_budget * 86_400) if pending else 0
|
|
return {
|
|
"channels_pending": len(pending),
|
|
"videos_pending_est": sum(c.video_count or 0 for c in pending),
|
|
"units_est": total_units,
|
|
"daily_budget": daily_budget,
|
|
"eta_seconds": eta_seconds,
|
|
}
|