feat(channels): channel-explore backend — about metadata, ephemeral provenance, cleanup

Adds the backend for a dedicated channel page + ephemeral browsing of un-subscribed
channels:
- migration 0033: channels.{total_view_count,published_at,banner_url,external_links,
  from_explore}, videos.via_explore, explored_channels (per-user, grace-clocked).
- enrich channels with part=brandingSettings (banner/links) + statistics.viewCount +
  snippet.publishedAt — no extra quota.
- GET /api/channels/{id}: About detail + this user's relationship; lazy-enriches the new
  About fields (published_at sentinel).
- POST /api/channels/{id}/explore (require_human, quota-reserve guarded): records the
  exploration, flags the channel ephemeral unless followed, ingests one page of uploads
  (via_explore) + enriches; returns next_page_token for load-more.
- feed: per-explorer gate — a via_explore video is visible only to users who explored or
  subscribe to its channel, so exploration never leaks into everyone's catalog.
- subscribe = keep: clears from_explore + via_explore on the channel's videos.
- scheduler explore_cleanup job + explore_grace_days config: hard-delete explored-but-
  unkept channels and their untouched ephemeral videos after a grace period.
This commit is contained in:
npeter83 2026-06-30 02:52:44 +02:00
parent bb7b9972fd
commit bc4c362423
11 changed files with 481 additions and 4 deletions

View file

@ -18,6 +18,29 @@ def _to_int(value) -> int | None:
return None
def _parse_dt(value: str | None) -> datetime | None:
if not value:
return None
try:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
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": l.get("title"), "url": l.get("url")}
for l in links
if isinstance(l, dict) and l.get("url")
]
return out or None
def apply_channel_details(db: Session, items: list[dict]) -> None:
now = datetime.now(timezone.utc)
for item in items:
@ -28,6 +51,7 @@ def apply_channel_details(db: Session, items: list[dict]) -> None:
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")
@ -40,6 +64,10 @@ def apply_channel_details(db: Session, items: list[dict]) -> None:
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.details_synced_at = now