siftlode/backend/app/routes/channels.py

657 lines
27 KiB
Python
Raw Normal View History

"""Channel manager: the user's subscriptions with their personal overrides (priority,
hidden, tags) and per-channel sync state. Read + local-write only YouTube-side writes
(unsubscribe) live behind the optional write scope and are added in a later phase."""
import logging
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func, select, update
from sqlalchemy.orm import Session
from app import quota, sysconfig
from app.auth import current_user, has_write_scope, require_human
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.admin import admin_user
from app.sync.explore import explore_ingest_page
from app.sync.runner import run_recent_backfill
from app.sync.subscriptions import apply_channel_details
from app.youtube.client import YouTubeClient, YouTubeError
router = APIRouter(prefix="/api/channels", tags=["channels"])
log = logging.getLogger("siftlode.api")
@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(), func.lower(Channel.title))
).all()
# Per-channel aggregates over the shared catalog in ONE grouped pass: stored count, last
# upload, total duration, and a Normal/Shorts/Live breakdown. Cheap — a single grouped scan
# (~35 ms over the full 233k-row catalog on the dev DB), so computed on read rather than
# denormalised (measure-first: reads aren't frequent/hot enough to warrant cached columns).
channel_ids = [c.id for _, c in rows]
agg: dict[str, dict] = {}
if channel_ids:
for cid, n, last_up, total_dur, shorts, live in db.execute(
select(
Video.channel_id,
func.count(Video.id),
func.max(Video.published_at),
func.coalesce(func.sum(Video.duration_seconds), 0),
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.in_(channel_ids))
.group_by(Video.channel_id)
).all():
agg[cid] = {
"stored": n,
"last_video_at": last_up.isoformat() if last_up else None,
"total_duration_seconds": int(total_dur or 0),
"count_short": int(shorts),
"count_live": int(live),
"count_normal": max(0, int(n) - int(shorts) - int(live)),
}
# 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),
"count_normal": (agg.get(ch.id) or {}).get("count_normal", 0),
"count_short": (agg.get(ch.id) or {}).get("count_short", 0),
"count_live": (agg.get(ch.id) or {}).get("count_live", 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(),
func.lower(Channel.title),
)
).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 demo / token-less users just skip it.
if user.yt_channel_id is None and user.token is not None and not user.is_demo:
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:
log.warning("discovery: own-channel resolve failed (user %s): %s", user.id, exc)
rows = _discovery_rows(db, user)
fix(deferred): close backend correctness/efficiency tails from the hygiene sweep Verified each deferred per-subsystem item against current source, then fixed the real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout): - search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment failure/omitted rows) — restores the scheduler's enriched_at guard so a real Short can't leak into the feed and never get reclassified. - downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets (they carry no TTL, so the expiry passes never cleared them). - downloads B6: quota/edit errors now return a structured {code,reason,limit} the trilingual client localises, instead of hardcoded English + dot-decimal GB. - channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update) instead of 404ing when the admin isn't personally subscribed. - channels BB2: enrich the stub on BOTH the normal and "already exists" desync path (nest the insert try inside the client `with`). - channels CB3: drop the redundant second _discovery_rows read (identity-mapped rows are already enriched; re-sort in Python for the title tie-break). - playlists PC1: read the live playlist once in push() and share it with plan_push + push_playlist (halves read-quota, closes the second TOCTOU window). - playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1); rename combines count+duration into one aggregate + a single cover lookup. - messages MB2: get_thread only resolves a messageable partner OR one we already share a thread with (closes the user-enumeration oracle). - messages MB3: push a live unread-count to the user's other tabs after mark-read. - messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of loading the whole message history into memory. - config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked to disable them rather than snapping back to the env default.
2026-07-12 05:59:03 +02:00
# 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))
db.commit()
# Match _discovery_rows' ORDER BY exactly: video-count DESC, then title (NULLs LAST,
# as Postgres orders them — coercing NULL→"" would wrongly float untitled channels up).
rows = sorted(
rows, key=lambda r: (-int(r[1]), r[0].title is None, (r[0].title or "").lower())
)
except YouTubeError as exc:
log.warning("discovery: channel enrich failed (user %s): %s", user.id, exc)
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,
}
def _channel_detail_dict(
channel: Channel, *, subscribed: bool, explored: bool, blocked: bool, stored: 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": channel.published_at.isoformat() if channel.published_at else None,
"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,
"from_explore": channel.from_explore,
"details_synced": channel.details_synced_at is not None,
}
@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, or a channel last enriched before those fields existed (published_at is a
reliable sentinel: every real channel has a creation date). The enrich uses the API key
(1 unit, cached after), so it's skipped for the quota-less demo account."""
channel = db.get(Channel, channel_id)
if channel is None:
raise HTTPException(status_code=404, detail="Unknown channel")
if channel.published_at is None and not user.is_demo:
try:
with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER), YouTubeClient(db, user) as yt:
apply_channel_details(db, yt.get_channels([channel_id]))
db.commit()
channel = db.get(Channel, channel_id)
except YouTubeError as exc:
log.warning("channel detail enrich failed (%s): %s", channel_id, exc)
db.rollback()
subscribed = db.execute(
select(Subscription.id).where(
Subscription.user_id == user.id, Subscription.channel_id == channel_id
)
).first() 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 = db.scalar(select(func.count(Video.id)).where(Video.channel_id == channel_id)) or 0
return _channel_detail_dict(
channel, subscribed=subscribed, explored=explored, blocked=blocked, stored=int(stored)
)
@router.post("/{channel_id}/explore")
def explore_channel(
channel_id: str,
page_token: str | None = None,
user: User = Depends(require_human),
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; blocked for the demo account (it spends shared quota)."""
channel = db.get(Channel, channel_id)
if channel is None:
raise HTTPException(status_code=404, detail="Unknown channel")
if _is_blocked(db, user, channel_id):
raise HTTPException(status_code=403, detail="You've blocked this channel.")
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
raise HTTPException(
status_code=429,
detail="The shared daily quota reserve is reached — try exploring later.",
)
# 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).
if (
channel.details_synced_at is None
or channel.uploads_playlist_id is None
or channel.published_at is None
):
apply_channel_details(db, yt.get_channels([channel_id]))
db.commit()
channel = db.get(Channel, channel_id)
result = explore_ingest_page(db, yt, channel, page_token)
except YouTubeError as exc:
raise HTTPException(
status_code=422, detail=f"YouTube couldn't load this channel: {exc}"
)
return {
"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 HTTPException(status_code=404, detail="Not subscribed to this channel")
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
)
@router.patch("/{channel_id}")
def update_channel(
channel_id: str,
payload: dict,
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 "priority" in payload:
sub.priority = int(payload["priority"])
if "hidden" in payload:
sub.hidden = bool(payload["hidden"])
if "deep_requested" in payload:
# 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 = bool(payload["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. Deep paging still follows on the scheduler's next run (recent-then-deep).
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)
return {
"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 = db.get(Channel, channel_id)
if channel is None:
raise HTTPException(status_code=404, detail="Unknown channel")
channel.backfill_done = False
channel.backfill_cursor = None
channel.recent_synced_at = None
fix(deferred): close backend correctness/efficiency tails from the hygiene sweep Verified each deferred per-subsystem item against current source, then fixed the real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout): - search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment failure/omitted rows) — restores the scheduler's enriched_at guard so a real Short can't leak into the feed and never get reclassified. - downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets (they carry no TTL, so the expiry passes never cleared them). - downloads B6: quota/edit errors now return a structured {code,reason,limit} the trilingual client localises, instead of hardcoded English + dot-decimal GB. - channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update) instead of 404ing when the admin isn't personally subscribed. - channels BB2: enrich the stub on BOTH the normal and "already exists" desync path (nest the insert try inside the client `with`). - channels CB3: drop the redundant second _discovery_rows read (identity-mapped rows are already enriched; re-sort in Python for the title tie-break). - playlists PC1: read the live playlist once in push() and share it with plan_push + push_playlist (halves read-quota, closes the second TOCTOU window). - playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1); rename combines count+duration into one aggregate + a single cover lookup. - messages MB2: get_thread only resolves a messageable partner OR one we already share a thread with (closes the user-enumeration oracle). - messages MB3: push a live unread-count to the user's other tabs after mark-read. - messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of loading the whole message history into memory. - config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked to disable them rather than snapping back to the env default.
2026-07-12 05:59:03 +02:00
# 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 {"id": channel_id, "reset": True}
@router.post("/{channel_id}/subscribe")
def subscribe(
channel_id: str,
user: User = Depends(current_user),
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). 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."""
if not has_write_scope(user):
raise HTTPException(
status_code=403,
detail="Enable playlist editing in Settings to subscribe on YouTube.",
)
channel = db.get(Channel, channel_id)
if channel is None:
raise HTTPException(status_code=404, detail="Unknown 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 HTTPException(status_code=409, detail="Already subscribed to this channel.")
fix(deferred): close backend correctness/efficiency tails from the hygiene sweep Verified each deferred per-subsystem item against current source, then fixed the real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout): - search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment failure/omitted rows) — restores the scheduler's enriched_at guard so a real Short can't leak into the feed and never get reclassified. - downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets (they carry no TTL, so the expiry passes never cleared them). - downloads B6: quota/edit errors now return a structured {code,reason,limit} the trilingual client localises, instead of hardcoded English + dot-decimal GB. - channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update) instead of 404ing when the admin isn't personally subscribed. - channels BB2: enrich the stub on BOTH the normal and "already exists" desync path (nest the insert try inside the client `with`). - channels CB3: drop the redundant second _discovery_rows read (identity-mapped rows are already enriched; re-sort in Python for the title tie-break). - playlists PC1: read the live playlist once in push() and share it with plan_push + push_playlist (halves read-quota, closes the second TOCTOU window). - playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1); rename combines count+duration into one aggregate + a single cover lookup. - messages MB2: get_thread only resolves a messageable partner OR one we already share a thread with (closes the user-enumeration oracle). - messages MB3: push a live unread-count to the user's other tabs after mark-read. - messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of loading the whole message history into memory. - config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked to disable them rather than snapping back to the env default.
2026-07-12 05:59:03 +02:00
with quota.attribute(user.id, quota.QuotaAction.CHANNELS_SUBSCRIBE), YouTubeClient(db, user) as yt:
try:
yt_sub_id = yt.insert_subscription(channel_id)
fix(deferred): close backend correctness/efficiency tails from the hygiene sweep Verified each deferred per-subsystem item against current source, then fixed the real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout): - search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment failure/omitted rows) — restores the scheduler's enriched_at guard so a real Short can't leak into the feed and never get reclassified. - downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets (they carry no TTL, so the expiry passes never cleared them). - downloads B6: quota/edit errors now return a structured {code,reason,limit} the trilingual client localises, instead of hardcoded English + dot-decimal GB. - channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update) instead of 404ing when the admin isn't personally subscribed. - channels BB2: enrich the stub on BOTH the normal and "already exists" desync path (nest the insert try inside the client `with`). - channels CB3: drop the redundant second _discovery_rows read (identity-mapped rows are already enriched; re-sort in Python for the title tie-break). - playlists PC1: read the live playlist once in push() and share it with plan_push + push_playlist (halves read-quota, closes the second TOCTOU window). - playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1); rename combines count+duration into one aggregate + a single cover lookup. - messages MB2: get_thread only resolves a messageable partner OR one we already share a thread with (closes the user-enumeration oracle). - messages MB3: push a live unread-count to the user's other tabs after mark-read. - messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of loading the whole message history into memory. - config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked to disable them rather than snapping back to the env default.
2026-07-12 05:59:03 +02:00
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").
msg = str(exc).lower()
if "already exists" in msg or "duplicate" in msg:
yt_sub_id = ""
else:
raise HTTPException(status_code=422, detail=f"YouTube couldn't subscribe you: {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]))
fix(deferred): close backend correctness/efficiency tails from the hygiene sweep Verified each deferred per-subsystem item against current source, then fixed the real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout): - search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment failure/omitted rows) — restores the scheduler's enriched_at guard so a real Short can't leak into the feed and never get reclassified. - downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets (they carry no TTL, so the expiry passes never cleared them). - downloads B6: quota/edit errors now return a structured {code,reason,limit} the trilingual client localises, instead of hardcoded English + dot-decimal GB. - channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update) instead of 404ing when the admin isn't personally subscribed. - channels BB2: enrich the stub on BOTH the normal and "already exists" desync path (nest the insert try inside the client `with`). - channels CB3: drop the redundant second _discovery_rows read (identity-mapped rows are already enriched; re-sort in Python for the title tie-break). - playlists PC1: read the live playlist once in push() and share it with plan_push + push_playlist (halves read-quota, closes the second TOCTOU window). - playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1); rename combines count+duration into one aggregate + a single cover lookup. - messages MB2: get_thread only resolves a messageable partner OR one we already share a thread with (closes the user-enumeration oracle). - messages MB3: push a live unread-count to the user's other tabs after mark-read. - messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of loading the whole message history into memory. - config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked to disable them rather than snapping back to the env default.
2026-07-12 05:59:03 +02:00
except YouTubeError as exc:
log.warning("subscribe: stub enrich failed (%s): %s", channel_id, exc)
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 {"subscribed": channel_id}
@router.delete("/{channel_id}/subscription")
def unsubscribe(
channel_id: str,
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> dict:
"""Unsubscribe from this channel on YouTube and drop the local subscription. Gated
behind the optional write scope; read-only users get a 403 and should Hide instead."""
if not has_write_scope(user):
raise HTTPException(
status_code=403,
detail="Enable playlist editing in Settings to unsubscribe on YouTube.",
)
sub = _user_subscription(db, user, channel_id)
if not sub.yt_subscription_id:
raise HTTPException(
status_code=400,
detail="No YouTube subscription id on record — sync subscriptions first.",
)
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 HTTPException(status_code=422, detail=f"YouTube couldn't unsubscribe you: {exc}")
db.delete(sub)
db.commit()
return {"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(func.lower(Channel.title))
).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."""
if db.get(Channel, channel_id) is None:
raise HTTPException(status_code=404, detail="Unknown 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 {"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 {"unblocked": channel_id}
@router.post("/{channel_id}/tags")
def attach_tag(
channel_id: str,
payload: dict,
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> dict:
_user_subscription(db, user, channel_id)
tag_id = payload.get("tag_id")
tag = db.get(Tag, tag_id) if tag_id is not None else None
# Only the user's own tags may be attached as personal links.
if tag is None or tag.user_id != user.id:
raise HTTPException(status_code=404, detail="Unknown tag")
exists = 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 exists is None:
db.add(
ChannelTag(
channel_id=channel_id,
tag_id=tag_id,
user_id=user.id,
source="user",
)
)
db.commit()
return {"channel_id": channel_id, "tag_id": 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 {"channel_id": channel_id, "tag_id": tag_id}