Compare commits
13 Commits
6a841afaa0
...
ec797b8074
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ec797b8074 | ||
|
|
06f6529cbe | ||
|
|
949783e76c | ||
|
|
61bcaec7a3 | ||
|
|
1ae1e90ae3 | ||
|
|
a0b1aba50e | ||
|
|
5ba346cc84 | ||
|
|
e3666c8c64 | ||
|
|
cbc888660f | ||
|
|
ba6935420d | ||
|
|
a93bc7907d | ||
|
|
1ace4ad98c | ||
|
|
8cc4ef5279 |
2
.gitignore
vendored
2
.gitignore
vendored
@ -37,3 +37,5 @@ converted/
|
||||
*.log
|
||||
specs.md
|
||||
*.AppImage
|
||||
squashfs-root/
|
||||
iOpenPod/
|
||||
|
||||
@ -39,10 +39,6 @@ source venv/bin/activate
|
||||
echo "Installing dependencies..."
|
||||
pip3 install -r requirements.txt
|
||||
|
||||
# Create necessary directories
|
||||
echo "Creating necessary directories..."
|
||||
mkdir -p temp downloads converted
|
||||
|
||||
echo ""
|
||||
echo "Installation complete!"
|
||||
echo ""
|
||||
|
||||
@ -10,3 +10,4 @@ tqdm>=4.65.0
|
||||
musicbrainzngs>=0.7.1
|
||||
PyQt6>=6.5.0
|
||||
numpy>=1.24.0
|
||||
iopenpod>=1.0.53
|
||||
|
||||
@ -54,8 +54,6 @@ install_pyinstaller() {
|
||||
"$VENV_DIR/bin/pip" install pyinstaller
|
||||
# Install project dependencies for bundling
|
||||
"$VENV_DIR/bin/pip" install -r "$PROJECT_DIR/requirements.txt"
|
||||
# Install wasmtime for HASHAB signing (iPod Nano 6G/7G)
|
||||
"$VENV_DIR/bin/pip" install wasmtime
|
||||
}
|
||||
|
||||
download_appimagetool() {
|
||||
|
||||
@ -9,21 +9,33 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def xdg_cache_path(subdir: str) -> str:
|
||||
"""Return path inside XDG cache directory for neo-pod-desktop.
|
||||
Creates the directory structure if missing.
|
||||
"""
|
||||
base = os.environ.get(
|
||||
"XDG_CACHE_HOME",
|
||||
os.path.join(os.path.expanduser("~"), ".cache"),
|
||||
)
|
||||
path = os.path.join(base, "neo-pod-desktop", subdir)
|
||||
os.makedirs(path, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
class CoverCache:
|
||||
"""Manages a local cache of cover art images for the player UI.
|
||||
|
||||
The cache stores JPEG thumbnails keyed by (artist, album) hash.
|
||||
Covers are stored in $XDG_CACHE_HOME/neo-pod-desktop/covers/
|
||||
(default ~/.cache/neo-pod-desktop/covers/).
|
||||
"""
|
||||
|
||||
def __init__(self, cache_dir: str | None = None):
|
||||
self.cache_dir = cache_dir or os.path.join(
|
||||
tempfile.gettempdir(), "neo-pod-covers"
|
||||
)
|
||||
self.cache_dir = cache_dir or xdg_cache_path("covers")
|
||||
os.makedirs(self.cache_dir, exist_ok=True)
|
||||
|
||||
def _key(self, artist: str, album: str) -> str:
|
||||
|
||||
@ -60,11 +60,10 @@ class ConfigLoader:
|
||||
if os.path.exists(self.config_file):
|
||||
logger.info(f"Loading configuration from {self.config_file}")
|
||||
self.config.read(self.config_file)
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"Configuration file {self.config_file} not found, using defaults")
|
||||
self._set_defaults()
|
||||
return False
|
||||
self._set_defaults()
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load configuration: {e}")
|
||||
self._set_defaults()
|
||||
@ -95,39 +94,93 @@ class ConfigLoader:
|
||||
if not self.config.has_section('General'):
|
||||
self.config.add_section('General')
|
||||
|
||||
self.config.set('General', 'output_dir', os.path.join(os.path.expanduser("~"), "Music", "iPod"))
|
||||
self.config.set('General', 'format', 'm4a')
|
||||
self.config.set('General', 'quality', '256')
|
||||
self.config.set('General', 'clean_temp', 'true')
|
||||
if not self.config.has_option('General', 'output_dir'):
|
||||
self.config.set('General', 'output_dir', os.path.join(os.path.expanduser("~"), "Music", "iPod"))
|
||||
if not self.config.has_option('General', 'format'):
|
||||
self.config.set('General', 'format', 'm4a')
|
||||
if not self.config.has_option('General', 'quality'):
|
||||
self.config.set('General', 'quality', '256')
|
||||
|
||||
# Video section
|
||||
if not self.config.has_section('Video'):
|
||||
self.config.add_section('Video')
|
||||
|
||||
self.config.set('Video', 'enable_video', 'false')
|
||||
self.config.set('Video', 'resolution', '640x480')
|
||||
if not self.config.has_option('Video', 'enable_video'):
|
||||
self.config.set('Video', 'enable_video', 'false')
|
||||
if not self.config.has_option('Video', 'resolution'):
|
||||
self.config.set('Video', 'resolution', '640x480')
|
||||
|
||||
# Metadata section
|
||||
if not self.config.has_section('Metadata'):
|
||||
self.config.add_section('Metadata')
|
||||
|
||||
self.config.set('Metadata', 'embed_artwork', 'true')
|
||||
self.config.set('Metadata', 'use_musicbrainz', 'false')
|
||||
if not self.config.has_option('Metadata', 'embed_artwork'):
|
||||
self.config.set('Metadata', 'embed_artwork', 'true')
|
||||
if not self.config.has_option('Metadata', 'use_musicbrainz'):
|
||||
self.config.set('Metadata', 'use_musicbrainz', 'false')
|
||||
|
||||
# Device section
|
||||
if not self.config.has_section('Device'):
|
||||
self.config.add_section('Device')
|
||||
|
||||
self.config.set('Device', 'auto_detect', 'true')
|
||||
self.config.set('Device', 'mount_point', '')
|
||||
if not self.config.has_option('Device', 'auto_detect'):
|
||||
self.config.set('Device', 'auto_detect', 'true')
|
||||
if not self.config.has_option('Device', 'mount_point'):
|
||||
self.config.set('Device', 'mount_point', '')
|
||||
|
||||
# Advanced section
|
||||
if not self.config.has_section('Advanced'):
|
||||
self.config.add_section('Advanced')
|
||||
|
||||
self.config.set('Advanced', 'temp_dir', 'temp')
|
||||
self.config.set('Advanced', 'concurrent_downloads', '2')
|
||||
self.config.set('Advanced', 'debug', 'false')
|
||||
if not self.config.has_option('Advanced', 'temp_dir'):
|
||||
self.config.set('Advanced', 'temp_dir', 'temp')
|
||||
if not self.config.has_option('Advanced', 'concurrent_downloads'):
|
||||
self.config.set('Advanced', 'concurrent_downloads', '2')
|
||||
if not self.config.has_option('Advanced', 'debug'):
|
||||
self.config.set('Advanced', 'debug', 'false')
|
||||
if not self.config.has_option('Advanced', 'clean_temp'):
|
||||
self.config.set('Advanced', 'clean_temp', 'true')
|
||||
if not self.config.has_option('Advanced', 'show_download'):
|
||||
self.config.set('Advanced', 'show_download', 'false')
|
||||
|
||||
# Hotkeys section
|
||||
if not self.config.has_section('Hotkeys'):
|
||||
self.config.add_section('Hotkeys')
|
||||
|
||||
hotkey_defaults = {
|
||||
'play_pause': 'Space',
|
||||
'prev_track': 'Ctrl+Left',
|
||||
'next_track': 'Ctrl+Right',
|
||||
'volume_up': 'Ctrl+Up',
|
||||
'volume_down': 'Ctrl+Down',
|
||||
'seek_forward': 'Right',
|
||||
'seek_backward': 'Left',
|
||||
'tab_download': 'Ctrl+1',
|
||||
'tab_library': 'Ctrl+2',
|
||||
'tab_ipod': 'Ctrl+3',
|
||||
'tab_settings': 'Ctrl+4',
|
||||
'search_focus': 'Ctrl+F',
|
||||
'select_all': 'Ctrl+A',
|
||||
'library_refresh': 'F5',
|
||||
'ipod_refresh': 'Ctrl+R',
|
||||
'download': 'Ctrl+Enter',
|
||||
'delete_selected': 'Delete',
|
||||
}
|
||||
for key, value in hotkey_defaults.items():
|
||||
if not self.config.has_option('Hotkeys', key):
|
||||
self.config.set('Hotkeys', key, value)
|
||||
|
||||
# Playback section
|
||||
if not self.config.has_section('Playback'):
|
||||
self.config.add_section('Playback')
|
||||
if not self.config.has_option('Playback', 'last_track_path'):
|
||||
self.config.set('Playback', 'last_track_path', '')
|
||||
if not self.config.has_option('Playback', 'last_position_ms'):
|
||||
self.config.set('Playback', 'last_position_ms', '0')
|
||||
if not self.config.has_option('Playback', 'last_volume'):
|
||||
self.config.set('Playback', 'last_volume', '80')
|
||||
if not self.config.has_option('Playback', 'last_playing'):
|
||||
self.config.set('Playback', 'last_playing', 'false')
|
||||
|
||||
def get(self, section: str, option: str, fallback: Any = None) -> Any:
|
||||
"""
|
||||
|
||||
312
src/hotkeys.py
Normal file
312
src/hotkeys.py
Normal file
@ -0,0 +1,312 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Hotkey Manager for neo-pod-desktop
|
||||
Provides configurable keyboard shortcuts with scope awareness.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Callable, Dict, Optional
|
||||
|
||||
from PyQt6.QtCore import Qt, QObject
|
||||
from PyQt6.QtGui import QKeySequence, QShortcut, QAction
|
||||
from PyQt6.QtWidgets import QApplication, QWidget, QDialog, QVBoxLayout, QLabel, QPushButton
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
QSeq = QKeySequence
|
||||
|
||||
# Key names that QKeySequence doesn't expose as enum values
|
||||
_KEY_MAP = {
|
||||
"Space": QSeq(Qt.Key.Key_Space),
|
||||
"Delete": QSeq(Qt.Key.Key_Delete),
|
||||
"Up": QSeq(Qt.Key.Key_Up),
|
||||
"Down": QSeq(Qt.Key.Key_Down),
|
||||
"Left": QSeq(Qt.Key.Key_Left),
|
||||
"Right": QSeq(Qt.Key.Key_Right),
|
||||
"Enter": QSeq(Qt.Key.Key_Enter),
|
||||
"Return": QSeq(Qt.Key.Key_Return),
|
||||
"Esc": QSeq(Qt.Key.Key_Escape),
|
||||
"Escape": QSeq(Qt.Key.Key_Escape),
|
||||
"Tab": QSeq(Qt.Key.Key_Tab),
|
||||
"Backspace": QSeq(Qt.Key.Key_Backspace),
|
||||
"Plus": QSeq(Qt.Key.Key_Plus),
|
||||
"Minus": QSeq(Qt.Key.Key_Minus),
|
||||
"F1": QSeq(Qt.Key.Key_F1),
|
||||
"F2": QSeq(Qt.Key.Key_F2),
|
||||
"F3": QSeq(Qt.Key.Key_F3),
|
||||
"F4": QSeq(Qt.Key.Key_F4),
|
||||
"F5": QSeq(Qt.Key.Key_F5),
|
||||
"F6": QSeq(Qt.Key.Key_F6),
|
||||
"F7": QSeq(Qt.Key.Key_F7),
|
||||
"F8": QSeq(Qt.Key.Key_F8),
|
||||
"F9": QSeq(Qt.Key.Key_F9),
|
||||
"F10": QSeq(Qt.Key.Key_F10),
|
||||
"F11": QSeq(Qt.Key.Key_F11),
|
||||
"F12": QSeq(Qt.Key.Key_F12),
|
||||
}
|
||||
|
||||
|
||||
def _build_seq(text: str) -> QKeySequence:
|
||||
"""Build a QKeySequence from a human-readable shortcut string."""
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return QSeq()
|
||||
known = _KEY_MAP.get(text)
|
||||
if known is not None:
|
||||
return known
|
||||
return QSeq(text)
|
||||
|
||||
|
||||
def _seq_to_string(seq: QKeySequence) -> str:
|
||||
"""Convert a QKeySequence to a human-readable string."""
|
||||
return seq.toString()
|
||||
|
||||
|
||||
def _is_editable_focused() -> bool:
|
||||
"""Return True if an editable widget currently has keyboard focus."""
|
||||
from PyQt6.QtWidgets import QLineEdit, QTextEdit, QPlainTextEdit, QSpinBox, QDoubleSpinBox
|
||||
w = QApplication.focusWidget()
|
||||
if w is None:
|
||||
return False
|
||||
return isinstance(w, (QLineEdit, QTextEdit, QPlainTextEdit, QSpinBox, QDoubleSpinBox))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Default shortcut definitions
|
||||
# (action_name, default_key_string, display_label, scope)
|
||||
# scope: "global" — always active; "tab" — handler decides based on current tab
|
||||
# ---------------------------------------------------------------------------
|
||||
DEFAULTS = [
|
||||
("play_pause", "Space", "Play / Pause", "global"),
|
||||
("prev_track", "Ctrl+Left", "Previous Track", "global"),
|
||||
("next_track", "Ctrl+Right", "Next Track", "global"),
|
||||
("volume_up", "Ctrl+Up", "Volume Up", "global"),
|
||||
("volume_down", "Ctrl+Down", "Volume Down", "global"),
|
||||
("seek_forward", "Right", "Seek Forward 5s", "global"),
|
||||
("seek_backward", "Left", "Seek Backward 5s", "global"),
|
||||
("tab_download", "Ctrl+1", "Switch to Download", "global"),
|
||||
("tab_library", "Ctrl+2", "Switch to Library", "global"),
|
||||
("tab_ipod", "Ctrl+3", "Switch to iPod", "global"),
|
||||
("tab_settings", "Ctrl+4", "Switch to Settings", "global"),
|
||||
("search_focus", "Ctrl+F", "Focus Search", "global"),
|
||||
("select_all", "Ctrl+A", "Select All", "global"),
|
||||
("library_refresh", "F5", "Refresh Library", "global"),
|
||||
("ipod_refresh", "Ctrl+R", "Refresh Devices", "global"),
|
||||
("download", "Ctrl+Enter", "Start Download", "global"),
|
||||
("delete_selected", "Delete", "Delete Selected", "global"),
|
||||
("edit_metadata", "F2", "Edit Metadata", "global"),
|
||||
]
|
||||
|
||||
|
||||
class KeyCaptureDialog(QDialog):
|
||||
"""Modal dialog that captures a single key combination."""
|
||||
|
||||
def __init__(self, parent=None, current: str = ""):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("Press New Shortcut")
|
||||
self.setMinimumWidth(320)
|
||||
self.setModal(True)
|
||||
|
||||
self._captured_seq: QKeySequence = QSeq()
|
||||
self._modifiers = Qt.KeyboardModifier.NoModifier
|
||||
self._key = Qt.Key.Key_unknown
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
self._label = QLabel(f"Press new shortcut for this action...\n(current: {current or 'none'})")
|
||||
self._label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self._label.setWordWrap(True)
|
||||
layout.addWidget(self._label)
|
||||
|
||||
clear_btn = QPushButton("Clear (Disable)")
|
||||
clear_btn.clicked.connect(self._on_clear)
|
||||
layout.addWidget(clear_btn)
|
||||
|
||||
layout.addWidget(QPushButton("Cancel", clicked=self.reject))
|
||||
|
||||
self.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
|
||||
self.setFocus()
|
||||
|
||||
def keyPressEvent(self, event):
|
||||
key = event.key()
|
||||
modifiers = event.modifiers()
|
||||
mod_mask = modifiers & (
|
||||
Qt.KeyboardModifier.ControlModifier
|
||||
| Qt.KeyboardModifier.ShiftModifier
|
||||
| Qt.KeyboardModifier.AltModifier
|
||||
| Qt.KeyboardModifier.MetaModifier
|
||||
)
|
||||
|
||||
# Ignore pure modifier presses
|
||||
ignore = {
|
||||
Qt.Key.Key_Control, Qt.Key.Key_Shift, Qt.Key.Key_Alt, Qt.Key.Key_Meta,
|
||||
Qt.Key.Key_Super_L, Qt.Key.Key_Super_R,
|
||||
}
|
||||
if key in ignore:
|
||||
return
|
||||
|
||||
# Build the sequence — handle both old (int) and new (enum) PyQt6 types
|
||||
def _to_int(val):
|
||||
return val if isinstance(val, int) else val.value
|
||||
|
||||
keys = _to_int(mod_mask) | _to_int(key)
|
||||
self._captured_seq = QSeq(keys)
|
||||
self._label.setText(f"Captured: {self._captured_seq.toString()}\nPress Close to accept")
|
||||
self.accept()
|
||||
|
||||
def _on_clear(self):
|
||||
self._captured_seq = QSeq()
|
||||
self._label.setText("Shortcut cleared (disabled)")
|
||||
self.accept()
|
||||
|
||||
def captured_sequence(self) -> QKeySequence:
|
||||
return self._captured_seq
|
||||
|
||||
|
||||
class HotkeyManager:
|
||||
"""Manages all keyboard shortcuts for the application.
|
||||
|
||||
Loads key bindings from config, creates QShortcut instances,
|
||||
and supports live reassignment at runtime.
|
||||
"""
|
||||
|
||||
def __init__(self, parent: QWidget, config_loader=None):
|
||||
self._parent = parent
|
||||
self._config_loader = config_loader
|
||||
|
||||
# action_name → (QShortcut, callback, default_seq, scope)
|
||||
self._shortcuts: Dict[str, QShortcut] = {}
|
||||
self._callbacks: Dict[str, Callable[[], None]] = {}
|
||||
self._defaults: Dict[str, QKeySequence] = {}
|
||||
self._labels: Dict[str, str] = {}
|
||||
self._scopes: Dict[str, str] = {}
|
||||
|
||||
for name, key_str, label, scope in DEFAULTS:
|
||||
self._defaults[name] = _build_seq(key_str)
|
||||
self._labels[name] = label
|
||||
self._scopes[name] = scope
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Registration
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def register(self, action_name: str, callback: Callable[[], None]) -> None:
|
||||
"""Register a shortcut with its callback.
|
||||
|
||||
If a saved key exists in config it is used; otherwise the default.
|
||||
"""
|
||||
saved_key = None
|
||||
if self._config_loader:
|
||||
saved_key = self._config_loader.get("Hotkeys", action_name, fallback=None)
|
||||
|
||||
seq = _build_seq(saved_key) if saved_key else self._defaults.get(action_name, QSeq())
|
||||
self._callbacks[action_name] = callback
|
||||
|
||||
scope = self._scopes.get(action_name, "global")
|
||||
context = Qt.ShortcutContext.WindowShortcut
|
||||
|
||||
sc = QShortcut(seq, self._parent, context=context)
|
||||
|
||||
def _make_handler(name):
|
||||
def handler():
|
||||
self._dispatch(name)
|
||||
return handler
|
||||
|
||||
sc.activated.connect(_make_handler(action_name))
|
||||
self._shortcuts[action_name] = sc
|
||||
|
||||
def register_all(self, action_map: Dict[str, Callable[[], None]]) -> None:
|
||||
"""Register all known shortcuts from an action_name→callback mapping."""
|
||||
for name, callback in action_map.items():
|
||||
self.register(name, callback)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Dispatch
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _dispatch(self, action_name: str):
|
||||
"""Check if the shortcut should fire (not blocked by editable focus),
|
||||
then invoke the callback.
|
||||
"""
|
||||
# For seek arrows and space, don't fire when text editing
|
||||
editable_blocked = {"seek_forward", "seek_backward", "play_pause"}
|
||||
if action_name in editable_blocked and _is_editable_focused():
|
||||
return
|
||||
|
||||
cb = self._callbacks.get(action_name)
|
||||
if cb:
|
||||
cb()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Configuration persistence
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def save_to_config(self) -> None:
|
||||
"""Write all current key bindings to the config loader."""
|
||||
if self._config_loader is None:
|
||||
return
|
||||
for name, sc in self._shortcuts.items():
|
||||
key_str = sc.key().toString()
|
||||
self._config_loader.set("Hotkeys", name, key_str if key_str else "")
|
||||
self._config_loader.save()
|
||||
|
||||
def load_from_config(self) -> None:
|
||||
"""Re-read key bindings from config and update all shortcuts."""
|
||||
if self._config_loader is None:
|
||||
return
|
||||
for name, sc in self._shortcuts.items():
|
||||
saved = self._config_loader.get("Hotkeys", name, fallback=None)
|
||||
if saved is not None:
|
||||
seq = _build_seq(saved) if saved != "" else QSeq()
|
||||
else:
|
||||
seq = self._defaults.get(name, QSeq())
|
||||
sc.setKey(seq)
|
||||
|
||||
def reset_to_defaults(self) -> None:
|
||||
"""Reset all shortcuts to their default keys and save."""
|
||||
for name, sc in self._shortcuts.items():
|
||||
default_seq = self._defaults.get(name, QSeq())
|
||||
sc.setKey(default_seq)
|
||||
self.save_to_config()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Query API (for Settings UI)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_label(self, action_name: str) -> str:
|
||||
return self._labels.get(action_name, action_name)
|
||||
|
||||
def get_current_key_string(self, action_name: str) -> str:
|
||||
sc = self._shortcuts.get(action_name)
|
||||
if sc is None:
|
||||
return ""
|
||||
return sc.key().toString()
|
||||
|
||||
def get_default_key_string(self, action_name: str) -> str:
|
||||
return self._defaults.get(action_name, QSeq()).toString()
|
||||
|
||||
def all_action_names(self):
|
||||
return list(self._shortcuts.keys())
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Live reassignment (called from Settings UI)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def set_shortcut(self, action_name: str, key_string: str) -> None:
|
||||
"""Change the key binding for an action at runtime and persist."""
|
||||
sc = self._shortcuts.get(action_name)
|
||||
if sc is None:
|
||||
return
|
||||
seq = _build_seq(key_string)
|
||||
sc.setKey(seq)
|
||||
if self._config_loader:
|
||||
self._config_loader.set("Hotkeys", action_name, key_string)
|
||||
self._config_loader.save()
|
||||
|
||||
def set_shortcut_seq(self, action_name: str, seq: QKeySequence) -> None:
|
||||
"""Set a shortcut directly from a QKeySequence (from key capture dialog)."""
|
||||
sc = self._shortcuts.get(action_name)
|
||||
if sc is not None:
|
||||
sc.setKey(seq)
|
||||
if self._config_loader:
|
||||
self._config_loader.set("Hotkeys", action_name, seq.toString())
|
||||
self._config_loader.save()
|
||||
@ -21,6 +21,7 @@ Public API preserved for main.py compatibility:
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import hashlib
|
||||
@ -38,7 +39,34 @@ from mutagen.easyid3 import EasyID3
|
||||
|
||||
from artwork import prepare_artwork_for_track, ArtworkEntry, write_artworkdb
|
||||
|
||||
_IOP_ROOT = "/tmp/iOpenPod"
|
||||
def _find_iop_root() -> str:
|
||||
"""Find iOpenPod's installed location.
|
||||
|
||||
Tries in order:
|
||||
1. pip-installed package (via importlib.util.find_spec)
|
||||
2. /tmp/iOpenPod (backward compat / development)
|
||||
"""
|
||||
for pkg in ("iTunesDB_Writer", "SQLiteDB_Writer", "ipod_device"):
|
||||
spec = importlib.util.find_spec(pkg)
|
||||
if spec and spec.origin:
|
||||
parent = os.path.dirname(spec.origin)
|
||||
if os.path.basename(parent) == pkg:
|
||||
return os.path.dirname(parent)
|
||||
|
||||
fallbacks = [
|
||||
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "iOpenPod"),
|
||||
"/tmp/iOpenPod",
|
||||
os.path.join(os.path.expanduser("~"), "iOpenPod"),
|
||||
]
|
||||
for path in fallbacks:
|
||||
if os.path.isdir(path):
|
||||
return path
|
||||
|
||||
raise ImportError(
|
||||
"iOpenPod not found. Install with: pip install iopenpod"
|
||||
)
|
||||
|
||||
_IOP_ROOT = _find_iop_root()
|
||||
_IOP_CACHE: dict[str, Any] = {}
|
||||
_IOP_READY = False
|
||||
|
||||
@ -60,19 +88,30 @@ def _import_iop(module_name: str):
|
||||
return _IOP_CACHE[module_name]
|
||||
|
||||
if not _IOP_READY:
|
||||
# Evict local shadows ONCE so iOpenPod modules resolve correctly.
|
||||
sys.modules.pop("ipod_device", None)
|
||||
_IOP_READY = True
|
||||
logger.debug("iOpenPod boot: evicted local ipod_device")
|
||||
|
||||
src_paths = [p for p in sys.path if p.endswith('/src') or p == 'src']
|
||||
for p in src_paths:
|
||||
sys.path.remove(p)
|
||||
|
||||
had_iop = _IOP_ROOT in sys.path
|
||||
if not had_iop:
|
||||
sys.path.insert(0, _IOP_ROOT)
|
||||
importlib.invalidate_caches()
|
||||
|
||||
try:
|
||||
mod = importlib.import_module(module_name)
|
||||
finally:
|
||||
for p in src_paths:
|
||||
if p not in sys.path:
|
||||
sys.path.insert(0, p)
|
||||
if not had_iop:
|
||||
sys.path.remove(_IOP_ROOT)
|
||||
try:
|
||||
sys.path.remove(_IOP_ROOT)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
_IOP_CACHE[module_name] = mod
|
||||
return mod
|
||||
|
||||
111
src/library_cache.py
Normal file
111
src/library_cache.py
Normal file
@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Library metadata cache for neo-pod-desktop.
|
||||
Caches extracted tag data per file path with mtime-based invalidation.
|
||||
Stored as JSON in XDG cache directory.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from artwork.cache import xdg_cache_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CACHE_FILE = "library_cache.json"
|
||||
|
||||
|
||||
class LibraryCache:
|
||||
"""JSON-based cache of audio file metadata.
|
||||
|
||||
Keys are absolute file paths. Values are dicts of metadata
|
||||
plus an internal ``_mtime`` field for change detection.
|
||||
|
||||
Cached entries are returned only when the file still exists
|
||||
and its ``os.path.getmtime`` matches the stored value.
|
||||
"""
|
||||
|
||||
def __init__(self, cache_path: Optional[str] = None):
|
||||
if cache_path is None:
|
||||
cache_path = os.path.join(xdg_cache_path(""), _CACHE_FILE)
|
||||
self._path = cache_path
|
||||
self._data: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
def load(self) -> None:
|
||||
self._data = {}
|
||||
if not os.path.exists(self._path):
|
||||
return
|
||||
try:
|
||||
with open(self._path, "r", encoding="utf-8") as fh:
|
||||
self._data = json.load(fh)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to load library cache: %s", exc)
|
||||
self._data = {}
|
||||
|
||||
def save(self) -> None:
|
||||
self._data = {k: v for k, v in self._data.items()
|
||||
if os.path.exists(k)}
|
||||
os.makedirs(os.path.dirname(self._path), exist_ok=True)
|
||||
try:
|
||||
with open(self._path, "w", encoding="utf-8") as fh:
|
||||
json.dump(self._data, fh, ensure_ascii=False)
|
||||
except OSError as exc:
|
||||
logger.warning("Failed to save library cache: %s", exc)
|
||||
|
||||
def get(self, file_path: str) -> Optional[Dict[str, Any]]:
|
||||
entry = self._data.get(file_path)
|
||||
if entry is None:
|
||||
return None
|
||||
if not os.path.isfile(file_path):
|
||||
self._data.pop(file_path, None)
|
||||
return None
|
||||
actual_mtime = os.path.getmtime(file_path)
|
||||
if actual_mtime != entry.get("_mtime", 0):
|
||||
return None
|
||||
return {k: v for k, v in entry.items() if k != "_mtime"}
|
||||
|
||||
def put(self, file_path: str, data: Dict[str, Any]) -> None:
|
||||
entry = dict(data)
|
||||
try:
|
||||
entry["_mtime"] = os.path.getmtime(file_path)
|
||||
except OSError:
|
||||
return
|
||||
self._data[file_path] = entry
|
||||
|
||||
def remove(self, file_path: str) -> None:
|
||||
self._data.pop(file_path, None)
|
||||
|
||||
def clear(self) -> None:
|
||||
self._data.clear()
|
||||
if os.path.exists(self._path):
|
||||
try:
|
||||
os.remove(self._path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._data)
|
||||
|
||||
|
||||
_library_cache: Optional[LibraryCache] = None
|
||||
|
||||
|
||||
def get_library_cache() -> LibraryCache:
|
||||
global _library_cache
|
||||
if _library_cache is None:
|
||||
_library_cache = LibraryCache()
|
||||
_library_cache.load()
|
||||
return _library_cache
|
||||
|
||||
|
||||
def save_library_cache() -> None:
|
||||
global _library_cache
|
||||
if _library_cache is not None:
|
||||
_library_cache.save()
|
||||
|
||||
|
||||
def invalidate_cache_for_path(file_path: str) -> None:
|
||||
cache = get_library_cache()
|
||||
cache.remove(file_path)
|
||||
497
src/main.py
497
src/main.py
@ -15,7 +15,7 @@ from PyQt6.QtWidgets import (
|
||||
QPushButton, QLabel, QLineEdit, QProgressBar, QFileDialog,
|
||||
QComboBox, QCheckBox, QTabWidget, QListWidget, QListWidgetItem,
|
||||
QMessageBox, QSpinBox, QGroupBox, QHeaderView,
|
||||
QTableWidget, QTableWidgetItem, QSlider, QMenu
|
||||
QTableWidget, QTableWidgetItem, QSlider, QMenu, QDialog
|
||||
)
|
||||
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QUrl
|
||||
from PyQt6.QtGui import QPixmap
|
||||
@ -24,8 +24,12 @@ from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
|
||||
from youtube_downloader import YouTubeDownloader, TrackInfo
|
||||
from audio_converter import AudioConverter
|
||||
from metadata_handler import MetadataHandler
|
||||
from metadata_editor import MetadataEditorDialog
|
||||
from ipod_device import IPodDevice
|
||||
from library_cache import get_library_cache, save_library_cache, invalidate_cache_for_path
|
||||
from artwork.cache import cover_cache
|
||||
from config_loader import ConfigLoader
|
||||
from hotkeys import HotkeyManager, KeyCaptureDialog
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -281,15 +285,27 @@ class MainWindow(QMainWindow):
|
||||
self.ipod_devices: List[Dict] = []
|
||||
self.current_mount_point: Optional[str] = None
|
||||
self.show_download_tab = False
|
||||
self.ipod_tab_index = 2
|
||||
|
||||
self.config_loader = ConfigLoader()
|
||||
self.hotkey_manager: Optional[HotkeyManager] = None
|
||||
|
||||
self.player: Optional[QMediaPlayer] = None
|
||||
self.audio_output: Optional[QAudioOutput] = None
|
||||
self.current_playback_index: int = -1
|
||||
self._is_playing = False
|
||||
self._saved_volume: int = 80
|
||||
self._saved_track_path: str = ""
|
||||
self._saved_position_ms: int = 0
|
||||
self._saved_playing: bool = False
|
||||
self._pending_seek_ms: int = 0
|
||||
|
||||
self._setup_ui()
|
||||
self._load_settings()
|
||||
self._setup_player()
|
||||
self._setup_hotkeys()
|
||||
self._scan_library()
|
||||
self._resume_playback_if_saved()
|
||||
self._on_refresh_devices_clicked()
|
||||
|
||||
def _cleanup_worker(self):
|
||||
@ -304,6 +320,102 @@ class MainWindow(QMainWindow):
|
||||
self.worker_thread.deleteLater()
|
||||
self.worker_thread = None
|
||||
|
||||
def _load_settings(self):
|
||||
config = self.config_loader
|
||||
self.output_dir_input.setText(
|
||||
config.get("General", "output_dir",
|
||||
fallback=os.path.join(os.path.expanduser("~"), "Music", "iPod"))
|
||||
)
|
||||
self.format_combo.setCurrentText(config.get("General", "format", fallback="m4a"))
|
||||
self.quality_spin.setValue(config.get_int("General", "quality", fallback=256))
|
||||
self.enable_video_check.setChecked(
|
||||
config.get_boolean("Video", "enable_video", fallback=False)
|
||||
)
|
||||
self.video_resolution_combo.setCurrentText(
|
||||
config.get("Video", "resolution", fallback="640x480")
|
||||
)
|
||||
self.embed_artwork_check.setChecked(
|
||||
config.get_boolean("Metadata", "embed_artwork", fallback=True)
|
||||
)
|
||||
self.use_musicbrainz_check.setChecked(
|
||||
config.get_boolean("Metadata", "use_musicbrainz", fallback=False)
|
||||
)
|
||||
self.clean_temp_check.setChecked(
|
||||
config.get_boolean("Advanced", "clean_temp", fallback=True)
|
||||
)
|
||||
self.auto_detect_check.setChecked(
|
||||
config.get_boolean("Device", "auto_detect", fallback=True)
|
||||
)
|
||||
show_download = config.get_boolean("Advanced", "show_download", fallback=False)
|
||||
self.show_download_check.setChecked(show_download)
|
||||
self.show_download_tab = show_download
|
||||
self.tab_widget.setTabVisible(self.download_tab_index, show_download)
|
||||
|
||||
self._saved_volume = config.get_int("Playback", "last_volume", fallback=80)
|
||||
self._saved_track_path = config.get("Playback", "last_track_path", fallback="")
|
||||
self._saved_position_ms = config.get_int("Playback", "last_position_ms", fallback=0)
|
||||
self._saved_playing = config.get_boolean("Playback", "last_playing", fallback=False)
|
||||
|
||||
def _save_settings(self):
|
||||
config = self.config_loader
|
||||
config.set("General", "output_dir", self.output_dir_input.text())
|
||||
config.set("General", "format", self.format_combo.currentText())
|
||||
config.set("General", "quality", str(self.quality_spin.value()))
|
||||
config.set("Video", "enable_video", str(self.enable_video_check.isChecked()).lower())
|
||||
config.set("Video", "resolution", self.video_resolution_combo.currentText())
|
||||
config.set("Metadata", "embed_artwork", str(self.embed_artwork_check.isChecked()).lower())
|
||||
config.set("Metadata", "use_musicbrainz", str(self.use_musicbrainz_check.isChecked()).lower())
|
||||
config.set("Advanced", "clean_temp", str(self.clean_temp_check.isChecked()).lower())
|
||||
config.set("Device", "auto_detect", str(self.auto_detect_check.isChecked()).lower())
|
||||
config.set("Advanced", "show_download", str(self.show_download_check.isChecked()).lower())
|
||||
|
||||
config.set("Playback", "last_volume", str(self.volume_slider.value()))
|
||||
if self.current_playback_index >= 0 and self.player.source().isValid():
|
||||
track_path = self.player.source().toLocalFile()
|
||||
if track_path and os.path.exists(track_path):
|
||||
config.set("Playback", "last_track_path", track_path)
|
||||
position = self.player.position()
|
||||
duration = self.player.duration()
|
||||
if duration > 0 and position > duration - 2000:
|
||||
position = 0
|
||||
config.set("Playback", "last_position_ms", str(position))
|
||||
config.set("Playback", "last_playing", str(self._is_playing).lower())
|
||||
else:
|
||||
config.set("Playback", "last_track_path", "")
|
||||
config.set("Playback", "last_position_ms", "0")
|
||||
else:
|
||||
config.set("Playback", "last_track_path", "")
|
||||
config.set("Playback", "last_position_ms", "0")
|
||||
config.save()
|
||||
|
||||
def _setup_hotkeys(self):
|
||||
self.hotkey_manager = HotkeyManager(self, self.config_loader)
|
||||
self.hotkey_manager.register_all({
|
||||
"play_pause": self._on_play_pause_clicked,
|
||||
"prev_track": self._on_prev_clicked,
|
||||
"next_track": self._on_next_clicked,
|
||||
"volume_up": lambda: self._adjust_volume(5),
|
||||
"volume_down": lambda: self._adjust_volume(-5),
|
||||
"seek_forward": self._seek_forward,
|
||||
"seek_backward": self._seek_backward,
|
||||
"tab_download": lambda: self.tab_widget.setCurrentIndex(self.download_tab_index),
|
||||
"tab_library": lambda: self.tab_widget.setCurrentIndex(1),
|
||||
"tab_ipod": lambda: self.tab_widget.setCurrentIndex(self.ipod_tab_index),
|
||||
"tab_settings": lambda: self.tab_widget.setCurrentIndex(3),
|
||||
"search_focus": self._focus_search,
|
||||
"select_all": self._select_all_current,
|
||||
"library_refresh": self._on_refresh_library_clicked,
|
||||
"ipod_refresh": self._on_refresh_devices_clicked,
|
||||
"download": self._on_download_clicked,
|
||||
"delete_selected": self._delete_selected_current,
|
||||
"edit_metadata": self._on_edit_metadata,
|
||||
})
|
||||
self._refresh_shortcuts_table()
|
||||
|
||||
def _adjust_volume(self, delta: int):
|
||||
new_val = max(0, min(100, self.volume_slider.value() + delta))
|
||||
self.volume_slider.setValue(new_val)
|
||||
|
||||
def _setup_ui(self):
|
||||
main_widget = QWidget()
|
||||
main_layout = QVBoxLayout(main_widget)
|
||||
@ -329,6 +441,7 @@ class MainWindow(QMainWindow):
|
||||
self._setup_settings_tab(settings_tab)
|
||||
|
||||
self.tab_widget.setTabVisible(self.download_tab_index, self.show_download_tab)
|
||||
self.tab_widget.setTabVisible(self.ipod_tab_index, False)
|
||||
|
||||
self.statusBar().showMessage("Ready")
|
||||
self.setCentralWidget(main_widget)
|
||||
@ -480,9 +593,11 @@ class MainWindow(QMainWindow):
|
||||
|
||||
self.library_progress = QProgressBar()
|
||||
self.library_progress.setRange(0, 100)
|
||||
self.library_progress.setVisible(False)
|
||||
layout.addWidget(self.library_progress)
|
||||
|
||||
self.library_status = QLabel("Ready")
|
||||
self.library_status.setVisible(False)
|
||||
layout.addWidget(self.library_status)
|
||||
|
||||
def _setup_library_toolbar(self, layout):
|
||||
@ -508,6 +623,11 @@ class MainWindow(QMainWindow):
|
||||
self.library_remove_btn.clicked.connect(self._on_library_remove_selected)
|
||||
toolbar.addWidget(self.library_remove_btn)
|
||||
|
||||
self.library_edit_btn = QPushButton("\u270F\uFE0F Edit")
|
||||
self.library_edit_btn.setEnabled(False)
|
||||
self.library_edit_btn.clicked.connect(self._on_edit_metadata)
|
||||
toolbar.addWidget(self.library_edit_btn)
|
||||
|
||||
refresh_btn = QPushButton("\uD83D\uDD04")
|
||||
refresh_btn.setToolTip("Refresh Library")
|
||||
refresh_btn.clicked.connect(self._on_refresh_library_clicked)
|
||||
@ -668,6 +788,28 @@ class MainWindow(QMainWindow):
|
||||
|
||||
layout.addWidget(advanced_group)
|
||||
|
||||
shortcuts_group = QGroupBox("Keyboard Shortcuts")
|
||||
shortcuts_layout = QVBoxLayout(shortcuts_group)
|
||||
|
||||
self.shortcuts_table = QTableWidget()
|
||||
self.shortcuts_table.setColumnCount(2)
|
||||
self.shortcuts_table.setHorizontalHeaderLabels(["Action", "Shortcut"])
|
||||
self.shortcuts_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
|
||||
self.shortcuts_table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection)
|
||||
self.shortcuts_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
|
||||
self.shortcuts_table.setAlternatingRowColors(True)
|
||||
self.shortcuts_table.cellDoubleClicked.connect(self._on_shortcut_double_clicked)
|
||||
shortcuts_header = self.shortcuts_table.horizontalHeader()
|
||||
shortcuts_header.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
|
||||
shortcuts_header.setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
|
||||
shortcuts_layout.addWidget(self.shortcuts_table)
|
||||
|
||||
reset_shortcuts_btn = QPushButton("Reset to Defaults")
|
||||
reset_shortcuts_btn.clicked.connect(self._on_reset_shortcuts_clicked)
|
||||
shortcuts_layout.addWidget(reset_shortcuts_btn)
|
||||
|
||||
layout.addWidget(shortcuts_group)
|
||||
|
||||
layout.addStretch()
|
||||
|
||||
about_label = QLabel(
|
||||
@ -686,14 +828,14 @@ class MainWindow(QMainWindow):
|
||||
"""Initialize QMediaPlayer and QAudioOutput for playback."""
|
||||
self.player = QMediaPlayer()
|
||||
self.audio_output = QAudioOutput()
|
||||
self.audio_output.setVolume(0.8)
|
||||
self.audio_output.setVolume(self._saved_volume / 100.0)
|
||||
self.player.setAudioOutput(self.audio_output)
|
||||
|
||||
self.player.positionChanged.connect(self._on_media_position_changed)
|
||||
self.player.durationChanged.connect(self._on_media_duration_changed)
|
||||
self.player.mediaStatusChanged.connect(self._on_media_status_changed)
|
||||
|
||||
def _play_track_from_index(self, index: int):
|
||||
def _play_track_from_index(self, index: int, start_position_ms: int = 0, auto_play: bool = True):
|
||||
"""Start playing a track from the library table by row index."""
|
||||
if index < 0 or index >= self.library_table.rowCount():
|
||||
return
|
||||
@ -706,9 +848,13 @@ class MainWindow(QMainWindow):
|
||||
|
||||
self.current_playback_index = index
|
||||
self.player.setSource(QUrl.fromLocalFile(track_data["path"]))
|
||||
self.player.play()
|
||||
self._is_playing = True
|
||||
self.play_btn.setText("\u23F8")
|
||||
if auto_play:
|
||||
self.player.play()
|
||||
self._is_playing = True
|
||||
self.play_btn.setText("\u23F8")
|
||||
else:
|
||||
self._is_playing = False
|
||||
self.play_btn.setText("\u25B6")
|
||||
|
||||
self.now_playing_label.setText(f"{track_data['artist']} \u2014 {track_data['title']}")
|
||||
self.now_playing_label.setStyleSheet(
|
||||
@ -721,7 +867,6 @@ class MainWindow(QMainWindow):
|
||||
self.cover_label.setPixmap(pixmap)
|
||||
self.cover_label.setToolTip(track_data.get("album", ""))
|
||||
else:
|
||||
# Fallback: try looking up by artist/album in cover cache
|
||||
artist = track_data.get("artist", "")
|
||||
album = track_data.get("album", "")
|
||||
cached = cover_cache.get(artist, album) if artist else None
|
||||
@ -733,10 +878,14 @@ class MainWindow(QMainWindow):
|
||||
self.cover_label.clear()
|
||||
self.cover_label.setToolTip("")
|
||||
|
||||
# Highlight the current row
|
||||
self.library_table.clearSelection()
|
||||
self.library_table.selectRow(index)
|
||||
|
||||
if start_position_ms > 0:
|
||||
self._pending_seek_ms = start_position_ms
|
||||
else:
|
||||
self._pending_seek_ms = 0
|
||||
|
||||
def _on_play_pause_clicked(self):
|
||||
if self.player is None:
|
||||
return
|
||||
@ -770,6 +919,37 @@ class MainWindow(QMainWindow):
|
||||
idx = min(self.library_table.rowCount() - 1, self.current_playback_index + 1)
|
||||
self._play_track_from_index(idx)
|
||||
|
||||
def _seek_forward(self):
|
||||
if self.player is None or self.player.duration() <= 0:
|
||||
return
|
||||
new_pos = min(self.player.duration(), self.player.position() + 5000)
|
||||
self.player.setPosition(new_pos)
|
||||
|
||||
def _seek_backward(self):
|
||||
if self.player is None or self.player.duration() <= 0:
|
||||
return
|
||||
new_pos = max(0, self.player.position() - 5000)
|
||||
self.player.setPosition(new_pos)
|
||||
|
||||
def _focus_search(self):
|
||||
self.tab_widget.setCurrentIndex(1)
|
||||
self.library_search_input.setFocus()
|
||||
self.library_search_input.selectAll()
|
||||
|
||||
def _select_all_current(self):
|
||||
idx = self.tab_widget.currentIndex()
|
||||
if idx == 1:
|
||||
self.library_table.selectAll()
|
||||
elif idx == 2:
|
||||
self.transferred_list.selectAll()
|
||||
|
||||
def _delete_selected_current(self):
|
||||
idx = self.tab_widget.currentIndex()
|
||||
if idx == 1:
|
||||
self._on_library_remove_selected()
|
||||
elif idx == 2:
|
||||
self._on_remove_tracks_clicked()
|
||||
|
||||
def _on_slider_pressed(self):
|
||||
if self.player:
|
||||
self.player.setPosition(
|
||||
@ -802,6 +982,10 @@ class MainWindow(QMainWindow):
|
||||
elif status == QMediaPlayer.MediaStatus.NoMedia:
|
||||
self.play_btn.setText("\u25B6")
|
||||
self._is_playing = False
|
||||
elif status == QMediaPlayer.MediaStatus.LoadedMedia:
|
||||
if self._pending_seek_ms > 0:
|
||||
self.player.setPosition(self._pending_seek_ms)
|
||||
self._pending_seek_ms = 0
|
||||
|
||||
def _on_table_double_clicked(self, row: int):
|
||||
self._play_track_from_index(row)
|
||||
@ -823,8 +1007,13 @@ class MainWindow(QMainWindow):
|
||||
return f"{size_bytes / 1024:.0f} KB"
|
||||
return f"{size_bytes / (1024 * 1024):.1f} MB"
|
||||
|
||||
def _scan_library(self):
|
||||
"""Scan Output Directory for ready-to-transfer and source files."""
|
||||
def _scan_library(self, force: bool = False):
|
||||
"""Scan Output Directory for ready-to-transfer and source files.
|
||||
|
||||
Uses a JSON cache to skip metadata extraction for files whose mtime
|
||||
hasn't changed since they were last cached. Pass ``force=True`` to
|
||||
bypass the cache and re-read every file (refresh / F5).
|
||||
"""
|
||||
output_dir = self.output_dir_input.text().strip()
|
||||
if not output_dir or not os.path.isdir(output_dir):
|
||||
self.library_ready.clear()
|
||||
@ -834,7 +1023,10 @@ class MainWindow(QMainWindow):
|
||||
|
||||
ready = []
|
||||
sources = []
|
||||
cache_hits = 0
|
||||
cache_misses = 0
|
||||
|
||||
cache = get_library_cache()
|
||||
handler = MetadataHandler()
|
||||
|
||||
for root, _dirs, files in os.walk(output_dir):
|
||||
@ -844,60 +1036,35 @@ class MainWindow(QMainWindow):
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
|
||||
cached = None if force else cache.get(fpath)
|
||||
|
||||
if cached is not None:
|
||||
cache_hits += 1
|
||||
if ext in AUDIO_EXTS:
|
||||
cached.pop("_mtime", None)
|
||||
ready.append(cached)
|
||||
elif ext in SOURCE_EXTS:
|
||||
cached.pop("_mtime", None)
|
||||
sources.append(cached)
|
||||
continue
|
||||
|
||||
cache_misses += 1
|
||||
|
||||
if ext in AUDIO_EXTS:
|
||||
size = os.path.getsize(fpath)
|
||||
cover_path = None
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
duration = info.duration if info else 0
|
||||
artist = info.artist if info else "Unknown"
|
||||
title = info.title if info else fname
|
||||
album = info.album if info else ""
|
||||
genre = info.genre if info else ""
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
duration = 0
|
||||
artist = "Unknown"
|
||||
title = fname
|
||||
album = ""
|
||||
genre = ""
|
||||
track_num = 0
|
||||
|
||||
ready.append({
|
||||
"path": fpath,
|
||||
"title": title,
|
||||
"artist": artist,
|
||||
"album": album,
|
||||
"duration_s": duration,
|
||||
"size": size,
|
||||
"genre": genre,
|
||||
"track_num": track_num,
|
||||
"cover_path": cover_path,
|
||||
})
|
||||
track_data = self._extract_ready_track(fpath, fname, handler)
|
||||
ready.append(track_data)
|
||||
cache.put(fpath, track_data)
|
||||
elif ext in SOURCE_EXTS:
|
||||
size = os.path.getsize(fpath)
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
title = info.title if info else fname
|
||||
artist = info.artist if info else "Unknown"
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
title = fname
|
||||
artist = "Unknown"
|
||||
track_num = 0
|
||||
cover_path = None
|
||||
sources.append({
|
||||
"path": fpath,
|
||||
"title": title,
|
||||
"artist": artist,
|
||||
"size": size,
|
||||
"track_num": track_num,
|
||||
"cover_path": cover_path,
|
||||
})
|
||||
track_data = self._extract_source_track(fpath, fname, handler)
|
||||
sources.append(track_data)
|
||||
cache.put(fpath, track_data)
|
||||
|
||||
cache.save()
|
||||
|
||||
if cache_misses:
|
||||
self.library_status.setText(
|
||||
f"Scanned {cache_misses} file(s) — {cache_hits} from cache")
|
||||
|
||||
# Deduplicate by file path (defensive — os.walk shouldn't produce dupes)
|
||||
seen_paths = set()
|
||||
deduped_ready = []
|
||||
for track in ready:
|
||||
@ -910,6 +1077,61 @@ class MainWindow(QMainWindow):
|
||||
self.library_scanned_sources = sources
|
||||
self._refresh_library_ui()
|
||||
|
||||
def _resume_playback_if_saved(self):
|
||||
if not self._saved_track_path or not os.path.exists(self._saved_track_path):
|
||||
return
|
||||
for row in range(self.library_table.rowCount()):
|
||||
data_item = self.library_table.item(row, 0)
|
||||
if data_item is None:
|
||||
continue
|
||||
_is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
||||
if track_data and track_data.get("path", "") == self._saved_track_path:
|
||||
self._play_track_from_index(row, start_position_ms=self._saved_position_ms, auto_play=self._saved_playing)
|
||||
return
|
||||
|
||||
def _extract_ready_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
||||
size = os.path.getsize(fpath)
|
||||
cover_path = None
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
duration = info.duration if info else 0
|
||||
artist = info.artist if info else "Unknown"
|
||||
title = info.title if info else fname
|
||||
album = info.album if info else ""
|
||||
genre = info.genre if info else ""
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
duration = 0
|
||||
artist = "Unknown"
|
||||
title = fname
|
||||
album = ""
|
||||
genre = ""
|
||||
track_num = 0
|
||||
return {
|
||||
"path": fpath, "title": title, "artist": artist,
|
||||
"album": album, "duration_s": duration, "size": size,
|
||||
"genre": genre, "track_num": track_num, "cover_path": cover_path,
|
||||
}
|
||||
|
||||
def _extract_source_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
||||
size = os.path.getsize(fpath)
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
title = info.title if info else fname
|
||||
artist = info.artist if info else "Unknown"
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
title = fname
|
||||
artist = "Unknown"
|
||||
track_num = 0
|
||||
cover_path = None
|
||||
return {
|
||||
"path": fpath, "title": title, "artist": artist,
|
||||
"size": size, "track_num": track_num, "cover_path": cover_path,
|
||||
}
|
||||
|
||||
def _refresh_library_ui(self):
|
||||
self.library_table.setSortingEnabled(False)
|
||||
self.library_table.setRowCount(0)
|
||||
@ -1005,9 +1227,10 @@ class MainWindow(QMainWindow):
|
||||
has_ready = ready_count > 0
|
||||
has_source = source_count > 0
|
||||
|
||||
self.library_transfer_btn.setEnabled(has_ready)
|
||||
self.library_transfer_btn.setEnabled(has_ready and self.current_mount_point is not None)
|
||||
self.library_convert_btn.setEnabled(has_source)
|
||||
self.library_remove_btn.setEnabled(len(all_tracks) > 0)
|
||||
self.library_edit_btn.setEnabled(len(all_tracks) > 0)
|
||||
|
||||
parts = []
|
||||
if ready_count:
|
||||
@ -1077,7 +1300,7 @@ class MainWindow(QMainWindow):
|
||||
|
||||
def _on_refresh_library_clicked(self):
|
||||
self.library_status.setText("Scanning library...")
|
||||
self._scan_library()
|
||||
self._scan_library(force=True)
|
||||
|
||||
def _on_library_add_files(self):
|
||||
source_filter = "Audio Files (*.flac *.wav *.ogg *.opus *.wma *.ape *.aiff *.alac)"
|
||||
@ -1094,6 +1317,71 @@ class MainWindow(QMainWindow):
|
||||
if new_files:
|
||||
self._scan_library()
|
||||
|
||||
def _incremental_update_after_edit(self, file_paths: list):
|
||||
"""Re-extract metadata for edited files, update cache, refresh UI
|
||||
without walking the entire library directory."""
|
||||
cache = get_library_cache()
|
||||
handler = MetadataHandler()
|
||||
|
||||
paths_set = set(file_paths)
|
||||
|
||||
for fpath in file_paths:
|
||||
ext = os.path.splitext(fpath)[1].lower()
|
||||
if ext in AUDIO_EXTS:
|
||||
data = self._extract_ready_track(fpath, os.path.basename(fpath), handler)
|
||||
cache.put(fpath, data)
|
||||
for item in self.library_ready:
|
||||
if item["path"] == fpath:
|
||||
item.update(data)
|
||||
break
|
||||
else:
|
||||
self.library_ready.append(data)
|
||||
elif ext in SOURCE_EXTS:
|
||||
data = self._extract_source_track(fpath, os.path.basename(fpath), handler)
|
||||
cache.put(fpath, data)
|
||||
for item in self.library_scanned_sources:
|
||||
if item["path"] == fpath:
|
||||
item.update(data)
|
||||
break
|
||||
else:
|
||||
self.library_scanned_sources.append(data)
|
||||
|
||||
cache.save()
|
||||
self._refresh_library_ui()
|
||||
|
||||
def _on_edit_metadata(self):
|
||||
"""Open metadata editor for selected track(s)."""
|
||||
selected_rows = set()
|
||||
for item in self.library_table.selectedItems():
|
||||
selected_rows.add(item.row())
|
||||
|
||||
if not selected_rows:
|
||||
QMessageBox.information(self, "Info", "No track selected")
|
||||
return
|
||||
|
||||
file_paths = []
|
||||
for row in sorted(selected_rows):
|
||||
data_item = self.library_table.item(row, 0)
|
||||
if data_item is None:
|
||||
continue
|
||||
is_ready, track_data = data_item.data(Qt.ItemDataRole.UserRole)
|
||||
file_path = track_data.get("path", "")
|
||||
if file_path and os.path.exists(file_path):
|
||||
file_paths.append(file_path)
|
||||
|
||||
if not file_paths:
|
||||
QMessageBox.warning(self, "Error", "No valid files found on disk")
|
||||
return
|
||||
|
||||
dialog = MetadataEditorDialog(file_paths, self)
|
||||
if dialog.exec() == MetadataEditorDialog.DialogCode.Accepted:
|
||||
self._incremental_update_after_edit(file_paths)
|
||||
|
||||
def _get_first_selected_row(self):
|
||||
for item in self.library_table.selectedItems():
|
||||
return item.row()
|
||||
return None
|
||||
|
||||
def _on_library_remove_selected(self):
|
||||
"""Delete selected files from disk and clean up empty directories."""
|
||||
files_to_delete = []
|
||||
@ -1148,13 +1436,24 @@ class MainWindow(QMainWindow):
|
||||
except OSError as e:
|
||||
logger.error(f"Failed to delete {path}: {e}")
|
||||
|
||||
for f in files_to_delete:
|
||||
invalidate_cache_for_path(f["path"])
|
||||
|
||||
self._scan_library()
|
||||
QMessageBox.information(self, "Done", f"Deleted {deleted} file(s)")
|
||||
|
||||
def _on_search_text_changed(self, text: str):
|
||||
query = text.lower()
|
||||
for row in range(self.library_table.rowCount()):
|
||||
item = self.library_table.item(row, 3)
|
||||
match = not text or text.lower() in item.text().lower() if item else True
|
||||
if not query:
|
||||
self.library_table.setRowHidden(row, False)
|
||||
continue
|
||||
match = False
|
||||
for col in (1, 2, 3): # Artist, Title, Album
|
||||
item = self.library_table.item(row, col)
|
||||
if item and query in item.text().lower():
|
||||
match = True
|
||||
break
|
||||
self.library_table.setRowHidden(row, not match)
|
||||
|
||||
def _on_library_table_context_menu(self, pos):
|
||||
@ -1172,7 +1471,7 @@ class MainWindow(QMainWindow):
|
||||
play_action = menu.addAction("\u25B6 Play")
|
||||
play_action.triggered.connect(lambda: self._play_track_from_index(row))
|
||||
|
||||
if is_ready:
|
||||
if is_ready and self.current_mount_point is not None:
|
||||
transfer_action = menu.addAction("\uD83D\uDCE4 Transfer to iPod")
|
||||
transfer_action.triggered.connect(self._on_library_transfer_clicked)
|
||||
else:
|
||||
@ -1181,6 +1480,9 @@ class MainWindow(QMainWindow):
|
||||
|
||||
menu.addSeparator()
|
||||
|
||||
edit_action = menu.addAction("\u270F\uFE0F Edit Metadata")
|
||||
edit_action.triggered.connect(lambda: self._on_edit_metadata())
|
||||
|
||||
show_action = menu.addAction("\uD83D\uDCC2 Show in File Manager")
|
||||
show_action.triggered.connect(lambda: self._on_show_in_file_manager(track_data))
|
||||
|
||||
@ -1437,9 +1739,15 @@ class MainWindow(QMainWindow):
|
||||
self.mount_button.setEnabled(False)
|
||||
self.eject_button.setEnabled(True)
|
||||
|
||||
self.tab_widget.setTabVisible(self.ipod_tab_index, True)
|
||||
|
||||
self.library_progress.setVisible(True)
|
||||
self.library_status.setVisible(True)
|
||||
|
||||
self._load_ipod_tracks()
|
||||
|
||||
def _set_device_unmounted(self, device: dict):
|
||||
self.current_mount_point = None
|
||||
self.device_info_label.setText(
|
||||
f"Device: {device['name']}\n"
|
||||
f"Click 'Mount' to connect"
|
||||
@ -1452,8 +1760,15 @@ class MainWindow(QMainWindow):
|
||||
self.export_button.setEnabled(False)
|
||||
self.export_progress.setVisible(False)
|
||||
self.export_status.setVisible(False)
|
||||
self.library_transfer_btn.setEnabled(False)
|
||||
|
||||
self.library_progress.setVisible(False)
|
||||
self.library_status.setVisible(False)
|
||||
|
||||
self.tab_widget.setTabVisible(self.ipod_tab_index, True)
|
||||
|
||||
def _set_device_not_found(self):
|
||||
self.current_mount_point = None
|
||||
self.device_info_label.setText("No iPod devices found")
|
||||
self.device_status_label.setText("Status: Not connected")
|
||||
self.device_combo.clear()
|
||||
@ -1464,6 +1779,12 @@ class MainWindow(QMainWindow):
|
||||
self.export_button.setEnabled(False)
|
||||
self.export_progress.setVisible(False)
|
||||
self.export_status.setVisible(False)
|
||||
self.library_transfer_btn.setEnabled(False)
|
||||
|
||||
self.library_progress.setVisible(False)
|
||||
self.library_status.setVisible(False)
|
||||
|
||||
self.tab_widget.setTabVisible(self.ipod_tab_index, False)
|
||||
|
||||
def _on_mount_clicked(self):
|
||||
if not self.ipod_devices:
|
||||
@ -1510,6 +1831,9 @@ class MainWindow(QMainWindow):
|
||||
success = ipod.unmount_device()
|
||||
if success or True:
|
||||
self.current_mount_point = None
|
||||
self.library_transfer_btn.setEnabled(False)
|
||||
self.library_progress.setVisible(False)
|
||||
self.library_status.setVisible(False)
|
||||
if self.ipod_devices:
|
||||
self.ipod_devices[0]["mounted"] = False
|
||||
self.ipod_devices[0]["mount_point"] = None
|
||||
@ -1745,11 +2069,52 @@ class MainWindow(QMainWindow):
|
||||
QMessageBox.information(self, "Duplicates" if success else "Error", message)
|
||||
self._cleanup_worker()
|
||||
|
||||
def _refresh_shortcuts_table(self):
|
||||
if self.hotkey_manager is None:
|
||||
return
|
||||
names = self.hotkey_manager.all_action_names()
|
||||
self.shortcuts_table.setRowCount(len(names))
|
||||
for row, name in enumerate(names):
|
||||
label = self.hotkey_manager.get_label(name)
|
||||
key = self.hotkey_manager.get_current_key_string(name)
|
||||
self.shortcuts_table.setItem(row, 0, QTableWidgetItem(label))
|
||||
item = QTableWidgetItem(key if key else "\u2014")
|
||||
self.shortcuts_table.setItem(row, 1, item)
|
||||
|
||||
def _on_shortcut_double_clicked(self, row: int, _col: int):
|
||||
if self.hotkey_manager is None:
|
||||
return
|
||||
names = self.hotkey_manager.all_action_names()
|
||||
if row < 0 or row >= len(names):
|
||||
return
|
||||
action_name = names[row]
|
||||
current = self.hotkey_manager.get_current_key_string(action_name)
|
||||
|
||||
dialog = KeyCaptureDialog(self, current=current)
|
||||
if dialog.exec() == QDialog.DialogCode.Accepted:
|
||||
seq = dialog.captured_sequence()
|
||||
self.hotkey_manager.set_shortcut_seq(action_name, seq)
|
||||
self._refresh_shortcuts_table()
|
||||
|
||||
def _on_reset_shortcuts_clicked(self):
|
||||
if self.hotkey_manager is None:
|
||||
return
|
||||
reply = QMessageBox.question(
|
||||
self, "Reset Shortcuts",
|
||||
"Reset all keyboard shortcuts to their default values?",
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
|
||||
)
|
||||
if reply == QMessageBox.StandardButton.Yes:
|
||||
self.hotkey_manager.reset_to_defaults()
|
||||
self._refresh_shortcuts_table()
|
||||
QMessageBox.information(self, "Done", "Shortcuts reset to defaults.")
|
||||
|
||||
def closeEvent(self, event):
|
||||
if self.worker_thread and self.worker_thread.isRunning():
|
||||
self.worker_thread.stop()
|
||||
self.worker_thread.wait(5000)
|
||||
self.worker_thread.deleteLater()
|
||||
self._save_settings()
|
||||
event.accept()
|
||||
|
||||
|
||||
|
||||
659
src/metadata_editor.py
Normal file
659
src/metadata_editor.py
Normal file
@ -0,0 +1,659 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Metadata Editor Dialog for neo-pod-desktop
|
||||
Allows editing tags and cover art of audio files (MP3, M4A, FLAC, OGG, OPUS).
|
||||
Supports single-track and batch (multi-track) editing.
|
||||
"""
|
||||
|
||||
import os
|
||||
import base64
|
||||
import struct
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any, List, Tuple
|
||||
from io import BytesIO
|
||||
|
||||
from PIL import Image
|
||||
from mutagen import File as MutagenFile
|
||||
from mutagen.mp4 import MP4
|
||||
from mutagen.id3 import ID3
|
||||
from mutagen.flac import FLAC
|
||||
from mutagen.oggvorbis import OggVorbis
|
||||
from mutagen.oggopus import OggOpus
|
||||
|
||||
from PyQt6.QtWidgets import (
|
||||
QDialog, QVBoxLayout, QHBoxLayout, QFormLayout,
|
||||
QPushButton, QLabel, QLineEdit, QSpinBox, QCheckBox,
|
||||
QFileDialog, QMessageBox, QFrame, QDialogButtonBox,
|
||||
)
|
||||
from PyQt6.QtCore import Qt, QSize
|
||||
from PyQt6.QtGui import QPixmap, QDragEnterEvent, QDropEvent
|
||||
|
||||
from metadata_handler import MetadataHandler
|
||||
from artwork.cache import cover_cache
|
||||
|
||||
|
||||
COVER_SIZE = 250
|
||||
SENTINEL_MULTIPLE = "__MULTIPLE__"
|
||||
|
||||
|
||||
class MetadataEditorDialog(QDialog):
|
||||
"""Dialog for viewing and editing audio file metadata and cover art."""
|
||||
|
||||
def __init__(self, file_paths: List[str], parent=None):
|
||||
super().__init__(parent)
|
||||
self.file_paths = file_paths
|
||||
self._batch_mode = len(file_paths) > 1
|
||||
self._current_file = file_paths[0] if file_paths else ""
|
||||
self._original_tags: Dict[str, Any] = {}
|
||||
self._mixed_fields: set = set()
|
||||
self._new_cover_path: Optional[str] = None
|
||||
self._cover_removed = False
|
||||
self._existing_cover_data: Optional[bytes] = None
|
||||
|
||||
title = f"Edit Metadata ({len(file_paths)} files)" if self._batch_mode else "Edit Metadata"
|
||||
self.setWindowTitle(title)
|
||||
self.setMinimumSize(620, 440)
|
||||
self.setModal(True)
|
||||
|
||||
self._load_tags()
|
||||
self._setup_ui()
|
||||
self._populate_form()
|
||||
|
||||
if self._batch_mode:
|
||||
self._apply_batch_mode()
|
||||
|
||||
def _load_tags(self):
|
||||
"""Load metadata from file(s). In batch mode, finds common values."""
|
||||
if not self.file_paths:
|
||||
return
|
||||
|
||||
first_tags = self._load_single_file(self.file_paths[0])
|
||||
self._original_tags = dict(first_tags)
|
||||
self._existing_cover_data = self._extract_cover_from_path(self.file_paths[0])
|
||||
|
||||
if self._batch_mode:
|
||||
all_tags = [first_tags]
|
||||
all_different = False
|
||||
for path in self.file_paths[1:]:
|
||||
tags = self._load_single_file(path, skip_cover=True)
|
||||
all_tags.append(tags)
|
||||
if all_different:
|
||||
continue
|
||||
for key in list(self._original_tags.keys()):
|
||||
if tags.get(key) != self._original_tags.get(key):
|
||||
self._mixed_fields.add(key)
|
||||
|
||||
def _load_single_file(self, file_path: str, skip_cover: bool = False) -> Dict[str, Any]:
|
||||
"""Load tags from a single file. Returns dict of tag_key -> value."""
|
||||
result: Dict[str, Any] = {}
|
||||
try:
|
||||
audio = MutagenFile(file_path)
|
||||
if audio is None:
|
||||
return result
|
||||
tags = audio.tags
|
||||
|
||||
ext = Path(file_path).suffix.lower()
|
||||
|
||||
field_map = {
|
||||
'title': ('title', '\xa9nam'),
|
||||
'artist': ('artist', '\xa9ART'),
|
||||
'album': ('album', '\xa9alb'),
|
||||
'album_artist': ('albumartist', 'aART'),
|
||||
'genre': ('genre', '\xa9gen'),
|
||||
'date': ('date', '\xa9day'),
|
||||
'comment': ('comment', '\xa9cmt'),
|
||||
}
|
||||
for our_key, (easy_key, mp4_key) in field_map.items():
|
||||
val = self._safe_get(tags, easy_key, mp4_key)
|
||||
if val is not None:
|
||||
result[our_key] = val
|
||||
|
||||
result.update(self._load_track_disc(tags))
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def _load_track_disc(self, tags) -> Dict[str, Any]:
|
||||
result = {}
|
||||
if tags is None:
|
||||
return result
|
||||
|
||||
if 'trkn' in tags:
|
||||
trkn = tags['trkn']
|
||||
if isinstance(trkn, list) and trkn:
|
||||
t = trkn[0]
|
||||
if isinstance(t, (tuple, list)) and len(t) >= 1:
|
||||
result['track_number'] = int(t[0]) if t[0] else 0
|
||||
if len(t) >= 2:
|
||||
result['track_total'] = int(t[1]) if t[1] else 0
|
||||
elif 'TRCK' in tags:
|
||||
tn, tt = self._parse_slash(str(tags['TRCK']))
|
||||
if tn:
|
||||
result['track_number'] = tn
|
||||
if tt:
|
||||
result['track_total'] = tt
|
||||
elif hasattr(tags, 'getall'):
|
||||
try:
|
||||
frames = tags.getall('TRCK')
|
||||
if frames:
|
||||
tn, tt = self._parse_slash(str(frames[0]))
|
||||
if tn:
|
||||
result['track_number'] = tn
|
||||
if tt:
|
||||
result['track_total'] = tt
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
for mp4_key, (vorbis_key, num_key, total_key) in (
|
||||
('disk', ('DISCNUMBER', 'disc_number', 'disc_total')),
|
||||
):
|
||||
if mp4_key in tags:
|
||||
disk_val = tags[mp4_key]
|
||||
if isinstance(disk_val, list) and disk_val:
|
||||
d = disk_val[0]
|
||||
if isinstance(d, (tuple, list)) and len(d) >= 1:
|
||||
result[num_key] = int(d[0]) if d[0] else 0
|
||||
if len(d) >= 2:
|
||||
result[total_key] = int(d[1]) if d[1] else 0
|
||||
elif vorbis_key in tags:
|
||||
dn, dt = self._parse_slash(str(tags[vorbis_key]))
|
||||
if dn:
|
||||
result[num_key] = dn
|
||||
if dt:
|
||||
result[total_key] = dt
|
||||
elif hasattr(tags, 'getall'):
|
||||
try:
|
||||
frames = tags.getall('TPOS')
|
||||
if frames:
|
||||
dn, dt = self._parse_slash(str(frames[0]))
|
||||
if dn:
|
||||
result[num_key] = dn
|
||||
if dt:
|
||||
result[total_key] = dt
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def _extract_cover_from_path(self, file_path: str) -> Optional[bytes]:
|
||||
try:
|
||||
audio = MutagenFile(file_path)
|
||||
if audio is None:
|
||||
return None
|
||||
return self._extract_cover_raw(audio, audio.tags)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _safe_get(self, tags, easy_key: str, mp4_key: str) -> Optional[str]:
|
||||
if tags is None:
|
||||
return None
|
||||
try:
|
||||
if mp4_key in tags:
|
||||
val = tags[mp4_key]
|
||||
if isinstance(val, list) and val:
|
||||
return str(val[0])
|
||||
return str(val)
|
||||
except (ValueError, KeyError):
|
||||
pass
|
||||
try:
|
||||
easy_lower = easy_key.lower()
|
||||
for k in tags.keys():
|
||||
if k.lower() == easy_lower:
|
||||
val = tags[k]
|
||||
if isinstance(val, list) and val:
|
||||
return str(val[0])
|
||||
return str(val)
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
id3_map = {
|
||||
'title': 'TIT2', 'artist': 'TPE1', 'album': 'TALB',
|
||||
'albumartist': 'TPE2', 'genre': 'TCON', 'date': 'TDRC', 'comment': 'COMM',
|
||||
}
|
||||
frame_id = id3_map.get(easy_key.lower())
|
||||
if frame_id and hasattr(tags, 'getall'):
|
||||
try:
|
||||
frames = tags.getall(frame_id)
|
||||
if frames:
|
||||
return str(frames[0])
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def _parse_slash(self, val: str):
|
||||
try:
|
||||
parts = str(val).split('/')
|
||||
first = int(parts[0]) if parts[0].strip().isdigit() else 0
|
||||
second = int(parts[1]) if len(parts) > 1 and parts[1].strip().isdigit() else 0
|
||||
return first, second
|
||||
except Exception:
|
||||
return 0, 0
|
||||
|
||||
def _extract_cover_raw(self, audio, tags) -> Optional[bytes]:
|
||||
if hasattr(tags, 'getall') and tags is not None:
|
||||
try:
|
||||
for frame in tags.getall('APIC'):
|
||||
return frame.data
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
if tags and 'covr' in tags:
|
||||
cover_list = tags['covr']
|
||||
if isinstance(cover_list, list) and cover_list:
|
||||
return bytes(cover_list[0])
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
if tags and 'metadata_block_picture' in tags:
|
||||
pic_data = tags['metadata_block_picture']
|
||||
if isinstance(pic_data, list):
|
||||
pic_data = pic_data[0]
|
||||
pic_data = str(pic_data)
|
||||
decoded = base64.b64decode(pic_data)
|
||||
pos = 0
|
||||
pos += 4
|
||||
mime_len = struct.unpack('>I', decoded[pos:pos + 4])[0]
|
||||
pos += 4 + mime_len
|
||||
desc_len = struct.unpack('>I', decoded[pos:pos + 4])[0]
|
||||
pos += 4 + desc_len
|
||||
pos += 20
|
||||
return decoded[pos:]
|
||||
except Exception:
|
||||
pass
|
||||
if hasattr(audio, 'pictures') and audio.pictures:
|
||||
return audio.pictures[0].data
|
||||
return None
|
||||
|
||||
def _setup_ui(self):
|
||||
outer = QVBoxLayout(self)
|
||||
outer.setSpacing(12)
|
||||
|
||||
body = QHBoxLayout()
|
||||
body.setSpacing(16)
|
||||
|
||||
cover_layout = QVBoxLayout()
|
||||
cover_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
|
||||
|
||||
self.cover_label = QLabel()
|
||||
self.cover_label.setFixedSize(COVER_SIZE, COVER_SIZE)
|
||||
self.cover_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.cover_label.setStyleSheet(
|
||||
"QLabel { border: 2px dashed #888; border-radius: 6px; background: palette(window); }"
|
||||
)
|
||||
self.cover_label.setAcceptDrops(True)
|
||||
self.cover_label.setText("Drag cover\nhere\nor click")
|
||||
self.cover_label.setToolTip("Click to choose cover image, drag & drop to replace")
|
||||
self.cover_label.mousePressEvent = self._on_cover_clicked
|
||||
cover_layout.addWidget(self.cover_label)
|
||||
|
||||
self.cover_apply_check = QCheckBox("Apply cover to all selected")
|
||||
self.cover_apply_check.setVisible(False)
|
||||
self.cover_apply_check.setChecked(True)
|
||||
cover_layout.addWidget(self.cover_apply_check)
|
||||
|
||||
clear_btn = QPushButton("Remove Cover")
|
||||
clear_btn.clicked.connect(self._on_clear_cover)
|
||||
cover_layout.addWidget(clear_btn)
|
||||
|
||||
cover_layout.addStretch()
|
||||
body.addLayout(cover_layout)
|
||||
|
||||
form_frame = QFrame()
|
||||
form_layout = QFormLayout(form_frame)
|
||||
form_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
|
||||
form_layout.setFieldGrowthPolicy(
|
||||
QFormLayout.FieldGrowthPolicy.ExpandingFieldsGrow
|
||||
)
|
||||
|
||||
self.title_edit = QLineEdit()
|
||||
self.title_edit.setPlaceholderText("Track Title")
|
||||
self.artist_edit = QLineEdit()
|
||||
self.artist_edit.setPlaceholderText("Artist")
|
||||
self.album_edit = QLineEdit()
|
||||
self.album_edit.setPlaceholderText("Album")
|
||||
self.album_artist_edit = QLineEdit()
|
||||
self.album_artist_edit.setPlaceholderText("Album Artist")
|
||||
self.genre_edit = QLineEdit()
|
||||
self.genre_edit.setPlaceholderText("Genre")
|
||||
self.comment_edit = QLineEdit()
|
||||
self.comment_edit.setPlaceholderText("Comment")
|
||||
|
||||
self.year_spin = QSpinBox()
|
||||
self.year_spin.setRange(0, 2100)
|
||||
self.year_spin.setSpecialValueText("")
|
||||
|
||||
self.track_num_spin = QSpinBox()
|
||||
self.track_num_spin.setRange(0, 999)
|
||||
self.track_num_spin.setSpecialValueText("")
|
||||
self.track_total_spin = QSpinBox()
|
||||
self.track_total_spin.setRange(0, 999)
|
||||
self.track_total_spin.setSpecialValueText("")
|
||||
|
||||
self.disc_num_spin = QSpinBox()
|
||||
self.disc_num_spin.setRange(0, 99)
|
||||
self.disc_num_spin.setSpecialValueText("")
|
||||
self.disc_total_spin = QSpinBox()
|
||||
self.disc_total_spin.setRange(0, 99)
|
||||
self.disc_total_spin.setSpecialValueText("")
|
||||
|
||||
self._title_label = QLabel("Title:")
|
||||
form_layout.addRow(self._title_label, self.title_edit)
|
||||
|
||||
self._artist_label = QLabel("Artist:")
|
||||
form_layout.addRow(self._artist_label, self.artist_edit)
|
||||
|
||||
self._album_label = QLabel("Album:")
|
||||
form_layout.addRow(self._album_label, self.album_edit)
|
||||
|
||||
self._album_artist_label = QLabel("Album Artist:")
|
||||
form_layout.addRow(self._album_artist_label, self.album_artist_edit)
|
||||
|
||||
self._genre_label = QLabel("Genre:")
|
||||
form_layout.addRow(self._genre_label, self.genre_edit)
|
||||
|
||||
self._year_label = QLabel("Year:")
|
||||
form_layout.addRow(self._year_label, self.year_spin)
|
||||
|
||||
self._track_label = QLabel("Track #:")
|
||||
self._track_row_layout = QHBoxLayout()
|
||||
self._track_row_layout.addWidget(self.track_num_spin)
|
||||
self._track_row_layout.addWidget(QLabel("/"))
|
||||
self._track_row_layout.addWidget(self.track_total_spin)
|
||||
self._track_row_layout.addStretch()
|
||||
form_layout.addRow(self._track_label, self._track_row_layout)
|
||||
|
||||
self._disc_label = QLabel("Disc #:")
|
||||
self._disc_row_layout = QHBoxLayout()
|
||||
self._disc_row_layout.addWidget(self.disc_num_spin)
|
||||
self._disc_row_layout.addWidget(QLabel("/"))
|
||||
self._disc_row_layout.addWidget(self.disc_total_spin)
|
||||
self._disc_row_layout.addStretch()
|
||||
form_layout.addRow(self._disc_label, self._disc_row_layout)
|
||||
|
||||
self._comment_label = QLabel("Comment:")
|
||||
form_layout.addRow(self._comment_label, self.comment_edit)
|
||||
|
||||
body.addWidget(form_frame, stretch=1)
|
||||
|
||||
outer.addLayout(body, stretch=1)
|
||||
|
||||
self._track_count_label = QLabel("")
|
||||
self._track_count_label.setVisible(False)
|
||||
self._track_count_label.setStyleSheet("color: #888; font-size: 11px;")
|
||||
outer.addWidget(self._track_count_label)
|
||||
|
||||
buttons = QDialogButtonBox(
|
||||
QDialogButtonBox.StandardButton.Save |
|
||||
QDialogButtonBox.StandardButton.Cancel
|
||||
)
|
||||
buttons.accepted.connect(self._on_save)
|
||||
buttons.rejected.connect(self.reject)
|
||||
outer.addWidget(buttons)
|
||||
|
||||
def _apply_batch_mode(self):
|
||||
title_text = f"\uD83D\uDCCB Editing {len(self.file_paths)} selected tracks"
|
||||
self._track_count_label.setText(title_text)
|
||||
self._track_count_label.setVisible(True)
|
||||
|
||||
self._title_label.setText("Title: (hidden in batch)")
|
||||
self.title_edit.setVisible(False)
|
||||
|
||||
for w in (self.track_num_spin, self.track_total_spin):
|
||||
w.setVisible(False)
|
||||
self._track_label.setText("Track #: (hidden in batch)")
|
||||
|
||||
for widget in self.findChildren(QLabel):
|
||||
if widget.text() == "/" and widget.parent() in (self._track_row_layout.parent(), None):
|
||||
pass
|
||||
|
||||
self.cover_apply_check.setVisible(True)
|
||||
|
||||
for key in self._mixed_fields:
|
||||
self._set_mixed(key)
|
||||
|
||||
def _set_mixed(self, key: str):
|
||||
placeholders = {
|
||||
'title': "(different values)",
|
||||
'artist': "(different values)",
|
||||
'album': "(different values)",
|
||||
'album_artist': "(different values)",
|
||||
'genre': "(different values)",
|
||||
'date': "",
|
||||
'comment': "(different values)",
|
||||
'track_number': 0,
|
||||
'track_total': 0,
|
||||
'disc_number': 0,
|
||||
'disc_total': 0,
|
||||
}
|
||||
edit_map = {
|
||||
'title': self.title_edit,
|
||||
'artist': self.artist_edit,
|
||||
'album': self.album_edit,
|
||||
'album_artist': self.album_artist_edit,
|
||||
'genre': self.genre_edit,
|
||||
'comment': self.comment_edit,
|
||||
}
|
||||
spin_map = {
|
||||
'date': self.year_spin,
|
||||
'track_number': self.track_num_spin,
|
||||
'track_total': self.track_total_spin,
|
||||
'disc_number': self.disc_num_spin,
|
||||
'disc_total': self.disc_total_spin,
|
||||
}
|
||||
|
||||
if key in edit_map:
|
||||
edit_map[key].setPlaceholderText(placeholders.get(key, "(different values)"))
|
||||
edit_map[key].setText("")
|
||||
elif key in spin_map:
|
||||
spin_map[key].setValue(0)
|
||||
spin_map[key].setSpecialValueText("\u2014")
|
||||
|
||||
def _populate_form(self):
|
||||
t = self._original_tags
|
||||
|
||||
self.title_edit.setText(t.get('title', ''))
|
||||
self.artist_edit.setText(t.get('artist', ''))
|
||||
self.album_edit.setText(t.get('album', ''))
|
||||
self.album_artist_edit.setText(t.get('album_artist', ''))
|
||||
self.genre_edit.setText(t.get('genre', ''))
|
||||
self.comment_edit.setText(t.get('comment', ''))
|
||||
|
||||
date_str = t.get('date', '')
|
||||
try:
|
||||
year = int(str(date_str)[:4])
|
||||
except (ValueError, TypeError):
|
||||
year = 0
|
||||
self.year_spin.setValue(year)
|
||||
|
||||
self.track_num_spin.setValue(t.get('track_number', 0))
|
||||
self.track_total_spin.setValue(t.get('track_total', 0))
|
||||
self.disc_num_spin.setValue(t.get('disc_number', 0))
|
||||
self.disc_total_spin.setValue(t.get('disc_total', 0))
|
||||
|
||||
self._update_cover_display()
|
||||
|
||||
def _update_cover_display(self):
|
||||
if self._cover_removed:
|
||||
self.cover_label.setText("No cover")
|
||||
self.cover_label.setPixmap(QPixmap())
|
||||
return
|
||||
|
||||
if self._batch_mode and not self._new_cover_path and self._existing_cover_data is None:
|
||||
self.cover_label.setText("Multiple\ncovers\n\nClick to replace")
|
||||
self.cover_label.setPixmap(QPixmap())
|
||||
return
|
||||
|
||||
pixmap = None
|
||||
if self._new_cover_path and os.path.exists(self._new_cover_path):
|
||||
pixmap = QPixmap(self._new_cover_path)
|
||||
elif self._existing_cover_data:
|
||||
pixmap = QPixmap()
|
||||
pixmap.loadFromData(self._existing_cover_data)
|
||||
|
||||
if pixmap and not pixmap.isNull():
|
||||
scaled = pixmap.scaled(
|
||||
COVER_SIZE - 16, COVER_SIZE - 16,
|
||||
Qt.AspectRatioMode.KeepAspectRatio,
|
||||
Qt.TransformationMode.SmoothTransformation,
|
||||
)
|
||||
self.cover_label.setPixmap(scaled)
|
||||
else:
|
||||
self.cover_label.setText("No cover")
|
||||
self.cover_label.setPixmap(QPixmap())
|
||||
|
||||
def _on_cover_clicked(self, event):
|
||||
path, _ = QFileDialog.getOpenFileName(
|
||||
self, "Select Cover Image", "",
|
||||
"Images (*.png *.jpg *.jpeg *.bmp *.gif *.tiff *.webp);;All Files (*)"
|
||||
)
|
||||
if path:
|
||||
self._new_cover_path = path
|
||||
self._cover_removed = False
|
||||
self._update_cover_display()
|
||||
|
||||
def _on_clear_cover(self):
|
||||
self._new_cover_path = None
|
||||
self._cover_removed = True
|
||||
self._update_cover_display()
|
||||
|
||||
def dragEnterEvent(self, event: QDragEnterEvent):
|
||||
if event.mimeData().hasUrls():
|
||||
event.acceptProposedAction()
|
||||
|
||||
def dropEvent(self, event: QDropEvent):
|
||||
urls = event.mimeData().urls()
|
||||
if urls:
|
||||
path = urls[0].toLocalFile()
|
||||
if os.path.isfile(path):
|
||||
ext = Path(path).suffix.lower()
|
||||
if ext in ('.png', '.jpg', '.jpeg', '.bmp', '.gif', '.tiff', '.webp'):
|
||||
self._new_cover_path = path
|
||||
self._cover_removed = False
|
||||
self._update_cover_display()
|
||||
|
||||
def _collect_current(self) -> Dict[str, Any]:
|
||||
return {
|
||||
'title': self.title_edit.text().strip(),
|
||||
'artist': self.artist_edit.text().strip(),
|
||||
'album': self.album_edit.text().strip(),
|
||||
'album_artist': self.album_artist_edit.text().strip(),
|
||||
'genre': self.genre_edit.text().strip(),
|
||||
'date': str(self.year_spin.value()) if self.year_spin.value() > 0 else '',
|
||||
'track_number': self.track_num_spin.value(),
|
||||
'track_total': self.track_total_spin.value(),
|
||||
'disc_number': self.disc_num_spin.value(),
|
||||
'disc_total': self.disc_total_spin.value(),
|
||||
'comment': self.comment_edit.text().strip(),
|
||||
}
|
||||
|
||||
def _collect_changes(self) -> Dict[str, Any]:
|
||||
if self._batch_mode:
|
||||
return self._collect_batch_changes()
|
||||
return self._collect_single_changes()
|
||||
|
||||
def _collect_single_changes(self) -> Dict[str, Any]:
|
||||
changes = {}
|
||||
current = self._collect_current()
|
||||
for key, val in current.items():
|
||||
orig = self._original_tags.get(key)
|
||||
if isinstance(val, int) and isinstance(orig, int):
|
||||
if val != orig:
|
||||
changes[key] = val
|
||||
elif isinstance(val, int) and isinstance(orig, str):
|
||||
try:
|
||||
if int(str(orig)[:4]) != val:
|
||||
changes[key] = val
|
||||
except ValueError:
|
||||
if val:
|
||||
changes[key] = val
|
||||
elif isinstance(val, int) and orig is None:
|
||||
if val != 0:
|
||||
changes[key] = val
|
||||
elif str(val) != str(orig or ''):
|
||||
changes[key] = val
|
||||
return changes
|
||||
|
||||
def _collect_batch_changes(self) -> Dict[str, Any]:
|
||||
current = self._collect_current()
|
||||
exclude = {'title', 'track_number', 'track_total'}
|
||||
changes = {}
|
||||
for key, val in current.items():
|
||||
if key in exclude:
|
||||
continue
|
||||
orig = self._original_tags.get(key)
|
||||
if key in self._mixed_fields:
|
||||
if isinstance(val, int) and val == 0:
|
||||
continue
|
||||
if isinstance(val, str) and not val:
|
||||
continue
|
||||
changes[key] = val
|
||||
else:
|
||||
if isinstance(val, int) and isinstance(orig, int):
|
||||
if val != orig:
|
||||
changes[key] = val
|
||||
elif isinstance(val, str) and orig is not None:
|
||||
if str(val) != str(orig):
|
||||
changes[key] = val
|
||||
elif val and not orig:
|
||||
changes[key] = val
|
||||
|
||||
return changes
|
||||
|
||||
def _on_save(self):
|
||||
changes = self._collect_changes()
|
||||
|
||||
cover_arg = None
|
||||
if self._cover_removed:
|
||||
cover_arg = ''
|
||||
elif self._new_cover_path:
|
||||
cover_arg = self._new_cover_path
|
||||
|
||||
if not changes and cover_arg is None:
|
||||
self.accept()
|
||||
return
|
||||
|
||||
paths = self.file_paths if self._batch_mode else [self.file_paths[0]]
|
||||
|
||||
try:
|
||||
handler = MetadataHandler()
|
||||
failed = 0
|
||||
for path in paths:
|
||||
ok = handler.update_tags(path, changes, cover_arg)
|
||||
if not ok:
|
||||
failed += 1
|
||||
else:
|
||||
self._invalidate_cover_cache(path, changes, cover_arg)
|
||||
|
||||
if failed == len(paths):
|
||||
QMessageBox.warning(self, "Error", "Failed to save metadata changes.")
|
||||
elif failed:
|
||||
QMessageBox.warning(
|
||||
self, "Partial Success",
|
||||
f"Saved {len(paths) - failed}/{len(paths)} files.\n{failed} file(s) failed."
|
||||
)
|
||||
self.accept()
|
||||
else:
|
||||
self.accept()
|
||||
except Exception as e:
|
||||
QMessageBox.critical(self, "Error", f"Failed to save metadata:\n{e}")
|
||||
|
||||
def _invalidate_cover_cache(self, file_path: str, changes: Dict[str, Any],
|
||||
cover_arg: Optional[str]) -> None:
|
||||
"""Update cover cache after successful save."""
|
||||
try:
|
||||
audio = MutagenFile(file_path)
|
||||
new_cover = self._extract_cover_raw(audio, audio.tags) if audio else None
|
||||
|
||||
artist = changes.get('artist') or self._original_tags.get('artist', 'Unknown Artist')
|
||||
album = changes.get('album') or self._original_tags.get('album', '')
|
||||
|
||||
if new_cover:
|
||||
cover_cache.put(artist, album, new_cover)
|
||||
elif cover_arg == '':
|
||||
key = cover_cache._key(artist, album)
|
||||
cache_path = os.path.join(cover_cache.cache_dir, f"{key}.jpg")
|
||||
if os.path.exists(cache_path):
|
||||
os.remove(cache_path)
|
||||
except Exception:
|
||||
pass
|
||||
@ -7,7 +7,6 @@ Handles metadata injection and album art embedding
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
import tempfile
|
||||
import struct
|
||||
import base64
|
||||
from typing import Optional, Dict, Any
|
||||
@ -22,7 +21,7 @@ from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK
|
||||
from mutagen.easyid3 import EasyID3
|
||||
|
||||
from youtube_downloader import TrackInfo
|
||||
from artwork.cache import cover_cache
|
||||
from artwork.cache import cover_cache, xdg_cache_path
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
@ -39,7 +38,7 @@ class MetadataHandler:
|
||||
Args:
|
||||
temp_dir: Directory for temporary files (optional)
|
||||
"""
|
||||
self.temp_dir = temp_dir or tempfile.gettempdir()
|
||||
self.temp_dir = temp_dir or xdg_cache_path("thumbnails")
|
||||
os.makedirs(self.temp_dir, exist_ok=True)
|
||||
|
||||
def download_thumbnail(self, thumbnail_url: str) -> Optional[str]:
|
||||
@ -557,6 +556,307 @@ class MetadataHandler:
|
||||
# MusicBrainz integration would go here
|
||||
return track_info
|
||||
|
||||
def update_tags(self, file_path: str, tags: Dict[str, Any],
|
||||
new_cover_path: Optional[str] = None) -> bool:
|
||||
"""
|
||||
Update metadata tags in an audio file.
|
||||
|
||||
Args:
|
||||
file_path: Path to audio file
|
||||
tags: Dict of tag_name -> value
|
||||
Supported keys: title, artist, album, album_artist, genre,
|
||||
date, track_number, track_total, disc_number, disc_total, comment
|
||||
new_cover_path: Path to new cover image, None to keep existing,
|
||||
empty string '' to remove cover art
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
from mutagen.flac import FLAC, Picture as FLACPicture
|
||||
from mutagen.oggvorbis import OggVorbis
|
||||
from mutagen.oggopus import OggOpus
|
||||
|
||||
file_ext = Path(file_path).suffix.lower()
|
||||
|
||||
try:
|
||||
if file_ext in ('.m4a', '.aac', '.mp4'):
|
||||
return self._update_mp4_tags(file_path, tags, new_cover_path)
|
||||
elif file_ext == '.mp3':
|
||||
return self._update_mp3_tags(file_path, tags, new_cover_path)
|
||||
elif file_ext == '.flac':
|
||||
return self._update_vorbis_tags(file_path, tags, new_cover_path, use_picture_block=True)
|
||||
elif file_ext in ('.ogg', '.opus'):
|
||||
return self._update_vorbis_tags(file_path, tags, new_cover_path, use_picture_block=False)
|
||||
else:
|
||||
logger.warning(f"Unsupported format for tag update: {file_ext}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update tags in {file_path}: {e}")
|
||||
return False
|
||||
|
||||
def _load_cover_data(self, cover_path: str) -> Optional[bytes]:
|
||||
"""Load cover image from path, convert to JPEG if needed."""
|
||||
try:
|
||||
img = Image.open(cover_path)
|
||||
if img.mode != 'RGB':
|
||||
img = img.convert('RGB')
|
||||
buf = BytesIO()
|
||||
img.save(buf, 'JPEG', quality=90)
|
||||
return buf.getvalue()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load cover image {cover_path}: {e}")
|
||||
return None
|
||||
|
||||
def _update_mp4_tags(self, file_path: str, tags: Dict[str, Any],
|
||||
new_cover_path: Optional[str]) -> bool:
|
||||
audio = MP4(file_path)
|
||||
|
||||
text_fields = {
|
||||
'title': '\xa9nam',
|
||||
'artist': '\xa9ART',
|
||||
'album': '\xa9alb',
|
||||
'album_artist': 'aART',
|
||||
'genre': '\xa9gen',
|
||||
'date': '\xa9day',
|
||||
'comment': '\xa9cmt',
|
||||
}
|
||||
for key, mp4_key in text_fields.items():
|
||||
if key in tags:
|
||||
val = tags[key]
|
||||
if val:
|
||||
audio[mp4_key] = [str(val)]
|
||||
elif mp4_key in audio:
|
||||
del audio[mp4_key]
|
||||
|
||||
if 'track_number' in tags or 'track_total' in tags:
|
||||
trkn_val = list(audio.get('trkn', [(0, 0)])[0]) if 'trkn' in audio else [0, 0]
|
||||
if 'track_number' in tags:
|
||||
trkn_val[0] = tags['track_number'] or 0
|
||||
if 'track_total' in tags:
|
||||
trkn_val[1] = tags['track_total'] or 0
|
||||
if trkn_val[0] or trkn_val[1]:
|
||||
audio['trkn'] = [tuple(trkn_val)]
|
||||
elif 'trkn' in audio:
|
||||
del audio['trkn']
|
||||
|
||||
if 'disc_number' in tags or 'disc_total' in tags:
|
||||
disk_val = list(audio.get('disk', [(0, 0)])[0]) if 'disk' in audio else [0, 0]
|
||||
if 'disc_number' in tags:
|
||||
disk_val[0] = tags['disc_number'] or 0
|
||||
if 'disc_total' in tags:
|
||||
disk_val[1] = tags['disc_total'] or 0
|
||||
if disk_val[0] or disk_val[1]:
|
||||
audio['disk'] = [tuple(disk_val)]
|
||||
elif 'disk' in audio:
|
||||
del audio['disk']
|
||||
|
||||
if new_cover_path == '':
|
||||
if 'covr' in audio:
|
||||
del audio['covr']
|
||||
elif new_cover_path is not None:
|
||||
data = self._load_cover_data(new_cover_path)
|
||||
if data:
|
||||
cover = MP4Cover(data, imageformat=MP4Cover.FORMAT_JPEG)
|
||||
audio['covr'] = [cover]
|
||||
|
||||
audio.save()
|
||||
return True
|
||||
|
||||
def _update_mp3_tags(self, file_path: str, tags: Dict[str, Any],
|
||||
new_cover_path: Optional[str]) -> bool:
|
||||
try:
|
||||
audio = EasyID3(file_path)
|
||||
except Exception:
|
||||
audio = ID3()
|
||||
audio.save(file_path)
|
||||
audio = EasyID3(file_path)
|
||||
|
||||
easy_fields = {
|
||||
'title': 'title',
|
||||
'artist': 'artist',
|
||||
'album': 'album',
|
||||
'album_artist': 'albumartist',
|
||||
'genre': 'genre',
|
||||
'date': 'date',
|
||||
}
|
||||
for key, easy_key in easy_fields.items():
|
||||
if key in tags:
|
||||
val = tags[key]
|
||||
if val:
|
||||
audio[easy_key] = str(val)
|
||||
elif easy_key in audio:
|
||||
del audio[easy_key]
|
||||
|
||||
if 'comment' in tags:
|
||||
try:
|
||||
audio['comment'] = str(tags['comment']) if tags['comment'] else ''
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if 'track_number' in tags or 'track_total' in tags:
|
||||
current = audio.get('tracknumber', [''])[0] if 'tracknumber' in audio else ''
|
||||
tn, tt = self._parse_slash_value(current)
|
||||
if 'track_number' in tags:
|
||||
tn = tags['track_number'] or 0
|
||||
if 'track_total' in tags:
|
||||
tt = tags['track_total'] or 0
|
||||
if tn or tt:
|
||||
audio['tracknumber'] = [f"{tn}/{tt}" if tt else str(tn)]
|
||||
elif 'tracknumber' in audio:
|
||||
del audio['tracknumber']
|
||||
|
||||
if 'disc_number' in tags or 'disc_total' in tags:
|
||||
current = audio.get('discnumber', [''])[0] if 'discnumber' in audio else ''
|
||||
dn, dt = self._parse_slash_value(current)
|
||||
if 'disc_number' in tags:
|
||||
dn = tags['disc_number'] or 0
|
||||
if 'disc_total' in tags:
|
||||
dt = tags['disc_total'] or 0
|
||||
if dn or dt:
|
||||
audio['discnumber'] = [f"{dn}/{dt}" if dt else str(dn)]
|
||||
elif 'discnumber' in audio:
|
||||
del audio['discnumber']
|
||||
|
||||
audio.save()
|
||||
|
||||
if new_cover_path is not None:
|
||||
audio = ID3(file_path)
|
||||
if new_cover_path == '':
|
||||
audio.delall('APIC')
|
||||
else:
|
||||
data = self._load_cover_data(new_cover_path)
|
||||
if data:
|
||||
audio.delall('APIC')
|
||||
audio.add(APIC(encoding=3, mime='image/jpeg', type=3,
|
||||
desc='Cover', data=data))
|
||||
audio.save()
|
||||
|
||||
return True
|
||||
|
||||
def _parse_slash_value(self, current: str) -> Tuple[int, int]:
|
||||
"""Parse '5/12' format into (5, 12)."""
|
||||
try:
|
||||
parts = str(current).split('/')
|
||||
first = int(parts[0]) if parts[0].strip().isdigit() else 0
|
||||
second = int(parts[1]) if len(parts) > 1 and parts[1].strip().isdigit() else 0
|
||||
return first, second
|
||||
except Exception:
|
||||
return 0, 0
|
||||
|
||||
def _update_vorbis_tags(self, file_path: str, tags: Dict[str, Any],
|
||||
new_cover_path: Optional[str],
|
||||
use_picture_block: bool = False) -> bool:
|
||||
from mutagen.flac import FLAC, Picture as FLACPicture
|
||||
from mutagen.oggvorbis import OggVorbis
|
||||
from mutagen.oggopus import OggOpus
|
||||
|
||||
ext = Path(file_path).suffix.lower()
|
||||
if ext == '.flac':
|
||||
audio = FLAC(file_path)
|
||||
elif ext == '.opus':
|
||||
audio = OggOpus(file_path)
|
||||
else:
|
||||
audio = OggVorbis(file_path)
|
||||
|
||||
if audio.tags is None:
|
||||
audio.add_tags()
|
||||
|
||||
vorbis_fields = {
|
||||
'title': 'TITLE',
|
||||
'artist': 'ARTIST',
|
||||
'album': 'ALBUM',
|
||||
'album_artist': 'ALBUMARTIST',
|
||||
'genre': 'GENRE',
|
||||
'date': 'DATE',
|
||||
'comment': 'COMMENT',
|
||||
}
|
||||
for key, vorbis_key in vorbis_fields.items():
|
||||
if key in tags:
|
||||
val = tags[key]
|
||||
if val:
|
||||
audio.tags[vorbis_key] = str(val)
|
||||
elif vorbis_key in audio.tags:
|
||||
try:
|
||||
del audio.tags[vorbis_key]
|
||||
except (KeyError, TypeError):
|
||||
pass
|
||||
|
||||
if 'track_number' in tags or 'track_total' in tags:
|
||||
current = audio.tags.get('TRACKNUMBER', [''])[0] if 'TRACKNUMBER' in audio.tags else ''
|
||||
tn, tt = self._parse_slash_value(current)
|
||||
if 'track_number' in tags:
|
||||
tn = tags['track_number'] or 0
|
||||
if 'track_total' in tags:
|
||||
tt = tags['track_total'] or 0
|
||||
if tn or tt:
|
||||
audio.tags['TRACKNUMBER'] = f"{tn}/{tt}" if tt else str(tn)
|
||||
elif 'TRACKNUMBER' in audio.tags:
|
||||
try:
|
||||
del audio.tags['TRACKNUMBER']
|
||||
except (KeyError, TypeError):
|
||||
pass
|
||||
|
||||
if 'disc_number' in tags or 'disc_total' in tags:
|
||||
current = audio.tags.get('DISCNUMBER', [''])[0] if 'DISCNUMBER' in audio.tags else ''
|
||||
dn, dt = self._parse_slash_value(current)
|
||||
if 'disc_number' in tags:
|
||||
dn = tags['disc_number'] or 0
|
||||
if 'disc_total' in tags:
|
||||
dt = tags['disc_total'] or 0
|
||||
if dn or dt:
|
||||
audio.tags['DISCNUMBER'] = f"{dn}/{dt}" if dt else str(dn)
|
||||
elif 'DISCNUMBER' in audio.tags:
|
||||
try:
|
||||
del audio.tags['DISCNUMBER']
|
||||
except (KeyError, TypeError):
|
||||
pass
|
||||
|
||||
if new_cover_path is not None:
|
||||
self._clear_cover_vorbis(audio)
|
||||
if new_cover_path != '':
|
||||
data = self._load_cover_data(new_cover_path)
|
||||
if data:
|
||||
self._embed_cover_vorbis(audio, data)
|
||||
elif use_picture_block:
|
||||
pass
|
||||
else:
|
||||
pass
|
||||
|
||||
audio.save()
|
||||
return True
|
||||
|
||||
def _clear_cover_vorbis(self, audio) -> None:
|
||||
"""Remove all cover art from Vorbis/FLAC file."""
|
||||
if audio.tags is None:
|
||||
return
|
||||
if 'METADATA_BLOCK_PICTURE' in audio.tags:
|
||||
del audio.tags['METADATA_BLOCK_PICTURE']
|
||||
if 'COVERART' in audio.tags:
|
||||
del audio.tags['COVERART']
|
||||
if hasattr(audio, 'clear_pictures'):
|
||||
audio.clear_pictures()
|
||||
|
||||
def _embed_cover_vorbis(self, audio, data: bytes) -> None:
|
||||
"""Embed JPEG cover art into Vorbis/FLAC file."""
|
||||
from mutagen.flac import Picture as FLACPicture
|
||||
|
||||
pic = FLACPicture()
|
||||
pic.type = 3 # Cover (front)
|
||||
pic.mime = 'image/jpeg'
|
||||
pic.desc = 'Cover'
|
||||
pic.width = 0
|
||||
pic.height = 0
|
||||
pic.depth = 24
|
||||
pic.data = data
|
||||
pic_data = pic.write()
|
||||
|
||||
if hasattr(audio, 'add_picture'):
|
||||
audio.clear_pictures()
|
||||
audio.add_picture(pic)
|
||||
else:
|
||||
encoded = base64.b64encode(pic_data).decode('ascii')
|
||||
audio.tags['METADATA_BLOCK_PICTURE'] = encoded
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user