peter
bd2340c204
The import asks YouTube how long the watched videos of the candidate channels are (five or more watched, none the user follows, blocked, put out or owns), keeps the length on the Takeout row and names the channels that make the list: five videos of five minutes or more. The shorts job asks what an import could not. A channel's counted videos come in as an exploration's when its list is opened, ready to play; a channel can be put out of the list and back.
716 lines
28 KiB
Python
716 lines
28 KiB
Python
"""Reader's writes: a user's labels and saved views, where they are in a video, their playlists'
|
|
items and order — and the system tags the catalog's auto-tagger hands over.
|
|
|
|
Every function here flushes and leaves the commit to its caller (the handler, or the job that owns
|
|
its session). A write that must not double up keeps its guard where it is written: an upsert on the
|
|
unique key, a row lock taken before a presence check. The comfort shelf's writes are Home's
|
|
(`discover.pins`); a playlist's YouTube sync is `sync.playlists`.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
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
|
|
|
|
from app.models import User
|
|
from app.reader import queries
|
|
from app.reader.models import (
|
|
ChannelTag,
|
|
Playlist,
|
|
PlaylistItem,
|
|
SavedView,
|
|
Tag,
|
|
TakeoutChannelDismissal,
|
|
TakeoutItem,
|
|
TakeoutKind,
|
|
TakeoutOutcome,
|
|
VideoState,
|
|
VideoUnmark,
|
|
)
|
|
from app.sync.playlists import lock_playlists, recompute_dirty
|
|
from app.userscope import owned_by
|
|
|
|
TAG_CATEGORIES = ("language", "topic", "other")
|
|
VIEW_NAME_MAX = 80
|
|
|
|
# --- tags ------------------------------------------------------------------------------------
|
|
|
|
|
|
def create_tag(db: Session, user: User, name: str, category: str, color: str | None) -> Tag:
|
|
"""A tag of the user's own. An unknown category falls back to "other" rather than refusing. A
|
|
name the user already has raises `IntegrityError` at the flush."""
|
|
tag = Tag(
|
|
user_id=user.id,
|
|
name=name,
|
|
color=color,
|
|
category=category if category in TAG_CATEGORIES else "other",
|
|
)
|
|
db.add(tag)
|
|
db.flush()
|
|
return tag
|
|
|
|
|
|
def update_tag(tag: Tag, *, sent: set[str], name: str | None, color: str | None) -> None:
|
|
"""Change the fields the caller sent: `color` sent as None clears it."""
|
|
if "name" in sent and name is not None:
|
|
tag.name = name
|
|
if "color" in sent:
|
|
tag.color = color
|
|
|
|
|
|
def delete_tag(db: Session, tag: Tag) -> None:
|
|
db.delete(tag) # its channel links cascade
|
|
db.flush()
|
|
|
|
|
|
def ensure_system_tag(db: Session, name: str, category: str, color: str | None) -> Tag:
|
|
"""The shared system tag of this name, made the first time a channel earns it."""
|
|
tag = db.execute(
|
|
select(Tag).where(Tag.user_id.is_(None), Tag.name == name)
|
|
).scalar_one_or_none()
|
|
if tag is None:
|
|
tag = Tag(user_id=None, name=name, category=category, color=color)
|
|
db.add(tag)
|
|
db.flush()
|
|
return tag
|
|
|
|
|
|
def set_auto_tags(db: Session, channel_id: str, desired: dict[int, float | None]) -> None:
|
|
"""Make a channel's automatic tag links exactly `desired` (tag id → confidence). Only the
|
|
auto-tagger's own links move; a user's links on the channel are never touched."""
|
|
existing = {
|
|
ct.tag_id: ct
|
|
for ct in db.execute(
|
|
select(ChannelTag).where(
|
|
ChannelTag.channel_id == channel_id,
|
|
ChannelTag.user_id.is_(None),
|
|
ChannelTag.source == "auto",
|
|
)
|
|
).scalars()
|
|
}
|
|
for tag_id, ct in existing.items():
|
|
if tag_id not in desired:
|
|
db.delete(ct)
|
|
for tag_id, conf in desired.items():
|
|
ct = existing.get(tag_id)
|
|
if ct is None:
|
|
db.add(
|
|
ChannelTag(
|
|
channel_id=channel_id,
|
|
tag_id=tag_id,
|
|
user_id=None,
|
|
source="auto",
|
|
confidence=conf,
|
|
)
|
|
)
|
|
else:
|
|
ct.confidence = conf
|
|
db.flush()
|
|
|
|
|
|
def prune_orphan_system_tags(db: Session) -> int:
|
|
"""Delete the system tags no channel links to any more (after a mapping change) — but not one a
|
|
saved feed view filters on. Views hold tag IDS, and a deleted tag comes back under a new id when
|
|
a channel earns it again, so the view's filter would point at nothing; the feed drops an unknown
|
|
id, and the view showed the whole feed under its old name. The kept ids are filtered here rather
|
|
than in SQL: a `NOT IN` over every view's ids could pass the driver's bind-parameter limit."""
|
|
in_views = queries.tags_in_feed_views(db)
|
|
orphans = [
|
|
tag
|
|
for tag in db.execute(
|
|
select(Tag).where(Tag.user_id.is_(None), ~exists().where(ChannelTag.tag_id == Tag.id))
|
|
).scalars()
|
|
if tag.id not in in_views
|
|
]
|
|
for tag in orphans:
|
|
db.delete(tag)
|
|
db.flush()
|
|
return len(orphans)
|
|
|
|
|
|
# --- saved views -----------------------------------------------------------------------------
|
|
|
|
|
|
def create_view(db: Session, user: User, source: str, name: str, filters: dict) -> SavedView:
|
|
"""A view appended after the user's others of the same module (each module numbers its own
|
|
0..N). A name the user already has there raises `IntegrityError` at the flush."""
|
|
maxpos = db.execute(
|
|
select(func.coalesce(func.max(SavedView.position), -1)).where(
|
|
owned_by(SavedView.user_id, user), SavedView.source == source
|
|
)
|
|
).scalar_one()
|
|
view = SavedView(
|
|
user_id=user.id,
|
|
source=source,
|
|
name=name[:VIEW_NAME_MAX],
|
|
filters=filters,
|
|
position=maxpos + 1,
|
|
)
|
|
db.add(view)
|
|
db.flush()
|
|
return view
|
|
|
|
|
|
def update_view(
|
|
db: Session,
|
|
view: SavedView,
|
|
*,
|
|
sent: set[str],
|
|
name: str | None = None,
|
|
filters: dict | None = None,
|
|
is_default: bool | None = None,
|
|
) -> None:
|
|
"""Change the fields the caller sent. Making a view the default clears the flag on the user's
|
|
other views of the same module first: at most one default per (user, source)."""
|
|
if "name" in sent and name is not None:
|
|
view.name = name[:VIEW_NAME_MAX]
|
|
if "filters" in sent and filters is not None:
|
|
view.filters = filters
|
|
if "is_default" in sent:
|
|
make_default = bool(is_default)
|
|
if make_default:
|
|
db.execute(
|
|
update(SavedView)
|
|
.where(
|
|
owned_by(SavedView.user_id, view.user_id),
|
|
SavedView.source == view.source,
|
|
SavedView.id != view.id,
|
|
)
|
|
.values(is_default=False)
|
|
)
|
|
view.is_default = make_default
|
|
db.flush()
|
|
|
|
|
|
def delete_view(db: Session, view: SavedView) -> None:
|
|
db.delete(view)
|
|
db.flush()
|
|
|
|
|
|
def reorder_views(db: Session, user: User, ids: list[int]) -> None:
|
|
"""Number the listed views 0..n in the listed order; ids that are not the user's are skipped."""
|
|
by_id = {v.id: v for v in queries.all_views(db, user)}
|
|
pos = 0
|
|
for vid in ids:
|
|
v = by_id.get(vid)
|
|
if v is not None:
|
|
v.position = pos
|
|
pos += 1
|
|
db.flush()
|
|
|
|
|
|
# --- watch state -----------------------------------------------------------------------------
|
|
|
|
|
|
def unmark(db: Session, user: User, video_id: str) -> None:
|
|
"""Back to "new". A row that still holds a resume point keeps it (un-marking "watched" restores
|
|
the in-progress state, not where the user left off); otherwise the row goes. The player fires
|
|
this beside a checkpoint that may delete the same row: the caller's commit then raises
|
|
`StaleDataError`, which is that lost race and nothing worse."""
|
|
lock_takeout(db, user) # an import settling meanwhile reads the un-mark below, or comes first
|
|
row = queries.video_state(db, user, video_id)
|
|
if row is None:
|
|
return
|
|
if row.position_seconds:
|
|
row.status = "new"
|
|
row.watched_at = None
|
|
else:
|
|
db.delete(row)
|
|
_remember_unmark(db, user, video_id)
|
|
db.flush()
|
|
|
|
|
|
def _remember_unmark(db: Session, user: User, video_id: str) -> None:
|
|
"""Note that the user took the video back to "new" now: the state they set leaves no row, and
|
|
an import of their YouTube history must not mark it watched again (`apply_takeout_watched`)."""
|
|
now = datetime.now(timezone.utc)
|
|
db.execute(
|
|
pg_insert(VideoUnmark)
|
|
.values(user_id=user.id, video_id=video_id, at=now)
|
|
.on_conflict_do_update(constraint="uq_video_unmark", set_={"at": now})
|
|
)
|
|
|
|
|
|
def mark(db: Session, user: User, video_id: str, status: str) -> None:
|
|
""" "watched" or "hidden", as one atomic upsert on (user, video). The player marks "watched" at
|
|
the end of playback at the instant a checkpoint may delete the in-progress row; a SELECT then
|
|
UPDATE raised `StaleDataError` on that race, and an INSERT … ON CONFLICT is immune to it.
|
|
"hidden" keeps any earlier `watched_at`."""
|
|
now = datetime.now(timezone.utc)
|
|
set_: dict = {"status": status, "updated_at": now}
|
|
if status == "watched":
|
|
set_["watched_at"] = now
|
|
db.execute(
|
|
pg_insert(VideoState)
|
|
.values(
|
|
user_id=user.id,
|
|
video_id=video_id,
|
|
status=status,
|
|
watched_at=now if status == "watched" else None,
|
|
)
|
|
.on_conflict_do_update(constraint="uq_user_video", set_=set_)
|
|
)
|
|
|
|
|
|
def clear_state(db: Session, user: User, video_id: str) -> None:
|
|
"""The video back to pristine for this user: status, resume point and `watched_at` all go.
|
|
Playlist membership lives elsewhere and stays."""
|
|
lock_takeout(db, user) # as `unmark`
|
|
row = queries.video_state(db, user, video_id)
|
|
if row is not None:
|
|
db.delete(row)
|
|
_remember_unmark(db, user, video_id)
|
|
db.flush()
|
|
|
|
|
|
def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
|
|
"""The player's resume point. 0 clears it: a row holding nothing else goes, one with a status
|
|
(watched, hidden) keeps it with the position zeroed. The status itself is never changed here.
|
|
A position is one upsert on (user, video): the first checkpoint and the end-of-playback mark
|
|
beside it both read "no row", and a plain add let the second one fail on `uq_user_video`."""
|
|
now = datetime.now(timezone.utc)
|
|
if position:
|
|
db.execute(
|
|
pg_insert(VideoState)
|
|
.values(
|
|
user_id=user.id,
|
|
video_id=video_id,
|
|
position_seconds=position,
|
|
progress_updated_at=now,
|
|
)
|
|
.on_conflict_do_update(
|
|
constraint="uq_user_video",
|
|
set_={"position_seconds": position, "progress_updated_at": now, "updated_at": now},
|
|
)
|
|
)
|
|
return
|
|
row = queries.video_state(db, user, video_id)
|
|
if row is None:
|
|
return
|
|
if row.status == "new":
|
|
db.delete(row)
|
|
else:
|
|
row.position_seconds = 0
|
|
row.progress_updated_at = now
|
|
db.flush()
|
|
|
|
|
|
# --- playlists -------------------------------------------------------------------------------
|
|
|
|
|
|
class MoveRefused(Exception):
|
|
"""A move between playlists that cannot happen: `reason` is the code the UI translates."""
|
|
|
|
def __init__(self, reason: str):
|
|
super().__init__(reason)
|
|
self.reason = reason
|
|
|
|
|
|
def touch(pl: Playlist) -> None:
|
|
"""Bump the playlist's `updated_at` for an edit that only writes `playlist_items`.
|
|
|
|
`UpdatedAtMixin`'s `onupdate` fires on an UPDATE of the PLAYLIST row, and add / remove /
|
|
reorder / move write the item rows instead — so without this the rail's "recently changed"
|
|
sort would only ever see renames. The assignment is what makes the column mean "last time
|
|
this list changed", which is what the rail promises the reader.
|
|
|
|
The DATABASE's clock, like the mixin's own `onupdate`: one sort key written from two clocks
|
|
(this process's and Postgres's) could order a rename and an item edit the wrong way round."""
|
|
pl.updated_at = func.now()
|
|
|
|
|
|
def _next_playlist_position(db: Session, user_id: int) -> int:
|
|
"""Append position for a new playlist in this user's rail."""
|
|
return (
|
|
db.execute(
|
|
select(func.coalesce(func.max(Playlist.position), -1)).where(
|
|
owned_by(Playlist.user_id, user_id)
|
|
)
|
|
).scalar_one()
|
|
+ 1
|
|
)
|
|
|
|
|
|
def next_item_position(db: Session, playlist_id: int) -> int:
|
|
"""Append position for a new item at the end of a playlist. Locks the playlist row first: two
|
|
adds (two tabs, the rail and the page) otherwise both read the same max and land on one
|
|
position, and the list's order between them is then arbitrary."""
|
|
lock_playlists(db, playlist_id)
|
|
return (
|
|
db.execute(
|
|
select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
|
|
PlaylistItem.playlist_id == playlist_id
|
|
)
|
|
).scalar_one()
|
|
+ 1
|
|
)
|
|
|
|
|
|
def create_playlist(db: Session, user: User, name: str) -> Playlist:
|
|
pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id))
|
|
db.add(pl)
|
|
db.flush()
|
|
return pl
|
|
|
|
|
|
def watch_later(db: Session, user: User) -> Playlist:
|
|
"""The user's built-in Watch later playlist, created on first use. Created by the insert itself
|
|
(ON CONFLICT DO NOTHING on the one-per-user index) and then read: the first two saves at once
|
|
both read "none", and each used to make a list of its own."""
|
|
pl = queries.watch_later(db, user)
|
|
if pl is None:
|
|
db.execute(
|
|
pg_insert(Playlist)
|
|
.values(
|
|
user_id=user.id,
|
|
name="Watch later",
|
|
kind="watch_later",
|
|
position=_next_playlist_position(db, user.id),
|
|
)
|
|
.on_conflict_do_nothing(
|
|
index_elements=["user_id"], index_where=text("kind = 'watch_later'")
|
|
)
|
|
)
|
|
pl = queries.watch_later(db, user)
|
|
assert pl is not None # inserted now, or committed by the request that won
|
|
return pl
|
|
|
|
|
|
def add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
|
|
"""Append a video to a playlist if not already present (idempotent). Returns whether it
|
|
was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).
|
|
Locked BEFORE the presence check: two adds of one video both saw "absent" and the second one's
|
|
insert hit uq_playlist_video as a 500."""
|
|
lock_playlists(db, pl.id)
|
|
if queries.item(db, pl.id, video_id) is not None:
|
|
return False
|
|
db.add(
|
|
PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_item_position(db, pl.id))
|
|
)
|
|
if mark_dirty:
|
|
recompute_dirty(db, pl)
|
|
touch(pl)
|
|
db.flush()
|
|
return True
|
|
|
|
|
|
def remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
|
|
"""Remove a video from a playlist if present (idempotent). Returns whether it was removed.
|
|
`mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later)."""
|
|
lock_playlists(db, pl.id)
|
|
found = queries.item(db, pl.id, video_id)
|
|
if found is None:
|
|
return False
|
|
db.delete(found)
|
|
if mark_dirty:
|
|
recompute_dirty(db, pl)
|
|
touch(pl)
|
|
db.flush()
|
|
return True
|
|
|
|
|
|
def place_item(db: Session, pl: Playlist, video_id: str, position: int) -> None:
|
|
"""Put a video already in the list at `position` (clamped to the end), renumbering the whole
|
|
list: positions can have gaps after removals, so "index N" is only meaningful over the
|
|
ordered rows."""
|
|
rows = queries.items(db, pl.id)
|
|
new = next(r for r in rows if r.video_id == video_id)
|
|
rows.remove(new)
|
|
rows.insert(min(position, len(rows)), new)
|
|
for pos, row in enumerate(rows):
|
|
row.position = pos
|
|
db.flush()
|
|
|
|
|
|
def add_to_playlist(db: Session, pl: Playlist, video_id: str, position: int | None) -> None:
|
|
"""Add a video to one of the user's lists, at `position` when given (the list page's "Undo" of
|
|
a removal puts the row back where it was in ONE request), else at the end. A video already in
|
|
the list stays where it is."""
|
|
added = add_item(db, pl, video_id, mark_dirty=False)
|
|
if added and position is not None:
|
|
place_item(db, pl, video_id, position)
|
|
recompute_dirty(db, pl)
|
|
db.flush()
|
|
|
|
|
|
def rename_playlist(db: Session, pl: Playlist, name: str) -> None:
|
|
"""A new name moves `dirty`; the caller holds the playlist's lock (the scheduled mirror refresh
|
|
reads `dirty` under it)."""
|
|
if name != pl.name:
|
|
pl.name = name
|
|
recompute_dirty(db, pl)
|
|
db.flush()
|
|
|
|
|
|
def delete_playlist(db: Session, pl: Playlist) -> None:
|
|
"""The playlist first, as every item write: the cascade deletes the items after the list row
|
|
is locked."""
|
|
lock_playlists(db, pl.id)
|
|
db.delete(pl) # items cascade
|
|
db.flush()
|
|
|
|
|
|
def move_item(db: Session, src: Playlist, dst: Playlist, video_id: str) -> None:
|
|
"""Move one video from `src` into `dst`, in the caller's one transaction.
|
|
|
|
The client-side shape — add to the target, then remove from the source — leaves the video in
|
|
BOTH lists whenever the second call fails. Idempotent: a video already in the target is simply
|
|
dropped from the source (moving into a list that has it is a removal, not a duplicate). Raises
|
|
`MoveRefused` for the same list twice, or a video a concurrent edit already took out."""
|
|
if src.id == dst.id:
|
|
raise MoveRefused("same_playlist")
|
|
# Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise.
|
|
lock_playlists(db, src.id, dst.id)
|
|
row = queries.item(db, src.id, video_id)
|
|
if row is None:
|
|
raise MoveRefused("not_in_playlist")
|
|
dst_changed = False
|
|
if queries.item(db, dst.id, video_id) is None:
|
|
# Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this")
|
|
# survives the move, and there is no window where the video belongs to neither list. The
|
|
# check above holds: both lists are locked, and an insert into the target (another tab's
|
|
# add, a YouTube merge) waits for the lock — its foreign key needs the target row.
|
|
row.playlist_id = dst.id
|
|
row.position = next_item_position(db, dst.id)
|
|
dst_changed = True
|
|
if not dst_changed:
|
|
db.delete(row)
|
|
# The source always changed; the target only when the row landed in it. Touching a target
|
|
# that already had the video promoted an untouched list in the rail's "recently changed" sort.
|
|
recompute_dirty(db, src)
|
|
touch(src)
|
|
if dst_changed:
|
|
recompute_dirty(db, dst)
|
|
touch(dst)
|
|
db.flush()
|
|
|
|
|
|
def reorder_items(db: Session, pl: Playlist, order: list[str]) -> int:
|
|
"""Number the listed videos 0..n in the listed order. Items the list left out (added in another
|
|
tab meanwhile) keep their relative order after them, so no two items share a position. The
|
|
caller holds the playlist's lock. Returns how many items the list has."""
|
|
by_video = {it.video_id: it for it in queries.items(db, pl.id)}
|
|
pos = 0
|
|
seen: set[str] = set()
|
|
for vid in order:
|
|
it = by_video.get(vid)
|
|
if it is not None:
|
|
it.position = pos
|
|
pos += 1
|
|
seen.add(vid)
|
|
for it in sorted(
|
|
(it for vid, it in by_video.items() if vid not in seen), key=lambda i: i.position
|
|
):
|
|
it.position = pos
|
|
pos += 1
|
|
recompute_dirty(db, pl)
|
|
touch(pl)
|
|
db.flush()
|
|
return pos
|
|
|
|
|
|
def attach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None:
|
|
"""The user's own label on a channel (idempotent). The caller has checked the tag is theirs."""
|
|
if queries.user_tag_link(db, user, channel_id, tag_id) is None:
|
|
db.add(ChannelTag(channel_id=channel_id, tag_id=tag_id, user_id=user.id, source="user"))
|
|
db.flush()
|
|
|
|
|
|
def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None:
|
|
row = queries.user_tag_link(db, user, channel_id, tag_id)
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.flush()
|
|
|
|
|
|
# --- Google Takeout --------------------------------------------------------------------------
|
|
|
|
_TAKEOUT_BATCH = 1000
|
|
# The advisory-lock namespace of one user's Takeout rows ("TO"), so a key cannot meet another lock's.
|
|
_TAKEOUT_LOCK_NS = 0x544F
|
|
|
|
|
|
def lock_takeout(db: Session, user: User) -> None:
|
|
"""Hold the user's Takeout rows for the rest of the transaction. An upload's staging and an
|
|
import's settling each lock many of the same rows, one statement after another and in no common
|
|
order, so without one lock over both the two can deadlock. An un-mark takes it too: the import
|
|
reads the un-marks before it marks, and one landing between the two would be overwritten."""
|
|
db.execute(select(func.pg_advisory_xact_lock(_TAKEOUT_LOCK_NS, user.id)))
|
|
|
|
|
|
def stage_takeout(
|
|
db: Session, user: User, kind: TakeoutKind, rows: dict[str, tuple[datetime, str | None]]
|
|
) -> None:
|
|
"""Keep the videos of one Takeout file, `{video id: (time, channel id)}`, for the import job.
|
|
A second upload of the same export, or a newer one, adds what is new: a video already kept keeps
|
|
the later time, and a row the import has settled keeps its outcome."""
|
|
lock_takeout(db, user)
|
|
values = [
|
|
{"user_id": user.id, "kind": kind, "video_id": vid, "at": at, "channel_id": channel}
|
|
for vid, (at, channel) in rows.items()
|
|
]
|
|
for start in range(0, len(values), _TAKEOUT_BATCH):
|
|
stmt = pg_insert(TakeoutItem).values(values[start : start + _TAKEOUT_BATCH])
|
|
db.execute(
|
|
stmt.on_conflict_do_update(
|
|
constraint="uq_takeout_item",
|
|
set_={
|
|
"at": func.greatest(TakeoutItem.at, stmt.excluded.at),
|
|
"channel_id": func.coalesce(stmt.excluded.channel_id, TakeoutItem.channel_id),
|
|
},
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
|
|
def _settle(db: Session, ids: list[int], outcome: TakeoutOutcome) -> None:
|
|
for start in range(0, len(ids), _TAKEOUT_BATCH):
|
|
db.execute(
|
|
update(TakeoutItem)
|
|
.where(TakeoutItem.id.in_(ids[start : start + _TAKEOUT_BATCH]))
|
|
.values(outcome=outcome)
|
|
)
|
|
|
|
|
|
def apply_takeout_watched(db: Session, user: User) -> dict[str, int]:
|
|
"""Mark watched what the user watched on YouTube, for the videos the catalog knows and has probed
|
|
for Shorts; the rest wait. `watched_at` is the Takeout time. A Short is not marked (swiping one
|
|
past is not watching it), and a state the user set here wins: watched or hidden, anything
|
|
changed after the Takeout time, or an un-mark since then (`VideoUnmark`). The upsert's own
|
|
condition is that rule, so a state set while the import runs wins too."""
|
|
lock_takeout(db, user)
|
|
short: list[int] = []
|
|
todo: dict[str, tuple[int, datetime]] = {}
|
|
ready = queries.takeout_watched_ready(db, user)
|
|
unmarked = queries.unmarked_at(db, user, [video_id for _, video_id, _, _ in ready])
|
|
kept: list[int] = []
|
|
for row_id, video_id, at, is_short in ready:
|
|
if is_short:
|
|
short.append(row_id)
|
|
elif video_id in unmarked and unmarked[video_id] >= at:
|
|
# Un-marked here since they watched it there: theirs, as a state they set is.
|
|
kept.append(row_id)
|
|
else:
|
|
todo[video_id] = (row_id, at)
|
|
marked: set[str] = set()
|
|
items = list(todo.items())
|
|
now = datetime.now(timezone.utc)
|
|
for start in range(0, len(items), _TAKEOUT_BATCH):
|
|
stmt = pg_insert(VideoState).values(
|
|
[
|
|
{"user_id": user.id, "video_id": vid, "status": "watched", "watched_at": at}
|
|
for vid, (_, at) in items[start : start + _TAKEOUT_BATCH]
|
|
]
|
|
)
|
|
marked.update(
|
|
db.execute(
|
|
stmt.on_conflict_do_update(
|
|
constraint="uq_user_video",
|
|
set_={
|
|
"status": "watched",
|
|
"watched_at": stmt.excluded.watched_at,
|
|
"updated_at": now,
|
|
},
|
|
where=(VideoState.status == "new")
|
|
& (VideoState.updated_at < stmt.excluded.watched_at),
|
|
).returning(VideoState.video_id)
|
|
).scalars()
|
|
)
|
|
kept += [row_id for vid, (row_id, _) in items if vid not in marked]
|
|
_settle(db, [todo[vid][0] for vid in marked], TakeoutOutcome.WATCHED)
|
|
_settle(db, kept, TakeoutOutcome.KEPT)
|
|
_settle(db, short, TakeoutOutcome.SHORT)
|
|
db.flush()
|
|
return {"watched": len(marked), "kept": len(kept), "shorts": len(short)}
|
|
|
|
|
|
def apply_takeout_watch_later(
|
|
db: Session, user: User, rows: list[tuple[int, str]], asked: set[str]
|
|
) -> dict[str, int]:
|
|
"""Put the videos of the user's waiting Watch later `rows`, (row id, video id) oldest first, into
|
|
their list here, after what it already holds. `asked` is the videos YouTube was just asked for: one
|
|
of those the catalog still lacks is gone from YouTube; any other unknown one keeps waiting."""
|
|
if not rows:
|
|
return {}
|
|
lock_takeout(db, user)
|
|
ids = [vid for _, vid in rows]
|
|
known = queries.known_video_ids(db, ids)
|
|
pl = watch_later(db, user)
|
|
# What add_item does a video at a time, for the whole list at once: one presence read and one
|
|
# insert, not four statements a video (a full list is thousands). Locked before the read, as there.
|
|
next_position = next_item_position(db, pl.id)
|
|
held = queries.in_playlist(db, pl.id, [vid for vid in ids if vid in known])
|
|
done: dict[TakeoutOutcome, list[int]] = {o: [] for o in TakeoutOutcome}
|
|
for row_id, vid in rows:
|
|
if vid not in known:
|
|
if vid in asked:
|
|
done[TakeoutOutcome.MISSING].append(row_id)
|
|
elif vid in held:
|
|
done[TakeoutOutcome.PRESENT].append(row_id)
|
|
else:
|
|
db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=next_position))
|
|
next_position += 1
|
|
held.add(vid)
|
|
done[TakeoutOutcome.ADDED].append(row_id)
|
|
if done[TakeoutOutcome.ADDED]:
|
|
touch(pl)
|
|
for outcome, ids in done.items():
|
|
_settle(db, ids, outcome)
|
|
db.flush()
|
|
return {
|
|
"watch_later_added": len(done[TakeoutOutcome.ADDED]),
|
|
"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()
|