Add smart playlists with full iTunes 12 import (Phase 1)

Auto-populating, criteria-driven playlists that import faithfully from iTunes.

- lintunes/smart.py: recursive SmartCriteria/SmartGroup/SmartRule/SmartLimit
  model (JSON round-tripping), a shared FIELD_REGISTRY used by both the
  evaluator and the editor, and evaluate() (match all/any, nested groups,
  string/int/duration/rating/date/bool ops, "in the last N", limit by
  items/time/size). Null play/skip dates are treated as the distant past,
  matching iTunes.
- iTunes import via a vendored MIT-licensed binary parser
  (lintunes/itunes_smart/, from cvzi/itunes_smartplaylist). Nested groups
  parse and evaluate; blobs we can't represent (MediaKind/iCloud/etc.) flag
  unsupported and keep the imported snapshot. "loved" is dropped per user pref.
- library_manager: create/set/recompute smart playlists (undoable), field-scoped
  coalesced live recompute hooked into the edit/play/skip/add funnels, a no-op
  equality guard to avoid Syncthing churn, and manual-edit guards. main.py
  recomputes on load; conflict_resolver keeps newest criteria for smart lists.
- GUI: ❧ glyph painted in the sidebar branch column, read-only track table for
  smart playlists, New/Edit Smart Playlist menus, and SmartPlaylistEditorDialog
  (per-field rule rows, match all/any, limits, live updating).

Tests: tests/test_round14.py (real captured blobs in tests/smart_blobs.json).
This commit is contained in:
2026-07-01 16:43:39 -04:00
parent 1274e12aeb
commit 2917a42b75
18 changed files with 2856 additions and 30 deletions
+135 -4
View File
@@ -4,7 +4,7 @@ from pathlib import Path
from PyQt6.QtCore import QObject, QTimer, pyqtSignal
from lintunes import tagging
from lintunes import smart, tagging
from lintunes.models import Library, Playlist, PlaylistType
from lintunes.storage import json_storage
from lintunes.undo import Command, UndoStack
@@ -47,6 +47,17 @@ class LibraryManager(QObject):
self._save_timer.setInterval(SAVE_DEBOUNCE_MS)
self._save_timer.timeout.connect(self.flush)
# Smart-playlist recompute is coalesced through a 0 ms timer so a burst
# of track edits / play bumps triggers a single pass. ``_recomputing``
# guards against re-entrancy; ``_pending_recompute_fields`` is the union
# of changed Track attributes (None means "recompute everything").
self._recomputing = False
self._pending_recompute_fields: set[str] | None = set()
self._recompute_timer = QTimer(self)
self._recompute_timer.setSingleShot(True)
self._recompute_timer.setInterval(0)
self._recompute_timer.timeout.connect(self._flush_recompute)
# ---- id generation ----
def new_track_id(self) -> int:
@@ -88,6 +99,42 @@ class LibraryManager(QObject):
redo=lambda: self._insert_playlists([folder])))
return folder
def create_smart_playlist(self, name: str, criteria, parent_pid: str = "") -> Playlist:
playlist = Playlist(
name=name,
persistent_id=self.new_persistent_id(),
parent_persistent_id=parent_pid,
playlist_type=PlaylistType.SMART,
smart_criteria=criteria,
)
self._insert_playlists([playlist])
self.recompute_smart_playlist(playlist.persistent_id, force=True)
self.undo_stack.push(Command(
"New Smart Playlist",
undo=lambda: self._remove_playlists([playlist]),
redo=lambda: self._insert_playlists([playlist])))
return playlist
def set_smart_criteria(self, pid: str, criteria):
playlist = self.library.playlists.get(pid)
if not playlist or not playlist.is_smart:
return
old = playlist.smart_criteria
self._apply_smart_criteria(pid, criteria)
self.undo_stack.push(Command(
"Edit Smart Playlist",
undo=lambda: self._apply_smart_criteria(pid, old),
redo=lambda: self._apply_smart_criteria(pid, criteria)))
def _apply_smart_criteria(self, pid: str, criteria):
playlist = self.library.playlists.get(pid)
if not playlist:
return
playlist.smart_criteria = criteria
self._mark_playlist(pid)
self.recompute_smart_playlist(pid, force=True)
self.playlists_changed.emit()
def rename_playlist(self, pid: str, name: str):
playlist = self.library.playlists.get(pid)
if playlist and name and playlist.name != name:
@@ -190,7 +237,8 @@ class LibraryManager(QObject):
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:
if not playlist or playlist.playlist_type == PlaylistType.FOLDER \
or playlist.is_smart:
return
track_ids = [tid for tid in track_ids if tid in self.library.tracks]
if not track_ids:
@@ -206,7 +254,7 @@ class LibraryManager(QObject):
def remove_tracks_from_playlist(self, pid: str, rows: list[int]):
playlist = self.library.playlists.get(pid)
if not playlist:
if not playlist or playlist.is_smart:
return
before = list(playlist.track_ids)
after = list(before)
@@ -219,7 +267,7 @@ class LibraryManager(QObject):
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:
if not playlist or playlist.is_smart:
return
before = list(playlist.track_ids)
rows = sorted(set(r for r in rows if 0 <= r < len(before)))
@@ -260,6 +308,7 @@ class LibraryManager(QObject):
self._dirty_tracks = True
self._dirty_metadata = True
self._schedule_save()
self._touch_smart(None)
return track.track_id
def update_track_fields(self, track_id: int, fields: dict):
@@ -278,6 +327,7 @@ class LibraryManager(QObject):
self._schedule_save()
self.track_updated.emit(track_id)
self.track_fields_edited.emit(track_id, changed_keys)
self._touch_smart(set(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
@@ -366,6 +416,7 @@ class LibraryManager(QObject):
self._schedule_save()
self.track_updated.emit(track_id)
self.track_fields_edited.emit(track_id, list(field_map.keys()))
self._touch_smart(set(field_map.keys()))
def _revert_track_fields(self, track_id: int, field_map: dict):
track = self.library.tracks.get(track_id)
@@ -382,6 +433,7 @@ class LibraryManager(QObject):
self._dirty_tracks = True
self._schedule_save()
self.track_updated.emit(track_id)
self._touch_smart({"play_count", "play_date_utc"})
def record_skip(self, track_id: int):
track = self.library.tracks.get(track_id)
@@ -392,6 +444,85 @@ class LibraryManager(QObject):
self._dirty_tracks = True
self._schedule_save()
self.track_updated.emit(track_id)
self._touch_smart({"skip_count", "skip_date"})
# ---- smart playlist recompute ----
def recompute_smart_playlist(self, pid: str, force: bool = False) -> bool:
"""Recompute a smart playlist's membership from its criteria. Returns
True if the membership actually changed. A no-op recompute does not mark
the file dirty or emit (keeps startup recompute from churning every
smart playlist file, which Syncthing would propagate)."""
playlist = self.library.playlists.get(pid)
if not playlist or not playlist.is_smart:
return False
criteria = playlist.smart_criteria
if criteria is None or criteria.unsupported:
return False # keep the imported snapshot
if not force and not criteria.live_update:
return False
try:
seed = int(playlist.persistent_id, 16)
except ValueError:
seed = 0
ids = smart.evaluate(criteria, self.library.tracks.values(), seed=seed)
if ids == playlist.track_ids:
return False
self._set_track_ids(pid, ids) # marks dirty + emits; never undoable
return True
def recompute_all_smart(self, force: bool = False) -> list[str]:
"""Recompute every smart playlist (used on library load). Respects each
playlist's live_update flag unless force=True."""
changed = []
self._recomputing = True
try:
for pid in list(self.library.playlists):
if self.recompute_smart_playlist(pid, force=force):
changed.append(pid)
finally:
self._recomputing = False
return changed
def _touch_smart(self, fields: set[str] | None):
"""Schedule a coalesced recompute after a track change. ``fields`` is the
set of changed Track attribute names (None = structural/everything)."""
if self._recomputing:
return
if not any(p.is_smart for p in self.library.playlists.values()):
return
if fields is None or self._pending_recompute_fields is None:
self._pending_recompute_fields = None
else:
self._pending_recompute_fields |= fields
self._recompute_timer.start()
def _flush_recompute(self):
fields = self._pending_recompute_fields
self._pending_recompute_fields = set()
self._recomputing = True
try:
for pid, playlist in list(self.library.playlists.items()):
if not playlist.is_smart:
continue
criteria = playlist.smart_criteria
if criteria is None or criteria.unsupported or not criteria.live_update:
continue
if fields is not None and not self._criteria_touched(criteria, fields):
continue
self.recompute_smart_playlist(pid)
finally:
self._recomputing = False
@staticmethod
def _criteria_touched(criteria, fields: set[str]) -> bool:
if criteria.referenced_attrs() & fields:
return True
if criteria.limit.enabled:
attr, _ = smart.SELECTION_SORT.get(criteria.limit.selection, (None, False))
if attr and attr in fields:
return True
return False
# ---- settings persistence (no signals; UI-originated) ----