peter
4aaaaa7860
The code cited its plan instead of saying why: `E28 review J4`, `(R6 S3b)`, `S1 justified…`,
`deep-review round 5`. That plan lives in the private ops repo, so for every reader of this one the
citation was noise, and a sentence whose subject was the plan step ("round 2 showed…") hid its
lesson behind an id. Each one is gone and the "why" kept: a bare citation is deleted, a sentence
carrying a lesson is reworded around it ("a later attempt showed…").
Measured (`scripts/audit_metrics.py`, product code): 652 lines in 270 files carried an epic code by
the definition the next commit holds (507 / 241 by the old, two-digit-only one) -> 0. Across all
code, tests, scripts and configs: ~1100 epic-code lines plus ~320 sprint/round ids.
Behaviour unchanged, checked mechanically: every changed .py file has the same AST once docstrings
are blanked; css/sh/toml/yml/json files are identical once comments are stripped; the only
non-comment changes in TS/JS are test titles, one description string in a contract table, and five
ESLint messages, whose text is all the gates snapshot diff holds. Not touched: Markdown (docs and
the changelog keep their history), Alembic migrations, TV episode notation (`S2 · E5`), DESIGN.md
ids (`C1`, `B4`, `S3h`), tracker ids (`SEC-16`, `DX-17`).
One stale claim fixed on the way: test_hls_reaper's docstring said the player checkpoints "every
10 s whether or not it is playing", which plex/stream.py calls wrong; it now says what the code does.
772 lines
32 KiB
Python
772 lines
32 KiB
Python
"""Channel manager: the user's subscriptions with their personal overrides (priority,
|
||
hidden, tags) and per-channel sync state. Local reads/writes plus YouTube-side writes
|
||
(subscribe / unsubscribe) that require the optional write scope."""
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, Depends
|
||
from pydantic import BaseModel
|
||
from sqlalchemy import Select, func, select, update
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.collation import collated, sorted_words
|
||
from app.mailtext import lang_for_user
|
||
from app import logs, quota
|
||
from app.auth import admin_user, current_user, require_write_scope
|
||
from app.db import get_db
|
||
from app.models import (
|
||
LIVE_OR_UPCOMING,
|
||
BlockedChannel,
|
||
Channel,
|
||
ChannelTag,
|
||
ExploredChannel,
|
||
Playlist,
|
||
PlaylistItem,
|
||
Subscription,
|
||
Tag,
|
||
User,
|
||
Video,
|
||
)
|
||
from app.routes._common import get_or_404, not_found, owned_or_404, youtube_refusal
|
||
from app.sync.explore import explore_ingest_page
|
||
from app.sync.runner import backfill_channel_now, run_recent_backfill
|
||
from app.sync.subscriptions import apply_channel_details, details_stale
|
||
from app.utils import iso
|
||
from app.youtube.client import YouTubeClient, YouTubeError
|
||
from app.refusal import refused
|
||
|
||
router = APIRouter(prefix="/api/channels", tags=["channels"])
|
||
|
||
log = logs.get_logger(__name__)
|
||
|
||
|
||
def channel_aggregates(channel_ids: list[str]) -> Select:
|
||
"""Per-channel stored count, last upload and total duration, in one grouped pass. Every column it
|
||
reads is in `ix_videos_channel_pub_cover` (migration 0095), so Postgres answers it from the index
|
||
alone; a test holds that (test_channel_list_index_only)."""
|
||
return (
|
||
select(
|
||
Video.channel_id,
|
||
func.count(Video.id),
|
||
func.max(Video.published_at),
|
||
func.coalesce(func.sum(Video.duration_seconds), 0),
|
||
)
|
||
.where(Video.channel_id.in_(channel_ids))
|
||
.group_by(Video.channel_id)
|
||
)
|
||
|
||
|
||
@router.get("")
|
||
def list_channels(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
|
||
rows = db.execute(
|
||
select(Subscription, Channel)
|
||
.join(Channel, Channel.id == Subscription.channel_id)
|
||
.where(Subscription.user_id == user.id)
|
||
.order_by(Subscription.priority.desc(), collated(Channel.title, lang_for_user(user)))
|
||
).all()
|
||
|
||
# Per-channel aggregates over the shared catalog (channel_aggregates), computed on read rather than
|
||
# denormalised: an Index Only Scan over a ~20 MB index, where it used to read every video's heap page
|
||
# (681 MB on prod, a 2–6 s first open once something else had pushed it out of memory).
|
||
# (The Normal/Shorts/Live breakdown moved to the channel-detail page, so it is no longer
|
||
# computed here — the manager table dropped its N/S/L column.)
|
||
channel_ids = [c.id for _, c in rows]
|
||
agg: dict[str, dict] = {}
|
||
if channel_ids:
|
||
for cid, n, last_up, total_dur in db.execute(channel_aggregates(channel_ids)).all():
|
||
agg[cid] = {
|
||
"stored": n,
|
||
"last_video_at": iso(last_up),
|
||
"total_duration_seconds": int(total_dur or 0),
|
||
}
|
||
|
||
# Channels already in the *global* deep-backfill queue — i.e. at least one user (any
|
||
# user) has requested full history and it isn't done yet. Their whole back-catalogue is
|
||
# coming for everyone (video data is shared), so the UI shouldn't offer "get full
|
||
# history" to a user who simply hasn't personally opted in.
|
||
deep_queued: set[str] = set()
|
||
if channel_ids:
|
||
for (cid,) in db.execute(
|
||
select(Subscription.channel_id)
|
||
.where(
|
||
Subscription.channel_id.in_(channel_ids),
|
||
Subscription.deep_requested.is_(True),
|
||
)
|
||
.distinct()
|
||
).all():
|
||
deep_queued.add(cid)
|
||
|
||
# The user's personal tag links, grouped by channel.
|
||
tags_by_channel: dict[str, list[int]] = {}
|
||
for cid, tag_id in db.execute(
|
||
select(ChannelTag.channel_id, ChannelTag.tag_id).where(ChannelTag.user_id == user.id)
|
||
).all():
|
||
tags_by_channel.setdefault(cid, []).append(tag_id)
|
||
|
||
return [
|
||
{
|
||
**_channel_summary(ch),
|
||
"stored_videos": (agg.get(ch.id) or {}).get("stored", 0),
|
||
"last_video_at": (agg.get(ch.id) or {}).get("last_video_at"),
|
||
"total_duration_seconds": (agg.get(ch.id) or {}).get("total_duration_seconds", 0),
|
||
"priority": sub.priority,
|
||
"hidden": sub.hidden,
|
||
"deep_requested": sub.deep_requested,
|
||
"deep_in_queue": ch.id in deep_queued and not ch.backfill_done,
|
||
"tag_ids": tags_by_channel.get(ch.id, []),
|
||
"details_synced": ch.details_synced_at is not None,
|
||
"recent_synced": ch.recent_synced_at is not None,
|
||
"backfill_done": ch.backfill_done,
|
||
}
|
||
for sub, ch in rows
|
||
]
|
||
|
||
|
||
# Cap how many stub channels we enrich per discovery call, to bound first-load latency and
|
||
# quota (channels.list = 1 unit / 50). Anything beyond is enriched on subsequent loads.
|
||
DISCOVERY_ENRICH_CAP = 200
|
||
|
||
|
||
def _discovery_rows(db: Session, user: User) -> list:
|
||
"""The user's playlist channels they don't subscribe to (and that aren't their own),
|
||
each with how many playlist videos come from it and across how many playlists."""
|
||
subscribed = select(Subscription.channel_id).where(Subscription.user_id == user.id)
|
||
q = (
|
||
select(
|
||
Channel,
|
||
func.count(func.distinct(PlaylistItem.video_id)),
|
||
func.count(func.distinct(Playlist.id)),
|
||
)
|
||
.select_from(PlaylistItem)
|
||
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
|
||
.join(Video, Video.id == PlaylistItem.video_id)
|
||
.join(Channel, Channel.id == Video.channel_id)
|
||
.where(Playlist.user_id == user.id, Channel.id.not_in(subscribed))
|
||
)
|
||
if user.yt_channel_id:
|
||
q = q.where(Channel.id != user.yt_channel_id)
|
||
return db.execute(
|
||
q.group_by(Channel.id).order_by(
|
||
func.count(func.distinct(PlaylistItem.video_id)).desc(),
|
||
collated(Channel.title, lang_for_user(user)),
|
||
)
|
||
).all()
|
||
|
||
|
||
@router.get("/discovery")
|
||
def discover_channels(
|
||
user: User = Depends(current_user), db: Session = Depends(get_db)
|
||
) -> list[dict]:
|
||
"""Channels that appear in the user's playlists but that they don't subscribe to (their
|
||
own channel excluded). The base list is a local join over the shared catalog. We also
|
||
enrich each channel's metadata up front (title/thumbnail/subscriber count) so the user
|
||
can judge a channel BEFORE subscribing — but we do NOT pull its videos. Most-present
|
||
channels come first; subscribing is a separate action."""
|
||
# Resolve & cache the user's own channel id so it can be excluded — you don't subscribe
|
||
# to yourself. Lazy + cached; needs OAuth, so token-less users just skip it.
|
||
if user.yt_channel_id is None and user.token is not None:
|
||
try:
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
own = yt.get_my_channel_id()
|
||
if own:
|
||
user.yt_channel_id = own
|
||
db.commit()
|
||
except YouTubeError as exc:
|
||
# No traceback anywhere in this file's YouTube handlers: for an upstream answer
|
||
# `_upstream_failure` already logged the structured `youtube.api.error`, and the
|
||
# message names the call, its status and its reason (or, with no answer at all, the
|
||
# transport failure). The stack under it is this `except` and the client call.
|
||
log.warning(
|
||
"discovery: own-channel resolve failed: %s",
|
||
exc,
|
||
extra=logs.event("channels.discovery.own_channel_failed", reason=exc.reason),
|
||
)
|
||
|
||
rows = _discovery_rows(db, user)
|
||
|
||
# Enrich stub channels' metadata (uses the API key — no write scope needed). The Channel
|
||
# objects are identity-mapped, so apply_channel_details mutates the very rows we already hold
|
||
# (fresh subscriber counts/thumbnails carry through without a re-query). Only the title can
|
||
# change, which affects the tie-break order → re-sort in Python instead of re-hitting the DB.
|
||
need = [ch.id for ch, _, _ in rows if ch.details_synced_at is None][:DISCOVERY_ENRICH_CAP]
|
||
if need:
|
||
try:
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
apply_channel_details(db, yt.get_channels(need), need)
|
||
db.commit()
|
||
# Match _discovery_rows' ORDER BY exactly: video-count DESC, then title in the reader's
|
||
# alphabet (NULLs LAST, as Postgres orders them — coercing NULL→"" would wrongly float
|
||
# untitled channels up).
|
||
titles = [r[0].title for r in rows if r[0].title is not None]
|
||
alpha = {w: i for i, w in enumerate(sorted_words(db, titles, lang_for_user(user)))}
|
||
rows = sorted(
|
||
rows,
|
||
key=lambda r: (-int(r[1]), r[0].title is None, alpha.get(r[0].title or "", 0)),
|
||
)
|
||
except YouTubeError as exc:
|
||
log.warning(
|
||
"discovery: channel enrich failed: %s",
|
||
exc,
|
||
extra=logs.event("channels.discovery.enrich_failed", reason=exc.reason),
|
||
)
|
||
db.rollback()
|
||
|
||
return [
|
||
{
|
||
**_channel_summary(ch),
|
||
"playlist_video_count": int(vid_count),
|
||
"playlist_count": int(pl_count),
|
||
"details_synced": ch.details_synced_at is not None,
|
||
}
|
||
for ch, vid_count, pl_count in rows
|
||
]
|
||
|
||
|
||
def _channel_summary(ch: Channel) -> dict:
|
||
"""The channel fields common to the list and discovery projections."""
|
||
return {
|
||
"id": ch.id,
|
||
"title": ch.title,
|
||
"handle": ch.handle,
|
||
"thumbnail_url": ch.thumbnail_url,
|
||
"subscriber_count": ch.subscriber_count,
|
||
"video_count": ch.video_count,
|
||
"description": ch.description,
|
||
}
|
||
|
||
|
||
def _channel_detail_dict(
|
||
channel: Channel,
|
||
*,
|
||
subscribed: bool,
|
||
explored: bool,
|
||
blocked: bool,
|
||
stored: int,
|
||
count_normal: int,
|
||
count_short: int,
|
||
count_live: int,
|
||
deep_requested: bool,
|
||
priority: int,
|
||
hidden: bool,
|
||
tag_ids: list[int],
|
||
) -> dict:
|
||
return {
|
||
"id": channel.id,
|
||
"title": channel.title,
|
||
"handle": channel.handle,
|
||
"description": channel.description,
|
||
"thumbnail_url": channel.thumbnail_url,
|
||
"banner_url": channel.banner_url,
|
||
"subscriber_count": channel.subscriber_count,
|
||
"video_count": channel.video_count,
|
||
"total_view_count": channel.total_view_count,
|
||
"published_at": iso(channel.published_at),
|
||
"external_links": channel.external_links or [],
|
||
"country": channel.country,
|
||
"default_language": channel.default_language,
|
||
"topic_categories": channel.topic_categories or [],
|
||
"keywords": channel.keywords,
|
||
"subscribed": subscribed,
|
||
"explored": explored,
|
||
"blocked": blocked,
|
||
"stored_videos": stored,
|
||
# Per-type breakdown (Normal / Shorts / Live) — moved off the manager table onto the
|
||
# channel page. Same aggregation the manager list uses (count_normal = stored − shorts − live).
|
||
"count_normal": count_normal,
|
||
"count_short": count_short,
|
||
"count_live": count_live,
|
||
# Deep-backfill state, so the channel page can offer "get full history" when the catalog
|
||
# holds only the recent-backfill/RSS subset (video_count is YouTube's raw total).
|
||
"deep_requested": deep_requested,
|
||
# This user's per-subscription feed priority (0 when not subscribed) so the channel page can
|
||
# offer the same priority stepper as the manager. Editable only while subscribed.
|
||
"priority": priority,
|
||
# This user's per-subscription "hide from feed" flag (False when not subscribed), so the
|
||
# channel page can offer the same Unhide toggle the manager has. Editable only while subscribed.
|
||
"hidden": hidden,
|
||
"backfill_done": channel.backfill_done,
|
||
"from_explore": channel.from_explore,
|
||
"details_synced": channel.details_synced_at is not None,
|
||
# This user's personal tag links for the channel (same shape as the manager list), so the
|
||
# channel page can offer the same tag editor next to Subscribe.
|
||
"tag_ids": tag_ids,
|
||
}
|
||
|
||
|
||
@router.get("/{channel_id}")
|
||
def channel_detail(
|
||
channel_id: str,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""One channel's "About" detail for the channel page header/About tab, plus this user's
|
||
relationship to it (subscribed / explored) and how many of its videos are in the catalog.
|
||
Lazily fills the About metadata (view count, join date, banner, links) the first time it's
|
||
needed (a stub) and refreshes it once it is stale (`DETAILS_MAX_AGE`), which is how a channel nobody subscribes to (an explored one, outside the
|
||
subscription import) gets a banner that still resolves. The enrich uses the API key (1 unit,
|
||
cached after)."""
|
||
channel = get_or_404(db, Channel, channel_id, what="channel")
|
||
# No `published_at is None` sentinel: a terminated channel never gets one, so it re-fetched on
|
||
# every view; rows enriched before that field existed are long past DETAILS_MAX_AGE anyway.
|
||
stale = details_stale(channel, datetime.now(timezone.utc))
|
||
if stale:
|
||
try:
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
apply_channel_details(db, yt.get_channels([channel_id]), [channel_id])
|
||
db.commit()
|
||
channel = db.get(Channel, channel_id)
|
||
except YouTubeError as exc:
|
||
log.warning(
|
||
"channel detail enrich failed: %s",
|
||
exc,
|
||
extra=logs.event(
|
||
"channels.detail.enrich_failed", channel_id=channel_id, reason=exc.reason
|
||
),
|
||
)
|
||
db.rollback()
|
||
sub = db.execute(
|
||
select(Subscription).where(
|
||
Subscription.user_id == user.id, Subscription.channel_id == channel_id
|
||
)
|
||
).scalar_one_or_none()
|
||
subscribed = sub is not None
|
||
explored = (
|
||
db.execute(
|
||
select(ExploredChannel.id).where(
|
||
ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id
|
||
)
|
||
).first()
|
||
is not None
|
||
)
|
||
blocked = _is_blocked(db, user, channel_id)
|
||
# Stored total + the Normal/Shorts/Live split in one grouped pass (same filters as the manager
|
||
# list), so the channel page can show the per-type breakdown that moved off the table.
|
||
n_all, n_short, n_live = db.execute(
|
||
select(
|
||
func.count(Video.id),
|
||
func.count(Video.id).filter(Video.is_short.is_(True)),
|
||
func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)),
|
||
).where(Video.channel_id == channel_id)
|
||
).one()
|
||
stored = int(n_all or 0)
|
||
count_short = int(n_short or 0)
|
||
count_live = int(n_live or 0)
|
||
count_normal = max(0, stored - count_short - count_live)
|
||
tag_ids = list(
|
||
db.scalars(
|
||
select(ChannelTag.tag_id).where(
|
||
ChannelTag.user_id == user.id, ChannelTag.channel_id == channel_id
|
||
)
|
||
).all()
|
||
)
|
||
return _channel_detail_dict(
|
||
channel,
|
||
subscribed=subscribed,
|
||
explored=explored,
|
||
blocked=blocked,
|
||
stored=int(stored),
|
||
count_normal=count_normal,
|
||
count_short=count_short,
|
||
count_live=count_live,
|
||
deep_requested=bool(sub.deep_requested) if sub else False,
|
||
priority=int(sub.priority) if sub else 0,
|
||
hidden=bool(sub.hidden) if sub else False,
|
||
tag_ids=tag_ids,
|
||
)
|
||
|
||
|
||
@router.post("/{channel_id}/explore")
|
||
def explore_channel(
|
||
channel_id: str,
|
||
page_token: str | None = None,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""Open an un-subscribed channel for browsing. Records the exploration (a per-user,
|
||
grace-clocked row that gates this user's access to the channel's ephemeral videos and
|
||
drives the later cleanup), flags the channel ephemeral unless someone organically follows
|
||
it, then ingests one page of its uploads (newest-first) marked via_explore + enriched.
|
||
Returns the next page token so the page can "load more". Guarded against the shared
|
||
backfill quota reserve."""
|
||
channel = get_or_404(db, Channel, channel_id, what="channel")
|
||
if _is_blocked(db, user, channel_id):
|
||
raise refused(403, "channel_blocked")
|
||
if quota.backfill_reserve_reached():
|
||
raise refused(429, "quota_reserve")
|
||
|
||
# Record/refresh the exploration (DB only) before ingesting, so the explorer can see the
|
||
# videos as soon as they land. Flag the channel ephemeral unless it's organically followed.
|
||
existing = db.execute(
|
||
select(ExploredChannel).where(
|
||
ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is None:
|
||
db.add(ExploredChannel(user_id=user.id, channel_id=channel_id))
|
||
else:
|
||
existing.created_at = datetime.now(timezone.utc) # refresh the grace clock
|
||
followed = db.execute(
|
||
select(Subscription.id).where(Subscription.channel_id == channel_id).limit(1)
|
||
).first()
|
||
if followed is None and not channel.from_explore:
|
||
channel.from_explore = True
|
||
db.commit()
|
||
|
||
try:
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_EXPLORE),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
# Enrich for the About panel + the uploads playlist id (needed to ingest videos).
|
||
# Stale, not "a field is missing": a terminated channel never gets an uploads playlist
|
||
# or a creation date, and asking again on every explore would spend quota on a miss.
|
||
if details_stale(channel, datetime.now(timezone.utc)):
|
||
apply_channel_details(db, yt.get_channels([channel_id]), [channel_id])
|
||
db.commit()
|
||
channel = db.get(Channel, channel_id)
|
||
result = explore_ingest_page(db, yt, channel, page_token)
|
||
except YouTubeError as exc:
|
||
raise youtube_refusal(exc) from exc
|
||
return {
|
||
"ok": True,
|
||
"channel_id": channel_id,
|
||
"ingested": result["ingested"],
|
||
"next_page_token": result["next_page_token"],
|
||
}
|
||
|
||
|
||
def _user_subscription(db: Session, user: User, channel_id: str) -> Subscription:
|
||
sub = db.execute(
|
||
select(Subscription).where(
|
||
Subscription.user_id == user.id, Subscription.channel_id == channel_id
|
||
)
|
||
).scalar_one_or_none()
|
||
if sub is None:
|
||
raise not_found("subscription")
|
||
return sub
|
||
|
||
|
||
def _is_blocked(db: Session, user: User, channel_id: str) -> bool:
|
||
"""Whether this user has blocked this channel."""
|
||
return (
|
||
db.execute(
|
||
select(BlockedChannel.id).where(
|
||
BlockedChannel.user_id == user.id,
|
||
BlockedChannel.channel_id == channel_id,
|
||
)
|
||
).first()
|
||
is not None
|
||
)
|
||
|
||
|
||
class ChannelUpdateIn(BaseModel):
|
||
# PATCH semantics: a field left unset (None) is untouched. Every field is optional so the
|
||
# UI can send just the one control it changed. int/bool coercion + rejection of garbage
|
||
# (a non-numeric priority, a non-bool flag) is now Pydantic's job → 422, not a 500.
|
||
priority: int | None = None
|
||
hidden: bool | None = None
|
||
deep_requested: bool | None = None
|
||
|
||
|
||
@router.patch("/{channel_id}")
|
||
def update_channel(
|
||
channel_id: str,
|
||
body: ChannelUpdateIn,
|
||
background: BackgroundTasks,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
sub = _user_subscription(db, user, channel_id)
|
||
channel = db.get(Channel, channel_id)
|
||
deep_turned_on = False
|
||
if body.priority is not None:
|
||
sub.priority = body.priority
|
||
if body.hidden is not None:
|
||
sub.hidden = body.hidden
|
||
if body.deep_requested is not None:
|
||
# Opt this channel into (or out of) full-history backfill. The deep scheduler
|
||
# picks the flag up on its next run; turning it off won't undo already-fetched
|
||
# videos, it just stops further deep paging if nobody else still wants it.
|
||
want = body.deep_requested
|
||
deep_turned_on = want and not sub.deep_requested
|
||
sub.deep_requested = want
|
||
db.commit()
|
||
|
||
# Opting a channel into full history should give an immediate feed: if we haven't even
|
||
# fetched its recent uploads yet, do that one channel now (cheap) instead of waiting for
|
||
# the scheduler.
|
||
if deep_turned_on and channel is not None and channel.recent_synced_at is None:
|
||
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
|
||
run_recent_backfill(db, [channel], max_channels=1)
|
||
|
||
# …and kick the FULL deep backfill for this one channel immediately in the background, so the
|
||
# whole catalogue lands in seconds instead of over the scheduler's paced 10-minute ticks. The
|
||
# request returns right away; the channel page's live poll shows the count climb. The scheduler
|
||
# still finishes a very large channel's tail on its normal runs (deep_requested stays set).
|
||
if deep_turned_on:
|
||
background.add_task(backfill_channel_now, channel_id, user.id)
|
||
|
||
return {
|
||
"ok": True,
|
||
"id": channel_id,
|
||
"priority": sub.priority,
|
||
"hidden": sub.hidden,
|
||
"deep_requested": sub.deep_requested,
|
||
"recent_synced": channel.recent_synced_at is not None if channel else None,
|
||
}
|
||
|
||
|
||
@router.post("/{channel_id}/reset-backfill")
|
||
def reset_backfill(
|
||
channel_id: str,
|
||
user: User = Depends(admin_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""Admin-only reset: re-fetch this channel from scratch regardless of its current sync
|
||
state (a "reset" trigger). Clears the channel's backfill markers, opts it back into deep
|
||
backfill, and re-runs its recent pull immediately; the deep scheduler re-pages the full
|
||
back-catalog on its next run. Idempotent — videos upsert by id, so nothing duplicates."""
|
||
channel = get_or_404(db, Channel, channel_id, what="channel")
|
||
channel.backfill_done = False
|
||
channel.backfill_cursor = None
|
||
channel.recent_synced_at = None
|
||
# Re-arm the demand-driven deep scheduler for this channel. run_deep_backfill gates on an
|
||
# EXISTS over subscriptions with deep_requested=True, so flip it on EVERY subscriber rather
|
||
# than requiring the admin to be personally subscribed (they reset from the manager, not
|
||
# their own feed). No-op row count if the channel has no subscribers.
|
||
db.execute(
|
||
update(Subscription)
|
||
.where(Subscription.channel_id == channel_id)
|
||
.values(deep_requested=True)
|
||
)
|
||
db.commit()
|
||
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
|
||
run_recent_backfill(db, [channel], max_channels=1)
|
||
return {"ok": True, "id": channel_id, "reset": True}
|
||
|
||
|
||
@router.post("/{channel_id}/subscribe")
|
||
def subscribe(
|
||
channel_id: str,
|
||
user: User = Depends(require_write_scope),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""Subscribe to a channel discovered from the user's playlists. Gated behind the
|
||
optional write scope (YouTube subscriptions.insert) via require_write_scope. Creates the
|
||
local subscription with YouTube's returned resource id and enriches the channel's metadata
|
||
if it's still a stub, so the new row renders properly in the manager. It does NOT pull the
|
||
channel's videos — the scheduler picks those up on its next run, like any subscribed channel."""
|
||
channel = get_or_404(db, Channel, channel_id, what="channel")
|
||
existing = db.execute(
|
||
select(Subscription).where(
|
||
Subscription.user_id == user.id, Subscription.channel_id == channel_id
|
||
)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
raise refused(409, "already_subscribed")
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_SUBSCRIBE),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
try:
|
||
yt_sub_id = yt.insert_subscription(channel_id)
|
||
except YouTubeError as exc:
|
||
# Already following on YouTube but not recorded here (a desync — e.g. a stale import
|
||
# dropped the local row, so it resurfaced in discovery). That's not a failure: the end
|
||
# state the user wanted already holds, so record it locally and move on. The next
|
||
# subscription resync fills in the resource id. Any other YouTube error is a real
|
||
# problem — surface it with a clear message (422, not a 502 that the client would
|
||
# mistake for a transient "connection lost").
|
||
# On the reason code, never on the prose: the error no longer carries the body (it is
|
||
# the request quoted back), and a message YouTube rewords would silently turn "already
|
||
# following" back into a 422 the user cannot act on. `reason` is comma-joined when the
|
||
# body listed several.
|
||
if "subscriptionDuplicate" in exc.reason.split(","):
|
||
yt_sub_id = ""
|
||
else:
|
||
raise youtube_refusal(exc) from exc
|
||
# Enrich a stub channel (title/thumbnail/subscriber count) so it shows up properly right
|
||
# away, on BOTH the normal and the desync path — a resurfaced stub still needs metadata to
|
||
# render. Keep the client open (we're still inside the `with`); a metadata failure here
|
||
# must not block recording the subscription the user asked for.
|
||
if channel.details_synced_at is None:
|
||
try:
|
||
apply_channel_details(db, yt.get_channels([channel_id]), [channel_id])
|
||
except YouTubeError as exc:
|
||
log.warning(
|
||
"subscribe: stub enrich failed: %s",
|
||
exc,
|
||
extra=logs.event(
|
||
"channels.subscribe.enrich_failed",
|
||
channel_id=channel_id,
|
||
reason=exc.reason,
|
||
),
|
||
)
|
||
db.add(
|
||
Subscription(
|
||
user_id=user.id,
|
||
channel_id=channel_id,
|
||
yt_subscription_id=yt_sub_id or None,
|
||
subscribed_at=datetime.now(timezone.utc),
|
||
)
|
||
)
|
||
# Subscribing is the "keep" signal: a previously-explored channel becomes a normal organic
|
||
# one. Drop its ephemeral flags so the cleanup job leaves it alone and its videos stop being
|
||
# per-user private (they join the shared Library like any subscribed channel's uploads).
|
||
if channel.from_explore:
|
||
channel.from_explore = False
|
||
db.execute(
|
||
update(Video)
|
||
.where(Video.channel_id == channel_id, Video.via_explore.is_(True))
|
||
.values(via_explore=False)
|
||
)
|
||
db.commit()
|
||
return {"ok": True, "subscribed": channel_id}
|
||
|
||
|
||
@router.delete("/{channel_id}/subscription")
|
||
def unsubscribe(
|
||
channel_id: str,
|
||
user: User = Depends(require_write_scope),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""Unsubscribe from this channel on YouTube and drop the local subscription. Gated
|
||
behind the optional write scope (require_write_scope); read-only users get a 403 and
|
||
should Hide instead."""
|
||
sub = _user_subscription(db, user, channel_id)
|
||
if not sub.yt_subscription_id:
|
||
raise refused(400, "no_subscription_id")
|
||
try:
|
||
with (
|
||
quota.attribute(user.id, quota.QuotaAction.CHANNELS_UNSUBSCRIBE),
|
||
YouTubeClient(db, user) as yt,
|
||
):
|
||
yt.delete_subscription(sub.yt_subscription_id)
|
||
except YouTubeError as exc:
|
||
# 422 (a real, explainable error → speaking modal), not 502 which the client treats as
|
||
# a transient "connection lost" gateway blip.
|
||
raise youtube_refusal(exc) from exc
|
||
db.delete(sub)
|
||
db.commit()
|
||
return {"ok": True, "unsubscribed": channel_id}
|
||
|
||
|
||
@router.get("/blocked/list")
|
||
def list_blocked(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
|
||
"""The user's blocked channels (videos from these are excluded from their search/feed/explore)."""
|
||
rows = (
|
||
db.execute(
|
||
select(Channel)
|
||
.join(BlockedChannel, BlockedChannel.channel_id == Channel.id)
|
||
.where(BlockedChannel.user_id == user.id)
|
||
.order_by(collated(Channel.title, lang_for_user(user)))
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
return [
|
||
{"id": c.id, "title": c.title, "handle": c.handle, "thumbnail_url": c.thumbnail_url}
|
||
for c in rows
|
||
]
|
||
|
||
|
||
@router.post("/{channel_id}/block")
|
||
def block_channel(
|
||
channel_id: str,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""Block a channel for this user: its videos won't appear in their live search (and won't be
|
||
ingested for them) and are hidden from their feed / explore. Idempotent. The now-hidden
|
||
un-kept search/explore videos are reclaimed by the discovery-cleanup job in due course."""
|
||
get_or_404(db, Channel, channel_id, what="channel")
|
||
if not _is_blocked(db, user, channel_id):
|
||
db.add(BlockedChannel(user_id=user.id, channel_id=channel_id))
|
||
# Stop any active exploration of it so it can be cleaned up.
|
||
db.execute(
|
||
ExploredChannel.__table__.delete().where(
|
||
(ExploredChannel.user_id == user.id) & (ExploredChannel.channel_id == channel_id)
|
||
)
|
||
)
|
||
db.commit()
|
||
return {"ok": True, "blocked": channel_id}
|
||
|
||
|
||
@router.delete("/{channel_id}/block")
|
||
def unblock_channel(
|
||
channel_id: str,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
db.execute(
|
||
BlockedChannel.__table__.delete().where(
|
||
(BlockedChannel.user_id == user.id) & (BlockedChannel.channel_id == channel_id)
|
||
)
|
||
)
|
||
db.commit()
|
||
return {"ok": True, "unblocked": channel_id}
|
||
|
||
|
||
class AttachTagIn(BaseModel):
|
||
tag_id: int
|
||
|
||
|
||
@router.post("/{channel_id}/tags")
|
||
def attach_tag(
|
||
channel_id: str,
|
||
body: AttachTagIn,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
_user_subscription(db, user, channel_id)
|
||
# Only the user's own tags may be attached as personal links (owned_or_404 → 404 if the tag
|
||
# is missing OR belongs to someone else, without leaking which).
|
||
owned_or_404(db, Tag, body.tag_id, user, what="tag")
|
||
exists = db.execute(
|
||
select(ChannelTag).where(
|
||
ChannelTag.channel_id == channel_id,
|
||
ChannelTag.tag_id == body.tag_id,
|
||
ChannelTag.user_id == user.id,
|
||
)
|
||
).scalar_one_or_none()
|
||
if exists is None:
|
||
db.add(
|
||
ChannelTag(
|
||
channel_id=channel_id,
|
||
tag_id=body.tag_id,
|
||
user_id=user.id,
|
||
source="user",
|
||
)
|
||
)
|
||
db.commit()
|
||
return {"ok": True, "channel_id": channel_id, "tag_id": body.tag_id}
|
||
|
||
|
||
@router.delete("/{channel_id}/tags/{tag_id}")
|
||
def detach_tag(
|
||
channel_id: str,
|
||
tag_id: int,
|
||
user: User = Depends(current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
row = db.execute(
|
||
select(ChannelTag).where(
|
||
ChannelTag.channel_id == channel_id,
|
||
ChannelTag.tag_id == tag_id,
|
||
ChannelTag.user_id == user.id,
|
||
)
|
||
).scalar_one_or_none()
|
||
if row is not None:
|
||
db.delete(row)
|
||
db.commit()
|
||
return {"ok": True, "channel_id": channel_id, "tag_id": tag_id}
|