peter
9c5f81fd94
- An ad-hoc request made while its target's run is under way is the run after it: the scheduler dropped a second fire of a running id, so a second Takeout file was never imported and its key stayed queued until a restart. - Un-marking or resetting a video is remembered (video_unmarks): an import of an older YouTube view no longer marks it watched again; un-marks take the user's Takeout lock. - One user's failure in the Shorts job's Takeout pass is theirs: the job keeps its own result, the others are settled, and a failed lookup spends none of the daily share. - The upload reads the user's id before giving the read connection back. - A watched row waits on a discovery video (search or explore) until the catalog keeps it, so a mark never pins it against the cleanup jobs. - The history is decoded a window at a time, a Watch later file over 2 MiB is refused unread, and two uploads at once across all users (takeout_server_busy). - A Watch later list goes in as one presence read and one insert.
536 lines
20 KiB
Python
536 lines
20 KiB
Python
"""Reader's reads: a user's labels, saved views, watch state, playlists and Takeout import, for the HTTP
|
|
layer and for the other units that ask about them.
|
|
|
|
Every read of a user's row goes through `userscope`, so "someone else's row" is the same answer as
|
|
a missing one. Reader reads Catalog's videos along its declared edge (a playlist's cover and
|
|
length, whether a video exists); another unit asks the questions at the end of this module rather
|
|
than joining Reader's tables itself.
|
|
"""
|
|
|
|
from collections.abc import Iterable
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import ColumnElement, and_, func, or_, select
|
|
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.reader.models import (
|
|
ChannelTag,
|
|
Playlist,
|
|
PlaylistItem,
|
|
SavedView,
|
|
Tag,
|
|
TakeoutItem,
|
|
TakeoutKind,
|
|
VideoState,
|
|
VideoUnmark,
|
|
)
|
|
from app.userscope import owned_by, owned_row
|
|
|
|
# --- catalog, read along the edge ------------------------------------------------------------
|
|
|
|
|
|
def video(db: Session, video_id: str) -> Video | None:
|
|
return db.get(Video, video_id)
|
|
|
|
|
|
# --- tags ------------------------------------------------------------------------------------
|
|
|
|
|
|
def tags_with_counts(db: Session, user: User) -> list[tuple[Tag, int]]:
|
|
"""The tags this user sees (the shared system ones and their own), each with how many channel
|
|
links it has, by category then name."""
|
|
return [
|
|
(tag, count)
|
|
for tag, count in db.execute(
|
|
select(Tag, func.count(ChannelTag.id))
|
|
.outerjoin(ChannelTag, ChannelTag.tag_id == Tag.id)
|
|
.where(or_(Tag.user_id.is_(None), owned_by(Tag.user_id, user)))
|
|
.group_by(Tag.id)
|
|
.order_by(Tag.category, Tag.name)
|
|
).tuples()
|
|
]
|
|
|
|
|
|
def own_tag(db: Session, user: User, tag_id: int) -> Tag | None:
|
|
"""One of the user's own tags. A system tag (no owner) is nobody's to edit, so it reads as
|
|
missing too."""
|
|
return owned_row(db, Tag, tag_id, user)
|
|
|
|
|
|
# --- saved views -----------------------------------------------------------------------------
|
|
|
|
|
|
def views(db: Session, user: User, source: str) -> list[SavedView]:
|
|
return list(
|
|
db.execute(
|
|
select(SavedView)
|
|
.where(owned_by(SavedView.user_id, user), SavedView.source == source)
|
|
.order_by(SavedView.position, SavedView.id)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def all_views(db: Session, user: User) -> list[SavedView]:
|
|
return list(db.execute(select(SavedView).where(owned_by(SavedView.user_id, user))).scalars())
|
|
|
|
|
|
def own_view(db: Session, user: User, view_id: int) -> SavedView | None:
|
|
return owned_row(db, SavedView, view_id, user)
|
|
|
|
|
|
# --- watch state -----------------------------------------------------------------------------
|
|
|
|
|
|
def video_state(db: Session, user: User, video_id: str) -> VideoState | None:
|
|
return db.execute(
|
|
select(VideoState).where(
|
|
owned_by(VideoState.user_id, user), VideoState.video_id == video_id
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
# --- playlists -------------------------------------------------------------------------------
|
|
|
|
|
|
def own_playlist(
|
|
db: Session, user: User, playlist_id: int, *, lock: bool = False
|
|
) -> Playlist | None:
|
|
return owned_row(db, Playlist, playlist_id, user, lock=lock)
|
|
|
|
|
|
def playlists_for(db: Session, user: User) -> list[Playlist]:
|
|
"""The user's playlists in their rail order."""
|
|
return list(
|
|
db.execute(
|
|
select(Playlist)
|
|
.where(owned_by(Playlist.user_id, user))
|
|
.order_by(Playlist.position, Playlist.created_at)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def watch_later(db: Session, user: User) -> Playlist | None:
|
|
return db.scalar(
|
|
select(Playlist).where(owned_by(Playlist.user_id, user), Playlist.kind == "watch_later")
|
|
)
|
|
|
|
|
|
def item_counts(db: Session, ids: list[int]) -> dict[int, int]:
|
|
return dict(
|
|
db.execute(
|
|
select(PlaylistItem.playlist_id, func.count())
|
|
.where(PlaylistItem.playlist_id.in_(ids))
|
|
.group_by(PlaylistItem.playlist_id)
|
|
)
|
|
.tuples()
|
|
.all()
|
|
)
|
|
|
|
|
|
def durations(db: Session, ids: list[int]) -> dict[int, int]:
|
|
"""Each playlist's running time: the sum of its videos' lengths."""
|
|
return {
|
|
pid: int(total or 0)
|
|
for pid, total in db.execute(
|
|
select(PlaylistItem.playlist_id, func.coalesce(func.sum(Video.duration_seconds), 0))
|
|
.join(Video, Video.id == PlaylistItem.video_id)
|
|
.where(PlaylistItem.playlist_id.in_(ids))
|
|
.group_by(PlaylistItem.playlist_id)
|
|
).tuples()
|
|
}
|
|
|
|
|
|
def containing(db: Session, ids: list[int], video_id: str) -> set[int]:
|
|
"""Which of these playlists already hold the video (the add-to-playlist popover's ticks)."""
|
|
return set(
|
|
db.execute(
|
|
select(PlaylistItem.playlist_id).where(
|
|
PlaylistItem.playlist_id.in_(ids), PlaylistItem.video_id == video_id
|
|
)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def covers(db: Session, ids: list[int]) -> dict[int, str | None]:
|
|
"""Each playlist's cover: its first item's thumbnail, for every list in ONE window query rather
|
|
than a lookup per playlist."""
|
|
rn = (
|
|
func.row_number()
|
|
.over(partition_by=PlaylistItem.playlist_id, order_by=PlaylistItem.position)
|
|
.label("rn")
|
|
)
|
|
first_items = (
|
|
select(PlaylistItem.playlist_id.label("pid"), Video.thumbnail_url.label("thumb"), rn)
|
|
.join(Video, Video.id == PlaylistItem.video_id)
|
|
.where(PlaylistItem.playlist_id.in_(ids))
|
|
.subquery()
|
|
)
|
|
return dict(
|
|
db.execute(select(first_items.c.pid, first_items.c.thumb).where(first_items.c.rn == 1))
|
|
.tuples()
|
|
.all()
|
|
)
|
|
|
|
|
|
def playlist_totals(db: Session, playlist_id: int) -> tuple[int, int, str | None]:
|
|
"""One playlist's item count, running time and cover. An outer join, so an empty list still
|
|
answers (0, 0, None)."""
|
|
count, total = db.execute(
|
|
select(func.count(PlaylistItem.id), func.coalesce(func.sum(Video.duration_seconds), 0))
|
|
.select_from(PlaylistItem)
|
|
.outerjoin(Video, Video.id == PlaylistItem.video_id)
|
|
.where(PlaylistItem.playlist_id == playlist_id)
|
|
).one()
|
|
cover = db.scalar(
|
|
select(Video.thumbnail_url)
|
|
.join(PlaylistItem, PlaylistItem.video_id == Video.id)
|
|
.where(PlaylistItem.playlist_id == playlist_id)
|
|
.order_by(PlaylistItem.position)
|
|
.limit(1)
|
|
)
|
|
return count or 0, total or 0, cover
|
|
|
|
|
|
def playlist_rows(db: Session, user: User, playlist_id: int) -> list:
|
|
"""A playlist's videos as feed rows (the same columns the feed serializes), in list order."""
|
|
state = aliased(VideoState)
|
|
return list(
|
|
db.execute(
|
|
select(*feed_columns(user, state))
|
|
.join(PlaylistItem, PlaylistItem.video_id == Video.id)
|
|
.join(Channel, Channel.id == Video.channel_id)
|
|
.outerjoin(state, and_(state.video_id == Video.id, owned_by(state.user_id, user)))
|
|
.where(PlaylistItem.playlist_id == playlist_id)
|
|
# `id` breaks a position tie the same way on every read (the renumbering add does too).
|
|
.order_by(PlaylistItem.position, PlaylistItem.id)
|
|
).all()
|
|
)
|
|
|
|
|
|
def items(db: Session, playlist_id: int) -> list[PlaylistItem]:
|
|
"""A playlist's item rows, in list order."""
|
|
return list(
|
|
db.execute(
|
|
select(PlaylistItem)
|
|
.where(PlaylistItem.playlist_id == playlist_id)
|
|
.order_by(PlaylistItem.position, PlaylistItem.id)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def item(db: Session, playlist_id: int, video_id: str) -> PlaylistItem | None:
|
|
return db.scalar(
|
|
select(PlaylistItem).where(
|
|
PlaylistItem.playlist_id == playlist_id, PlaylistItem.video_id == video_id
|
|
)
|
|
)
|
|
|
|
|
|
# --- for the other units ---------------------------------------------------------------------
|
|
|
|
|
|
def kept_by_a_reader(video_id: InstrumentedAttribute[str]) -> ColumnElement[bool]:
|
|
"""A video someone holds: watched, in progress or hidden (a `VideoState`), or in one of their
|
|
playlists. The catalog's cleanup jobs keep such a video; as a clause, so their DELETE stays
|
|
one statement."""
|
|
return or_(
|
|
select(VideoState.id).where(VideoState.video_id == video_id).exists(),
|
|
select(PlaylistItem.id).where(PlaylistItem.video_id == video_id).exists(),
|
|
)
|
|
|
|
|
|
def playlisted_by(db: Session, video_ids: list[str]) -> list[tuple[int, str]]:
|
|
"""(user id, video id) for every user who has one of these videos in a playlist."""
|
|
return list(
|
|
db.execute(
|
|
select(Playlist.user_id, PlaylistItem.video_id)
|
|
.join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id)
|
|
.where(PlaylistItem.video_id.in_(video_ids))
|
|
).tuples()
|
|
)
|
|
|
|
|
|
def has_auto_tags(channel_id: InstrumentedAttribute[str]) -> ColumnElement[bool]:
|
|
"""A channel the auto-tagger has already labelled, as a clause for its "only the missing" scan."""
|
|
return (
|
|
select(ChannelTag.id)
|
|
.where(
|
|
ChannelTag.channel_id == channel_id,
|
|
ChannelTag.source == "auto",
|
|
ChannelTag.user_id.is_(None),
|
|
)
|
|
.exists()
|
|
)
|
|
|
|
|
|
def view_tag_ids(filters: object) -> list[int]:
|
|
"""The tag ids a stored view's filters name; anything else in that slot names none."""
|
|
tags = filters.get("tags") if isinstance(filters, dict) else None
|
|
if not isinstance(tags, list):
|
|
return []
|
|
return [t for t in tags if type(t) is int] # `type`, not isinstance: True is not a tag id
|
|
|
|
|
|
def tags_in_feed_views(db: Session) -> set[int]:
|
|
"""Every tag id any user's saved feed view filters on. Read defensively (`view_tag_ids`): a view
|
|
stored before its route checked the shape must not raise for everyone's job."""
|
|
return {
|
|
t
|
|
for (filters,) in db.execute(select(SavedView.filters).where(SavedView.source == "feed"))
|
|
for t in view_tag_ids(filters)
|
|
}
|
|
|
|
|
|
def held_by(db: Session, user: User, video_ids: list[str]) -> set[str]:
|
|
"""Which of these videos the user already holds: watched, in progress or hidden, or in one of
|
|
their playlists (a live search sinks them below the new finds)."""
|
|
if not video_ids:
|
|
return set()
|
|
has_state = (
|
|
select(VideoState.id)
|
|
.where(VideoState.video_id == Video.id, owned_by(VideoState.user_id, user))
|
|
.exists()
|
|
)
|
|
in_playlist = (
|
|
select(PlaylistItem.id)
|
|
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
|
|
.where(PlaylistItem.video_id == Video.id, owned_by(Playlist.user_id, user))
|
|
.exists()
|
|
)
|
|
return set(
|
|
db.scalars(select(Video.id).where(Video.id.in_(video_ids), or_(has_state, in_playlist)))
|
|
)
|
|
|
|
|
|
def feed_rows_by_ids(db: Session, user: User, video_ids: list[str]) -> list[dict]:
|
|
"""These videos serialized with the feed's projection (their channel, this user's watch state,
|
|
saved), in the order given."""
|
|
if not video_ids:
|
|
return []
|
|
state = aliased(VideoState)
|
|
rows = db.execute(
|
|
select(*feed_columns(user, state))
|
|
.join(Channel, Channel.id == Video.channel_id)
|
|
.outerjoin(state, and_(state.video_id == Video.id, owned_by(state.user_id, user)))
|
|
.where(Video.id.in_(video_ids))
|
|
).all()
|
|
by_id = {r.id: serialize(r) for r in rows}
|
|
return [by_id[i] for i in video_ids if i in by_id]
|
|
|
|
|
|
def playlist_channels(db: Session, user: User, exclude: str | None, lang: str) -> list:
|
|
"""The channels of the videos in the user's playlists that they don't subscribe to (`exclude`:
|
|
their own channel), each with how many of those videos it has and across how many playlists;
|
|
most present first, then by title in the reader's alphabet."""
|
|
subscribed = select(Subscription.channel_id).where(owned_by(Subscription.user_id, user))
|
|
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(owned_by(Playlist.user_id, user), Channel.id.not_in(subscribed))
|
|
)
|
|
if exclude:
|
|
q = q.where(Channel.id != exclude)
|
|
return list(
|
|
db.execute(
|
|
q.group_by(Channel.id).order_by(
|
|
func.count(func.distinct(PlaylistItem.video_id)).desc(),
|
|
collated(Channel.title, lang),
|
|
)
|
|
).all()
|
|
)
|
|
|
|
|
|
def user_tag_links(db: Session, user: User, channel_id: str | None = None) -> dict[str, list[int]]:
|
|
"""The user's own tag links by channel (one channel's, when given)."""
|
|
stmt = select(ChannelTag.channel_id, ChannelTag.tag_id).where(
|
|
owned_by(ChannelTag.user_id, user)
|
|
)
|
|
if channel_id is not None:
|
|
stmt = stmt.where(ChannelTag.channel_id == channel_id)
|
|
out: dict[str, list[int]] = {}
|
|
for cid, tag_id in db.execute(stmt).tuples():
|
|
out.setdefault(cid, []).append(tag_id)
|
|
return out
|
|
|
|
|
|
def user_tag_link(db: Session, user: User, channel_id: str, tag_id: int) -> ChannelTag | None:
|
|
return db.execute(
|
|
select(ChannelTag).where(
|
|
ChannelTag.channel_id == channel_id,
|
|
ChannelTag.tag_id == tag_id,
|
|
owned_by(ChannelTag.user_id, user),
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
# --- Google Takeout --------------------------------------------------------------------------
|
|
|
|
|
|
def _settled_here() -> ColumnElement[bool]:
|
|
"""A video a waiting "watched" row may be marked on: one the catalog keeps of its own, not a
|
|
discovery video only a search or an explore brought in. A mark is a `VideoState`, which the
|
|
cleanup jobs read as a reader keeping it (`kept_by_a_reader`): on one of those it would keep the
|
|
video, and its explored channel, for good. Such a row waits until the video is the catalog's own
|
|
(its channel subscribed to) — or is gone."""
|
|
subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
|
|
return and_(Video.via_explore.is_(False), or_(Video.via_search.is_(False), subscribed))
|
|
|
|
|
|
def takeout_watched_ready(db: Session, user: User) -> list[tuple[int, str, datetime, bool]]:
|
|
"""The user's waiting "watched" rows whose video the catalog knows and has probed for Shorts:
|
|
(row id, video id, Takeout time, is Short)."""
|
|
return list(
|
|
db.execute(
|
|
select(TakeoutItem.id, TakeoutItem.video_id, TakeoutItem.at, Video.is_short)
|
|
.join(Video, Video.id == TakeoutItem.video_id)
|
|
.where(
|
|
owned_by(TakeoutItem.user_id, user),
|
|
TakeoutItem.kind == TakeoutKind.WATCHED,
|
|
TakeoutItem.outcome.is_(None),
|
|
Video.shorts_probed,
|
|
_settled_here(),
|
|
)
|
|
).tuples()
|
|
)
|
|
|
|
|
|
def takeout_waiting(db: Session, user: User, kind: TakeoutKind) -> list[tuple[int, str]]:
|
|
"""The user's rows of `kind` the import has not settled, oldest first: (row id, video id)."""
|
|
return list(
|
|
db.execute(
|
|
select(TakeoutItem.id, TakeoutItem.video_id)
|
|
.where(
|
|
owned_by(TakeoutItem.user_id, user),
|
|
TakeoutItem.kind == kind,
|
|
TakeoutItem.outcome.is_(None),
|
|
)
|
|
.order_by(TakeoutItem.at, TakeoutItem.id)
|
|
).tuples()
|
|
)
|
|
|
|
|
|
# Ids per IN list: a list is bound one parameter an id, and a statement takes at most 65,535.
|
|
_IN_CHUNK = 5000
|
|
|
|
|
|
def known_video_ids(db: Session, ids: list[str]) -> set[str]:
|
|
"""Which of `ids` the catalog holds."""
|
|
known: set[str] = set()
|
|
for start in range(0, len(ids), _IN_CHUNK):
|
|
chunk = ids[start : start + _IN_CHUNK]
|
|
known.update(db.execute(select(Video.id).where(Video.id.in_(chunk))).scalars())
|
|
return known
|
|
|
|
|
|
def unmarked_at(db: Session, user: User, ids: list[str]) -> dict[str, datetime]:
|
|
"""When the user last took each of `ids` back to "new" here, for those they ever did."""
|
|
out: dict[str, datetime] = {}
|
|
for start in range(0, len(ids), _IN_CHUNK):
|
|
rows = db.execute(
|
|
select(VideoUnmark.video_id, VideoUnmark.at).where(
|
|
owned_by(VideoUnmark.user_id, user),
|
|
VideoUnmark.video_id.in_(ids[start : start + _IN_CHUNK]),
|
|
)
|
|
).tuples()
|
|
out.update((vid, at) for vid, at in rows)
|
|
return out
|
|
|
|
|
|
def in_playlist(db: Session, playlist_id: int, ids: list[str]) -> set[str]:
|
|
"""Which of `ids` the playlist already holds."""
|
|
held: set[str] = set()
|
|
for start in range(0, len(ids), _IN_CHUNK):
|
|
chunk = ids[start : start + _IN_CHUNK]
|
|
held.update(
|
|
db.execute(
|
|
select(PlaylistItem.video_id).where(
|
|
PlaylistItem.playlist_id == playlist_id, PlaylistItem.video_id.in_(chunk)
|
|
)
|
|
).scalars()
|
|
)
|
|
return held
|
|
|
|
|
|
def takeout_rows_besides(
|
|
db: Session, user: User, kind: TakeoutKind, video_ids: Iterable[str]
|
|
) -> int:
|
|
"""How many Takeout rows the user keeps other than `video_ids` of `kind`: what an upload of
|
|
those videos would be added to."""
|
|
total = (
|
|
db.scalar(select(func.count(TakeoutItem.id)).where(owned_by(TakeoutItem.user_id, user)))
|
|
or 0
|
|
)
|
|
ids = list(video_ids)
|
|
for start in range(0, len(ids), _IN_CHUNK):
|
|
total -= (
|
|
db.scalar(
|
|
select(func.count(TakeoutItem.id)).where(
|
|
owned_by(TakeoutItem.user_id, user),
|
|
TakeoutItem.kind == kind,
|
|
TakeoutItem.video_id.in_(ids[start : start + _IN_CHUNK]),
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
return total
|
|
|
|
|
|
def takeout_outcomes(db: Session, user: User) -> dict[tuple[str, str | None], int]:
|
|
"""How many of the user's rows each kind and outcome holds (None = waiting)."""
|
|
rows = db.execute(
|
|
select(TakeoutItem.kind, TakeoutItem.outcome, func.count())
|
|
.where(owned_by(TakeoutItem.user_id, user))
|
|
.group_by(TakeoutItem.kind, TakeoutItem.outcome)
|
|
).all()
|
|
return {(kind, outcome): n for kind, outcome, n in rows}
|
|
|
|
|
|
def takeout_summary(db: Session, user: User) -> dict[str, dict[str, int]]:
|
|
"""The user's kept Takeout videos by kind and outcome, for their settings page: `{"watched":
|
|
{"watched": n, "short": n, "waiting": n, …}, "watch_later": {…}}` — the import's outcome words,
|
|
"waiting" for a video not known yet."""
|
|
out: dict[str, dict[str, int]] = {kind.value: {} for kind in TakeoutKind}
|
|
for (kind, outcome), n in takeout_outcomes(db, user).items():
|
|
out[kind][outcome or "waiting"] = n
|
|
return out
|
|
|
|
|
|
def takeout_users_ready(db: Session) -> list[int]:
|
|
"""Every user with a waiting "watched" row whose video is now known and probed for Shorts: the
|
|
rows a later run of the Shorts probe made ready."""
|
|
return list(
|
|
db.execute(
|
|
select(TakeoutItem.user_id)
|
|
.join(Video, Video.id == TakeoutItem.video_id)
|
|
.where(
|
|
TakeoutItem.kind == TakeoutKind.WATCHED,
|
|
TakeoutItem.outcome.is_(None),
|
|
Video.shorts_probed,
|
|
_settled_here(),
|
|
)
|
|
.distinct()
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def takeout_users_watch_later_waiting(db: Session) -> list[int]:
|
|
"""Every user with a Watch later row a run could not put in yet."""
|
|
return list(
|
|
db.execute(
|
|
select(TakeoutItem.user_id)
|
|
.where(TakeoutItem.kind == TakeoutKind.WATCH_LATER, TakeoutItem.outcome.is_(None))
|
|
.distinct()
|
|
).scalars()
|
|
)
|