peter
3a5212cd8a
The first checkpoint of an item and the final flush beside it both read no state and both inserted; the loser failed on uq_plex_user_item. Two first saves each made a Watch later list. Both are created by an ON CONFLICT DO NOTHING insert and then read; a partial unique index holds one Watch later per user, and the migration folds an existing second list into the first. Plex keeps one per-item state map (rollup.item_state_map).
489 lines
18 KiB
Python
489 lines
18 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, VideoState
|
|
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."""
|
|
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)
|
|
db.flush()
|
|
|
|
|
|
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."""
|
|
row = queries.video_state(db, user, video_id)
|
|
if row is not None:
|
|
db.delete(row)
|
|
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."""
|
|
row = queries.video_state(db, user, video_id)
|
|
now = datetime.now(timezone.utc)
|
|
if not position:
|
|
if row is None:
|
|
return
|
|
if row.status == "new":
|
|
db.delete(row)
|
|
else:
|
|
row.position_seconds = 0
|
|
row.progress_updated_at = now
|
|
db.flush()
|
|
return
|
|
if row is None:
|
|
row = VideoState(user_id=user.id, video_id=video_id)
|
|
db.add(row)
|
|
row.position_seconds = position
|
|
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()
|