15007250db
When a user requests full history for a channel whose recent uploads aren't fetched yet, run a one-channel recent backfill synchronously in the request so the feed populates at once instead of waiting for the scheduler. Deep paging still follows on the scheduler (recent-then-deep). The deep-toggle mutation now also refreshes my-status and the feed.
197 lines
7 KiB
Python
197 lines
7 KiB
Python
"""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."""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import and_, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import current_user, has_write_scope
|
|
from app.db import get_db
|
|
from app.models import Channel, ChannelTag, Subscription, Tag, User, Video
|
|
from app.sync.runner import run_recent_backfill
|
|
from app.youtube.client import YouTubeClient, YouTubeError
|
|
|
|
router = APIRouter(prefix="/api/channels", tags=["channels"])
|
|
|
|
|
|
@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 stored video count (shared data) in one grouped query.
|
|
channel_ids = [c.id for _, c in rows]
|
|
stored: dict[str, int] = {}
|
|
if channel_ids:
|
|
for cid, count in db.execute(
|
|
select(Video.channel_id, func.count(Video.id))
|
|
.where(Video.channel_id.in_(channel_ids))
|
|
.group_by(Video.channel_id)
|
|
).all():
|
|
stored[cid] = count
|
|
|
|
# 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 [
|
|
{
|
|
"id": ch.id,
|
|
"title": ch.title,
|
|
"handle": ch.handle,
|
|
"thumbnail_url": ch.thumbnail_url,
|
|
"subscriber_count": ch.subscriber_count,
|
|
"video_count": ch.video_count,
|
|
"stored_videos": stored.get(ch.id, 0),
|
|
"priority": sub.priority,
|
|
"hidden": sub.hidden,
|
|
"deep_requested": sub.deep_requested,
|
|
"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
|
|
]
|
|
|
|
|
|
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
|
|
|
|
|
|
@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:
|
|
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.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 YouTubeClient(db, user) as yt:
|
|
yt.delete_subscription(sub.yt_subscription_id)
|
|
except YouTubeError as exc:
|
|
raise HTTPException(status_code=502, detail=f"YouTube unsubscribe failed: {exc}")
|
|
db.delete(sub)
|
|
db.commit()
|
|
return {"unsubscribed": 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}
|