diff --git a/backend/alembic/versions/0115_takeout_channels.py b/backend/alembic/versions/0115_takeout_channels.py new file mode 100644 index 00000000..b7ef1b95 --- /dev/null +++ b/backend/alembic/versions/0115_takeout_channels.py @@ -0,0 +1,49 @@ +"""A watched Takeout video's length, and the channels a user put out of their list + +Revision ID: 0115_takeout_channels +Revises: 0114_video_unmarks +Create Date: 2026-10-24 + +The "channels you watch but do not follow" list counts a user's watched videos over five minutes +by channel. The length is asked of YouTube for that list and kept on the Takeout row; a channel the +user does not want on the list is kept apart. +""" + +import sqlalchemy as sa +from alembic import op + +revision: str = "0115_takeout_channels" +down_revision: str | None = "0114_video_unmarks" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.add_column( + "takeout_items", sa.Column("looked_up_at", sa.DateTime(timezone=True), nullable=True) + ) + op.add_column("takeout_items", sa.Column("duration_seconds", sa.Integer(), nullable=True)) + op.create_table( + "takeout_channel_dismissals", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "user_id", + sa.Integer(), + sa.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("channel_id", sa.String(32), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + sa.UniqueConstraint("user_id", "channel_id", name="uq_takeout_channel_dismissal"), + ) + + +def downgrade() -> None: + op.drop_table("takeout_channel_dismissals") + op.drop_column("takeout_items", "duration_seconds") + op.drop_column("takeout_items", "looked_up_at") diff --git a/backend/app/reader/models.py b/backend/app/reader/models.py index 7ff5d7b1..a146e058 100644 --- a/backend/app/reader/models.py +++ b/backend/app/reader/models.py @@ -254,8 +254,9 @@ class TakeoutItem(Base, TimestampMixin): """One video from a user's Google Takeout export: watched (with when) or in their YouTube Watch later list (with when it was added). The rows are both the import job's input and its memory: a video the catalog does not know yet waits here, with no outcome, until it enters the catalog. - The id is not a foreign key for that reason. The channel is kept for a later "channels you - watch but do not follow" list.""" + The id is not a foreign key for that reason. A watched row's channel and length feed the + "channels you watch but do not follow" list: the length is asked of YouTube for that list alone + and kept here, so its videos need not enter the shared catalog to be counted.""" __tablename__ = "takeout_items" __table_args__ = ( @@ -277,3 +278,21 @@ class TakeoutItem(Base, TimestampMixin): # Watched: the last time in the export; Watch later: when it was added. at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) outcome: Mapped[str | None] = mapped_column(String(12)) + # When YouTube was asked for the video's length (NULL = not yet), and the length it gave; a + # video it no longer has keeps a NULL length. + looked_up_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + duration_seconds: Mapped[int | None] = mapped_column(Integer) + + +class TakeoutChannelDismissal(Base, TimestampMixin): + """A channel the user put out of their "watched but not followed" list. The id is not a foreign + key: the channel may never have entered the catalog, or may leave it, and the choice stays.""" + + __tablename__ = "takeout_channel_dismissals" + __table_args__ = ( + UniqueConstraint("user_id", "channel_id", name="uq_takeout_channel_dismissal"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE")) + channel_id: Mapped[str] = mapped_column(String(32)) diff --git a/backend/app/reader/queries.py b/backend/app/reader/queries.py index b06eb402..fa3fd4fd 100644 --- a/backend/app/reader/queries.py +++ b/backend/app/reader/queries.py @@ -16,13 +16,14 @@ from sqlalchemy.orm import InstrumentedAttribute, Session, aliased from app.collation import collated from app.feedquery import feed_columns, serialize from app.models import User -from app.catalog.models import Channel, Subscription, Video +from app.catalog.models import BlockedChannel, Channel, Subscription, Video from app.reader.models import ( ChannelTag, Playlist, PlaylistItem, SavedView, Tag, + TakeoutChannelDismissal, TakeoutItem, TakeoutKind, VideoState, @@ -533,3 +534,164 @@ def takeout_users_watch_later_waiting(db: Session) -> list[int]: .distinct() ).scalars() ) + + +# --- channels watched but not followed --------------------------------------------------------- + +# A channel joins the list with this many watched videos of at least this length. A Short is three +# minutes at most, so the length alone keeps them out, and a video under five minutes says too +# little of what a channel is about to count towards it. +WATCHED_CHANNEL_MIN_VIDEOS = 5 +WATCHED_CHANNEL_MIN_SECONDS = 300 + + +def _unfollowed(user: User) -> ColumnElement[bool]: + """A watched row whose channel the user neither follows, blocked, put out of the list, nor is.""" + subscribed = select(Subscription.channel_id).where(owned_by(Subscription.user_id, user)) + blocked = select(BlockedChannel.channel_id).where(owned_by(BlockedChannel.user_id, user)) + dismissed = select(TakeoutChannelDismissal.channel_id).where( + owned_by(TakeoutChannelDismissal.user_id, user) + ) + clauses = [ + owned_by(TakeoutItem.user_id, user), + TakeoutItem.kind == TakeoutKind.WATCHED, + TakeoutItem.channel_id.is_not(None), + TakeoutItem.channel_id.not_in(subscribed), + TakeoutItem.channel_id.not_in(blocked), + TakeoutItem.channel_id.not_in(dismissed), + ] + if user.yt_channel_id: + clauses.append(TakeoutItem.channel_id != user.yt_channel_id) + return and_(*clauses) + + +def _candidate_channels(user: User): + """The unfollowed channels with enough watched videos to make the list once their lengths are + known: (channel id, how many watched).""" + return ( + select(TakeoutItem.channel_id, func.count().label("watched")) + .where(_unfollowed(user)) + .group_by(TakeoutItem.channel_id) + .having(func.count() >= WATCHED_CHANNEL_MIN_VIDEOS) + .subquery() + ) + + +def takeout_lengths_to_ask(db: Session, user: User) -> list[str]: + """The watched videos whose length the list still needs: those of the candidate channels not + asked about yet, the most watched channels first, so a lookup cut short fills the top.""" + cand = _candidate_channels(user) + return list( + db.execute( + select(TakeoutItem.video_id) + .join(cand, cand.c.channel_id == TakeoutItem.channel_id) + .where(_unfollowed(user), TakeoutItem.looked_up_at.is_(None)) + .order_by(cand.c.watched.desc(), TakeoutItem.channel_id, TakeoutItem.at.desc()) + ).scalars() + ) + + +def watched_channels_pending(db: Session, user: User) -> int: + """How many candidate channels still have a watched video whose length is not known: the + channels that may yet join the list.""" + cand = _candidate_channels(user) + return ( + db.scalar( + select(func.count(func.distinct(TakeoutItem.channel_id))) + .join(cand, cand.c.channel_id == TakeoutItem.channel_id) + .where(_unfollowed(user), TakeoutItem.looked_up_at.is_(None)) + ) + or 0 + ) + + +def watched_channels_to_describe(db: Session, user: User) -> list[str]: + """The channels on the list (by their counted videos) that the catalog lacks or has no details + of yet: what the list cannot name.""" + counted = ( + select(TakeoutItem.channel_id) + .where(_unfollowed(user), TakeoutItem.duration_seconds >= WATCHED_CHANNEL_MIN_SECONDS) + .group_by(TakeoutItem.channel_id) + .having(func.count() >= WATCHED_CHANNEL_MIN_VIDEOS) + .subquery() + ) + return list( + db.execute( + select(counted.c.channel_id) + .outerjoin(Channel, Channel.id == counted.c.channel_id) + .where(Channel.details_synced_at.is_(None)) + ).scalars() + ) + + +def watched_channels(db: Session, user: User) -> list[tuple[Channel, int, datetime]]: + """The channels the user watched often but does not follow: (channel, how many of its videos + over five minutes they watched, when last). Most watched first; on a tie the one watched more + lately, which is the likelier interest.""" + n = func.count(TakeoutItem.id) + last = func.max(TakeoutItem.at) + return list( + db.execute( + select(Channel, n, last) + .join(Channel, Channel.id == TakeoutItem.channel_id) + .where(_unfollowed(user), TakeoutItem.duration_seconds >= WATCHED_CHANNEL_MIN_SECONDS) + .group_by(Channel.id) + .having(n >= WATCHED_CHANNEL_MIN_VIDEOS) + .order_by(n.desc(), last.desc(), Channel.id) + ).tuples() + ) + + +def has_watch_history(db: Session, user: User) -> bool: + """Whether the user has uploaded a Takeout watch history at all.""" + return bool( + db.scalar( + select( + select(TakeoutItem.id) + .where(owned_by(TakeoutItem.user_id, user), TakeoutItem.kind == TakeoutKind.WATCHED) + .exists() + ) + ) + ) + + +def watched_channel_videos( + db: Session, user: User, channel_id: str +) -> tuple[list[tuple[str, datetime]], int]: + """The user's watched videos of one channel that the list counts, last watched first: (video + id, when). And how many YouTube no longer has.""" + rows = db.execute( + select(TakeoutItem.video_id, TakeoutItem.at, TakeoutItem.duration_seconds) + .where( + owned_by(TakeoutItem.user_id, user), + TakeoutItem.kind == TakeoutKind.WATCHED, + TakeoutItem.channel_id == channel_id, + TakeoutItem.looked_up_at.is_not(None), + ) + .order_by(TakeoutItem.at.desc(), TakeoutItem.video_id) + ).tuples() + counted: list[tuple[str, datetime]] = [] + gone = 0 + for vid, at, seconds in rows: + if seconds is None: + gone += 1 + elif seconds >= WATCHED_CHANNEL_MIN_SECONDS: + counted.append((vid, at)) + return counted, gone + + +def takeout_users_with_lengths_to_ask(db: Session) -> list[int]: + """Every user with a channel of enough watched videos not all asked about yet. It does not ask + whether they follow the channel: that is each user's own question, which + `takeout_lengths_to_ask` answers (and an empty answer costs a run nothing).""" + per_channel = ( + select(TakeoutItem.user_id) + .where(TakeoutItem.kind == TakeoutKind.WATCHED, TakeoutItem.channel_id.is_not(None)) + .group_by(TakeoutItem.user_id, TakeoutItem.channel_id) + .having( + func.count() >= WATCHED_CHANNEL_MIN_VIDEOS, + func.count().filter(TakeoutItem.looked_up_at.is_(None)) > 0, + ) + .subquery() + ) + return list(db.execute(select(per_channel.c.user_id).distinct()).scalars()) diff --git a/backend/app/reader/service.py b/backend/app/reader/service.py index 08bbabc2..5402e653 100644 --- a/backend/app/reader/service.py +++ b/backend/app/reader/service.py @@ -9,7 +9,9 @@ unique key, a row lock taken before a presence check. The comfort shelf's writes from datetime import datetime, timezone -from sqlalchemy import exists, func, select, text, update +from typing import cast + +from sqlalchemy import Table, bindparam, delete, exists, func, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session @@ -21,6 +23,7 @@ from app.reader.models import ( PlaylistItem, SavedView, Tag, + TakeoutChannelDismissal, TakeoutItem, TakeoutKind, TakeoutOutcome, @@ -662,3 +665,51 @@ def apply_takeout_watch_later( "watch_later_present": len(done[TakeoutOutcome.PRESENT]), "watch_later_missing": len(done[TakeoutOutcome.MISSING]), } + + +# --- channels watched but not followed --------------------------------------------------------- + + +def record_takeout_lengths( + db: Session, user: User, asked: list[str], lengths: dict[str, int | None] +) -> None: + """Keep what YouTube said of the user's watched videos `asked` about: each one's length, or none + for a video it no longer has (`lengths` leaves it out). Every asked row is marked asked, so a + video YouTube lost is not asked about again.""" + if not asked: + return + lock_takeout(db, user) # these rows are an upload's to update too + now = datetime.now(timezone.utc) + table = cast(Table, TakeoutItem.__table__) + db.execute( + table.update() + .where( + owned_by(TakeoutItem.user_id, user), + table.c.kind == TakeoutKind.WATCHED, + table.c.video_id == bindparam("vid"), + ) + .values(looked_up_at=now, duration_seconds=bindparam("seconds")), + [{"vid": vid, "seconds": lengths.get(vid)} for vid in asked], + ) + db.flush() + + +def dismiss_watched_channel(db: Session, user: User, channel_id: str) -> None: + """Put a channel out of the user's "watched but not followed" list (again is a no-op).""" + db.execute( + pg_insert(TakeoutChannelDismissal) + .values(user_id=user.id, channel_id=channel_id) + .on_conflict_do_nothing(constraint="uq_takeout_channel_dismissal") + ) + db.flush() + + +def restore_watched_channel(db: Session, user: User, channel_id: str) -> None: + """Let a channel the user put out of the list back on it.""" + db.execute( + delete(TakeoutChannelDismissal).where( + owned_by(TakeoutChannelDismissal.user_id, user), + TakeoutChannelDismissal.channel_id == channel_id, + ) + ) + db.flush() diff --git a/backend/app/reader/takeout.py b/backend/app/reader/takeout.py index 35d5fc1d..89f9c2f0 100644 --- a/backend/app/reader/takeout.py +++ b/backend/app/reader/takeout.py @@ -4,7 +4,9 @@ only way in. Two files of the export are read, each uploaded on its own: `history/watch-history.json` and `playlists/Watch later-videos.csv`. Their videos are kept (`TakeoutItem`) and the import job applies -what it can; a video the catalog does not know yet waits for it. +what it can; a video the catalog does not know yet waits for it. The watched videos' channels also +make the list of channels the user watches but does not follow (`queries.watched_channels`): the +job asks YouTube how long those videos are, and what the channels are, for it. The format, measured on a real export (2026-10): the history is a JSON list, one entry per view, newest first. A view of a video has a `titleUrl` with `watch?v=` (on www.youtube.com or @@ -37,8 +39,9 @@ from app.ratelimit import RateLimiter from app.reader import queries, service from app.reader.models import TakeoutKind from app.sync.runner import get_service_user -from app.sync.videos import ensure_videos -from app.youtube.client import YouTubeClient +from app.sync.subscriptions import describe_channels +from app.sync.videos import ensure_videos, parse_iso8601_duration +from app.youtube.client import YouTubeClient, YouTubeError log = logs.get_logger(__name__) @@ -351,10 +354,79 @@ def _put_in_watch_later(db: Session, user: User) -> dict: return service.apply_takeout_watch_later(db, user, waiting, asked) +def _ask_lengths(db: Session, user: User) -> dict: + """Ask YouTube how long the user's watched videos are that the "watched but not followed" list + still needs, and keep the answer — on the same terms and the same daily share as the Watch + later lookups. What is not asked yet (no YouTube-connected account, the reserve reached, the + share spent) waits for a later run, and the list says it is not complete. A YouTube failure is + logged and left for a later run too: it is no reason to fail the import it follows. The caller + commits.""" + ask = queries.takeout_lengths_to_ask(db, user) + fetcher = get_service_user(db) if ask else None + if fetcher is None or quota.backfill_reserve_reached(): + return {} + ask = ask[: int(_lookups.room(str(user.id)))] + if not ask: + return {} + try: + with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP): + with YouTubeClient(db, fetcher) as yt: + items = yt.get_videos(ask) + except YouTubeError as exc: + log.warning( + "Takeout: asking the watched videos' lengths failed: %s", + exc, + extra=logs.event("takeout.lengths.failed", user_id=user.id, reason=exc.reason), + ) + return {} + _lookups.allow(str(user.id), cost=len(ask)) + lengths = { + it["id"]: parse_iso8601_duration(it.get("contentDetails", {}).get("duration")) + for it in items + if it.get("id") + } + service.record_takeout_lengths(db, user, ask, lengths) + return {"lengths_asked": len(ask)} + + +def _describe_channels(db: Session, user: User) -> dict: + """Bring the channels the user's list holds into the catalog with their details (name, picture, + description), the ones it lacks: without them the list cannot show a channel. A YouTube failure + is logged and left for a later run. The caller commits.""" + need = queries.watched_channels_to_describe(db, user) + fetcher = get_service_user(db) if need else None + if fetcher is None or quota.backfill_reserve_reached(): + return {} + try: + with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER): + with YouTubeClient(db, fetcher) as yt: + describe_channels(db, yt, need) + except YouTubeError as exc: + db.rollback() + log.warning( + "Takeout: describing the watched channels failed: %s", + exc, + extra=logs.event("takeout.channels.failed", user_id=user.id, reason=exc.reason), + ) + return {} + return {"channels_described": len(need)} + + +def _channel_list(db: Session, user: User) -> dict: + """The "watched but not followed" list's part of a run: the lengths, landed, then the channels. + Commits each.""" + result = _ask_lengths(db, user) + db.commit() + result |= _describe_channels(db, user) + db.commit() + return result + + def run_import(user_id: int) -> dict: """The import job: apply what the user's kept Takeout rows can say now, as one recorded run. The watched part commits before the Watch later part asks YouTube for the videos the catalog - lacks, so a YouTube failure there loses nothing already marked. Returns the counts.""" + lacks, so a YouTube failure there loses nothing already marked; the "watched but not followed" + list's lookups come last. Returns the counts.""" with _one_import(user_id), SessionLocal() as db: user = db.get(User, user_id) if user is None: @@ -363,6 +435,7 @@ def run_import(user_id: int) -> dict: db.commit() result |= _put_in_watch_later(db, user) db.commit() + result |= _channel_list(db, user) outcomes = queries.takeout_outcomes(db, user) result["waiting"] = outcomes.get((TakeoutKind.WATCHED, None), 0) if outcomes.get((TakeoutKind.WATCH_LATER, None)): @@ -373,13 +446,15 @@ def run_import(user_id: int) -> dict: def apply_waiting(db: Session) -> dict: """Settle what every user's import left waiting and can be settled now, one commit each: a watched video the catalog has come to know and the Shorts probe has cleared (by its job or a - search's), and the Watch later videos a run could not put in yet. Run after each run of the + search's), the Watch later videos a run could not put in yet, and what the "watched but not + followed" list still needs asked of YouTube. Run after each run of the `shorts` job, which is what clears the most of them. Returns its counts, or an empty dict when nothing was waiting that could move.""" - marked = added = failed = 0 + marked = added = asked = failed = 0 ready = set(queries.takeout_users_ready(db)) later = set(queries.takeout_users_watch_later_waiting(db)) - for user_id in sorted(ready | later): + lengths = set(queries.takeout_users_with_lengths_to_ask(db)) + for user_id in sorted(ready | later | lengths): user = db.get(User, user_id) if user is None: continue @@ -393,6 +468,8 @@ def apply_waiting(db: Session) -> dict: if user_id in later: added += _put_in_watch_later(db, user).get("watch_later_added", 0) db.commit() + if user_id in lengths: + asked += _channel_list(db, user).get("lengths_asked", 0) except Exception: db.rollback() failed += 1 @@ -406,6 +483,8 @@ def apply_waiting(db: Session) -> dict: out["takeout_watched"] = marked if added: out["takeout_watch_later_added"] = added + if asked: + out["takeout_lengths_asked"] = asked if failed: out["takeout_failed"] = failed return out diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 491a104b..95f9941f 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -3,12 +3,13 @@ from collections.abc import AsyncIterator from datetime import date from functools import partial -from fastapi import APIRouter, Depends, Query, Request +from fastapi import APIRouter, Depends, Path, Query, Request from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy.orm.exc import StaleDataError from app import jobruns, quota, scheduler +from app.catalog import queries as catalog_queries, service as catalog_service from app.auth import current_user from app.db import get_db, release_read_connection from app.mailtext import lang_for_user @@ -24,7 +25,7 @@ from app.routes._common import ( not_found, youtube_refusal, ) -from app.sync.videos import parse_iso8601_duration +from app.sync.videos import ensure_videos, parse_iso8601_duration from app.utils import iso from app.youtube.client import YouTubeClient, YouTubeError from app.refusal import refused @@ -417,3 +418,95 @@ def takeout_status(user: User = Depends(current_user), db: Session = Depends(get }, **queries.takeout_summary(db, user), } + + +# --- channels watched but not followed ----------------------------------------------------------- + + +@router.get("/takeout/channels") +def watched_channels(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: + """The channels the user's Takeout history shows them watching often but not following: each + with how many of its videos over five minutes they watched, and when last. `pending` is how many + more channels may join once the import has asked YouTube for their videos' lengths, so a list + still being made does not read as the whole of it; `imported` tells "no history uploaded" from + "nothing to suggest".""" + return { + "imported": queries.has_watch_history(db, user), + "pending": queries.watched_channels_pending(db, user), + "channels": [ + { + "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, + "watched": int(n), + "last_watched_at": iso(last), + } + for ch, n, last in queries.watched_channels(db, user) + ], + } + + +@router.post("/takeout/channels/{channel_id}/videos") +def watched_channel_videos( + channel_id: str, user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + """The videos of a channel that the user's list counts, ready to play, last watched first. The + channel is explored for the user (as its page does), and the videos the catalog lacks come in as + the exploration's: the player needs them in the catalog, the shared Library does not get them, + and the cleanup reclaims them once nobody keeps them. `gone`: how many YouTube no longer has.""" + channel = catalog_queries.channel(db, channel_id) + if channel is None: + raise not_found("channel") + if catalog_queries.is_blocked(db, user, channel_id): + raise refused(403, "channel_blocked") + counted, gone = queries.watched_channel_videos(db, user, channel_id) + ids = [vid for vid, _ in counted] + known = queries.known_video_ids(db, ids) + missing = [vid for vid in ids if vid not in known] + if missing and quota.backfill_reserve_reached(): + raise refused(429, "quota_reserve") + catalog_service.explore(db, user, channel) + db.commit() + if missing: + try: + with ( + quota.attribute(user.id, quota.QuotaAction.CHANNELS_EXPLORE), + YouTubeClient(db, user) as yt, + ): + ensure_videos(db, yt, missing, explore=True) + except YouTubeError as exc: + raise youtube_refusal(exc) from exc + items = queries.feed_rows_by_ids(db, user, ids) + watched_at = dict(counted) + for item in items: + item["takeout_watched_at"] = iso(watched_at[item["id"]]) + # A video YouTube still had when its length was asked may have gone since. + return {"ok": True, "items": items, "gone": gone + len(ids) - len(items)} + + +@router.put("/takeout/channels/{channel_id}/dismissal") +def dismiss_watched_channel( + channel_id: str = Path(max_length=32), + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Put a channel out of the user's list (it stays out until restored).""" + service.dismiss_watched_channel(db, user, channel_id) + db.commit() + return {"ok": True, "channel_id": channel_id} + + +@router.delete("/takeout/channels/{channel_id}/dismissal") +def restore_watched_channel( + channel_id: str = Path(max_length=32), + user: User = Depends(current_user), + db: Session = Depends(get_db), +) -> dict: + """Let a channel the user put out of the list back on it.""" + service.restore_watched_channel(db, user, channel_id) + db.commit() + return {"ok": True, "channel_id": channel_id} diff --git a/backend/app/sync/subscriptions.py b/backend/app/sync/subscriptions.py index 301cd019..495af93e 100644 --- a/backend/app/sync/subscriptions.py +++ b/backend/app/sync/subscriptions.py @@ -109,6 +109,21 @@ def apply_channel_details(db: Session, items: list[dict], asked: Iterable[str]) channel.details_synced_at = now +def describe_channels(db: Session, yt: YouTubeClient, channel_ids: list[str]) -> None: + """Make sure each given channel is in the catalog with its details: a channel it lacks is added + as a stub, then all of them are asked for (`channels.list`, 1 unit a 50). Flushes; the caller + commits.""" + if not channel_ids: + return + have = set(db.execute(select(Channel.id).where(Channel.id.in_(channel_ids))).scalars()) + for cid in dict.fromkeys(channel_ids): + if cid not in have: + db.add(Channel(id=cid)) + db.flush() + apply_channel_details(db, yt.get_channels(channel_ids), channel_ids) + db.flush() + + def should_prune(fetched_count: int, existing_count: int) -> bool: """Whether an import may delete the local subscriptions missing from this fetch. diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 6ae40983..2f77b9df 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -420,11 +420,15 @@ def run_shorts_classification(db: Session, limit: int | None = None) -> dict: return {"probed": probed, "shorts": shorts} -def ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None: +def ensure_videos( + db: Session, yt: YouTubeClient, video_ids: list[str], *, explore: bool = False +) -> None: """Make sure every given video id exists in the catalog, fetching + ingesting the missing ones (and a stub channel for any unknown channel). Deleted/private videos that the API won't return are simply left out. The catalog's write for a playlist sync that names - videos it has never seen; commits.""" + videos it has never seen; commits. `explore` adds the new ones as an exploration's + (`via_explore`): visible to who explores or follows their channel, and reclaimed by the cleanup + once nobody keeps them.""" if not video_ids: return have = set(db.execute(select(Video.id).where(Video.id.in_(video_ids))).scalars().all()) @@ -458,7 +462,12 @@ def ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None: seen.add(vid) # published_at is NOT NULL (0068); apply_video_details re-fills a missing date, but seed a now() # fallback here too so the constructor never carries an explicit NULL. - v = Video(id=vid, channel_id=cid, published_at=parse_dt(sn.get("publishedAt")) or now) + v = Video( + id=vid, + channel_id=cid, + published_at=parse_dt(sn.get("publishedAt")) or now, + via_explore=explore, + ) apply_video_details(v, it) v.enriched_at = now db.add(v) diff --git a/backend/tests/test_unit_fences.py b/backend/tests/test_unit_fences.py index c189977e..985df0d2 100644 --- a/backend/tests/test_unit_fences.py +++ b/backend/tests/test_unit_fences.py @@ -45,6 +45,7 @@ UNIT_MODELS = { "SavedView", "ComfortPin", "TakeoutItem", + "TakeoutChannelDismissal", "VideoUnmark", "Tag", "ChannelTag", @@ -367,6 +368,8 @@ UNIT_COMMITS = { "reader/takeout.py:run_import", # Run inside the Shorts job, on its session: each user's waiting rows land on their own. "reader/takeout.py:apply_waiting", + # Inside either of those: the lengths YouTube gave land before the channels are asked for. + "reader/takeout.py:_channel_list", }, # The catalog's ingest jobs commit per page or batch, a YouTube call between each: a page # fetched and stored is not lost when the next call fails, and an enrich that fails rolls back diff --git a/backend/tests/test_watched_channels.py b/backend/tests/test_watched_channels.py new file mode 100644 index 00000000..1c4cc867 --- /dev/null +++ b/backend/tests/test_watched_channels.py @@ -0,0 +1,399 @@ +"""The channels a user's Takeout history shows them watching but not following: which channels make +the list (five videos of five minutes or more, none they follow, block, put out or own), what the +import asks YouTube for it and when, and the channel's watched videos made ready to play.""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import select + +from app import quota +from app.catalog.models import BlockedChannel, Channel, ExploredChannel, Subscription, Video +from app.reader import queries, service, takeout +from app.reader.models import TakeoutChannelDismissal, TakeoutItem, TakeoutKind +from app.routes import feed as feed_routes +from app.sync import videos as sync_videos +from app.youtube.client import YouTubeError + +T0 = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc) +LONG = 600 +SHORT = 120 + + +def _ch(n: int) -> str: + return f"UC{n:022d}" + + +def _vid(channel: int, n: int) -> str: + return f"c{channel:02d}v{n:07d}" + + +def _watched(db, user, channel: int, lengths: list[int | None], *, at=T0, asked=True): + """Stage `len(lengths)` watched videos of a channel, the newest first, each with its length + already asked (None: YouTube no longer has it) — or not asked yet.""" + service.stage_takeout( + db, + user, + TakeoutKind.WATCHED, + {_vid(channel, i): (at - timedelta(days=i), _ch(channel)) for i in range(len(lengths))}, + ) + db.flush() + if asked: + ids = [_vid(channel, i) for i in range(len(lengths))] + service.record_takeout_lengths( + db, user, ids, {v: s for v, s in zip(ids, lengths, strict=True) if s is not None} + ) + db.commit() + + +def _channel(db, n: int, title: str | None = None): + db.add(Channel(id=_ch(n), title=title or f"Channel {n}", details_synced_at=T0)) + db.commit() + + +def _listed(db, user) -> list[tuple[str, int]]: + return [(ch.id, n) for ch, n, _ in queries.watched_channels(db, user)] + + +# --- which channels make the list -------------------------------------------------------------- + + +def test_a_channel_makes_the_list_with_five_watched_videos_of_five_minutes_or_more(db, make_user): + user = make_user("list@example.com") + for n in (1, 2, 3): + _channel(db, n) + _watched(db, user, 1, [LONG] * 5 + [SHORT] * 3) # five long, three short: on it with 5 + _watched(db, user, 2, [LONG] * 4 + [SHORT] * 6) # ten watched, only four long: off it + _watched(db, user, 3, [299, 300, 300, 300, 300, 300]) # five at exactly five minutes: on it + # A tie: channel 1's counted videos were watched later (channel 3's newest is under five minutes). + assert _listed(db, user) == [(_ch(1), 5), (_ch(3), 5)] + + +def test_a_video_youtube_no_longer_has_does_not_count(db, make_user): + user = make_user("gone@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG] * 4 + [None] * 3) + assert _listed(db, user) == [] + + +def test_the_most_watched_lead_and_a_tie_goes_to_the_one_watched_more_lately(db, make_user): + user = make_user("order@example.com") + for n in (1, 2, 3): + _channel(db, n) + _watched(db, user, 1, [LONG] * 6, at=T0 - timedelta(days=100)) + _watched(db, user, 2, [LONG] * 9, at=T0 - timedelta(days=300)) + _watched(db, user, 3, [LONG] * 6, at=T0) + assert _listed(db, user) == [(_ch(2), 9), (_ch(3), 6), (_ch(1), 6)] + last = {ch.id: when for ch, _, when in queries.watched_channels(db, user)} + assert last[_ch(3)] == T0 # the last watched of the counted ones + + +def test_a_channel_the_user_follows_blocked_put_out_or_owns_is_not_on_the_list(db, make_user): + user = make_user("off@example.com", yt_channel_id=_ch(4)) + for n in (1, 2, 3, 4, 5): + _channel(db, n) + _watched(db, user, n, [LONG] * 5) + db.add(Subscription(user_id=user.id, channel_id=_ch(1))) + db.add(BlockedChannel(user_id=user.id, channel_id=_ch(2))) + db.commit() + service.dismiss_watched_channel(db, user, _ch(3)) + db.commit() + assert _listed(db, user) == [(_ch(5), 5)] + + +def test_one_users_follows_and_dismissals_leave_anothers_list_alone(db, make_user): + me, other = make_user("me@example.com"), make_user("other@example.com") + _channel(db, 1) + for user in (me, other): + _watched(db, user, 1, [LONG] * 5) + db.add(Subscription(user_id=other.id, channel_id=_ch(1))) + db.commit() + service.dismiss_watched_channel(db, other, _ch(1)) + db.commit() + assert _listed(db, me) == [(_ch(1), 5)] + assert _listed(db, other) == [] + + +def test_a_channel_whose_lengths_are_not_all_known_is_pending_until_they_are(db, make_user): + user = make_user("pending@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG] * 6, asked=False) + _watched(db, user, 2, [LONG] * 4, asked=False) # four watched: it can never make the list + assert queries.watched_channels_pending(db, user) == 1 + assert _listed(db, user) == [] + ids = [_vid(1, i) for i in range(6)] + service.record_takeout_lengths(db, user, ids, dict.fromkeys(ids, LONG)) + db.commit() + assert queries.watched_channels_pending(db, user) == 0 + assert _listed(db, user) == [(_ch(1), 6)] + + +def test_a_new_upload_of_the_history_keeps_the_lengths_already_asked(db, make_user): + user = make_user("again@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG] * 5) + _watched(db, user, 1, [LONG] * 5, at=T0 + timedelta(days=1), asked=False) + assert _listed(db, user) == [(_ch(1), 5)] + assert queries.takeout_lengths_to_ask(db, user) == [] + + +# --- what the import asks YouTube for ---------------------------------------------------------- + + +class FakeYouTube: + """Stands in for `YouTubeClient`: answers `videos.list` with the lengths it was given and + `channels.list` with a title per channel, and records what it was asked.""" + + def __init__(self, lengths: dict[str, int] | None = None, *, fail: bool = False): + self.lengths = lengths or {} + self.fail = fail + self.videos_asked: list[list[str]] = [] + self.channels_asked: list[list[str]] = [] + self.actions: list[tuple[int | None, str]] = [] + + def __call__(self, db, user): + return self + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def get_videos(self, ids): + self.videos_asked.append(list(ids)) + self.actions.append((quota._actor_id.get(), quota._action.get())) + if self.fail: + raise YouTubeError("refused", reason="quotaExceeded", path="videos") + return [ + {"id": v, "contentDetails": {"duration": f"PT{self.lengths[v]}S"}} + for v in ids + if v in self.lengths + ] + + def get_channels(self, ids): + self.channels_asked.append(list(ids)) + self.actions.append((quota._actor_id.get(), quota._action.get())) + return [{"id": c, "snippet": {"title": f"Named {c[-2:]}"}} for c in ids] + + +@pytest.fixture +def youtube(monkeypatch, db, make_user): + fetcher = make_user("fetcher@example.com") + fake = FakeYouTube() + monkeypatch.setattr(takeout, "YouTubeClient", fake) + monkeypatch.setattr(takeout, "get_service_user", lambda session: fetcher) + monkeypatch.setattr(takeout.quota, "backfill_reserve_reached", lambda: False) + return fake + + +def test_the_import_asks_the_lengths_the_list_needs_and_names_its_channels(db, make_user, youtube): + user = make_user("import@example.com") + _watched(db, user, 1, [None] * 7, asked=False) # the most watched channel + _watched(db, user, 2, [None] * 5, asked=False) + _watched(db, user, 3, [None] * 4, asked=False) # too few to ever make it: not asked about + youtube.lengths = {_vid(1, i): LONG for i in range(6)} | {_vid(2, i): SHORT for i in range(5)} + + result = takeout.run_import(user.id) + + asked = [_vid(1, i) for i in range(7)] + [_vid(2, i) for i in range(5)] + assert youtube.videos_asked == [asked] # the top channel first + assert result["lengths_asked"] == 12 + # Only channel 1 made it (six long; its seventh is gone, channel 2's are short): it is named. + assert youtube.channels_asked == [[_ch(1)]] + assert result["channels_described"] == 1 + assert { + (user.id, quota.QuotaAction.VIDEOS_LOOKUP), + (user.id, quota.QuotaAction.CHANNELS_DISCOVER), + } == set(youtube.actions) + db.expire_all() + assert [(ch.title, n) for ch, n, _ in queries.watched_channels(db, user)] == [("Named 01", 6)] + assert queries.watched_channels_pending(db, user) == 0 + # Asked once: a second run asks nothing again, the gone video included. + youtube.videos_asked.clear() + youtube.channels_asked.clear() + assert "lengths_asked" not in takeout.run_import(user.id) + assert youtube.videos_asked == [] and youtube.channels_asked == [] + + +def test_the_lookups_stop_at_the_users_daily_share_and_the_rest_wait( + db, make_user, youtube, monkeypatch +): + user = make_user("share@example.com") + _watched(db, user, 1, [None] * 8, asked=False) + _watched(db, user, 2, [None] * 6, asked=False) + monkeypatch.setattr(takeout, "_lookups", takeout.RateLimiter(10, 3600)) + takeout.run_import(user.id) + assert youtube.videos_asked == [[_vid(1, i) for i in range(8)] + [_vid(2, 0), _vid(2, 1)]] + db.expire_all() + assert queries.watched_channels_pending(db, user) == 1 # channel 2 still has some to ask + youtube.videos_asked.clear() + takeout.run_import(user.id) + assert youtube.videos_asked == [] # the share is spent for today + + +@pytest.mark.parametrize("why", ["no_fetcher", "reserve"]) +def test_nothing_is_asked_without_a_connected_account_or_past_the_reserve( + db, make_user, youtube, monkeypatch, why +): + user = make_user(f"{why}@example.com") + _watched(db, user, 1, [None] * 5, asked=False) + if why == "no_fetcher": + monkeypatch.setattr(takeout, "get_service_user", lambda session: None) + else: + monkeypatch.setattr(takeout.quota, "backfill_reserve_reached", lambda: True) + assert "lengths_asked" not in takeout.run_import(user.id) + assert youtube.videos_asked == [] + assert queries.watched_channels_pending(db, user) == 1 + + +def test_a_youtube_failure_asking_lengths_fails_neither_the_import_nor_its_marks( + db, make_user, youtube +): + user = make_user("fail@example.com") + _watched(db, user, 1, [None] * 5, asked=False) + youtube.fail = True + result = takeout.run_import(user.id) + assert "lengths_asked" not in result + assert queries.takeout_lengths_to_ask(db, user) == [_vid(1, i) for i in range(5)] + db.expire_all() + rows = db.execute(select(TakeoutItem).where(TakeoutItem.user_id == user.id)).scalars() + assert all(r.looked_up_at is None for r in rows) # left to ask again + + +def test_the_shorts_job_asks_what_an_import_could_not(db, make_user, youtube): + user = make_user("later@example.com") + _watched(db, user, 1, [None] * 5, asked=False) + youtube.lengths = {_vid(1, i): LONG for i in range(5)} + assert takeout.apply_waiting(db)["takeout_lengths_asked"] == 5 + db.expire_all() + assert [n for _, n, _ in queries.watched_channels(db, user)] == [5] + + +def test_a_channel_the_user_follows_is_not_asked_about(db, make_user, youtube): + user = make_user("follows@example.com") + _channel(db, 1) + _watched(db, user, 1, [None] * 5, asked=False) + db.add(Subscription(user_id=user.id, channel_id=_ch(1))) + db.commit() + assert takeout.apply_waiting(db) == {} + assert youtube.videos_asked == [] + + +# --- the channel's watched videos, ready to play ------------------------------------------------ + + +def _fake_ensure(calls): + def ensure(session, yt, ids, *, explore=False): + calls.append((list(ids), explore)) + for vid in ids: + if vid.endswith("4"): + continue # gone from YouTube since its length was asked + session.add(Video(id=vid, channel_id=_ch(1), published_at=T0, via_explore=explore)) + session.commit() + + return ensure + + +def test_a_channels_watched_videos_come_ready_to_play_and_explored( + db, make_user, as_user, monkeypatch +): + user = make_user("play@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG, SHORT, LONG, None, LONG, LONG, LONG]) + db.add(Video(id=_vid(1, 0), channel_id=_ch(1), published_at=T0)) # already in the catalog + db.commit() + calls = [] + monkeypatch.setattr(feed_routes, "ensure_videos", _fake_ensure(calls)) + monkeypatch.setattr(feed_routes.quota, "backfill_reserve_reached", lambda: False) + + resp = as_user(user).post(f"/api/takeout/channels/{_ch(1)}/videos") + + assert resp.status_code == 200, resp.text + body = resp.json() + # Only the counted ones, newest first; the short one and the gone ones are left out. + assert calls == [([_vid(1, 2), _vid(1, 4), _vid(1, 5), _vid(1, 6)], True)] + assert [it["id"] for it in body["items"]] == [_vid(1, 0), _vid(1, 2), _vid(1, 5), _vid(1, 6)] + assert body["items"][0]["takeout_watched_at"].startswith("2026-05-01") + assert body["gone"] == 2 # one YouTube had lost before, one since + assert body["ok"] is True + db.expire_all() + assert db.execute( + select(ExploredChannel).where( + ExploredChannel.user_id == user.id, ExploredChannel.channel_id == _ch(1) + ) + ).scalar_one() + assert ( + db.get(Channel, _ch(1)).from_explore is True + ) # nobody follows it: the cleanup's to reclaim + + +def test_a_blocked_or_unknown_channels_videos_are_refused(db, make_user, as_user): + user = make_user("refused@example.com") + _channel(db, 1) + db.add(BlockedChannel(user_id=user.id, channel_id=_ch(1))) + db.commit() + client = as_user(user) + assert client.post(f"/api/takeout/channels/{_ch(1)}/videos").status_code == 403 + assert client.post(f"/api/takeout/channels/{_ch(9)}/videos").status_code == 404 + + +def test_ensure_videos_adds_an_explorations_videos_as_its_own(db): + db.add(Channel(id=_ch(1), title="C")) + db.commit() + + class YT: + def get_videos(self, ids): + return [{"id": v, "snippet": {"channelId": _ch(1)}} for v in ids] + + sync_videos.ensure_videos(db, YT(), ["exploredvid"], explore=True) + sync_videos.ensure_videos(db, YT(), ["organicvid0"]) + db.expire_all() + assert db.get(Video, "exploredvid").via_explore is True + assert db.get(Video, "organicvid0").via_explore is False + + +# --- the routes ----------------------------------------------------------------------------------- + + +def test_the_list_route_says_what_it_holds_and_whether_a_history_was_uploaded( + db, make_user, as_user +): + user = make_user("route@example.com") + client = as_user(user) + assert client.get("/api/takeout/channels").json() == { + "imported": False, + "pending": 0, + "channels": [], + } + _channel(db, 1, "Some channel") + _watched(db, user, 1, [LONG] * 5) + _watched(db, user, 2, [None] * 5, asked=False) + body = client.get("/api/takeout/channels").json() + assert body["imported"] is True and body["pending"] == 1 + assert [(c["id"], c["title"], c["watched"]) for c in body["channels"]] == [ + (_ch(1), "Some channel", 5) + ] + assert body["channels"][0]["last_watched_at"].startswith("2026-05-01") + + +def test_a_channel_put_out_of_the_list_comes_back_when_restored(db, make_user, as_user): + user = make_user("dismiss@example.com") + _channel(db, 1) + _watched(db, user, 1, [LONG] * 5) + client = as_user(user) + url = f"/api/takeout/channels/{_ch(1)}/dismissal" + assert client.put(url).json() == {"ok": True, "channel_id": _ch(1)} + assert client.put(url).status_code == 200 # again: no error + assert client.get("/api/takeout/channels").json()["channels"] == [] + assert client.delete(url).json() == {"ok": True, "channel_id": _ch(1)} + assert [c["id"] for c in client.get("/api/takeout/channels").json()["channels"]] == [_ch(1)] + db.expire_all() + assert db.execute(select(TakeoutChannelDismissal)).scalars().all() == [] + + +def test_a_dismissal_of_an_overlong_channel_id_is_refused_before_the_database( + db, make_user, as_user +): + user = make_user("long@example.com") + assert as_user(user).put(f"/api/takeout/channels/{'U' * 33}/dismissal").status_code == 422