From 7e8266c14c6689bf74a83ffe2518e081f5cc0f9e Mon Sep 17 00:00:00 2001 From: trav Date: Wed, 1 Jul 2026 16:43:39 -0400 Subject: [PATCH] Add smart playlists with full iTunes 12 import (Phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). Co-Authored-By: Claude Opus 4.8 --- lintunes/gui/main_window.py | 2 + lintunes/gui/playlist_view.py | 22 +- lintunes/gui/sidebar.py | 71 +- lintunes/gui/smart_playlist_dialog.py | 423 ++++++++++++ lintunes/gui/track_table.py | 23 +- lintunes/importers/itunes_importer.py | 23 +- lintunes/itunes_smart/LICENSE | 21 + lintunes/itunes_smart/__init__.py | 19 + lintunes/itunes_smart/data_structure.py | 237 +++++++ lintunes/itunes_smart/parse.py | 843 ++++++++++++++++++++++++ lintunes/library_manager.py | 139 +++- lintunes/main.py | 3 + lintunes/models/playlist.py | 15 + lintunes/smart.py | 678 +++++++++++++++++++ lintunes/storage/conflict_resolver.py | 7 + tests/smart_blobs.json | 22 + tests/test_itunes_importer.py | 11 +- tests/test_round14.py | 327 +++++++++ 18 files changed, 2856 insertions(+), 30 deletions(-) create mode 100644 lintunes/gui/smart_playlist_dialog.py create mode 100644 lintunes/itunes_smart/LICENSE create mode 100644 lintunes/itunes_smart/__init__.py create mode 100644 lintunes/itunes_smart/data_structure.py create mode 100644 lintunes/itunes_smart/parse.py create mode 100644 lintunes/smart.py create mode 100644 tests/smart_blobs.json create mode 100644 tests/test_round14.py diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 2438eb3..5eb1631 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -146,6 +146,8 @@ class MainWindow(QMainWindow): file_menu = bar.addMenu("&File") self._add_action(file_menu, "New Playlist", "Ctrl+N", lambda: self._sidebar.tree.create_playlist_interactive()) + self._add_action(file_menu, "New Smart Playlist", "Ctrl+Alt+N", + lambda: self._sidebar.tree.create_smart_playlist_interactive()) self._add_action(file_menu, "New Playlist Folder", "Ctrl+Shift+N", lambda: self._manager.create_folder("untitled folder")) file_menu.addSeparator() diff --git a/lintunes/gui/playlist_view.py b/lintunes/gui/playlist_view.py index bed7007..8928856 100644 --- a/lintunes/gui/playlist_view.py +++ b/lintunes/gui/playlist_view.py @@ -62,6 +62,8 @@ class PlaylistView(QWidget): def show_playlist(self, playlist): self._playlist = playlist self.table.set_source_playlist(playlist.persistent_id) + # Smart playlists are auto-populated: no manual add/remove/reorder. + self.table.set_content_editable(not playlist.is_smart) self.table.apply_settings(playlist.settings) self._reload_tracks() @@ -73,7 +75,8 @@ class PlaylistView(QWidget): if tid in self._manager.library.tracks] self.table.set_tracks(tracks) # Count/time now live in the bottom status bar, not next to the name. - self._name_label.setText(self._playlist.name) + prefix = "❧ " if self._playlist.is_smart else "" + self._name_label.setText(prefix + self._playlist.name) def _on_content_changed(self, pid): if self._playlist is not None and pid == self._playlist.persistent_id: @@ -114,35 +117,40 @@ class PlaylistView(QWidget): # ---- content edits ---- + def _editable(self) -> bool: + """Manual content edits are blocked for smart playlists (membership is + derived from criteria).""" + return self._playlist is not None and not self._playlist.is_smart + def _on_reorder(self, rows, dest): - if self._playlist is not None: + if self._editable(): self._manager.move_tracks_in_playlist( self._playlist.persistent_id, rows, dest) def _on_tracks_dropped(self, track_ids, drop_row): - if self._playlist is not None: + if self._editable(): add_tracks_with_dup_check( self._manager, self, self._playlist.persistent_id, track_ids, drop_row) def _on_files_dropped(self, paths, drop_row): - if self._playlist is not None: + if self._editable(): self.files_dropped.emit(paths, self._playlist.persistent_id) def _on_remove(self, rows): - if self._playlist is not None: + if self._editable(): self._manager.remove_tracks_from_playlist( self._playlist.persistent_id, rows) def _on_cut(self, rows): # Cut removes immediately (so the track visibly leaves); the ids are # already on the clipboard for a later paste to re-insert. - if self._playlist is not None: + if self._editable(): self._manager.remove_tracks_from_playlist( self._playlist.persistent_id, rows) def _on_paste(self): - if self._playlist is None: + if not self._editable(): return payload = parse_tracks_mime(QApplication.clipboard().mimeData()) if not payload: diff --git a/lintunes/gui/sidebar.py b/lintunes/gui/sidebar.py index 5dd28c7..19d4dca 100644 --- a/lintunes/gui/sidebar.py +++ b/lintunes/gui/sidebar.py @@ -14,7 +14,11 @@ from lintunes.gui import drag_ghost PID_ROLE = Qt.ItemDataRole.UserRole -KIND_ROLE = Qt.ItemDataRole.UserRole + 1 # "folder" | "playlist" +KIND_ROLE = Qt.ItemDataRole.UserRole + 1 # "folder" | "playlist" | "smart" + +# Rotated Floral Heart Bullet, painted in the branch area (where the folder +# triangle sits) for smart playlists, so the name stays aligned with the others. +SMART_GLYPH = "❧" class SidebarPanel(QWidget): @@ -186,10 +190,12 @@ class PlaylistTree(QTreeWidget): def add_level(parent_item, parent_pid): for playlist in sorted(children.get(parent_pid, []), key=lambda p: p.name.lower()): + is_folder = playlist.playlist_type == PlaylistType.FOLDER + is_smart = playlist.is_smart item = QTreeWidgetItem([playlist.name]) item.setData(0, PID_ROLE, playlist.persistent_id) - is_folder = playlist.playlist_type == PlaylistType.FOLDER - item.setData(0, KIND_ROLE, "folder" if is_folder else "playlist") + kind = "folder" if is_folder else "smart" if is_smart else "playlist" + item.setData(0, KIND_ROLE, kind) item.setFlags(item.flags() | Qt.ItemFlag.ItemIsEditable) if parent_item is None: self.addTopLevelItem(item) @@ -227,6 +233,24 @@ class PlaylistTree(QTreeWidget): stack.extend(item.child(i) for i in range(item.childCount())) return None + def drawBranches(self, painter, rect, index): + """Paint the ❧ for smart playlists in the branch column (where a folder's + disclosure triangle sits), so the playlist name stays left-aligned with + the others instead of being pushed right by a text prefix.""" + super().drawBranches(painter, rect, index) + item = self.itemFromIndex(index) + if item is None or item.data(0, KIND_ROLE) != "smart": + return + selected = self.selectionModel().isSelected(index) + role = (self.palette().highlightedText() if selected + else self.palette().text()) + painter.save() + painter.setPen(role.color()) + painter.drawText(rect.adjusted(0, 0, -2, 0), + Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter, + SMART_GLYPH) + painter.restore() + def current_playlist_id(self) -> str: item = self.currentItem() if item is None: @@ -252,12 +276,38 @@ class PlaylistTree(QTreeWidget): self.setCurrentItem(item) # selects + navigates to the new playlist self.editItem(item, 0) # live inline rename; editor selects all + def create_smart_playlist_interactive(self, parent_pid: str = ""): + """Open the criteria editor; on accept, create the smart playlist and + drop into inline rename (mirrors create_playlist_interactive).""" + from lintunes.gui.smart_playlist_dialog import SmartPlaylistEditorDialog + from lintunes.smart import SmartCriteria + dialog = SmartPlaylistEditorDialog(SmartCriteria(), self, + title="New Smart Playlist") + if not dialog.exec(): + return + playlist = self._manager.create_smart_playlist( + "untitled smart playlist", dialog.criteria(), parent_pid) + item = self._find_item(playlist.persistent_id) + if item is not None: + self.setCurrentItem(item) + self.editItem(item, 0) + + def edit_smart_playlist(self, pid: str): + playlist = self._manager.library.playlists.get(pid) + if not playlist or not playlist.is_smart: + return + from lintunes.gui.smart_playlist_dialog import SmartPlaylistEditorDialog + dialog = SmartPlaylistEditorDialog(playlist.smart_criteria, self, + title="Edit Smart Playlist") + if dialog.exec(): + self._manager.set_smart_criteria(pid, dialog.criteria()) + # ---- selection / rename ---- def _on_selection_changed(self, current, _previous): if current is None: return - if current.data(0, KIND_ROLE) == "playlist": + if current.data(0, KIND_ROLE) in ("playlist", "smart"): self.playlist_selected.emit(current.data(0, PID_ROLE)) def _on_item_renamed(self, item, _column): @@ -275,17 +325,20 @@ class PlaylistTree(QTreeWidget): parent_pid = "" if kind == "folder": parent_pid = pid - elif kind == "playlist": + elif kind in ("playlist", "smart"): playlist = self._manager.library.playlists.get(pid) if playlist: parent_pid = playlist.parent_persistent_id menu = QMenu(self) new_playlist = menu.addAction("New Playlist") + new_smart = menu.addAction("New Smart Playlist") new_folder = menu.addAction("New Playlist Folder") - rename = delete = None - if kind in ("playlist", "folder"): + rename = delete = edit_smart = None + if kind in ("playlist", "folder", "smart"): menu.addSeparator() + if kind == "smart": + edit_smart = menu.addAction("Edit Smart Playlist…") rename = menu.addAction("Rename") delete = menu.addAction("Delete") @@ -294,6 +347,10 @@ class PlaylistTree(QTreeWidget): return if chosen is new_playlist: self.create_playlist_interactive(parent_pid) + elif chosen is new_smart: + self.create_smart_playlist_interactive(parent_pid) + elif edit_smart is not None and chosen is edit_smart: + self.edit_smart_playlist(pid) elif chosen is new_folder: name, ok = QInputDialog.getText(self, "New Playlist Folder", "Name:", text="untitled folder") diff --git a/lintunes/gui/smart_playlist_dialog.py b/lintunes/gui/smart_playlist_dialog.py new file mode 100644 index 0000000..06a970c --- /dev/null +++ b/lintunes/gui/smart_playlist_dialog.py @@ -0,0 +1,423 @@ +"""Editor dialog for smart-playlist criteria (the LinTunes equivalent of +iTunes' "Smart Playlist" sheet). + +Phase 1 edits a flat rule list (one top-level *Match all/any* + rules + an +optional limit). Imported playlists that use nested rule groups are shown +read-only with an explanatory banner; nested-group editing is Phase 2. +""" + +from PyQt6.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QLabel, QComboBox, QLineEdit, QSpinBox, + QDateEdit, QCheckBox, QToolButton, QDialogButtonBox, + QScrollArea, QWidget, QFrame, +) +from PyQt6.QtCore import Qt, QDate + +from lintunes.smart import ( + FIELD_REGISTRY, FieldType, SmartCriteria, SmartGroup, SmartRule, SmartLimit, +) + + +OP_LABELS = { + "is": "is", "is_not": "is not", "contains": "contains", + "not_contains": "does not contain", "starts_with": "starts with", + "ends_with": "ends with", "greater": "is greater than", + "less": "is less than", "in_range": "is in the range", + "after": "is after", "before": "is before", + "not_in_range": "is not in the range", "in_last": "is in the last", + "not_in_last": "is not in the last", +} + +_LIMIT_BY = [("items", "items"), ("minutes", "minutes"), ("hours", "hours"), + ("mb", "MB"), ("gb", "GB")] +_SELECTIONS = [ + ("random", "random"), ("name", "name"), ("album", "album"), + ("artist", "artist"), ("genre", "genre"), + ("highest_rating", "highest rating"), ("lowest_rating", "lowest rating"), + ("most_recently_played", "most recently played"), + ("least_recently_played", "least recently played"), + ("most_often_played", "most often played"), + ("least_often_played", "least often played"), + ("most_recently_added", "most recently added"), + ("least_recently_added", "least recently added"), +] +_DATE_UNITS = [("days", "days"), ("weeks", "weeks"), ("months", "months")] + +_INT_MAX = 2_000_000_000 + + +def _iso_to_qdate(value) -> QDate: + try: + d = QDate.fromString(str(value)[:10], "yyyy-MM-dd") + if d.isValid(): + return d + except (ValueError, TypeError): + pass + return QDate.currentDate() + + +def _qdate_to_iso(qdate: QDate) -> str: + return qdate.toString("yyyy-MM-dd") + "T00:00:00" + + +class RuleRow(QWidget): + """One condition: field combo -> operator combo -> value widget(s).""" + + def __init__(self, dialog: "SmartPlaylistEditorDialog", rule: SmartRule): + super().__init__(dialog) + self._dialog = dialog + layout = QHBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + self.field_combo = QComboBox() + for key, meta in FIELD_REGISTRY.items(): + self.field_combo.addItem(meta.label, key) + layout.addWidget(self.field_combo) + + self.op_combo = QComboBox() + layout.addWidget(self.op_combo) + + self.value_host = QWidget() + self.value_layout = QHBoxLayout(self.value_host) + self.value_layout.setContentsMargins(0, 0, 0, 0) + layout.addWidget(self.value_host, stretch=1) + + self.minus = QToolButton(); self.minus.setText("–") + self.plus = QToolButton(); self.plus.setText("+") + layout.addWidget(self.minus) + layout.addWidget(self.plus) + self.minus.clicked.connect(lambda: dialog.remove_row(self)) + self.plus.clicked.connect(lambda: dialog.add_row_after(self)) + + self.field_combo.currentIndexChanged.connect(self._on_field_changed) + self.op_combo.currentIndexChanged.connect(self._rebuild_value) + + self._load(rule) + + # ---- population ---- + + def _load(self, rule: SmartRule): + idx = self.field_combo.findData(rule.field) + self.field_combo.setCurrentIndex(idx if idx >= 0 else 0) + self._populate_ops() + opidx = self.op_combo.findData(rule.operator) + self.op_combo.setCurrentIndex(opidx if opidx >= 0 else 0) + self._rebuild_value() + self._set_values(rule) + + def _current_field(self) -> str: + return self.field_combo.currentData() + + def _current_meta(self): + return FIELD_REGISTRY[self._current_field()] + + def _populate_ops(self): + self.op_combo.blockSignals(True) + self.op_combo.clear() + for op in self._current_meta().operators: + self.op_combo.addItem(OP_LABELS.get(op, op), op) + self.op_combo.blockSignals(False) + + def _on_field_changed(self): + self._populate_ops() + self._rebuild_value() + + # ---- value widget(s) ---- + + def _clear_value(self): + while self.value_layout.count(): + item = self.value_layout.takeAt(0) + w = item.widget() + if w is not None: + w.deleteLater() + + def _new_spin(self, maximum=_INT_MAX, minimum=0) -> QSpinBox: + spin = QSpinBox() + spin.setRange(minimum, maximum) + return spin + + def _rebuild_value(self): + self._clear_value() + meta = self._current_meta() + op = self.op_combo.currentData() + self._kind = None + + if meta.type is FieldType.STRING: + self._kind = "string" + self.w_line = QLineEdit() + self.value_layout.addWidget(self.w_line) + + elif meta.type is FieldType.BOOL: + self._kind = "bool" + self.w_bool = QComboBox() + self.w_bool.addItem("true", True) + self.w_bool.addItem("false", False) + self.value_layout.addWidget(self.w_bool) + + elif meta.type is FieldType.RATING: + self._kind = "rating" + self.w_spin = self._new_spin(maximum=5) + self.w_spin.setSuffix(" stars") + self.value_layout.addWidget(self.w_spin) + if op == "in_range": + self.value_layout.addWidget(QLabel("to")) + self.w_spin2 = self._new_spin(maximum=5) + self.w_spin2.setSuffix(" stars") + self.value_layout.addWidget(self.w_spin2) + self._kind = "rating_range" + + elif meta.type in (FieldType.INT, FieldType.DURATION): + unit = " sec" if meta.type is FieldType.DURATION else "" + self._kind = "int" if meta.type is FieldType.INT else "duration" + self.w_spin = self._new_spin() + if unit: + self.w_spin.setSuffix(unit) + self.value_layout.addWidget(self.w_spin) + if op == "in_range": + self.value_layout.addWidget(QLabel("to")) + self.w_spin2 = self._new_spin() + if unit: + self.w_spin2.setSuffix(unit) + self.value_layout.addWidget(self.w_spin2) + self._kind += "_range" + + elif meta.type is FieldType.DATE: + if op in ("in_last", "not_in_last"): + self._kind = "date_last" + self.w_count = self._new_spin(maximum=100000, minimum=1) + self.w_unit = QComboBox() + for key, label in _DATE_UNITS: + self.w_unit.addItem(label, key) + self.value_layout.addWidget(self.w_count) + self.value_layout.addWidget(self.w_unit) + elif op in ("in_range", "not_in_range"): + self._kind = "date_range" + self.w_date = self._new_date() + self.w_date2 = self._new_date() + self.value_layout.addWidget(self.w_date) + self.value_layout.addWidget(QLabel("to")) + self.value_layout.addWidget(self.w_date2) + else: + self._kind = "date" + self.w_date = self._new_date() + self.value_layout.addWidget(self.w_date) + + def _new_date(self) -> QDateEdit: + d = QDateEdit() + d.setCalendarPopup(True) + d.setDisplayFormat("yyyy-MM-dd") + d.setDate(QDate.currentDate()) + return d + + def _set_values(self, rule: SmartRule): + k = self._kind + if k == "string": + self.w_line.setText("" if rule.value is None else str(rule.value)) + elif k == "bool": + self.w_bool.setCurrentIndex(0 if bool(rule.value) else 1) + elif k in ("rating", "rating_range"): + self.w_spin.setValue(int((rule.value or 0)) // 20) + if k == "rating_range": + self.w_spin2.setValue(int((rule.value2 or 0)) // 20) + elif k in ("int", "duration", "int_range", "duration_range"): + div = 1000 if k.startswith("duration") else 1 + self.w_spin.setValue(int((rule.value or 0)) // div) + if k.endswith("_range"): + self.w_spin2.setValue(int((rule.value2 or 0)) // div) + elif k == "date": + self.w_date.setDate(_iso_to_qdate(rule.value)) + elif k == "date_range": + self.w_date.setDate(_iso_to_qdate(rule.value)) + self.w_date2.setDate(_iso_to_qdate(rule.value2)) + elif k == "date_last": + self.w_count.setValue(int(rule.value or 1)) + uidx = self.w_unit.findData(rule.unit or "days") + self.w_unit.setCurrentIndex(uidx if uidx >= 0 else 0) + + # ---- read back ---- + + def rule(self) -> SmartRule: + field = self._current_field() + op = self.op_combo.currentData() + k = self._kind + if k == "string": + return SmartRule(field, op, self.w_line.text()) + if k == "bool": + return SmartRule(field, "is", self.w_bool.currentData()) + if k == "rating": + return SmartRule(field, op, self.w_spin.value() * 20) + if k == "rating_range": + return SmartRule(field, op, self.w_spin.value() * 20, + self.w_spin2.value() * 20) + if k in ("int", "duration"): + mul = 1000 if k == "duration" else 1 + return SmartRule(field, op, self.w_spin.value() * mul) + if k in ("int_range", "duration_range"): + mul = 1000 if k.startswith("duration") else 1 + return SmartRule(field, op, self.w_spin.value() * mul, + self.w_spin2.value() * mul) + if k == "date": + return SmartRule(field, op, _qdate_to_iso(self.w_date.date())) + if k == "date_range": + return SmartRule(field, op, _qdate_to_iso(self.w_date.date()), + _qdate_to_iso(self.w_date2.date())) + if k == "date_last": + return SmartRule(field, op, self.w_count.value(), + unit=self.w_unit.currentData()) + return SmartRule(field, op, "") + + +class SmartPlaylistEditorDialog(QDialog): + def __init__(self, criteria: SmartCriteria | None, parent=None, + title="Smart Playlist"): + super().__init__(parent) + self.setWindowTitle(title) + self.setMinimumWidth(560) + self._original = criteria or SmartCriteria() + self._read_only = self._original.has_nested or self._original.unsupported + self._rows: list[RuleRow] = [] + + outer = QVBoxLayout(self) + + if self._read_only: + banner = QLabel( + "This smart playlist was imported from iTunes and uses " + "conditions LinTunes can't edit yet" + + (" (nested rule groups)." if self._original.has_nested + else ".") + + " It still updates and plays; saving here is disabled.") + banner.setWordWrap(True) + banner.setStyleSheet("color: palette(highlight); padding: 4px;") + outer.addWidget(banner) + + # Match all/any header. + match_row = QHBoxLayout() + match_row.addWidget(QLabel("Match")) + self.match_combo = QComboBox() + self.match_combo.addItem("all", "all") + self.match_combo.addItem("any", "any") + self.match_combo.setCurrentIndex( + 1 if self._original.root.conjunction == "any" else 0) + match_row.addWidget(self.match_combo) + match_row.addWidget(QLabel("of the following rules:")) + match_row.addStretch(1) + outer.addLayout(match_row) + + # Scrollable rules area. + self._rules_container = QWidget() + self._rules_layout = QVBoxLayout(self._rules_container) + self._rules_layout.setContentsMargins(0, 0, 0, 0) + self._rules_layout.addStretch(1) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setWidget(self._rules_container) + scroll.setMinimumHeight(140) + outer.addWidget(scroll, stretch=1) + + # Limit + options. + outer.addWidget(self._build_limit_row()) + self.live_check = QCheckBox("Live updating") + self.live_check.setChecked(self._original.live_update) + outer.addWidget(self.live_check) + self.checked_check = QCheckBox("Match only checked items") + self.checked_check.setChecked(self._original.match_only_checked) + self.checked_check.setEnabled(False) + self.checked_check.setToolTip( + "LinTunes has no per-track checkbox, so this iTunes option is " + "ignored.") + outer.addWidget(self.checked_check) + + # Buttons. + buttons = QDialogButtonBox( + QDialogButtonBox.StandardButton.Ok + | QDialogButtonBox.StandardButton.Cancel) + buttons.accepted.connect(self.accept) + buttons.rejected.connect(self.reject) + self._ok_button = buttons.button(QDialogButtonBox.StandardButton.Ok) + outer.addWidget(buttons) + + # Populate rule rows (flat criteria only). + if not self._read_only: + rules = [c for c in self._original.root.children + if isinstance(c, SmartRule)] + if not rules: + rules = [SmartRule()] + for rule in rules: + self._append_row(rule) + else: + self._set_read_only() + + def _build_limit_row(self) -> QWidget: + box = QFrame() + row = QHBoxLayout(box) + row.setContentsMargins(0, 0, 0, 0) + limit = self._original.limit + self.limit_check = QCheckBox("Limit to") + self.limit_check.setChecked(limit.enabled) + row.addWidget(self.limit_check) + self.limit_count = QSpinBox() + self.limit_count.setRange(1, _INT_MAX) + self.limit_count.setValue(max(1, limit.count)) + row.addWidget(self.limit_count) + self.limit_by = QComboBox() + for key, label in _LIMIT_BY: + self.limit_by.addItem(label, key) + self.limit_by.setCurrentIndex( + max(0, self.limit_by.findData(limit.by))) + row.addWidget(self.limit_by) + row.addWidget(QLabel("selected by")) + self.limit_selection = QComboBox() + for key, label in _SELECTIONS: + self.limit_selection.addItem(label, key) + self.limit_selection.setCurrentIndex( + max(0, self.limit_selection.findData(limit.selection))) + row.addWidget(self.limit_selection) + row.addStretch(1) + return box + + def _set_read_only(self): + for w in (self.match_combo, self.limit_check, self.limit_count, + self.limit_by, self.limit_selection, self.live_check): + w.setEnabled(False) + self._ok_button.setEnabled(False) + + # ---- row management ---- + + def _append_row(self, rule: SmartRule): + row = RuleRow(self, rule) + # Insert before the trailing stretch. + self._rules_layout.insertWidget(self._rules_layout.count() - 1, row) + self._rows.append(row) + + def add_row_after(self, after: RuleRow): + row = RuleRow(self, SmartRule()) + idx = self._rules_layout.indexOf(after) + self._rules_layout.insertWidget(idx + 1, row) + pos = self._rows.index(after) + 1 + self._rows.insert(pos, row) + + def remove_row(self, row: RuleRow): + if row not in self._rows: + return + self._rows.remove(row) + self._rules_layout.removeWidget(row) + row.deleteLater() + + # ---- result ---- + + def criteria(self) -> SmartCriteria: + if self._read_only: + return self._original + root = SmartGroup( + conjunction=self.match_combo.currentData(), + children=[row.rule() for row in self._rows]) + limit = SmartLimit( + enabled=self.limit_check.isChecked(), + count=self.limit_count.value(), + by=self.limit_by.currentData(), + selection=self.limit_selection.currentData()) + return SmartCriteria( + root=root, limit=limit, + live_update=self.live_check.isChecked(), + match_only_checked=self._original.match_only_checked, + unsupported=False, has_nested=False) diff --git a/lintunes/gui/track_table.py b/lintunes/gui/track_table.py index a46bf35..3bd1935 100644 --- a/lintunes/gui/track_table.py +++ b/lintunes/gui/track_table.py @@ -342,6 +342,8 @@ class TrackTableView(QTableView): super().__init__(parent) self._playlist_mode = playlist_mode self._source_playlist_id = "" + # False for smart playlists: blocks manual add/remove/reorder/cut/paste. + self._content_editable = True self.model_ = TrackTableModel(self, with_index_column=playlist_mode) self.proxy = QSortFilterProxyModel(self) @@ -411,6 +413,12 @@ class TrackTableView(QTableView): def set_source_playlist(self, pid: str): self._source_playlist_id = pid + def set_content_editable(self, editable: bool): + """Smart playlists pass False: dropping, reordering, cut/paste and + remove are all disabled (membership is derived from criteria).""" + self._content_editable = editable + self.setAcceptDrops(editable) + def context_id(self) -> str: """Identifies what this table is currently showing, so the now-playing icon can be scoped to the context playback started from.""" @@ -515,7 +523,7 @@ class TrackTableView(QTableView): rows = self.selected_source_rows() QApplication.clipboard().setMimeData( make_tracks_mime(ids, self._source_playlist_id, rows)) - if self._playlist_mode and self._source_playlist_id: + if self._playlist_mode and self._source_playlist_id and self._content_editable: self.cut_requested.emit(rows) def paste_anchor_row(self): @@ -571,7 +579,7 @@ class TrackTableView(QTableView): copy_action = menu.addAction("Copy\tCtrl+C") cut_action = menu.addAction("Cut\tCtrl+X") paste_action = None - if (self._playlist_mode + if (self._playlist_mode and self._content_editable and parse_tracks_mime(QApplication.clipboard().mimeData())): paste_action = menu.addAction("Paste\tCtrl+V") @@ -596,7 +604,7 @@ class TrackTableView(QTableView): show_in_menu.addAction(name).setData(pid) remove_action = None - if self._playlist_mode: + if self._playlist_mode and self._content_editable: menu.addSeparator() remove_action = menu.addAction("Remove from Playlist\tDel") chosen = menu.exec(self.viewport().mapToGlobal(pos)) @@ -630,7 +638,8 @@ class TrackTableView(QTableView): def keyPressEvent(self, event): if (event.key() in (Qt.Key.Key_Delete, Qt.Key.Key_Backspace) - and self._playlist_mode and self.selected_source_rows()): + and self._playlist_mode and self._content_editable + and self.selected_source_rows()): self.remove_requested.emit(self.selected_source_rows()) return if event.key() in (Qt.Key.Key_Return, Qt.Key.Key_Enter): @@ -704,6 +713,10 @@ class TrackTableView(QTableView): event.ignore() def dragMoveEvent(self, event): + if not self._content_editable: # smart playlist: no drops + event.ignore() + self._set_drop_indicator(None) + return drag_ghost.move(self.viewport().mapToGlobal(event.position().toPoint())) mime = event.mimeData() accepted = False @@ -735,6 +748,8 @@ class TrackTableView(QTableView): super().dragLeaveEvent(event) def dropEvent(self, event): + if not self._content_editable: # smart playlist: no drops + return self._set_drop_indicator(None) self._stop_autoscroll() mime = event.mimeData() diff --git a/lintunes/importers/itunes_importer.py b/lintunes/importers/itunes_importer.py index fc49e89..5da48aa 100644 --- a/lintunes/importers/itunes_importer.py +++ b/lintunes/importers/itunes_importer.py @@ -19,7 +19,8 @@ class ImportReport: case_fixed: int = 0 # locations fixed by case/normalization matching playlist_count: int = 0 folder_count: int = 0 - skipped_smart: int = 0 + smart_count: int = 0 # smart playlists imported + smart_unsupported: int = 0 # of those, criteria kept as a static snapshot skipped_system: int = 0 def summary(self) -> str: @@ -31,7 +32,9 @@ class ImportReport: f"Missing files: {len(self.missing_files)}", f"Unmapped locations: {len(self.unmapped_locations)}", f"Playlists imported: {self.playlist_count} (+ {self.folder_count} folders)", - f"Playlists skipped: {self.skipped_smart} smart, {self.skipped_system} system", + f"Smart playlists: {self.smart_count} imported " + f"({self.smart_unsupported} kept as snapshot)", + f"Playlists skipped: {self.skipped_system} system", ] return "\n".join(lines) @@ -149,9 +152,16 @@ def import_itunes_xml(xml_path: Path, music_root: Path, report.skipped_system += 1 continue if playlist.playlist_type == PlaylistType.SMART: - report.skipped_smart += 1 - continue - # Drop references to tracks we didn't import + from lintunes import smart + playlist.smart_criteria = smart.parse_itunes_smart( + itunes_playlist.get("Smart Info"), + itunes_playlist.get("Smart Criteria")) + report.smart_count += 1 + if playlist.smart_criteria.unsupported: + report.smart_unsupported += 1 + # Drop references to tracks we didn't import. For smart playlists this + # snapshot is the fallback membership until the first recompute (and the + # permanent membership when the criteria are unsupported). playlist.track_ids = [tid for tid in playlist.track_ids if tid in tracks] playlists[playlist.persistent_id] = playlist if playlist.playlist_type == PlaylistType.FOLDER: @@ -163,7 +173,8 @@ def import_itunes_xml(xml_path: Path, music_root: Path, playlist.parent_persistent_id = "" report.folder_count = len(kept_folder_ids) - report.playlist_count = len(playlists) - report.folder_count + # Regular playlists only; folders and smart playlists are counted separately. + report.playlist_count = len(playlists) - report.folder_count - report.smart_count library = Library( tracks=tracks, diff --git a/lintunes/itunes_smart/LICENSE b/lintunes/itunes_smart/LICENSE new file mode 100644 index 0000000..3be7c6d --- /dev/null +++ b/lintunes/itunes_smart/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) cuzi 2018 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/lintunes/itunes_smart/__init__.py b/lintunes/itunes_smart/__init__.py new file mode 100644 index 0000000..66ffa3c --- /dev/null +++ b/lintunes/itunes_smart/__init__.py @@ -0,0 +1,19 @@ +"""Vendored iTunes smart-playlist criteria parser. + +This package is a vendored copy of `itunessmart` by cuzi +(https://github.com/cvzi/itunes_smartplaylist), used under the MIT License +(see the bundled LICENSE file). It decodes the base64 "Smart Info" / "Smart +Criteria" binary blobs from an iTunes XML library into a structured rule tree. + +Vendored rather than pip-installed so LinTunes stays self-contained for an +offline, Syncthing-synced desktop app. Only the import line in parse.py was +changed (to a package-relative import); the parsing logic is unmodified. + +LinTunes code should not use this package directly -- go through +`lintunes.smart.parse_itunes_smart()`, which converts the parser's output into +LinTunes' own `SmartCriteria` model. +""" + +from lintunes.itunes_smart.parse import SmartPlaylistParser, SmartPlaylist + +__all__ = ["SmartPlaylistParser", "SmartPlaylist"] diff --git a/lintunes/itunes_smart/data_structure.py b/lintunes/itunes_smart/data_structure.py new file mode 100644 index 0000000..18bf321 --- /dev/null +++ b/lintunes/itunes_smart/data_structure.py @@ -0,0 +1,237 @@ +""" +Module holding static data related to the iTunes playlist format. +""" + +from enum import IntEnum + + +class FileKind: + def __init__(self, name, extension): + self.name = name + self.extension = extension + + +FileKinds = [ + FileKind("Protected AAC audio file", ".m4p"), + FileKind("Purchased AAC audio file", ".m4a"), + FileKind("Apple Lossless audio file", ".m4a"), + FileKind("MPEG audio file", ".mp3"), + FileKind("AIFF audio file", ".aiff"), + FileKind("WAV audio file", ".wav"), + FileKind("QuickTime movie file", ".mov"), + FileKind("MPEG-4 video file", ".mp4"), + FileKind("Purchased MPEG-4 video file", ".mp4"), + FileKind("AAC audio file", ".m4a") +] + +MediaKinds = { + 0x01: "Music", + 0x02: "Movie", + 0x04: "Podcast", + 0x08: "Audiobook", + 0x20: "Music Video", + 0x40: "TV Show", + 0x400: "Home Video", + 0x10000: "iTunes Extras", + 0x100000: "Voice Memo", # TODO "Voice Memo" might be 0x00 on Apple Music + 0x200000: "iTunes U", + 0xC00000: "Book", + # TODO mediakinds of toplevel playlists + # The following are only used in the toplevel playlists: Music, TV Shows, Movies and Books + # (These playlists can be selected with the arrows or from a dropdown menu in iTunes 12) + # Theses Media Kinds cannot be selected by the user. I am unsure about their meaning. + 0xC00008: "Book or Audiobook", # TODO My guess: This contains Books and Audiobooks + 0x1021B1: "Music", # TODO My guess: This is similar to Music + 0x208004: "Undesired Music", # TODO My guess: This is some kind of Music that should not appear in the toplevel playlist + 0x20A004: "Undesired Other" # TODO My guess: This is something (other than music) that should not appear in the toplevel playlist +} + +iCloudStatus = { + 0x01: "Purchased", + 0x02: "Matched", + 0x03: "Uploaded", + 0x04: "Ineligible", + 0x05: "Local Only", + 0x07: "Duplicate", + 0x09: "No Longer Available", +} + +LoveStatus = { + 0x00: "None", + 0x02: "Loved", + 0x03: "Disliked", +} + +LocationKinds = { + 0x01: "Computer", + 0x10: "iCloud" +} + +SelectionMethodsStrings = { + "Random": ["random", "RANDOM()"], + "Name": ["name", "SortName"], + "Album": ["album", "SortAlbum"], + "Artist": ["artist", "SortArtist"], + "Genre": ["genre", "Genre"], + "HighestRating": ["highest rated", "Rating DESC"], + "LowestRating": ["lowest rated", "Rating ASC"], + "RecentlyPlayed": [["most recently played", "least recently played"], ["LastPlayed DESC", "LastPlayed ASC"]], + "OftenPlayed": [["most often played", "least often played"], ["Plays DESC", "Plays ASC"]], + "RecentlyAdded": [["most recently added", "least recently added"], ["DateAdded DESC", "DateAdded ASC"]] +} + + +DateStartFromUnix = -2082844800 # iTunes/Unix time stamp 0 difference + + +class LimitMethods(IntEnum): + """The methods by which the number of songs in a playlist are limited""" + Minutes = 0x01 + MB = 0x02 + Items = 0x03 + Hours = 0x04 + GB = 0x05 + + +class SelectionMethods(IntEnum): + """The methods by which songs are selected for inclusion in a limited playlist""" + Random = 0x02 + Name = 0x05 + Album = 0x06 + Artist = 0x07 + Genre = 0x09 + HighestRating = 0x1c + LowestRating = 0x01 + RecentlyPlayed = 0x1a + OftenPlayed = 0x19 + RecentlyAdded = 0x15 + + +class StringFields(IntEnum): + """The matching criteria which take string data""" + Album = 0x03 + AlbumArtist = 0x47 + Artist = 0x04 + Category = 0x37 + Comments = 0x0e + Composer = 0x12 + Description = 0x36 + Genre = 0x08 + Grouping = 0x27 + Kind = 0x09 + Name = 0x02 + Show = 0x3e + SortAlbum = 0x4f + SortAlbumartist = 0x51 + SortComposer = 0x52 + SortName = 0x4e + SortShow = 0x53 + VideoRating = 0x59 + + +class IntFields(IntEnum): + """The matching criteria which take integer data""" + BPM = 0x23 + BitRate = 0x05 + Compilation = 0x1f + DiskNumber = 0x18 + Plays = 0x16 + Rating = 0x19 + Podcast = 0x39 + SampleRate = 0x06 + Season = 0x3f + Size = 0x0c + Skips = 0x44 + Duration = 0x0d + TrackNumber = 0x0b + Year = 0x07 + + +class BooleanFields(IntEnum): + """The matching criteria which take boolean data""" + HasArtwork = 0x25 + Purchased = 0x29 + Checked = 0x1d + + +class DateFields(IntEnum): + """The matching criteria which take date data""" + DateAdded = 0x10 + DateModified = 0x0a + LastPlayed = 0x17 + LastSkipped = 0x45 + + +class MediaKindFields(IntEnum): + """The matching criteria which take a Media Kind, as defined above""" + MediaKind = 0x3c + + +class PlaylistFields(IntEnum): + """The matching criteria which take a Persistent Playlist ID (64bit)""" + PlaylistPersistentID = 0x28 + + +class LoveFields(IntEnum): + """The matching criteria which take a Persistent Playlist ID (64bit)""" + Love = 0x9a + + +class CloudFields(IntEnum): + """The matching criteria which take iCloud status , as defined above """ + iCloudStatus = 0x86 + + +class LocationFields(IntEnum): + """The matching criteria which take a Location, as defined above""" + Location = 0x85 + + +class LogicSign(IntEnum): + """The signs which apply to different kinds of logic (is vs. is not, contains vs. doesn't contain, etc.)""" + IntPositive = 0x00 + StringPositive = 0x01 + IntNegative = 0x02 + StringNegative = 0x03 + + +class LogicRule(IntEnum): + """The logical rules""" + Other = 0x00 + Is = 0x01 + Contains = 0x02 + Starts = 0x04 + Ends = 0x08 + Greater = 0x10 + Less = 0x40 + + +class Offset(IntEnum): + """Byte offsets for the fields""" + INTLENGTH = 67 # The length on a int criteria starting at the first int + SUBEXPRESSIONLENGTH = 192 # The length of a subexpression starting from FIELD + + # INFO OFFSETS + # Offsets for bytes which... + LIVEUPDATE = 0 # determine whether live updating is enabled - Absolute offset + MATCHBOOL = 1 # determine whether logical matching is to be performed - Absolute offset + LIMITBOOL = 2 # determine whether results are limited - Absolute offset + LIMITMETHOD = 3 # determine by what criteria the results are limited - Absolute offset + SELECTIONMETHOD = 7 # determine by what criteria limited playlists are populated - Absolute offset + LIMITINT = 8 # determine the limited - Absolute offset + LIMITCHECKED = 12 # determine whether to exclude unchecked items - Absolute offset + SELECTIONMETHODSIGN = 13 # determine whether certain selection methods are "most" or "least" - Absolute offset + + # CRITERIA OFFSETS + # Offsets for bytes which... + LOGICTYPE = 15 # determine whether all or any criteria must match - Absolute offset + FIELD = 139 # determine what is being matched (Artist, Album, &c) - Absolute offset + LOGICSIGN = 1 # determine whether the matching rule is positive or negative (e.g., is vs. is not) - Relative offset from FIELD + LOGICRULE = 4 # determine the kind of logic used (is, contains, begins, &c) - Relative offset from FIELD + STRING = 54 # begin string data - Relative offset from FIELD + INTA = 57 # begin the first int - Relative offset from FIELD + INTB = 24 # begin the second int - Relative offset from INTA + TIMEMULTIPLE = 73 # begin the int with the multiple of time - Relative offset from FIELD + TIMEVALUE = 65 # begin the inverse int with the value of time - Relative offset from FIELD + SUBLOGICTYPE = 68 # determine whether all or any criteria must match - Relative offset from FIELD + SUBINT = 61 # begin the first int - Relative offset from FIELD diff --git a/lintunes/itunes_smart/parse.py b/lintunes/itunes_smart/parse.py new file mode 100644 index 0000000..969e815 --- /dev/null +++ b/lintunes/itunes_smart/parse.py @@ -0,0 +1,843 @@ +""" +Module holding the parsing algorithm +""" + +import sys +import logging +import base64 +import datetime +import struct +import json +import re +from lintunes.itunes_smart.data_structure import * + + +class SmartPlaylist: + """ Parser result. Contains all decoded playlist data""" + + def __init__(self, parser): + self.output = parser.output + self.query = parser.query + self.queryTree = parser.queryTree + self.ignore = parser.ignore + + def __str__(self): + return "SmartPlaylist(`%s`)" % self.query + + def __repr__(self): + return "SmartPlaylist(%s)" % json.dumps({"queryTree": self.queryTree, "ignore": self.ignore}, indent=2) + + +class SmartPlaylistParser: + def __init__(self, datastr_info=None, datastr_criteria=None): + self.is_parsed = False + if datastr_info and datastr_criteria: + self.str_data(datastr_info, datastr_criteria) + self.parse() + + def str_data(self, datastr_info, datastr_criteria): + i = base64.standard_b64decode("".join(datastr_info.split())) + c = base64.standard_b64decode("".join(datastr_criteria.split())) + return self.data(i, c) + + def data(self, data_info, data_criteria): + self.is_parsed = False + self.info = data_info + self.criteria = data_criteria + self.query = "" + self.root = {} + self.queryTree = self.root + self.queryTreeCurrent = [] + self.fullTreeRoot = {} + self.fullTree = self.fullTreeRoot + self.fullTreeCurrent = [] + self.output = "" + self.ignore = "" + self.limit = {} + + self.subStack = [] + + def result(self): + if self.is_parsed: + return SmartPlaylist(self) + raise RuntimeError("Data not parsed yet. Call parse() before result()") + + def parse(self): + if self.is_parsed: + return + + if not hasattr(self, 'info') or not hasattr(self, 'criteria') or not self.info or not self.criteria: # pragma: no cover + raise RuntimeError("Set smart info with data() or strdata() before running parse()") + + self.offset = int(Offset.FIELD) + + if self.info[Offset.MATCHBOOL] == 1: + self.current_operator = self._operatorFromLogicType(self.criteria[Offset.LOGICTYPE]) + self.queryTree[self.current_operator] = self.queryTreeCurrent + self.fullTree[self.current_operator] = self.fullTreeCurrent + + while True: + self.again = False + + if len(self.subStack) > 0: + if self.subStack[-1]["N"] == 0: + old = self.subStack.pop() + + self.current_operator = old["operator"] + self.queryTreeCurrent = old["queryTreeCurrent"] + self.fullTreeCurrent = old["fullTreeCurrent"] + else: + self.subStack[-1]["N"] -= 1 + + self.logicSignOffset = self.offset + int(Offset.LOGICSIGN) + self.logicRulesOffset = self.offset + int(Offset.LOGICRULE) + self.stringOffset = self.offset + int(Offset.STRING) + self.intAOffset = self.offset + int(Offset.INTA) + self.intBOffset = self.intAOffset + int(Offset.INTB) + self.timeMultipleOffset = self.offset + \ + int(Offset.TIMEMULTIPLE) + self.timeValueOffset = self.offset + int(Offset.TIMEVALUE) + + if any(self.criteria[self.offset] == + e.value for e in StringFields): + self.ProcessStringField() + elif any(self.criteria[self.offset] == e.value for e in IntFields): + self.ProcessIntField() + elif any(self.criteria[self.offset] == e.value for e in DateFields): + self.ProcessDateField() + elif any(self.criteria[self.offset] == e.value for e in BooleanFields): + self.ProcessBooleanField() + elif any(self.criteria[self.offset] == e.value for e in MediaKindFields): + self.ProcessListField(MediaKindFields, MediaKinds, listtype="mediakind") + elif any(self.criteria[self.offset] == e.value for e in PlaylistFields): + self.ProcessPlaylistField() + elif any(self.criteria[self.offset] == e.value for e in CloudFields): + self.ProcessListField(CloudFields, iCloudStatus, listtype="cloud") + elif any(self.criteria[self.offset] == e.value for e in LoveFields): + self.ProcessListField(LoveFields, LoveStatus, listtype="love") + elif any(self.criteria[self.offset] == e.value for e in LocationFields): + self.ProcessListField(LocationFields, LocationKinds, listtype="location") + elif self.criteria[self.offset] == 0: + # Subexpression + + parent_operator = self.current_operator + subgroup_operator = self._operatorFromLogicType(self.criteria[self.offset + Offset.SUBLOGICTYPE]) + + numberOfSubExpression = self._iTunesUint( + self.criteria[self.offset + Offset.SUBINT:self.offset + Offset.SUBINT + 4]) + + self.subStack.append({ + "N": numberOfSubExpression, + "operator": parent_operator, + "queryTreeCurrent": self.queryTreeCurrent, + "fullTreeCurrent": self.fullTreeCurrent, + }) + + newtree = {} + newcurrent = [] + self.queryTreeCurrent.append(newtree) + + newfulltree = {} + newfulltreecurrent = [] + self.fullTreeCurrent.append(newfulltree) + + newtree[subgroup_operator] = newcurrent + newfulltree[subgroup_operator] = newfulltreecurrent + + self.current_operator = subgroup_operator + self.queryTreeCurrent = newcurrent + self.fullTreeCurrent = newfulltreecurrent + + self.offset += Offset.SUBEXPRESSIONLENGTH + self.again = True + else: # pragma: no cover + errormessage = "Unknown field: %s" % (hex(self.criteria[self.offset])) + logging.warning(errormessage) + self.ignore += "Not processed: %s " % errormessage + logging.debug(self.criteria[self.offset:self.offset + 100]) + + if not self.again: + break + + self.query = self._buildQuery(self._stripImplicitMediaKindFilter(self.root)) + self.output = self._buildOutputTree(self._stripImplicitMediaKindFilter(self.fullTreeRoot)) + + if self.info[Offset.LIMITBOOL] == 1: + # Limit + self.limit["number"] = self._iTunesUint( + self.info[Offset.LIMITINT:Offset.LIMITINT + 4]) + + self.limit["type"] = LimitMethods( + self.info[Offset.LIMITMETHOD]).name + + # Selection + selectionmethod = SelectionMethods( + self.info[Offset.SELECTIONMETHOD]).name + sign = int(self.info[Offset.SELECTIONMETHODSIGN] == 0) + self.limit["order"] = SelectionMethodsStrings[selectionmethod][1] if isinstance( + SelectionMethodsStrings[selectionmethod][1], + str) else SelectionMethodsStrings[selectionmethod][1][sign] + + if len(self.output) > 0: + self.output += '\n' + self.output += "Limited to %d %s selected by %s" % (self.limit["number"], + self.limit["type"], + SelectionMethodsStrings[selectionmethod][0] if isinstance( + SelectionMethodsStrings[selectionmethod][0], + str) else SelectionMethodsStrings[selectionmethod][0][sign]) + + if self.info[Offset.LIMITCHECKED] == 1: + # Exclude unchecked items + self.limit["onlychecked"] = True + + if len(self.output) > 0: + self.output += '\n' + self.output += "Exclude unchecked items" + else: + self.limit["onlychecked"] = False + + if self.info[Offset.LIVEUPDATE] == 0: + # Live Update disabled + self.limit["liveupdate"] = False + + if len(self.output) > 0: + self.output += '\n' + self.output += "Live updating disabled" + else: + self.limit["liveupdate"] = True + + t = self.limit + t["tree"] = self.queryTree + self.queryTree = t + + t2 = self.limit + t2["fulltree"] = self.fullTree + self.fullTree = t2 + + self.is_parsed = True + + def ProcessStringField(self): + end = False + self.fieldName = StringFields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "(lower(" + self.fieldName + ")" + self.workingFull = {"field": self.fieldName, "type": "string"} + + KindEval = None + + if self.criteria[self.logicRulesOffset] == LogicRule.Contains: + if self.criteria[self.logicSignOffset] == LogicSign.StringPositive: + self.workingOutput += " contains " + self.workingQuery += " LIKE '%" + self.workingFull["operator"] = "like" + else: + self.workingOutput += " does not contain " + self.workingQuery += " NOT LIKE '%" + self.workingFull["operator"] = "not like" + if self.criteria[self.offset] == StringFields.Kind: + KindEval = lambda kind, query: query.lower() in kind.name.lower() + end = True + + elif self.criteria[self.logicRulesOffset] == LogicRule.Is: + if self.criteria[self.logicSignOffset] == LogicSign.StringPositive: + self.workingOutput += " is " + self.workingQuery += " = '" + self.workingFull["operator"] = "is" + else: + self.workingOutput += " is not " + self.workingQuery += " != '" + self.workingFull["operator"] = "is not" + if self.criteria[self.offset] == StringFields.Kind: + KindEval = lambda kind, query: query.lower() == kind.name.lower() + + elif self.criteria[self.logicRulesOffset] == LogicRule.Starts: + self.workingOutput += " starts with " + self.workingQuery += " Like '" + self.workingFull["operator"] = "starts with" + if self.criteria[self.offset] == StringFields.Kind: + KindEval = lambda kind, query: kind.name.lower().startswith(query.lower()) + end = True + + elif self.criteria[self.logicRulesOffset] == LogicRule.Ends: + self.workingOutput += " ends with " + self.workingQuery += " Like '%" + self.workingFull["operator"] = "ends with" + if self.criteria[self.offset] == StringFields.Kind: + KindEval = lambda kind, query: kind.name.lower().endswith(query.lower()) + end = True + + self.workingOutput += '"' + self.content = "" + onByte = True + for i in range(self.stringOffset, len(self.criteria)): + if onByte: + if self.criteria[i] == 0 and i != len(self.criteria) - 1: + self.FinishStringField(end, KindEval) + self.offset = i + 2 + self.again = True + return + self.content += chr(self.criteria[i]) + onByte = not onByte + self.FinishStringField(end, KindEval) + return + + def _should_use_structured_kind_match(self, matched_kinds): + if not matched_kinds: + return False + + if self.criteria[self.logicRulesOffset] == LogicRule.Is: + return True + + query = self.content.strip().lower() + if len(query) < 3 or len(matched_kinds) > 3: + return False + + generic_tokens = {"audio", "video", "file", "movie", "protected", "purchased", "quicktime"} + if query in generic_tokens: + return False + + query_pattern = re.compile(r"\b%s\b" % re.escape(query)) + return all(query_pattern.search(kind.name.lower()) for kind in matched_kinds) + + def FinishStringField(self, end, KindEval): + self.workingOutput += self.content + self.workingOutput += '" ' + query_node = None + if self.criteria[self.offset] == StringFields.Kind: + self.workingFull["value"] = self.content + matched_kinds = [kind for kind in FileKinds if KindEval(kind, self.content)] + kindOperator = "like" if self.criteria[self.logicSignOffset] == LogicSign.StringPositive else "not like" + if self._should_use_structured_kind_match(matched_kinds): + kindNodes = [] + joiner = "or" if kindOperator == "like" else "and" + for kind in matched_kinds: + kindQuery = "(lower(Uri)" + kindQuery += (" LIKE '%" + + kind.extension + + "')") if kindOperator == "like" else (" NOT LIKE '%" + + kind.extension + + "%')") + kindNodes.append(("Kind", kindQuery)) + if len(kindNodes) == 1: + query_node = kindNodes[0] + else: + query_node = {joiner: kindNodes} + else: + logging.warning( + 'Treating Kind rule as free text instead of structured file-kind match: %s %s "%s"', + self.fieldName, + self.workingFull["operator"], + self.content, + ) + self.workingQuery += self.content.lower() + self.workingQuery += "%')" if end else "')" + query_node = (self.fieldName, self.workingQuery) + else: + self.workingQuery += self.content.lower() + self.workingQuery += "%')" if end else "')" + self.workingFull["value"] = self.content + query_node = (self.fieldName, self.workingQuery) + + if len(self.ignore) > 0: + self.ignore += " or\n" if self.current_operator == "or" else " and\n" + + self.queryTreeCurrent.append(query_node) + self.fullTreeCurrent.append(self.workingFull) + + def ProcessIntField(self): + self.fieldName = IntFields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "(" + self.fieldName + self.workingFull = {"field": self.fieldName, "type": "int"} + + if self.criteria[self.logicRulesOffset] == LogicRule.Is: + number = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is %d" % number + self.workingQuery += " = %d" % number + self.workingFull["operator"] = "is" + self.workingFull["value"] = number + else: + self.workingOutput += " is not %d" % number + self.workingQuery += " != %d" % number + self.workingFull["operator"] = "is not" + self.workingFull["value"] = number + + elif self.criteria[self.logicRulesOffset] == LogicRule.Greater: + number = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + self.workingOutput += " is greater than %d" % number + self.workingQuery += " > %d" % number + self.workingFull["operator"] = "greater than" + self.workingFull["value"] = number + + elif self.criteria[self.logicRulesOffset] == LogicRule.Less: + number = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + self.workingOutput += " is less than %d" % number + self.workingQuery += " < %d" % number + self.workingFull["operator"] = "less than" + self.workingFull["value"] = number + + elif self.criteria[self.logicRulesOffset] == LogicRule.Other: + if self.criteria[self.logicSignOffset + 2] == 1: + numberA = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + numberB = self._iTunesUint( + self.criteria[self.intBOffset:self.intBOffset + 4], self.criteria[self.offset] == IntFields.Rating) + self.workingOutput += " is in the range of %d to %d" % ( + numberA, numberB) + self.workingQuery += " BETWEEN %d AND %d" % (numberA, numberB) + self.workingFull["operator"] = "between" + self.workingFull["value"] = (numberA, numberB) + + else: + numberA = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + numberB = self._iTunesUint( + self.criteria[self.intBOffset:self.intBOffset + 4], self.criteria[self.offset] == IntFields.Rating) + if numberA == numberB: + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is %d" % numberA + self.workingQuery += " = %d" % numberA + self.workingFull["operator"] = "is" + self.workingFull["value"] = numberA + else: + self.workingOutput += " is not %d" % numberA + self.workingQuery += " != %d" % numberA + self.workingFull["operator"] = "is not" + self.workingFull["value"] = numberA + else: # pragma: no cover + errormessage = "Unkown case in ProcessIntField:LogicRule.Other: a=%d and b=%d" % (numberA, numberB) + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + + self.workingOutput += " ##UnkownCase IntField: LogicRule.Other##" + self.workingQuery += " ##UnkownCase IntField: LogicRule.Other##" + + self.workingQuery += ")" + + self.queryTreeCurrent.append((self.fieldName, self.workingQuery)) + self.fullTreeCurrent.append(self.workingFull) + + self.offset = self.intAOffset + Offset.INTLENGTH + if len(self.criteria) > self.offset: + self.again = True + + def ProcessPlaylistField(self): + self.fieldName = PlaylistFields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "(" + self.fieldName + self.workingFull = {"field": self.fieldName, "type": "playlist"} + + if self.criteria[self.logicRulesOffset] == LogicRule.Is: + idpart0 = self._iTunesUint( + self.criteria[self.intAOffset - 4:self.intAOffset], self.criteria[self.offset] == IntFields.Rating) + idpart1 = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is %s" % self._formatPersistentID(idpart0, idpart1) + self.workingQuery += " = '%s'" % self._formatPersistentID(idpart0, idpart1) + self.workingFull["operator"] = "is" + self.workingFull["value"] = "%s" % self._formatPersistentID(idpart0, idpart1) + else: + self.workingOutput += " is not %s" % self._formatPersistentID(idpart0, idpart1) + self.workingQuery += " != '%s'" % self._formatPersistentID(idpart0, idpart1) + self.workingFull["operator"] = "is not" + self.workingFull["value"] = "%s" % self._formatPersistentID(idpart0, idpart1) + + else: # pragma: no cover + errormessage = "Unkown logic rule in ProcessPlaylistField: LogicRule=%d" % self.criteria[self.logicRulesOffset] + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + self.workingOutput += " ##UnkownCase PlaylistField:LogicRule##" + self.workingQuery += " ##UnkownCase PlaylistField:LogicRule##" + + self.workingQuery += ")" + + self.queryTreeCurrent.append((self.fieldName, self.workingQuery)) + self.fullTreeCurrent.append(self.workingFull) + + self.offset = self.intAOffset + Offset.INTLENGTH + if len(self.criteria) > self.offset: + self.again = True + + def ProcessBooleanField(self): + self.fieldName = BooleanFields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "(" + self.fieldName + self.workingFull = {"field": self.fieldName, "type": "boolean"} + + if self.criteria[self.logicRulesOffset] == LogicRule.Is: + value = self.criteria[self.logicSignOffset] != LogicSign.IntPositive + boolstr = "True" if value == 1 else "False" + + self.workingOutput += " is %s" % (boolstr) + self.workingQuery += " = %d" % (value) + self.workingFull["operator"] = "is" + self.workingFull["value"] = value == 1 + + else: # pragma: no cover + errormessage = "Unkown logic rule in ProcessBooleanField: LogicRule=%d" % self.criteria[self.logicRulesOffset] + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + self.workingOutput += " ##UnkownCase BooleanField:LogicRule##" + self.workingQuery += " ##UnkownCase BooleanField:LogicRule##" + + self.workingQuery += ")" + + self.queryTreeCurrent.append((self.fieldName, self.workingQuery)) + self.fullTreeCurrent.append(self.workingFull) + + self.offset = self.intAOffset + Offset.INTLENGTH + if len(self.criteria) > self.offset: + self.again = True + + def ProcessDateField(self): + self.fieldName = DateFields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "TIMESTAMP(%s)" % self.fieldName + self.workingFull = {"field": self.fieldName, "type": "date"} + + if self.criteria[self.logicRulesOffset] == LogicRule.Greater: + timestamp = self._iTunesDate( + self.criteria[self.intAOffset:self.intAOffset + 4]) + self.workingOutput += " is after %s" % self._dateString(timestamp) + self.workingQuery += " > %d" % timestamp + self.workingFull["operator"] = "is after" + self.workingFull["value"] = timestamp + elif self.criteria[self.logicRulesOffset] == LogicRule.Less: + timestamp = self._iTunesDate( + self.criteria[self.intAOffset:self.intAOffset + 4]) + self.workingOutput += " is before %s" % self._dateString(timestamp) + self.workingQuery += " < %d" % timestamp + self.workingFull["operator"] = "is before" + self.workingFull["value"] = timestamp + elif self.criteria[self.logicRulesOffset] == LogicRule.Other: + if self.criteria[self.logicSignOffset + 2] == 1: + timestampA = self._iTunesDate( + self.criteria[self.intAOffset:self.intAOffset + 4]) + timestampB = self._iTunesDate( + self.criteria[self.intBOffset:self.intBOffset + 4]) + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is in the range of %s to %s" % ( + self._dateString(timestampA), self._dateString(timestampB)) + self.workingQuery += " BETWEEN %d AND %d" % ( + timestampA, timestampB) + self.workingFull["operator"] = "is in the range" + self.workingFull["value"] = (timestampA, timestampB) + self.workingFull["value_date"] = ( + self._dateString(timestampA), self._dateString(timestampB)) + else: + self.workingOutput += " is not in the range of %s to %s" % ( + self._dateString(timestampA), self._dateString(timestampB)) + self.workingQuery += " NOT BETWEEN %d AND %d" % ( + timestampA, timestampB) + self.workingFull["operator"] = "is not in the range" + self.workingFull["value"] = (timestampA, timestampB) + self.workingFull["value_date"] = ( + self._dateString(timestampA), self._dateString(timestampB)) + elif self.criteria[self.logicSignOffset + 2] == 2: + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is in the last " + self.workingQuery = "(TIMESTAMP(NOW()) - TIMESTAMP(%s)) < " % self.fieldName + self.workingFull["operator"] = "is in the last" + else: + self.workingOutput += " is not in the last " + self.workingQuery = "(TIMESTAMP(NOW()) - TIMESTAMP(%s)) > " % self.fieldName + self.workingFull["operator"] = "is not in the last" + + t = (self._iTunesUint(bytes( + [255 - c for c in self.criteria[self.timeValueOffset:self.timeValueOffset + 4]])) + 1) % 4294967296 + multiple = self._iTunesUint( + self.criteria[self.timeMultipleOffset:self.timeMultipleOffset + 4]) + self.workingQuery += "%d" % (t * multiple) + self.workingFull["value"] = t * multiple + if multiple == 86400: + self.workingOutput += "%d days" % t + self.workingFull["value_date"] = "%d days" % t + elif multiple == 604800: + self.workingOutput += "%d weeks" % t + self.workingFull["value_date"] = "%d weeks" % t + elif multiple == 2628000: + self.workingOutput += "%d months" % t + self.workingFull["value_date"] = "%d months" % t + else: # pragma: no cover + self.workingOutput += "%d*%d seconds" % (multiple, t) + self.workingFull["value_date"] = "%d*%d seconds" % ( + multiple, t) + errormessage = "##UnkownCase DateField: LogicRule.Other: multiple '%d' is unkown##" % multiple + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + + self.queryTreeCurrent.append((self.fieldName, self.workingQuery)) + self.fullTreeCurrent.append(self.workingFull) + + self.offset = self.intAOffset + Offset.INTLENGTH + if len(self.criteria) > self.offset: + self.again = True + + def ProcessListField(self, fields, valueDict, listtype="list"): + self.fieldName = fields(self.criteria[self.offset]).name + self.workingOutput = self.fieldName + self.workingQuery = "(" + self.fieldName + self.workingFull = {"field": self.fieldName, "type": listtype} + + if self.criteria[self.logicRulesOffset] == LogicRule.Is: + number = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + value = self._lookupListValue(valueDict, number, self.fieldName, listtype) + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is %s" % value + self.workingQuery += " = '%s'" % value + self.workingFull["operator"] = "is" + self.workingFull["value"] = value + + else: + self.workingOutput += " is not %s" % value + self.workingQuery += " != '%s'" % value + self.workingFull["operator"] = "is not" + self.workingFull["value"] = value + + elif self.criteria[self.logicRulesOffset] == LogicRule.Other: + numberA = self._iTunesUint( + self.criteria[self.intAOffset:self.intAOffset + 4], self.criteria[self.offset] == IntFields.Rating) + numberB = self._iTunesUint( + self.criteria[self.intBOffset:self.intBOffset + 4], self.criteria[self.offset] == IntFields.Rating) + if numberA == numberB: + value = self._lookupListValue(valueDict, numberA, self.fieldName, listtype) + if self.criteria[self.logicSignOffset] == LogicSign.IntPositive: + self.workingOutput += " is %s" % value + self.workingQuery += " = '%s'" % value + self.workingFull["operator"] = "is" + self.workingFull["value"] = value + else: + self.workingOutput += " is not %s" % value + self.workingQuery += " != '%s'" % value + self.workingFull["operator"] = "is not" + self.workingFull["value"] = value + + else: # pragma: no cover + errormessage = "Unkown case in ProcessListField %s:LogicRule.Other: %d != %d" % (self.fieldName, numberA, numberB) + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + + self.workingOutput += " ##UnkownCase ListField %s: LogicRule.Other##" % self.fieldName + self.workingQuery += " ##UnkownCase ListField %s: LogicRule.Other##" % self.fieldName + else: # pragma: no cover + errormessage = "Unkown logic rule in ProcessListField %s: LogicRule=%d" % (self.fieldName, self.criteria[self.logicRulesOffset]) + logging.warning(errormessage) + self.ignore += " Not processed: %s " % errormessage + + self.workingOutput += " ##UnkownCase ListField %s:LogicRule##" % self.fieldName + self.workingQuery += " ##UnkownCase ListField %s:LogicRule##" % self.fieldName + + self.workingQuery += ")" + + self.queryTreeCurrent.append((self.fieldName, self.workingQuery)) + self.fullTreeCurrent.append(self.workingFull) + + self.offset = self.intAOffset + Offset.INTLENGTH + if len(self.criteria) > self.offset: + self.again = True + + @staticmethod + def _iTunesUint(bytearr, divideby=False, denominator=20): + num = struct.unpack('>I', bytearr)[0] + if divideby: # For rating/stars by 20 + num = int(num / denominator) + return num + + @classmethod + def _iTunesDate(cls, bytearr): + timestamp = cls._iTunesUint(bytearr) + return timestamp + DateStartFromUnix + + @staticmethod + def _dateString(timestamp): + if sys.version_info < (3, 11): + return datetime.datetime.utcfromtimestamp( + int(timestamp)).strftime('%Y-%m-%dT%H:%M:%SZ') + return datetime.datetime.fromtimestamp( + int(timestamp), datetime.UTC).strftime('%Y-%m-%dT%H:%M:%SZ') + + @staticmethod + def _formatPersistentID(idpart0, idpart1): + return '{:08X}{:08X}'.format(idpart0, idpart1) + + @staticmethod + def _operatorFromLogicType(logictype): + if logictype == 1: + return "or" + if logictype == 0: + return "and" + + logging.error("Unknown logic type encountered: %d; defaulting to 'and'", logictype) + return "and" + + @staticmethod + def _lookupListValue(valueDict, number, fieldName="unknown", listtype="list"): + if number in valueDict: + return valueDict[number] + logging.warning("Unknown %s list value encountered for %s: %d", listtype, fieldName, number) + return "UnknownValue[%d]" % number + + def _buildQuery(self, node, nested=False): + if not node: + return "" + + if isinstance(node, tuple): + return node[1] + + if isinstance(node, dict): + if "and" in node: + rendered = " AND ".join( + filter(None, (self._buildQuery(child, nested=True) for child in node["and"])) + ) + elif "or" in node: + rendered = " OR ".join( + filter(None, (self._buildQuery(child, nested=True) for child in node["or"])) + ) + else: + return "" + if not rendered: + return "" + if nested: + return "( %s )" % rendered + return rendered + + return "" + + def _stripImplicitMediaKindFilter(self, node): + if not isinstance(node, dict) or "and" not in node: + return node + + children = list(node["and"]) + if len(children) < 2 or not self._isImplicitMediaKindFilter(children[0]): + return node + + remaining = children[1:] + if len(remaining) == 1: + return remaining[0] + return {"and": remaining} + + @staticmethod + def _isImplicitMediaKindFilter(node): + if isinstance(node, tuple): + return node[0] == "MediaKind" and node[1] in { + "(MediaKind = 'Music')", + "(MediaKind = 'Music Video')", + } + + if isinstance(node, dict) and "or" not in node: + return ( + node.get("field") == "MediaKind" and + node.get("operator") == "is" and + node.get("value") in {"Music", "Music Video"} + ) + + if not isinstance(node, dict) or "or" not in node: + return False + + values = [] + for child in node["or"]: + if isinstance(child, tuple): + if child[0] != "MediaKind": + return False + if child[1] == "(MediaKind = 'Music')": + values.append("Music") + elif child[1] == "(MediaKind = 'Music Video')": + values.append("Music Video") + else: + return False + elif isinstance(child, dict): + if child.get("field") != "MediaKind" or child.get("operator") != "is": + return False + if child.get("value") not in {"Music", "Music Video"}: + return False + values.append(child["value"]) + else: + return False + + return sorted(values) == ["Music", "Music Video"] + + def _buildOutputTree(self, node, nested=False): + if not node: + return "" + + if isinstance(node, dict): + if "and" in node: + rendered = ' and\n'.join( + filter(None, (self._buildOutputTree(child, nested=True) for child in node["and"])) + ) + return self._formatNestedOutput(rendered) if nested else rendered + if "or" in node: + rendered = ' or\n'.join( + filter(None, (self._buildOutputTree(child, nested=True) for child in node["or"])) + ) + return self._formatNestedOutput(rendered) if nested else rendered + return self._renderOutputLeaf(node) + + return "" + + @staticmethod + def _formatNestedOutput(rendered): + if not rendered: + return "" + return "[\n\t%s\n]" % "\n\t".join(rendered.split("\n")) + + def _renderOutputLeaf(self, node): + field = node["field"] + operator = node["operator"] + value = node.get("value_date", node.get("value")) + + if node["type"] == "string": + if operator == "like": + return '%s contains "%s" ' % (field, value) + if operator == "not like": + return '%s does not contain "%s" ' % (field, value) + if operator == "is": + return '%s is "%s" ' % (field, value) + if operator == "is not": + return '%s is not "%s" ' % (field, value) + if operator == "starts with": + return '%s starts with "%s" ' % (field, value) + if operator == "ends with": + return '%s ends with "%s" ' % (field, value) + + if operator == "is": + return "%s is %s" % (field, value) + + if operator == "is not": + return "%s is not %s" % (field, value) + + if operator in {"is in the range", "is not in the range"} and isinstance(value, tuple): + return "%s %s of %s to %s" % (field, operator, value[0], value[1]) + + if operator == "greater than": + return "%s is greater than %s" % (field, value) + + if operator == "less than": + return "%s is less than %s" % (field, value) + + if operator == "is after": + return "%s is after %s" % (field, value) + + if operator == "is before": + return "%s is before %s" % (field, value) + + if operator == "is in the last": + return "%s is in the last %s" % (field, value) + + if operator == "is not in the last": + return "%s is not in the last %s" % (field, value) + + if operator == "between" and isinstance(value, tuple): + return "%s is in the range of %s to %s" % (field, value[0], value[1]) + + if operator == "not between" and isinstance(value, tuple): + return "%s is not in the range of %s to %s" % (field, value[0], value[1]) + + return "%s %s %s" % (field, operator, value) diff --git a/lintunes/library_manager.py b/lintunes/library_manager.py index 40489ac..b1b8d32 100644 --- a/lintunes/library_manager.py +++ b/lintunes/library_manager.py @@ -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) ---- diff --git a/lintunes/main.py b/lintunes/main.py index 1b1c293..1ad52cb 100644 --- a/lintunes/main.py +++ b/lintunes/main.py @@ -121,6 +121,9 @@ def run_gui(data_dir: Path, files: list[Path], qt_args: list[str] | None = None) theme.apply_theme(app, prefs) _ensure_now_playing_font(prefs) manager = LibraryManager(library, data_dir) + # Rebuild smart-playlist membership from criteria now that the full library + # (and any merged sync conflicts) is loaded. Cheap when nothing changed. + manager.recompute_all_smart() lastfm = LastFm(prefs, data_dir) window = MainWindow(manager, prefs, lastfm) app.installEventFilter(window) diff --git a/lintunes/models/playlist.py b/lintunes/models/playlist.py index c106204..8dc2c1c 100644 --- a/lintunes/models/playlist.py +++ b/lintunes/models/playlist.py @@ -58,6 +58,14 @@ class Playlist: track_ids: list[int] = field(default_factory=list) settings: PlaylistSettings = field(default_factory=PlaylistSettings) is_system: bool = False + # Set only for SMART playlists. Membership (track_ids) is derived from this + # by the LibraryManager; for imported playlists track_ids also doubles as a + # snapshot fallback when the criteria couldn't be fully parsed. + smart_criteria: Optional["SmartCriteria"] = None + + @property + def is_smart(self) -> bool: + return self.playlist_type == PlaylistType.SMART def to_dict(self) -> dict: d = { @@ -70,10 +78,16 @@ class Playlist: } if self.parent_persistent_id: d["parent_persistent_id"] = self.parent_persistent_id + if self.smart_criteria is not None: + d["smart_criteria"] = self.smart_criteria.to_dict() return d @classmethod def from_dict(cls, d: dict) -> "Playlist": + smart_criteria = None + if d.get("smart_criteria") is not None: + from lintunes.smart import SmartCriteria + smart_criteria = SmartCriteria.from_dict(d["smart_criteria"]) return cls( name=d.get("name", ""), persistent_id=d.get("persistent_id", ""), @@ -82,6 +96,7 @@ class Playlist: track_ids=d.get("track_ids", []), settings=PlaylistSettings.from_dict(d.get("settings", {})), is_system=d.get("is_system", False), + smart_criteria=smart_criteria, ) @classmethod diff --git a/lintunes/smart.py b/lintunes/smart.py new file mode 100644 index 0000000..a8ddc26 --- /dev/null +++ b/lintunes/smart.py @@ -0,0 +1,678 @@ +"""Smart playlists: criteria model, evaluation, and iTunes import. + +A smart playlist's membership is *derived* from a `SmartCriteria` (a tree of +rules) rather than hand-curated. This module owns: + +* the data model (`SmartRule` / `SmartGroup` / `SmartLimit` / `SmartCriteria`), + recursive so nested rule groups round-trip through the playlist JSON; +* `FIELD_REGISTRY`, the single source of truth shared by the evaluator and the + editor dialog (field -> Track attribute, value type, allowed operators); +* `evaluate()`, which walks the tree against the library and returns the + ordered list of matching track ids; and +* `parse_itunes_smart()`, which converts the iTunes "Smart Info"/"Smart + Criteria" binary blobs into a `SmartCriteria` (via the vendored parser in + `lintunes.itunes_smart`). + +The evaluator reads tracks purely by attribute name (`getattr`), so this module +imports nothing from `lintunes.models` -- keeping it free of import cycles with +`playlist.py`. +""" + +from __future__ import annotations + +import logging +import random as _random +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from enum import Enum +from typing import Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- # +# Field metadata +# --------------------------------------------------------------------------- # + +class FieldType(Enum): + STRING = "string" + INT = "int" + DURATION = "duration" # milliseconds, like INT but rendered as time + DATE = "date" + BOOL = "bool" + RATING = "rating" # stored 0-100, edited as 0-5 stars + + +# Operator vocabularies per field type. +STRING_OPS = ("is", "is_not", "contains", "not_contains", "starts_with", "ends_with") +NUMERIC_OPS = ("is", "is_not", "greater", "less", "in_range") +DATE_OPS = ("after", "before", "in_range", "not_in_range", "in_last", "not_in_last") +BOOL_OPS = ("is",) + + +@dataclass(frozen=True) +class FieldMeta: + key: str + label: str + track_attr: str + type: FieldType + operators: tuple + + @property + def ops(self) -> tuple: + return self.operators + + +def _ops_for(ftype: FieldType) -> tuple: + if ftype is FieldType.STRING: + return STRING_OPS + if ftype in (FieldType.INT, FieldType.DURATION, FieldType.RATING): + return NUMERIC_OPS + if ftype is FieldType.DATE: + return DATE_OPS + return BOOL_OPS + + +def _meta(key, label, attr, ftype) -> FieldMeta: + return FieldMeta(key, label, attr, ftype, _ops_for(ftype)) + + +# The fields the editor offers and the evaluator understands. Only fields the +# LinTunes Track actually backs are present (the library is music-only). +FIELD_REGISTRY: dict[str, FieldMeta] = {m.key: m for m in [ + _meta("name", "Name", "name", FieldType.STRING), + _meta("artist", "Artist", "artist", FieldType.STRING), + _meta("album_artist", "Album Artist", "album_artist", FieldType.STRING), + _meta("album", "Album", "album", FieldType.STRING), + _meta("genre", "Genre", "genre", FieldType.STRING), + _meta("composer", "Composer", "composer", FieldType.STRING), + _meta("grouping", "Grouping", "grouping", FieldType.STRING), + _meta("comments", "Comments", "comments", FieldType.STRING), + _meta("kind", "Kind", "kind", FieldType.STRING), + _meta("sort_name", "Sort Name", "sort_name", FieldType.STRING), + _meta("sort_album", "Sort Album", "sort_album", FieldType.STRING), + _meta("year", "Year", "year", FieldType.INT), + _meta("bpm", "BPM", "bpm", FieldType.INT), + _meta("bit_rate", "Bit Rate", "bit_rate", FieldType.INT), + _meta("sample_rate", "Sample Rate", "sample_rate", FieldType.INT), + _meta("size", "Size", "size", FieldType.INT), + _meta("disc_number", "Disc Number", "disc_number", FieldType.INT), + _meta("track_number", "Track Number", "track_number", FieldType.INT), + _meta("play_count", "Plays", "play_count", FieldType.INT), + _meta("skip_count", "Skips", "skip_count", FieldType.INT), + _meta("rating", "Rating", "rating", FieldType.RATING), + _meta("total_time", "Time", "total_time", FieldType.DURATION), + _meta("compilation", "Compilation", "compilation", FieldType.BOOL), + _meta("date_added", "Date Added", "date_added", FieldType.DATE), + _meta("date_modified", "Date Modified", "date_modified", FieldType.DATE), + _meta("last_played", "Last Played", "play_date_utc", FieldType.DATE), + _meta("last_skipped", "Last Skipped", "skip_date", FieldType.DATE), +]} + + +# Limit "selected by" methods -> (Track attribute to sort on, reverse?). +# "random" is special-cased. Used by the evaluator to pick which N items. +SELECTION_SORT: dict[str, tuple] = { + "random": (None, False), + "name": ("sort_name", False), + "album": ("sort_album", False), + "artist": ("sort_artist", False), + "genre": ("genre", False), + "highest_rating": ("rating", True), + "lowest_rating": ("rating", False), + "most_recently_played": ("play_date_utc", True), + "least_recently_played": ("play_date_utc", False), + "most_often_played": ("play_count", True), + "least_often_played": ("play_count", False), + "most_recently_added": ("date_added", True), + "least_recently_added": ("date_added", False), +} + +_UNIT_SECONDS = {"days": 86400, "weeks": 604800, "months": 2628000} + + +# --------------------------------------------------------------------------- # +# Schema +# --------------------------------------------------------------------------- # + +@dataclass +class SmartRule: + field: str = "artist" + operator: str = "contains" + value: object = "" + value2: object = None # upper bound for in_range/not_in_range + unit: Optional[str] = None # "days"|"weeks"|"months" for in_last/not_in_last + + def to_dict(self) -> dict: + d = {"kind": "rule", "field": self.field, + "operator": self.operator, "value": self.value} + if self.value2 is not None: + d["value2"] = self.value2 + if self.unit is not None: + d["unit"] = self.unit + return d + + @classmethod + def from_dict(cls, d: dict) -> "SmartRule": + return cls(field=d.get("field", ""), operator=d.get("operator", ""), + value=d.get("value", ""), value2=d.get("value2"), + unit=d.get("unit")) + + +@dataclass +class SmartGroup: + conjunction: str = "all" # "all" (AND) | "any" (OR) + children: list = field(default_factory=list) # SmartRule | SmartGroup + + def to_dict(self) -> dict: + return {"kind": "group", "conjunction": self.conjunction, + "children": [c.to_dict() for c in self.children]} + + @classmethod + def from_dict(cls, d: dict) -> "SmartGroup": + return cls(conjunction=d.get("conjunction", "all"), + children=[_node_from_dict(c) for c in d.get("children", [])]) + + +def _node_from_dict(d: dict): + if d.get("kind") == "group" or "children" in d: + return SmartGroup.from_dict(d) + return SmartRule.from_dict(d) + + +@dataclass +class SmartLimit: + enabled: bool = False + count: int = 25 + by: str = "items" # items|minutes|hours|mb|gb + selection: str = "random" # key into SELECTION_SORT + + def to_dict(self) -> dict: + return {"enabled": self.enabled, "count": self.count, + "by": self.by, "selection": self.selection} + + @classmethod + def from_dict(cls, d: dict) -> "SmartLimit": + return cls(enabled=d.get("enabled", False), count=d.get("count", 25), + by=d.get("by", "items"), selection=d.get("selection", "random")) + + +@dataclass +class SmartCriteria: + root: SmartGroup = field(default_factory=SmartGroup) + limit: SmartLimit = field(default_factory=SmartLimit) + live_update: bool = True + match_only_checked: bool = False + # True when the iTunes blob couldn't be fully converted; the manager then + # keeps the imported track snapshot instead of recomputing membership. + unsupported: bool = False + # True when the rule tree contains a nested group; the editor is read-only + # for these in Phase 1 (nested-group editing UI is Phase 2). + has_nested: bool = False + + def to_dict(self) -> dict: + return { + "root": self.root.to_dict(), + "limit": self.limit.to_dict(), + "live_update": self.live_update, + "match_only_checked": self.match_only_checked, + "unsupported": self.unsupported, + "has_nested": self.has_nested, + } + + @classmethod + def from_dict(cls, d: dict) -> "SmartCriteria": + root_d = d.get("root") + return cls( + root=SmartGroup.from_dict(root_d) if root_d else SmartGroup(), + limit=SmartLimit.from_dict(d.get("limit", {})), + live_update=d.get("live_update", True), + match_only_checked=d.get("match_only_checked", False), + unsupported=d.get("unsupported", False), + has_nested=d.get("has_nested", False), + ) + + def references_attr(self, attr: str) -> bool: + """True if any rule matches on the given Track attribute.""" + return _group_references(self.root, attr) + + def referenced_attrs(self) -> set: + out: set = set() + _collect_attrs(self.root, out) + return out + + +def _group_references(group: SmartGroup, attr: str) -> bool: + for child in group.children: + if isinstance(child, SmartGroup): + if _group_references(child, attr): + return True + else: + meta = FIELD_REGISTRY.get(child.field) + if meta and meta.track_attr == attr: + return True + return False + + +def _collect_attrs(group: SmartGroup, out: set) -> None: + for child in group.children: + if isinstance(child, SmartGroup): + _collect_attrs(child, out) + else: + meta = FIELD_REGISTRY.get(child.field) + if meta: + out.add(meta.track_attr) + + +# --------------------------------------------------------------------------- # +# Evaluation +# --------------------------------------------------------------------------- # + +def _parse_dt(value) -> Optional[datetime]: + """Parse a Track ISO date string into a naive-UTC datetime.""" + if not value: + return None + if isinstance(value, datetime): + dt = value + else: + try: + dt = datetime.fromisoformat(str(value)) + except (ValueError, TypeError): + return None + if dt.tzinfo is not None: + dt = dt.replace(tzinfo=None) + return dt + + +def _match_string(op, track_val, qval) -> bool: + t = ("" if track_val is None else str(track_val)).casefold() + q = ("" if qval is None else str(qval)).casefold() + if op == "is": + return t == q + if op == "is_not": + return t != q + if op == "contains": + return q in t + if op == "not_contains": + return q not in t + if op == "starts_with": + return t.startswith(q) + if op == "ends_with": + return t.endswith(q) + return False + + +def _match_numeric(op, track_val, qval, qval2) -> bool: + try: + t = float(track_val or 0) + q = float(qval or 0) + except (ValueError, TypeError): + return False + if op == "is": + return t == q + if op == "is_not": + return t != q + if op == "greater": + return t > q + if op == "less": + return t < q + if op == "in_range": + try: + q2 = float(qval2 or 0) + except (ValueError, TypeError): + return False + lo, hi = (q, q2) if q <= q2 else (q2, q) + return lo <= t <= hi + return False + + +def _match_date(op, track_val, rule: "SmartRule", now: datetime) -> bool: + dt = _parse_dt(track_val) + if op in ("after", "before", "in_range"): + # iTunes treats a null date (never played/skipped/etc.) as the distant + # past: "before X" matches it; "after"/"in range" don't. + d = dt if dt is not None else datetime.min + if op == "in_range": + lo, hi = _parse_dt(rule.value), _parse_dt(rule.value2) + if lo is None or hi is None: + return False + if lo > hi: + lo, hi = hi, lo + return lo <= d <= hi + q = _parse_dt(rule.value) + if q is None: + return False + return d > q if op == "after" else d < q + if op == "not_in_range": + if dt is None: + return True + lo, hi = _parse_dt(rule.value), _parse_dt(rule.value2) + if lo is None or hi is None: + return True + if lo > hi: + lo, hi = hi, lo + return not (lo <= dt <= hi) + if op in ("in_last", "not_in_last"): + seconds = _rule_window_seconds(rule) + if op == "in_last": + return dt is not None and (now - dt) <= timedelta(seconds=seconds) + return dt is None or (now - dt) > timedelta(seconds=seconds) + return False + + +def _rule_window_seconds(rule: "SmartRule") -> float: + try: + count = float(rule.value or 0) + except (ValueError, TypeError): + count = 0 + return count * _UNIT_SECONDS.get(rule.unit or "days", 86400) + + +def _match_rule(rule: SmartRule, track, now: datetime) -> bool: + meta = FIELD_REGISTRY.get(rule.field) + if meta is None: + return False + track_val = getattr(track, meta.track_attr, None) + if meta.type is FieldType.STRING: + return _match_string(rule.operator, track_val, rule.value) + if meta.type in (FieldType.INT, FieldType.DURATION, FieldType.RATING): + return _match_numeric(rule.operator, track_val, rule.value, rule.value2) + if meta.type is FieldType.DATE: + return _match_date(rule.operator, track_val, rule, now) + if meta.type is FieldType.BOOL: + return bool(track_val) == bool(rule.value) + return False + + +def _match_group(group: SmartGroup, track, now: datetime) -> bool: + if not group.children: + return True + results = ( + _match_group(c, track, now) if isinstance(c, SmartGroup) + else _match_rule(c, track, now) + for c in group.children + ) + return all(results) if group.conjunction == "all" else any(results) + + +def _selection_key(selection: str, pid_seed: int): + attr, reverse = SELECTION_SORT.get(selection, (None, False)) + + if attr is None: # random -> deterministic shuffle (stable between runs) + rng = _random.Random(pid_seed) + + def keyfn(track): + return rng.random() + return keyfn, False + + def keyfn(track): + val = getattr(track, attr, None) + if attr in ("play_date_utc", "date_added"): + dt = _parse_dt(val) + # Tracks never played/added sort oldest. + return dt or datetime.min + return val if val is not None else "" + return keyfn, reverse + + +def _apply_limit(matches: list, limit: SmartLimit, seed: int) -> list: + keyfn, reverse = _selection_key(limit.selection, seed) + ordered = sorted(matches, key=keyfn, reverse=reverse) + if limit.by == "items": + return ordered[:max(0, limit.count)] + # Cumulative budget by time or size. + if limit.by in ("minutes", "hours"): + budget = limit.count * (60000 if limit.by == "minutes" else 3600000) + attr = "total_time" + else: # mb, gb + budget = limit.count * (1024 ** 2 if limit.by == "mb" else 1024 ** 3) + attr = "size" + out, total = [], 0 + for track in ordered: + amount = getattr(track, attr, 0) or 0 + if out and total + amount > budget: + break + out.append(track) + total += amount + return out + + +def evaluate(criteria: SmartCriteria, tracks, now: Optional[datetime] = None, + seed: int = 0) -> list: + """Return the ordered list of track ids matching `criteria`. + + `tracks` is any iterable of Track objects. `now` (naive UTC) is injectable + for deterministic date tests. `seed` makes "limit ... selected by random" + stable between recomputes (pass the playlist's persistent id hash). + """ + if now is None: + now = datetime.now(timezone.utc).replace(tzinfo=None) + matched = [t for t in tracks if _match_group(criteria.root, t, now)] + if criteria.limit.enabled and criteria.limit.count > 0: + matched = _apply_limit(matched, criteria.limit, seed) + return sorted(t.track_id for t in matched) + + +# --------------------------------------------------------------------------- # +# iTunes import +# --------------------------------------------------------------------------- # + +class _UnsupportedCriteria(Exception): + """Raised mid-conversion when a rule can't be represented in our model.""" + + +# cvzi field enum name -> our registry key. +_ITUNES_FIELD_TO_KEY = { + "Name": "name", "Artist": "artist", "AlbumArtist": "album_artist", + "Album": "album", "Genre": "genre", "Composer": "composer", + "Grouping": "grouping", "Comments": "comments", "Kind": "kind", + "SortName": "sort_name", "SortAlbum": "sort_album", + "Year": "year", "BPM": "bpm", "BitRate": "bit_rate", + "SampleRate": "sample_rate", "Size": "size", "DiskNumber": "disc_number", + "TrackNumber": "track_number", "Plays": "play_count", "Skips": "skip_count", + "Rating": "rating", "Duration": "total_time", "Compilation": "compilation", + "DateAdded": "date_added", "DateModified": "date_modified", + "LastPlayed": "last_played", "LastSkipped": "last_skipped", +} + +_ITUNES_STRING_OP = { + "like": "contains", "not like": "not_contains", "is": "is", + "is not": "is_not", "starts with": "starts_with", "ends with": "ends_with", +} +_ITUNES_INT_OP = { + "is": "is", "is not": "is_not", "greater than": "greater", + "less than": "less", "between": "in_range", +} +_ITUNES_DATE_OP = { + "is after": "after", "is before": "before", "is in the range": "in_range", + "is not in the range": "not_in_range", "is in the last": "in_last", + "is not in the last": "not_in_last", +} + + +def _unix_to_iso(ts) -> str: + return datetime.fromtimestamp(int(ts), timezone.utc).replace(tzinfo=None).isoformat() + + +def _convert_leaf(leaf: dict): + """Convert a cvzi fullTree leaf dict into a SmartRule (or None to drop).""" + fname = leaf.get("field", "") + ltype = leaf.get("type", "") + op = leaf.get("operator", "") + value = leaf.get("value") + + # Fields with no Track backing. + if fname == "Love": + return None # user does not use "loved"; drop the rule entirely + if fname == "MediaKind": + if op == "is" and value in ("Music", "Music Video"): + return None # implicit/no-op filter in a music-only library + raise _UnsupportedCriteria("MediaKind rule") + + key = _ITUNES_FIELD_TO_KEY.get(fname) + if key is None or key not in FIELD_REGISTRY: + raise _UnsupportedCriteria("unmapped field %r" % fname) + meta = FIELD_REGISTRY[key] + + if meta.type is FieldType.STRING: + our_op = _ITUNES_STRING_OP.get(op) + if our_op is None: + raise _UnsupportedCriteria("string op %r" % op) + return SmartRule(field=key, operator=our_op, value=str(value)) + + if meta.type is FieldType.BOOL: # compilation + truth = bool(value) + if op == "is not": + truth = not truth + return SmartRule(field=key, operator="is", value=truth) + + if meta.type in (FieldType.INT, FieldType.DURATION, FieldType.RATING): + our_op = _ITUNES_INT_OP.get(op) + if our_op is None: + raise _UnsupportedCriteria("int op %r" % op) + scale = 20 if meta.type is FieldType.RATING else 1 + if our_op == "in_range" and isinstance(value, (tuple, list)): + return SmartRule(field=key, operator="in_range", + value=int(value[0]) * scale, + value2=int(value[1]) * scale) + return SmartRule(field=key, operator=our_op, value=int(value) * scale) + + if meta.type is FieldType.DATE: + our_op = _ITUNES_DATE_OP.get(op) + if our_op is None: + raise _UnsupportedCriteria("date op %r" % op) + if our_op in ("in_last", "not_in_last"): + count, unit = _parse_relative_date(leaf.get("value_date"), value) + return SmartRule(field=key, operator=our_op, value=count, unit=unit) + if our_op in ("in_range", "not_in_range") and isinstance(value, (tuple, list)): + return SmartRule(field=key, operator=our_op, + value=_unix_to_iso(value[0]), + value2=_unix_to_iso(value[1])) + return SmartRule(field=key, operator=our_op, value=_unix_to_iso(value)) + + raise _UnsupportedCriteria("unhandled type %r" % meta.type) + + +def _parse_relative_date(value_date, raw_seconds): + """cvzi gives value_date like '12 months'; fall back to raw seconds.""" + if isinstance(value_date, str): + parts = value_date.split() + if len(parts) == 2 and parts[1] in _UNIT_SECONDS: + try: + return int(parts[0]), parts[1] + except ValueError: + pass + # Fallback: convert raw seconds to whole days. + try: + return max(1, int(int(raw_seconds) // 86400)), "days" + except (ValueError, TypeError): + return 1, "days" + + +def _convert_group(node: dict) -> SmartGroup: + op = "and" if "and" in node else "or" if "or" in node else None + if op is None: + return SmartGroup("all", []) + children = [] + for child in node[op]: + if isinstance(child, dict) and ("and" in child or "or" in child): + children.append(_convert_group(child)) + else: + rule = _convert_leaf(child) + if rule is not None: + children.append(rule) + return SmartGroup("all" if op == "and" else "any", children) + + +def _simplify(group: SmartGroup) -> SmartGroup: + """Drop empty groups (left behind by dropped MediaKind/Love rules) and + unwrap a group whose only child is itself a group. iTunes wraps every + playlist in an implicit "MediaKind is Music" group; stripping that keeps + ordinary playlists flat (and therefore editable).""" + new_children = [] + for child in group.children: + if isinstance(child, SmartGroup): + child = _simplify(child) + if not child.children: + continue + new_children.append(child) + group.children = new_children + if len(group.children) == 1 and isinstance(group.children[0], SmartGroup): + return group.children[0] + return group + + +def _contains_group(group: SmartGroup) -> bool: + return any(isinstance(c, SmartGroup) for c in group.children) + + +def parse_itunes_smart(info: Optional[bytes], + criteria: Optional[bytes]) -> SmartCriteria: + """Decode the iTunes "Smart Info"/"Smart Criteria" blobs into a + SmartCriteria. Never raises: on any problem it returns a criteria flagged + `unsupported=True`, so the importer falls back to the track snapshot. + """ + if not info or not criteria: + return SmartCriteria(unsupported=True) + info, criteria = bytes(info), bytes(criteria) + # Real iTunes criteria blobs always start with the "SLst" magic. Anything + # else is junk we shouldn't trust (and would otherwise parse to an empty + # "match everything" rule set), so fall back to the snapshot. + if not criteria.startswith(b"SLst") or len(info) < 14: + return SmartCriteria(unsupported=True) + try: + from lintunes.itunes_smart.parse import SmartPlaylistParser + parser = SmartPlaylistParser() + parser.data(info, criteria) + parser.parse() + except Exception as exc: # noqa: BLE001 - import must never abort + logger.warning("smart criteria parse failed: %r", exc) + return SmartCriteria(unsupported=True) + + limit = _read_limit(bytes(info)) + result = SmartCriteria(limit=limit, + live_update=bool(info[0]), + match_only_checked=bool(info[12])) + + if getattr(parser, "ignore", "").strip(): + # The parser hit a field/case it couldn't decode -> incomplete tree. + result.unsupported = True + return result + + try: + root = _simplify(_convert_group(parser.fullTreeRoot or {})) + except _UnsupportedCriteria as exc: + logger.info("smart criteria not fully supported: %s", exc) + result.unsupported = True + return result + result.root = root + result.has_nested = _contains_group(root) + return result + + +_LIMIT_METHOD = {1: "minutes", 2: "mb", 3: "items", 4: "hours", 5: "gb"} +# (iTunes selection-method byte, "is least" sign) -> our selection key. +_SELECTION_BASE = { + 0x02: "random", 0x05: "name", 0x06: "album", 0x07: "artist", 0x09: "genre", + 0x1c: "highest_rating", 0x01: "lowest_rating", +} +_SELECTION_SIGNED = { + 0x1a: ("most_recently_played", "least_recently_played"), + 0x19: ("most_often_played", "least_often_played"), + 0x15: ("most_recently_added", "least_recently_added"), +} + + +def _read_limit(info: bytes) -> SmartLimit: + if len(info) < 14 or info[2] != 1: + return SmartLimit(enabled=False) + count = int.from_bytes(info[8:12], "big") + by = _LIMIT_METHOD.get(info[3], "items") + method = info[7] + if method in _SELECTION_SIGNED: + # cvzi: sign == 1 (i.e. the sign byte is 0) selects the "least" variant. + least = (info[13] == 0) + selection = _SELECTION_SIGNED[method][1 if least else 0] + else: + selection = _SELECTION_BASE.get(method, "random") + return SmartLimit(enabled=True, count=count, by=by, selection=selection) diff --git a/lintunes/storage/conflict_resolver.py b/lintunes/storage/conflict_resolver.py index 940779a..e46478f 100644 --- a/lintunes/storage/conflict_resolver.py +++ b/lintunes/storage/conflict_resolver.py @@ -117,6 +117,13 @@ def _merge_playlist(original_path: Path, conflict_path: Path): original = _load_json(original_path) conflict = _load_json(conflict_path) + # Smart playlists: membership is derived, so unioning track_ids would + # resurrect tracks that no longer match. Keep the newer file's criteria; + # membership is recomputed from it at the next launch. + if original.get("playlist_type") == "smart" or conflict.get("playlist_type") == "smart": + _keep_newer(original_path, conflict_path) + return + # Union track_ids, order from the newer file orig_ids = set(original.get("track_ids", [])) conf_ids = set(conflict.get("track_ids", [])) diff --git a/tests/smart_blobs.json b/tests/smart_blobs.json new file mode 100644 index 0000000..2d3f273 --- /dev/null +++ b/tests/smart_blobs.json @@ -0,0 +1,22 @@ +{ + "convert these": { + "items": 0, + "info_b64": "AQEAAwAAAAIAAAAZAAAAAAAAAAcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==", + "crit_b64": "U0xzdAABAAEAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEEU0xzdAABAAEAAAABAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADwAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABEAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA1FNMc3QAAQABAAAAAQAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAJAQAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFABzAG8AdQBuAGQAIABmAGkAbABl" + }, + "sun": { + "items": 156, + "info_b64": "AQEAAwAAAAIAAAAZAAAAAAAAAAcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==", + "crit_b64": "U0xzdAABAAEAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGAU0xzdAABAAEAAAACAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADwAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABEAAAAAAAAAAEAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAEAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA8AAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAARAAAAAAAAAAgAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAAgAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQpTTHN0AAEAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAgEAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAYAcwB1AG4AAAACAwAAAgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADABzAHUAbgBkAGEAeQ==" + }, + "top 300 of the decade!": { + "items": 300, + "info_b64": "AQEBAwAAABkAAAEsAAAAAAAAAAcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==", + "crit_b64": "U0xzdAABAAEAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABELa4tri2uLa7/////////iAAAAAAAKBmgLa4tri2uLa4AAAAAAAAAAAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAA=" + }, + "unrated": { + "items": 4035, + "info_b64": "AQEAAwAAAAIAAAAZAAAAAAAAAAcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==", + "crit_b64": "U0xzdAABAAEAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABkAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAkAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAARAAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAABAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAA" + } +} \ No newline at end of file diff --git a/tests/test_itunes_importer.py b/tests/test_itunes_importer.py index f1a3e13..0c3bb43 100644 --- a/tests/test_itunes_importer.py +++ b/tests/test_itunes_importer.py @@ -114,11 +114,18 @@ class TestItunesImporter: xml_path = _make_itunes_xml(tracks, playlists) library, report = import_itunes_xml(xml_path, MUSIC_ROOT, check_files=False) - assert set(library.playlists.keys()) == {"FOLDER1", "PL1"} + # Smart playlists are now imported (the junk blobs can't be parsed, so + # the criteria are flagged unsupported and the iTunes snapshot is kept). + assert set(library.playlists.keys()) == {"FOLDER1", "PL1", "SMART1"} + smart = library.playlists["SMART1"] + assert smart.playlist_type.value == "smart" + assert smart.smart_criteria is not None and smart.smart_criteria.unsupported + assert smart.track_ids == [1] # snapshot fallback # Track 999 doesn't exist — dropped from the playlist assert library.playlists["PL1"].track_ids == [2] assert library.playlists["PL1"].parent_persistent_id == "FOLDER1" - assert report.skipped_smart == 1 + assert report.smart_count == 1 + assert report.smart_unsupported == 1 assert report.skipped_system == 2 assert report.playlist_count == 1 assert report.folder_count == 1 diff --git a/tests/test_round14.py b/tests/test_round14.py new file mode 100644 index 0000000..fb97a9f --- /dev/null +++ b/tests/test_round14.py @@ -0,0 +1,327 @@ +"""Round 14: smart playlists — criteria model, evaluation, iTunes import, +manager recompute/guards, and conflict resolution. + +The iTunes blobs in tests/smart_blobs.json are real "Smart Info"/"Smart +Criteria" data captured from the test library (gitignored), base64-encoded so +the parser is exercised against genuine iTunes 12 output. +""" +import base64 +import json +import os +from datetime import datetime +from pathlib import Path + +from lintunes.smart import ( + SmartCriteria, SmartGroup, SmartRule, SmartLimit, evaluate, parse_itunes_smart, +) +from lintunes.library_manager import LibraryManager +from lintunes.models import Library, Track, Playlist, PlaylistType + + +_BLOBS = json.loads((Path(__file__).parent / "smart_blobs.json").read_text()) + + +def _blob(name): + rec = _BLOBS[name] + return base64.b64decode(rec["info_b64"]), base64.b64decode(rec["crit_b64"]) + + +# --------------------------------------------------------------------------- # +# iTunes parser / converter (real captured blobs) +# --------------------------------------------------------------------------- # + +class TestParseItunes: + def test_sun_string_rules(self): + c = parse_itunes_smart(*_blob("sun")) + assert not c.unsupported and not c.has_nested + assert c.root.conjunction == "all" + rules = c.root.children + assert all(isinstance(r, SmartRule) for r in rules) + assert (rules[0].field, rules[0].operator, rules[0].value) == ("name", "contains", "sun") + assert (rules[1].field, rules[1].operator, rules[1].value) == ("name", "not_contains", "sunday") + + def test_unrated_numeric_rules(self): + c = parse_itunes_smart(*_blob("unrated")) + assert not c.unsupported + rating, plays = c.root.children + assert rating.field == "rating" and rating.operator == "in_range" + assert rating.value == 0 and rating.value2 == 0 + assert plays.field == "play_count" and plays.operator == "greater" and plays.value == 1 + + def test_top_relative_date_and_limit(self): + c = parse_itunes_smart(*_blob("top 300 of the decade!")) + assert not c.unsupported + rule = c.root.children[0] + assert rule.field == "date_added" and rule.operator == "in_last" + assert rule.value == 120 and rule.unit == "months" + assert c.limit.enabled and c.limit.count == 300 and c.limit.by == "items" + + def test_mediakind_is_unsupported(self): + # "convert these" matches on MediaKind (Podcast), which LinTunes can't + # represent -> falls back to the imported snapshot. + c = parse_itunes_smart(*_blob("convert these")) + assert c.unsupported is True + + def test_junk_blob_is_unsupported_no_raise(self): + c = parse_itunes_smart(b"not a real blob", b"x" * 200) + assert c.unsupported is True + + def test_missing_blobs_unsupported(self): + assert parse_itunes_smart(None, None).unsupported is True + + +# --------------------------------------------------------------------------- # +# Schema round-trip +# --------------------------------------------------------------------------- # + +def test_criteria_dict_roundtrip_with_nesting(): + crit = SmartCriteria( + root=SmartGroup("all", [ + SmartRule("genre", "contains", "Rock"), + SmartGroup("any", [ + SmartRule("rating", "in_range", 80, 100), + SmartRule("play_count", "greater", 10), + ]), + ]), + limit=SmartLimit(True, 25, "items", "random"), + live_update=False, has_nested=True) + back = SmartCriteria.from_dict(json.loads(json.dumps(crit.to_dict()))) + assert back.root.conjunction == "all" + assert isinstance(back.root.children[1], SmartGroup) + assert back.root.children[1].children[0].value2 == 100 + assert back.limit.selection == "random" and back.live_update is False + assert back.has_nested is True + + +# --------------------------------------------------------------------------- # +# Evaluator +# --------------------------------------------------------------------------- # + +def _tracks(*specs): + out = [] + for i, kw in enumerate(specs, start=1): + kw.setdefault("track_id", i) + out.append(Track(**kw)) + return out + + +class TestEvaluate: + def test_string_contains_casefold(self): + ts = _tracks({"name": "Sunshine"}, {"name": "Moon"}, {"name": "SUN city"}) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("name", "contains", "sun")])) + assert evaluate(c, ts) == [1, 3] + + def test_string_is_not(self): + ts = _tracks({"genre": "Rock"}, {"genre": "Jazz"}) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is_not", "rock")])) + assert evaluate(c, ts) == [2] + + def test_numeric_in_range_and_greater(self): + ts = _tracks({"play_count": 1}, {"play_count": 5}, {"play_count": 10}) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "in_range", 2, 9)])) + assert evaluate(c, ts) == [2] + c2 = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "greater", 4)])) + assert evaluate(c2, ts) == [2, 3] + + def test_rating_stars(self): + ts = _tracks({"rating": 60}, {"rating": 80}, {"rating": 100}) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("rating", "in_range", 80, 100)])) + assert evaluate(c, ts) == [2, 3] + + def test_date_in_last(self): + now = datetime(2026, 6, 26) + ts = _tracks( + {"date_added": "2026-06-01T00:00:00"}, # within + {"date_added": "2025-01-01T00:00:00"}, # outside + {"date_added": None}, # never -> excluded + ) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("date_added", "in_last", 3, unit="months")])) + assert evaluate(c, ts, now=now) == [1] + + def test_date_before_treats_null_as_distant_past(self): + # A never-played track (play_date_utc=None) must match "last played + # before X" — iTunes treats null dates as the distant past. + ts = _tracks( + {"play_date_utc": "2020-01-01T00:00:00"}, # before -> match + {"play_date_utc": "2025-01-01T00:00:00"}, # after -> no + {"play_date_utc": None}, # never played -> match + ) + c = SmartCriteria(root=SmartGroup("all", [ + SmartRule("last_played", "before", "2023-01-01T00:00:00")])) + assert evaluate(c, ts) == [1, 3] + # "after" must NOT match a null date. + c2 = SmartCriteria(root=SmartGroup("all", [ + SmartRule("last_played", "after", "2023-01-01T00:00:00")])) + assert evaluate(c2, ts) == [2] + + def test_bool_compilation(self): + ts = _tracks({"compilation": True}, {"compilation": False}) + c = SmartCriteria(root=SmartGroup("all", [SmartRule("compilation", "is", True)])) + assert evaluate(c, ts) == [1] + + def test_match_any_vs_all(self): + ts = _tracks({"genre": "Rock", "year": 1990}, {"genre": "Jazz", "year": 2020}) + rules = [SmartRule("genre", "is", "rock"), SmartRule("year", "greater", 2000)] + assert evaluate(SmartCriteria(root=SmartGroup("any", rules)), ts) == [1, 2] + assert evaluate(SmartCriteria(root=SmartGroup("all", rules)), ts) == [] + + def test_nested_group(self): + ts = _tracks( + {"genre": "Rock", "rating": 100}, + {"genre": "Rock", "rating": 20}, + {"genre": "Jazz", "rating": 100}, + ) + crit = SmartCriteria(root=SmartGroup("all", [ + SmartRule("genre", "is", "rock"), + SmartGroup("any", [SmartRule("rating", "greater", 79)]), + ])) + assert evaluate(crit, ts) == [1] + + def test_empty_rules_matches_all(self): + ts = _tracks({"name": "a"}, {"name": "b"}) + assert evaluate(SmartCriteria(root=SmartGroup("all", [])), ts) == [1, 2] + + def test_limit_items_by_most_played(self): + ts = _tracks({"play_count": 1}, {"play_count": 9}, {"play_count": 5}) + c = SmartCriteria(root=SmartGroup("all", []), + limit=SmartLimit(True, 2, "items", "most_often_played")) + # Highest two are tracks 2 (9) and 3 (5); returned sorted by id. + assert evaluate(c, ts) == [2, 3] + + def test_limit_by_minutes_budget(self): + ts = _tracks({"total_time": 120000}, {"total_time": 120000}, {"total_time": 120000}) + c = SmartCriteria(root=SmartGroup("all", []), + limit=SmartLimit(True, 3, "minutes", "least_recently_added")) + # Budget 3 minutes = 180000 ms -> only one 2-minute track fits. + assert len(evaluate(c, ts)) == 1 + + +# --------------------------------------------------------------------------- # +# LibraryManager +# --------------------------------------------------------------------------- # + +def _manager(tmp_path, tracks=()): + library = Library() + for t in tracks: + library.tracks[t.track_id] = t + return LibraryManager(library, tmp_path) + + +class TestManagerSmart: + def test_create_computes_membership(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks( + {"genre": "Rock"}, {"genre": "Jazz"}, {"genre": "Rock"})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) + pl = mgr.create_smart_playlist("rockers", crit) + assert pl.track_ids == [1, 3] + assert pl.is_smart + + def test_edit_field_updates_membership_and_emits(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Jazz"}, {"genre": "Rock"})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) + pl = mgr.create_smart_playlist("rockers", crit) + assert pl.track_ids == [2] + emitted = [] + mgr.playlist_content_changed.connect(emitted.append) + mgr.update_track_fields(1, {"genre": "Rock"}) # track 1 becomes Rock + mgr._flush_recompute() + assert pl.track_ids == [1, 2] + assert pl.persistent_id in emitted + + def test_record_play_updates_most_played(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"play_count": 0}, {"play_count": 0})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("play_count", "greater", 0)])) + pl = mgr.create_smart_playlist("played", crit) + assert pl.track_ids == [] + mgr.record_play(2) + mgr._flush_recompute() + assert pl.track_ids == [2] + + def test_noop_recompute_not_dirty(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Rock"})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) + pl = mgr.create_smart_playlist("rockers", crit) + mgr.flush() + assert pl.persistent_id not in mgr._dirty_playlists + assert mgr.recompute_smart_playlist(pl.persistent_id) is False + assert pl.persistent_id not in mgr._dirty_playlists + + def test_manual_edits_blocked(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) + pl = mgr.create_smart_playlist("rockers", crit) + assert pl.track_ids == [1] + mgr.add_tracks_to_playlist(pl.persistent_id, [2]) + mgr.remove_tracks_from_playlist(pl.persistent_id, [0]) + mgr.move_tracks_in_playlist(pl.persistent_id, [0], 0) + assert pl.track_ids == [1] # unchanged + + def test_unsupported_keeps_snapshot(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) + pl = Playlist(name="frozen", persistent_id="AB12CD34", + playlist_type=PlaylistType.SMART, + smart_criteria=SmartCriteria(unsupported=True), track_ids=[2]) + mgr.library.playlists[pl.persistent_id] = pl + assert mgr.recompute_smart_playlist(pl.persistent_id, force=True) is False + assert pl.track_ids == [2] # snapshot preserved + + def test_edit_criteria_undo_redo(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Rock"}, {"genre": "Jazz"})) + crit = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")])) + pl = mgr.create_smart_playlist("p", crit) + assert pl.track_ids == [1] + new = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "jazz")])) + mgr.set_smart_criteria(pl.persistent_id, new) + assert pl.track_ids == [2] + mgr.undo_stack.undo() + assert pl.smart_criteria.root.children[0].value == "rock" + assert pl.track_ids == [1] + mgr.undo_stack.redo() + assert pl.track_ids == [2] + + def test_recompute_all_respects_live_update(self, qapp, tmp_path): + mgr = _manager(tmp_path, _tracks({"genre": "Rock"})) + live = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")]), + live_update=True) + frozen = SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", "rock")]), + live_update=False) + a = Playlist(name="a", persistent_id="AAAA0001", + playlist_type=PlaylistType.SMART, smart_criteria=live, track_ids=[]) + b = Playlist(name="b", persistent_id="BBBB0002", + playlist_type=PlaylistType.SMART, smart_criteria=frozen, track_ids=[]) + mgr.library.playlists["AAAA0001"] = a + mgr.library.playlists["BBBB0002"] = b + changed = mgr.recompute_all_smart() # force=False + assert "AAAA0001" in changed and "BBBB0002" not in changed + assert a.track_ids == [1] and b.track_ids == [] + + +# --------------------------------------------------------------------------- # +# Conflict resolution +# --------------------------------------------------------------------------- # + +def test_conflict_resolver_smart_newest_wins(tmp_path): + from lintunes.storage.conflict_resolver import resolve_conflicts + pdir = tmp_path / "playlists" + pdir.mkdir() + + def crit_dict(genre): + return SmartCriteria(root=SmartGroup("all", [SmartRule("genre", "is", genre)])).to_dict() + + original = {"name": "p", "persistent_id": "PID1", "playlist_type": "smart", + "track_ids": [1, 2], "smart_criteria": crit_dict("rock"), "settings": {}} + conflict = {"name": "p", "persistent_id": "PID1", "playlist_type": "smart", + "track_ids": [3, 4], "smart_criteria": crit_dict("jazz"), "settings": {}} + orig_path = pdir / "PID1.json" + conf_path = pdir / "PID1.sync-conflict-20250101-120000-ABCDEFG.json" + orig_path.write_text(json.dumps(original)) + conf_path.write_text(json.dumps(conflict)) + os.utime(orig_path, (1000, 1000)) # original older + os.utime(conf_path, (2000, 2000)) # conflict newer + + resolve_conflicts(tmp_path) + + merged = json.loads(orig_path.read_text()) + # Newest criteria wins; track_ids are NOT unioned (no [1,2,3,4]). + assert merged["smart_criteria"]["root"]["children"][0]["value"] == "jazz" + assert merged["track_ids"] == [3, 4]