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.
665 lines
26 KiB
Python
665 lines
26 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 sqlalchemy import 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,
|
|
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]),
|
|
}
|