peter
663f30feeb
Channel metadata (banner, avatar, description, counts) was fetched only the first time a channel was seen: the subscription import asked only for channels with no details, and the channel page only for ones missing a join date. 394 of 397 subscribed channels had July metadata. Both now also refresh details older than DETAILS_MAX_AGE (7 days), through one shared staleness rule; channels.list costs 1 unit per 50 channels.
196 lines
7.9 KiB
Python
196 lines
7.9 KiB
Python
"""Import the authenticated user's YouTube subscriptions and channel metadata."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs
|
|
from app.models import Channel, Subscription, User
|
|
from app.utils import normalize_description
|
|
from app.youtube.client import YouTubeClient, best_thumbnail
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
def _to_int(value) -> int | None:
|
|
try:
|
|
return int(value) if value is not None else None
|
|
# silent-ok: an upstream count we cannot read is stored as no count, which the card renders as
|
|
# absent rather than as zero. Per-channel over a whole subscription list, so it must stay quiet.
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _parse_dt(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
# silent-ok: as `_to_int` above — an unreadable upstream date becomes NULL, and every caller
|
|
# already treats a missing date as unknown.
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _external_links(branding: dict) -> list | None:
|
|
"""Channel "About" links. The API exposes a few in brandingSettings.channel.profileColor/
|
|
/... varies; the durable spot is brandingSettings.channel.links is gone post-2023, so we
|
|
read what's present (settings/links) defensively and keep [{title, url}] pairs."""
|
|
channel = branding.get("channel", {})
|
|
links = channel.get("links") or []
|
|
out = [
|
|
{"title": item.get("title"), "url": item.get("url")}
|
|
for item in links
|
|
if isinstance(item, dict) and item.get("url")
|
|
]
|
|
return out or None
|
|
|
|
|
|
# How old a channel's metadata may get before it is fetched again. Banners, avatars, descriptions and
|
|
# counts change on YouTube, and an old banner URL stops resolving at all — without a refresh a channel
|
|
# kept whatever it had the day it was first seen. A week keeps counts roughly current at almost no
|
|
# cost: `channels.list` is 1 quota unit per 50 channels, and only the stale ones are asked for.
|
|
DETAILS_MAX_AGE = timedelta(days=7)
|
|
|
|
|
|
def details_stale(channel: Channel, now: datetime) -> bool:
|
|
"""Whether `channel`'s metadata is missing or older than DETAILS_MAX_AGE (see `details_stale_clause`)."""
|
|
synced = channel.details_synced_at
|
|
return synced is None or synced < now - DETAILS_MAX_AGE
|
|
|
|
|
|
def details_stale_clause(now: datetime):
|
|
"""`details_stale` as a WHERE clause, so the batch import and a single channel agree on "stale"."""
|
|
return or_(
|
|
Channel.details_synced_at.is_(None),
|
|
Channel.details_synced_at < now - DETAILS_MAX_AGE,
|
|
)
|
|
|
|
|
|
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", {})
|
|
branding = item.get("brandingSettings", {})
|
|
|
|
channel.title = snippet.get("title") or channel.title
|
|
channel.handle = snippet.get("customUrl")
|
|
channel.description = normalize_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.total_view_count = _to_int(stats.get("viewCount"))
|
|
channel.published_at = _parse_dt(snippet.get("publishedAt"))
|
|
channel.banner_url = branding.get("image", {}).get("bannerExternalUrl")
|
|
channel.external_links = _external_links(branding)
|
|
channel.topic_categories = topics.get("topicCategories")
|
|
channel.keywords = (branding.get("channel") or {}).get("keywords")
|
|
channel.details_synced_at = now
|
|
|
|
|
|
def should_prune(fetched_count: int, existing_count: int) -> bool:
|
|
"""Whether an import may delete the local subscriptions missing from this fetch.
|
|
|
|
NEVER on an empty fetch that contradicts local state: the API can answer 200 with no items (a
|
|
transient blip, a scope/token edge), and pruning against that wipes every subscription the user
|
|
has — taking each row's `deep_requested` flag with it — in a single run. Someone who genuinely
|
|
unsubscribed from everything simply keeps their rows until a sync that returns something.
|
|
|
|
Pure so it can be tested without a database; `import_subscriptions` is the only caller."""
|
|
return fetched_count > 0 or existing_count == 0
|
|
|
|
|
|
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 whose details are missing or stale (`DETAILS_MAX_AGE`)."""
|
|
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=normalize_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 — but only when this fetch is trustworthy
|
|
# (see `should_prune`).
|
|
removed = 0
|
|
existing = (
|
|
db.execute(select(Subscription).where(Subscription.user_id == user.id)).scalars().all()
|
|
)
|
|
prune_skipped = not should_prune(len(fetched), len(existing))
|
|
if prune_skipped:
|
|
log.warning(
|
|
"Subscription import (user %s): YouTube returned 0 subscriptions while %d exist "
|
|
"locally — skipping the prune (treating it as a transient empty response)",
|
|
user.id,
|
|
len(existing),
|
|
)
|
|
else:
|
|
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, or had them too long ago.
|
|
need_details = (
|
|
db.execute(
|
|
select(Channel).where(
|
|
Channel.id.in_(fetched.keys()),
|
|
details_stale_clause(datetime.now(timezone.utc)),
|
|
)
|
|
)
|
|
.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,
|
|
"prune_skipped": prune_skipped,
|
|
}
|
|
log.info("Subscription import (user %s): %s", user.id, result)
|
|
return result
|