trav
d1db60c057
Cut (Ctrl-X) marks the selection on the clipboard with a `cut` flag; a paste of a cut from a real playlist is a *move* (single-undo across playlists via move_tracks_between_playlists; same-playlist reorders), while a copy — or a cut from the library, which has nothing to remove — just adds. Paste now inserts *above the selected track* rather than appending. Cut/Paste added to the track context menu; parse_tracks_mime now tolerates a None/empty clipboard.
468 lines
18 KiB
Python
468 lines
18 KiB
Python
import secrets
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtCore import QObject, QTimer, pyqtSignal
|
|
|
|
from lintunes import tagging
|
|
from lintunes.models import Library, Playlist, PlaylistType
|
|
from lintunes.storage import json_storage
|
|
from lintunes.undo import Command, UndoStack
|
|
|
|
|
|
SAVE_DEBOUNCE_MS = 3000
|
|
|
|
|
|
class LibraryManager(QObject):
|
|
"""Owns the Library, applies mutations, and persists them (debounced).
|
|
|
|
Dirty state is tracked per area so a play-count bump only rewrites
|
|
library.json and a playlist edit only rewrites that playlist's file.
|
|
|
|
User-driven structural edits (playlist create/delete/rename/move, track
|
|
add/remove/reorder, metadata edits) are recorded on ``undo_stack`` so they
|
|
can be reversed with Ctrl+Z. The internal ``_apply_*``/``_set_*`` helpers do
|
|
the actual mutation (plus dirty-marking and the existing refresh signals)
|
|
and never touch the stack, so undo/redo reuse them without recursing.
|
|
"""
|
|
|
|
playlists_changed = pyqtSignal() # structure: add/remove/rename/move
|
|
playlist_content_changed = pyqtSignal(str) # persistent_id
|
|
track_updated = pyqtSignal(int) # track_id
|
|
track_fields_edited = pyqtSignal(int, list) # track_id, changed field names
|
|
|
|
def __init__(self, library: Library, data_dir: Path, parent=None):
|
|
super().__init__(parent)
|
|
self.library = library
|
|
self.data_dir = Path(data_dir)
|
|
self.undo_stack = UndoStack(parent=self)
|
|
|
|
self._dirty_tracks = False
|
|
self._dirty_metadata = False
|
|
self._dirty_playlists: set[str] = set()
|
|
self._deleted_playlists: set[str] = set()
|
|
|
|
self._save_timer = QTimer(self)
|
|
self._save_timer.setSingleShot(True)
|
|
self._save_timer.setInterval(SAVE_DEBOUNCE_MS)
|
|
self._save_timer.timeout.connect(self.flush)
|
|
|
|
# ---- id generation ----
|
|
|
|
def new_track_id(self) -> int:
|
|
return max(self.library.tracks.keys(), default=0) + 1
|
|
|
|
def new_persistent_id(self) -> str:
|
|
while True:
|
|
pid = secrets.token_hex(8).upper()
|
|
if pid not in self.library.playlists:
|
|
return pid
|
|
|
|
# ---- playlist structure ----
|
|
|
|
def create_playlist(self, name: str, parent_pid: str = "") -> Playlist:
|
|
playlist = Playlist(
|
|
name=name,
|
|
persistent_id=self.new_persistent_id(),
|
|
parent_persistent_id=parent_pid,
|
|
playlist_type=PlaylistType.REGULAR,
|
|
)
|
|
self._insert_playlists([playlist])
|
|
self.undo_stack.push(Command(
|
|
"New Playlist",
|
|
undo=lambda: self._remove_playlists([playlist]),
|
|
redo=lambda: self._insert_playlists([playlist])))
|
|
return playlist
|
|
|
|
def create_folder(self, name: str, parent_pid: str = "") -> Playlist:
|
|
folder = Playlist(
|
|
name=name,
|
|
persistent_id=self.new_persistent_id(),
|
|
parent_persistent_id=parent_pid,
|
|
playlist_type=PlaylistType.FOLDER,
|
|
)
|
|
self._insert_playlists([folder])
|
|
self.undo_stack.push(Command(
|
|
"New Folder",
|
|
undo=lambda: self._remove_playlists([folder]),
|
|
redo=lambda: self._insert_playlists([folder])))
|
|
return folder
|
|
|
|
def rename_playlist(self, pid: str, name: str):
|
|
playlist = self.library.playlists.get(pid)
|
|
if playlist and name and playlist.name != name:
|
|
old = playlist.name
|
|
self._apply_rename(pid, name)
|
|
self.undo_stack.push(Command(
|
|
"Rename Playlist",
|
|
undo=lambda: self._apply_rename(pid, old),
|
|
redo=lambda: self._apply_rename(pid, name)))
|
|
|
|
def delete_playlist(self, pid: str):
|
|
"""Delete a playlist, or a folder and everything inside it."""
|
|
playlist = self.library.playlists.get(pid)
|
|
if not playlist:
|
|
return
|
|
doomed_ids = [pid]
|
|
if playlist.playlist_type == PlaylistType.FOLDER:
|
|
doomed_ids.extend(self._descendant_ids(pid))
|
|
doomed = [self.library.playlists[d] for d in doomed_ids
|
|
if d in self.library.playlists]
|
|
self._remove_playlists(doomed)
|
|
self.undo_stack.push(Command(
|
|
"Delete Playlist",
|
|
undo=lambda: self._insert_playlists(doomed),
|
|
redo=lambda: self._remove_playlists(doomed)))
|
|
|
|
def move_playlist(self, pid: str, new_parent_pid: str):
|
|
playlist = self.library.playlists.get(pid)
|
|
if not playlist or pid == new_parent_pid:
|
|
return
|
|
if new_parent_pid and new_parent_pid in self._descendant_ids(pid):
|
|
return # no cycles
|
|
old_parent = playlist.parent_persistent_id
|
|
if old_parent == new_parent_pid:
|
|
return
|
|
self._apply_reparent(pid, new_parent_pid)
|
|
self.undo_stack.push(Command(
|
|
"Move Playlist",
|
|
undo=lambda: self._apply_reparent(pid, old_parent),
|
|
redo=lambda: self._apply_reparent(pid, new_parent_pid)))
|
|
|
|
def playlists_containing(self, track_id: int) -> list[tuple[str, str]]:
|
|
"""(persistent_id, name) of every REGULAR playlist holding this track,
|
|
sorted by name. Folders/smart/system playlists are excluded — they
|
|
aren't jump-to targets with static membership."""
|
|
out = [(p.persistent_id, p.name)
|
|
for p in self.library.playlists.values()
|
|
if p.playlist_type == PlaylistType.REGULAR
|
|
and track_id in p.track_ids]
|
|
return sorted(out, key=lambda pn: pn[1].casefold())
|
|
|
|
def _descendant_ids(self, pid: str) -> list[str]:
|
|
result = []
|
|
stack = [pid]
|
|
while stack:
|
|
current = stack.pop()
|
|
for p in self.library.playlists.values():
|
|
if p.parent_persistent_id == current:
|
|
result.append(p.persistent_id)
|
|
stack.append(p.persistent_id)
|
|
return result
|
|
|
|
# structure apply helpers (no undo push) --------------------------------
|
|
|
|
def _insert_playlists(self, playlists: list[Playlist]):
|
|
for playlist in playlists:
|
|
self.library.playlists[playlist.persistent_id] = playlist
|
|
self._dirty_playlists.add(playlist.persistent_id)
|
|
self._deleted_playlists.discard(playlist.persistent_id)
|
|
self._dirty_metadata = True
|
|
self._schedule_save()
|
|
self.playlists_changed.emit()
|
|
|
|
def _remove_playlists(self, playlists: list[Playlist]):
|
|
for playlist in playlists:
|
|
pid = playlist.persistent_id
|
|
self.library.playlists.pop(pid, None)
|
|
self._dirty_playlists.discard(pid)
|
|
self._deleted_playlists.add(pid)
|
|
self._dirty_metadata = True
|
|
self._schedule_save()
|
|
self.playlists_changed.emit()
|
|
|
|
def _apply_rename(self, pid: str, name: str):
|
|
playlist = self.library.playlists.get(pid)
|
|
if playlist:
|
|
playlist.name = name
|
|
self._mark_playlist(pid)
|
|
self.playlists_changed.emit()
|
|
|
|
def _apply_reparent(self, pid: str, parent_pid: str):
|
|
playlist = self.library.playlists.get(pid)
|
|
if playlist:
|
|
playlist.parent_persistent_id = parent_pid
|
|
self._mark_playlist(pid)
|
|
self.playlists_changed.emit()
|
|
|
|
# ---- playlist contents ----
|
|
|
|
def add_tracks_to_playlist(self, pid: str, track_ids: list[int],
|
|
position: int | None = None):
|
|
playlist = self.library.playlists.get(pid)
|
|
if not playlist or playlist.playlist_type == PlaylistType.FOLDER:
|
|
return
|
|
track_ids = [tid for tid in track_ids if tid in self.library.tracks]
|
|
if not track_ids:
|
|
return
|
|
before = list(playlist.track_ids)
|
|
after = list(before)
|
|
if position is None or position >= len(after):
|
|
after.extend(track_ids)
|
|
else:
|
|
after[position:position] = track_ids
|
|
self._set_track_ids(pid, after)
|
|
self._push_track_ids("Add to Playlist", pid, before, after)
|
|
|
|
def remove_tracks_from_playlist(self, pid: str, rows: list[int]):
|
|
playlist = self.library.playlists.get(pid)
|
|
if not playlist:
|
|
return
|
|
before = list(playlist.track_ids)
|
|
after = list(before)
|
|
for row in sorted(set(rows), reverse=True):
|
|
if 0 <= row < len(after):
|
|
del after[row]
|
|
self._set_track_ids(pid, after)
|
|
self._push_track_ids("Remove from Playlist", pid, before, after)
|
|
|
|
def move_tracks_in_playlist(self, pid: str, rows: list[int], dest: int):
|
|
"""Move the tracks at the given manual-order rows so the block starts at dest."""
|
|
playlist = self.library.playlists.get(pid)
|
|
if not playlist:
|
|
return
|
|
before = list(playlist.track_ids)
|
|
rows = sorted(set(r for r in rows if 0 <= r < len(before)))
|
|
if not rows:
|
|
return
|
|
moving = [before[r] for r in rows]
|
|
# Destination index counted in the list *after* removal
|
|
dest -= sum(1 for r in rows if r < dest)
|
|
remaining = [tid for i, tid in enumerate(before) if i not in rows]
|
|
dest = max(0, min(dest, len(remaining)))
|
|
after = remaining[:dest] + moving + remaining[dest:]
|
|
self._set_track_ids(pid, after)
|
|
self._push_track_ids("Reorder Playlist", pid, before, after)
|
|
|
|
def move_tracks_between_playlists(self, src_pid: str, src_rows: list[int],
|
|
dest_pid: str, dest_position: int | None = None):
|
|
"""Cut tracks out of one playlist and insert them into another as a
|
|
single undoable move (Ctrl-X / Ctrl-V across playlists). For the same
|
|
playlist, use move_tracks_in_playlist instead."""
|
|
src = self.library.playlists.get(src_pid)
|
|
dest = self.library.playlists.get(dest_pid)
|
|
if (not src or not dest or src_pid == dest_pid
|
|
or dest.playlist_type == PlaylistType.FOLDER):
|
|
return
|
|
src_before = list(src.track_ids)
|
|
dest_before = list(dest.track_ids)
|
|
rows = sorted(set(r for r in src_rows if 0 <= r < len(src_before)))
|
|
if not rows:
|
|
return
|
|
moving = [src_before[r] for r in rows]
|
|
src_after = [tid for i, tid in enumerate(src_before) if i not in rows]
|
|
dest_after = list(dest_before)
|
|
if dest_position is None or dest_position >= len(dest_after):
|
|
dest_after.extend(moving)
|
|
else:
|
|
dest_after[dest_position:dest_position] = moving
|
|
|
|
def apply(src_ids, dest_ids):
|
|
self._set_track_ids(src_pid, src_ids)
|
|
self._set_track_ids(dest_pid, dest_ids)
|
|
|
|
apply(src_after, dest_after)
|
|
self.undo_stack.push(Command(
|
|
"Move to Playlist",
|
|
undo=lambda: apply(src_before, dest_before),
|
|
redo=lambda: apply(src_after, dest_after)))
|
|
|
|
def _set_track_ids(self, pid: str, ids: list[int]):
|
|
playlist = self.library.playlists.get(pid)
|
|
if playlist is None:
|
|
return
|
|
playlist.track_ids = list(ids)
|
|
self._mark_playlist(pid)
|
|
self.playlist_content_changed.emit(pid)
|
|
|
|
def _push_track_ids(self, label, pid, before, after):
|
|
if before != after:
|
|
self.undo_stack.push(Command(
|
|
label,
|
|
undo=lambda: self._set_track_ids(pid, before),
|
|
redo=lambda: self._set_track_ids(pid, after)))
|
|
|
|
# ---- tracks ----
|
|
|
|
def add_track(self, track) -> int:
|
|
if not track.track_id or track.track_id in self.library.tracks:
|
|
track.track_id = self.new_track_id()
|
|
if not track.persistent_id:
|
|
track.persistent_id = secrets.token_hex(8).upper()
|
|
self.library.tracks[track.track_id] = track
|
|
self._dirty_tracks = True
|
|
self._dirty_metadata = True
|
|
self._schedule_save()
|
|
return track.track_id
|
|
|
|
def update_track_fields(self, track_id: int, fields: dict):
|
|
"""In-memory-only field update (no file write, not undoable)."""
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
return
|
|
changed_keys = []
|
|
for key, value in fields.items():
|
|
if hasattr(track, key) and getattr(track, key) != value:
|
|
setattr(track, key, value)
|
|
changed_keys.append(key)
|
|
if changed_keys:
|
|
track.date_modified = _utc_now_iso()
|
|
self._dirty_tracks = True
|
|
self._schedule_save()
|
|
self.track_updated.emit(track_id)
|
|
self.track_fields_edited.emit(track_id, changed_keys)
|
|
|
|
def set_track_location(self, track_id: int, location: str):
|
|
"""Repoint a track at a new on-disk path (e.g. user relocated a file
|
|
that was moved/renamed outside lintunes). Not a content edit: no tag
|
|
write, no date_modified bump, not undoable."""
|
|
track = self.library.tracks.get(track_id)
|
|
if not track or track.location == location:
|
|
return
|
|
track.location = location
|
|
self._dirty_tracks = True
|
|
self._schedule_save()
|
|
self.track_updated.emit(track_id)
|
|
|
|
def edit_track_fields(self, track_id: int, fields: dict):
|
|
"""Edit a track's metadata: write the changed tags to the audio file,
|
|
update the in-memory library, and record the change for undo.
|
|
|
|
Reverts text/number fields only — artwork is written separately by the
|
|
Info dialog and is not part of the undo history.
|
|
"""
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
return
|
|
changed = {k: v for k, v in fields.items()
|
|
if hasattr(track, k) and getattr(track, k) != v}
|
|
if not changed:
|
|
return
|
|
old = {k: getattr(track, k) for k in changed}
|
|
if not self._write_track_tags(track, changed):
|
|
# File write failed: leave the library untouched so memory and file
|
|
# don't diverge, and don't record an un-undoable phantom edit.
|
|
return
|
|
self._apply_track_fields(track_id, changed)
|
|
self.undo_stack.push(Command(
|
|
"Edit Info",
|
|
undo=lambda: self._revert_track_fields(track_id, old),
|
|
redo=lambda: self._revert_track_fields(track_id, changed)))
|
|
|
|
def edit_tracks_fields(self, track_ids: list[int], fields: dict):
|
|
"""Apply the same field edits to many tracks at once, writing each
|
|
file's tags, recorded as ONE undoable command.
|
|
|
|
Per track only fields that genuinely differ are written, so a value a
|
|
track already has costs nothing and a track that ends up unchanged is
|
|
skipped entirely (no file write, no undo entry).
|
|
"""
|
|
changes = [] # (track_id, new_fields, old_fields)
|
|
for track_id in track_ids:
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
continue
|
|
changed = {k: v for k, v in fields.items()
|
|
if hasattr(track, k) and getattr(track, k) != v}
|
|
if not changed or not self._write_track_tags(track, changed):
|
|
continue
|
|
old = {k: getattr(track, k) for k in changed}
|
|
self._apply_track_fields(track_id, changed)
|
|
changes.append((track_id, changed, old))
|
|
if not changes:
|
|
return
|
|
self.undo_stack.push(Command(
|
|
"Edit Info",
|
|
undo=lambda: [self._revert_track_fields(tid, old)
|
|
for tid, _new, old in changes],
|
|
redo=lambda: [self._revert_track_fields(tid, new)
|
|
for tid, new, _old in changes]))
|
|
|
|
def _write_track_tags(self, track, field_map: dict) -> bool:
|
|
if not track.location:
|
|
return True
|
|
try:
|
|
tagging.write_tags(track.location, field_map)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Could not write tags to {track.location}: {e}")
|
|
return False
|
|
|
|
def _apply_track_fields(self, track_id: int, field_map: dict):
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
return
|
|
for key, value in field_map.items():
|
|
setattr(track, key, value)
|
|
track.date_modified = _utc_now_iso()
|
|
self._dirty_tracks = True
|
|
self._schedule_save()
|
|
self.track_updated.emit(track_id)
|
|
self.track_fields_edited.emit(track_id, list(field_map.keys()))
|
|
|
|
def _revert_track_fields(self, track_id: int, field_map: dict):
|
|
track = self.library.tracks.get(track_id)
|
|
if track:
|
|
self._write_track_tags(track, field_map) # best-effort
|
|
self._apply_track_fields(track_id, field_map)
|
|
|
|
def record_play(self, track_id: int):
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
return
|
|
track.play_count += 1
|
|
track.play_date_utc = _utc_now_iso()
|
|
self._dirty_tracks = True
|
|
self._schedule_save()
|
|
self.track_updated.emit(track_id)
|
|
|
|
def record_skip(self, track_id: int):
|
|
track = self.library.tracks.get(track_id)
|
|
if not track:
|
|
return
|
|
track.skip_count += 1
|
|
track.skip_date = _utc_now_iso()
|
|
self._dirty_tracks = True
|
|
self._schedule_save()
|
|
self.track_updated.emit(track_id)
|
|
|
|
# ---- settings persistence (no signals; UI-originated) ----
|
|
|
|
def mark_playlist_settings_dirty(self, pid: str):
|
|
self._mark_playlist(pid)
|
|
|
|
def mark_library_settings_dirty(self):
|
|
self._dirty_metadata = True
|
|
self._schedule_save()
|
|
|
|
# ---- saving ----
|
|
|
|
def _mark_playlist(self, pid: str):
|
|
self._dirty_playlists.add(pid)
|
|
self._dirty_metadata = True
|
|
self._schedule_save()
|
|
|
|
def _schedule_save(self):
|
|
self._save_timer.start()
|
|
|
|
def flush(self):
|
|
self._save_timer.stop()
|
|
if self._dirty_tracks:
|
|
json_storage.save_tracks(self.library, self.data_dir)
|
|
self._dirty_tracks = False
|
|
for pid in list(self._dirty_playlists):
|
|
playlist = self.library.playlists.get(pid)
|
|
if playlist:
|
|
json_storage.save_playlist(playlist, self.data_dir)
|
|
self._dirty_playlists.clear()
|
|
for pid in list(self._deleted_playlists):
|
|
json_storage.delete_playlist_file(pid, self.data_dir)
|
|
self._deleted_playlists.clear()
|
|
if self._dirty_metadata:
|
|
json_storage.save_metadata(self.library, self.data_dir)
|
|
self._dirty_metadata = False
|
|
|
|
|
|
def _utc_now_iso() -> str:
|
|
return datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
|