f6c5488566
- uvicorn --log-config (log_config.json): timestamped formatters for uvicorn access/error and the app's own "subfeed" loggers (ms precision) - Log key activity: startup/shutdown, login/denied, token refresh, YouTube API errors, subscription imports, sync pause/resume - Background sync jobs no longer swallow errors silently — failures in RSS poll, backfill, enrichment, autotag and resync are logged with tracebacks
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
"""Import the authenticated user's YouTube subscriptions and channel metadata."""
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import Channel, Subscription, User
|
|
from app.youtube.client import YouTubeClient, best_thumbnail
|
|
|
|
log = logging.getLogger("subfeed.sync")
|
|
|
|
|
|
def _to_int(value) -> int | None:
|
|
try:
|
|
return int(value) if value is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def apply_channel_details(db: Session, items: list[dict]) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
for item in items:
|
|
channel = db.get(Channel, item["id"])
|
|
if channel is None:
|
|
continue
|
|
snippet = item.get("snippet", {})
|
|
content = item.get("contentDetails", {})
|
|
stats = item.get("statistics", {})
|
|
topics = item.get("topicDetails", {})
|
|
|
|
channel.title = snippet.get("title") or channel.title
|
|
channel.handle = snippet.get("customUrl")
|
|
channel.description = snippet.get("description")
|
|
channel.thumbnail_url = (
|
|
best_thumbnail(snippet.get("thumbnails")) or channel.thumbnail_url
|
|
)
|
|
channel.default_language = snippet.get("defaultLanguage")
|
|
channel.country = snippet.get("country")
|
|
channel.uploads_playlist_id = content.get("relatedPlaylists", {}).get("uploads")
|
|
channel.subscriber_count = _to_int(stats.get("subscriberCount"))
|
|
channel.video_count = _to_int(stats.get("videoCount"))
|
|
channel.topic_categories = topics.get("topicCategories")
|
|
channel.details_synced_at = now
|
|
|
|
|
|
def import_subscriptions(db: Session, user: User) -> dict:
|
|
"""Pull all subscriptions for `user`, upsert channels + subscription links, prune
|
|
channels the user has unsubscribed from on YouTube, and fetch channel details for
|
|
any channel we don't have details for yet."""
|
|
fetched: dict[str, dict] = {}
|
|
new_channels = 0
|
|
|
|
with YouTubeClient(db, user) as yt:
|
|
for sub in yt.iter_subscriptions():
|
|
cid = sub.get("channel_id")
|
|
if cid:
|
|
fetched[cid] = sub
|
|
|
|
for cid, sub in fetched.items():
|
|
channel = db.get(Channel, cid)
|
|
if channel is None:
|
|
channel = Channel(
|
|
id=cid,
|
|
title=sub.get("title"),
|
|
description=sub.get("description"),
|
|
thumbnail_url=sub.get("thumbnail_url"),
|
|
)
|
|
db.add(channel)
|
|
new_channels += 1
|
|
sub_row = db.execute(
|
|
select(Subscription).where(
|
|
Subscription.user_id == user.id, Subscription.channel_id == cid
|
|
)
|
|
).scalar_one_or_none()
|
|
if sub_row is None:
|
|
sub_row = Subscription(user_id=user.id, channel_id=cid)
|
|
db.add(sub_row)
|
|
sub_row.yt_subscription_id = sub.get("yt_subscription_id")
|
|
db.flush()
|
|
|
|
# Prune subscriptions removed on YouTube.
|
|
removed = 0
|
|
existing = (
|
|
db.execute(select(Subscription).where(Subscription.user_id == user.id))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for sub_row in existing:
|
|
if sub_row.channel_id not in fetched:
|
|
db.delete(sub_row)
|
|
removed += 1
|
|
db.commit()
|
|
|
|
# Fetch details for channels that don't have them yet.
|
|
need_details = (
|
|
db.execute(
|
|
select(Channel).where(
|
|
Channel.id.in_(fetched.keys()),
|
|
Channel.details_synced_at.is_(None),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
detailed = 0
|
|
if need_details:
|
|
items = yt.get_channels([c.id for c in need_details])
|
|
apply_channel_details(db, items)
|
|
detailed = len(items)
|
|
db.commit()
|
|
|
|
result = {
|
|
"subscriptions": len(fetched),
|
|
"channels_new": new_channels,
|
|
"channels_detailed": detailed,
|
|
"removed_stale": removed,
|
|
}
|
|
log.info("Subscription import (user %s): %s", user.id, result)
|
|
return result
|