f255728f75
Track who burned how much YouTube API quota. A QuotaEvent audit log (migration 0009) records every spend with the triggering user (NULL = background/system) and an action label, set via a request/job-scoped contextvar (quota.attribute) so no call signatures change. User-initiated work (sync subscriptions, unsubscribe, opt-in recent backfill, manual enrich) attributes to the user; scheduler work to System, split by action. - backend: QuotaEvent model + migration 0009; quota.attribute() contextvar; record_usage logs events; entry points wrapped (routes/sync, routes/channels, scheduler); GET /api/quota/my-usage + GET /api/quota/admin - frontend: admin-only Stats page (header nav, page=stats) with daily bars + per-user breakdown by action and range picker; 'Your API usage' in Settings -> Sync for every user Verified: attribution + endpoints compute correctly; events are per-user vs System.
126 lines
3.3 KiB
Python
126 lines
3.3 KiB
Python
"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
|
|
recent-first / deep backfill. Runs inside the API process (single worker)."""
|
|
import logging
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
from app import quota
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.state import is_sync_paused
|
|
from app.sync.autotag import run_autotag_all
|
|
from app.sync.runner import (
|
|
run_deep_backfill,
|
|
run_enrich,
|
|
run_recent_backfill,
|
|
run_rss_poll,
|
|
run_shorts,
|
|
run_subscription_resync,
|
|
)
|
|
|
|
logger = logging.getLogger("subfeed.scheduler")
|
|
|
|
_scheduler: BackgroundScheduler | None = None
|
|
|
|
|
|
def _job(name: str, fn) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
if is_sync_paused(db):
|
|
logger.info("job %s skipped (sync paused)", name)
|
|
return
|
|
result = fn(db)
|
|
logger.info("job %s -> %s", name, result)
|
|
except Exception:
|
|
db.rollback()
|
|
logger.exception("job %s failed", name)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _rss_job() -> None:
|
|
_job("rss_poll", lambda db: run_rss_poll(db))
|
|
|
|
|
|
def _enrich_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, "enrich"):
|
|
return run_enrich(db)
|
|
|
|
_job("enrich", work)
|
|
|
|
|
|
def _backfill_job() -> None:
|
|
# Recent-first for not-yet-synced channels, then deep backfill for the rest. All
|
|
# background spend is attributed to the system (no actor), split by action.
|
|
def work(db):
|
|
with quota.attribute(None, "backfill_recent"):
|
|
recent = run_recent_backfill(db, max_channels=25)
|
|
with quota.attribute(None, "backfill_deep"):
|
|
deep = run_deep_backfill(db, max_channels=10)
|
|
return {"recent": recent, "deep": deep}
|
|
|
|
_job("backfill", work)
|
|
|
|
|
|
def _autotag_job() -> None:
|
|
_job("autotag", lambda db: run_autotag_all(db, only_missing=True))
|
|
|
|
|
|
def _shorts_job() -> None:
|
|
_job("shorts", run_shorts)
|
|
|
|
|
|
def _subscriptions_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, "subscription_resync"):
|
|
return run_subscription_resync(db)
|
|
|
|
_job("subscriptions", work)
|
|
|
|
|
|
def start_scheduler() -> None:
|
|
global _scheduler
|
|
if not settings.scheduler_enabled or _scheduler is not None:
|
|
return
|
|
scheduler = BackgroundScheduler(timezone="UTC")
|
|
scheduler.add_job(
|
|
_rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll"
|
|
)
|
|
scheduler.add_job(
|
|
_enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich"
|
|
)
|
|
scheduler.add_job(
|
|
_backfill_job,
|
|
"interval",
|
|
minutes=settings.backfill_interval_minutes,
|
|
id="backfill",
|
|
)
|
|
scheduler.add_job(
|
|
_autotag_job,
|
|
"interval",
|
|
minutes=settings.autotag_interval_minutes,
|
|
id="autotag",
|
|
)
|
|
scheduler.add_job(
|
|
_shorts_job,
|
|
"interval",
|
|
minutes=settings.shorts_probe_interval_minutes,
|
|
id="shorts",
|
|
)
|
|
scheduler.add_job(
|
|
_subscriptions_job,
|
|
"interval",
|
|
minutes=settings.subscriptions_resync_minutes,
|
|
id="subscriptions",
|
|
)
|
|
scheduler.start()
|
|
_scheduler = scheduler
|
|
logger.info("scheduler started")
|
|
|
|
|
|
def shutdown_scheduler() -> None:
|
|
global _scheduler
|
|
if _scheduler is not None:
|
|
_scheduler.shutdown(wait=False)
|
|
_scheduler = None
|