en2zmax/neo-pod-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: en2zmax:199fb43f04a179dd2e3ec29d448a4d05d29e4078
Branches Tags
en2zmax:main
..
compare: en2zmax:ec797b80745262dd3e0b9b96dc6103f91d388d41
Branches Tags
en2zmax:main

No commits in common. "199fb43f04a179dd2e3ec29d448a4d05d29e4078" and "ec797b80745262dd3e0b9b96dc6103f91d388d41" have entirely different histories.
199fb43f04 ... ec797b8074

34 changed files with 3102 additions and 4127 deletions
Show Stats Expand all files Collapse all files

1
.gitignore vendored
View File

@ -15,6 +15,7 @@ ENV/
# Temporary files
temp/
downloads/
converted/
# IDE files

74
README.md
View File

@ -1,16 +1,17 @@
# neo-pod-desktop
Desktop application for converting, managing and transferring music to iPod Nano 7G. Includes format conversion, metadata management, local library with playback, cover art pipeline, and full iTunesDB/ArtworkDB generation.
Desktop application for downloading, converting, managing and transferring music to iPod Nano 7G. Includes YouTube audio extraction, format conversion, metadata management, local library with playback, cover art pipeline, and full iTunesDB/ArtworkDB generation.
## Features
- **Download** audio from YouTube videos/playlists
- **Convert** to iPod-compatible formats (AAC/M4A, MP3) with `-map 0:t?` cover preservation
- **Metadata** handling -- title, artist, album, track number, genre, embedded cover art
- **Cover art pipeline** -- extract from audio files (APIC/covr/FLAC pictures), encode to iPod Nano 7G formats (RGB565), write ArtworkDB + ithmb
- **iTunes-style player** -- play local library with cover art display (60x60), seek, volume
- **iPod Nano 7G database** -- full iTunesCDB sync, SQLite management, Locations.itdb with HASHAB checksums
- **Library management** -- sortable table, source/ready track separation, bulk transfer
- **Portable AppImage** -- `scripts/build-appimage.sh` builds standalone x86_64 AppImage
- **Metadata** handling — title, artist, album, track number, genre, embedded cover art
- **Cover art pipeline** — extract from audio files (APIC/covr/FLAC pictures), encode to iPod Nano 7G formats (RGB565), write ArtworkDB + ithmb
- **iTunes-style player** — play local library with cover art display (60×60), seek, volume
- **iPod Nano 7G database** — full iTunesCDB sync, SQLite management, Locations.itdb with HASHAB checksums
- **Library management** — sortable table, source/ready track separation, bulk transfer
- **Portable AppImage** — `scripts/build-appimage.sh` builds standalone x86_64 AppImage
- **Config** stored in `~/.config/neo-pod-desktop/config.ini`
## Installation
@ -38,6 +39,12 @@ Install FFmpeg:
python src/main.py
```
### Command Line Interface
```bash
python src/cli.py --url "https://www.youtube.com/watch?v=VIDEO_ID" --output-dir "/path/to/ipod"
```
### AppImage (portable)
```bash
@ -54,31 +61,42 @@ Build from source:
```
src/
main.py -- GUI application (PyQt6)
track_info.py -- TrackInfo data class
audio_converter.py -- FFmpeg audio conversion
metadata_handler.py -- ID3/MP4 tag reading and writing
metadata_editor.py -- GUI metadata editing widget
config_loader.py -- Config in ~/.config/
ipod_device.py -- iPod device detection and file management
ipod_nano7_db.py -- Nano 7G SQLite + iTunesCDB + ArtworkDB
ipod_mhbd_writer.py -- Binary iTunesDB writer
hashab.py -- HASHAB checksum generation
hotkeys.py -- Global keyboard shortcut management
library_cache.py -- Local library caching
main.py — GUI application (PyQt6)
cli.py — CLI interface
youtube_downloader.py — YouTube audio downloader
audio_converter.py — FFmpeg audio conversion
metadata_handler.py — ID3/MP4 tag reading and writing
config_loader.py — Config in ~/.config/
ipod_device.py — iPod device detection and file management
ipod_nano7_db.py — Nano 7G SQLite + iTunesCDB + ArtworkDB
hashab.py — HASHAB checksum generation
artwork/
__init__.py -- Public API
presets.py -- 50+ iPod artwork format definitions
types.py -- ArtworkEntry, EncodedFormatPayload, IthmbLocation
codecs.py -- RGB565/RGB555/UYVY/JPEG encode/decode (numpy)
extractor.py -- Cover extraction from audio files (mutagen)
chunks.py -- ArtworkDB binary parser/writer (MHFD->MHII->MHNI)
writer.py -- ithmb file + ArtworkDB writing
cache.py -- Local JPEG cover cache for player UI
__init__.py — Public API
presets.py — 50+ iPod artwork format definitions
types.py — ArtworkEntry, EncodedFormatPayload, IthmbLocation
codecs.py — RGB565/RGB555/UYVY/JPEG encode/decode (numpy)
extractor.py — Cover extraction from audio files (mutagen)
chunks.py — ArtworkDB binary parser/writer (MHFD→MHII→MHNI)
writer.py — ithmb file + ArtworkDB writing
cache.py — Local JPEG cover cache for player UI
scripts/
build-appimage.sh -- AppImage builder (PyInstaller + appimagetool)
build-appimage.sh — AppImage builder (PyInstaller + appimagetool)
```
## Cover Art Re-embed
One-time script to restore cover art in already-converted files:
```bash
python /tmp/reembed_covers.py /path/to/source_music /path/to/output_dir
```
Matches output files to source files by artist+album, detects embedded covers and separate `cover.jpg`/`folder.jpg` files.
## Legal Disclaimer
This tool is intended for personal use only. Users are responsible for complying with YouTube's Terms of Service and copyright laws.
## License
MIT License

8
install.bat
View File

@ -1,7 +1,7 @@
@echo off
REM Installation script for neo-pod-desktop (Windows)
REM Installation script for YouTube Music to iPod Nano Transfer Tool (Windows)
echo Installing neo-pod-desktop...
echo Installing YouTube Music to iPod Nano Transfer Tool...
REM Check if Python 3 is installed
python --version 2>NUL
@ -32,6 +32,7 @@ pip install -r requirements.txt
REM Create necessary directories
echo Creating necessary directories...
if not exist temp mkdir temp
if not exist downloads mkdir downloads
if not exist converted mkdir converted
echo.
@ -39,7 +40,8 @@ echo Installation complete!
echo.
echo To run the application:
echo 1. Activate the virtual environment: venv\Scripts\activate.bat
echo 2. Run the application: python run.py
echo 2. Run the GUI application: python run.py
echo 3. Or run the CLI application: python run.py --cli --url "https://www.youtube.com/watch?v=VIDEO_ID"
echo.
echo For more information, see the README.md file.

7
install.sh
View File

@ -1,9 +1,9 @@
#!/bin/bash
# Installation script for neo-pod-desktop
# Installation script for YouTube Music to iPod Nano Transfer Tool
set -e
echo "Installing neo-pod-desktop..."
echo "Installing YouTube Music to iPod Nano Transfer Tool..."
# Check if Python 3 is installed
if ! command -v python3 &> /dev/null; then
@ -44,6 +44,7 @@ echo "Installation complete!"
echo ""
echo "To run the application:"
echo " 1. Activate the virtual environment: source venv/bin/activate"
echo " 2. Run the application: python run.py"
echo " 2. Run the GUI application: python run.py"
echo " 3. Or run the CLI application: python run.py --cli --url \"https://www.youtube.com/watch?v=VIDEO_ID\""
echo ""
echo "For more information, see the README.md file."

4
requirements.txt
View File

@ -1,9 +1,13 @@
yt-dlp>=2023.3.4
pytube>=12.1.3
ffmpeg-python>=0.2.0
mutagen>=1.46.0
eyeD3>=0.9.7
pyusb>=1.2.1
pillow>=9.5.0
requests>=2.28.2
tqdm>=4.65.0
musicbrainzngs>=0.7.1
PyQt6>=6.5.0
numpy>=1.24.0
iopenpod>=1.0.53

26
run.py
View File

@ -5,16 +5,30 @@ Launcher script for neo-pod-desktop
import os
import sys
import argparse
def main():
"""Main entry point"""
parser = argparse.ArgumentParser(description="neo-pod-desktop")
parser.add_argument("--cli", action="store_true", help="Run in command-line mode")
args = parser.parse_args()
# Add src directory to path
src_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "src")
sys.path.insert(0, src_dir)
from src.app import main as app_main
app_main()
if args.cli:
# Run CLI version
from src.cli import YouTubeToIPodCLI
cli = YouTubeToIPodCLI()
return cli.run()
else:
# Run GUI version
from src.main import QApplication, MainWindow
app = QApplication(sys.argv)
window = MainWindow()
window.show()
return app.exec()
if __name__ == "__main__":
main()
sys.exit(main())

2
scripts/build-appimage.sh
View File

@ -98,7 +98,7 @@ build_appdir() {
--exclude-module PyQt6.QtQuick3D \
--exclude-module PyQt6.QtShaderTools \
--exclude-module PyQt6.QtSpatialAudio \
src/app.py
src/main.py
}
create_desktop_file() {

4
setup.py
View File

@ -17,7 +17,7 @@ with open('README.md', encoding='utf-8') as f:
setup(
name="neo-pod-desktop",
version="1.0.0",
description="Desktop application for converting, managing and transferring music to iPod Nano devices",
description="Desktop application for downloading, converting, managing and transferring music to iPod Nano devices",
long_description=long_description,
long_description_content_type="text/markdown",
author="neo-pod-desktop",
@ -27,7 +27,7 @@ setup(
install_requires=requirements,
entry_points={
'console_scripts': [
'neo-pod-desktop=src.app:main',
'neo-pod-desktop=src.cli:main',
],
},
classifiers=[

478
src/app.py
View File

@ -1,478 +0,0 @@
#!/usr/bin/env python3
"""
Main Application Module for neo-pod-desktop
Provides a GUI for converting, managing and transferring music to iPod Nano devices.
"""
import os
import sys
import logging
from typing import Optional
from PyQt6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout,
QSplitter, QStackedWidget,
)
from PyQt6.QtCore import Qt, QUrl
from PyQt6.QtGui import QPixmap
from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
from config_loader import ConfigLoader
from device_monitor import DeviceMonitor
from hotkeys import HotkeyManager
from playlist_manager import PlaylistManager
from ui.library_tab import LibraryTab
from ui.ipod_tab import iPodTab
from ui.settings_tab import SettingsTab
from ui.sidebar_widget import SIDEBAR_DEFAULT_WIDTH
from ui.player_header import PlayerHeader
from artwork.cache import cover_cache
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class MainWindow(QMainWindow):
"""Main application window"""
def __init__(self):
super().__init__()
self.setWindowTitle("neo-pod-desktop")
self.setMinimumSize(800, 600)
self.config_loader = ConfigLoader()
self.hotkey_manager: Optional[HotkeyManager] = None
self.device_monitor: Optional[DeviceMonitor] = None
self.ipod_tab_index = 1
self.playlist_manager = PlaylistManager()
self._sidebar_visible = True
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_player()
self._setup_ui()
self.library_tab.set_playlist_manager(self.playlist_manager)
self._load_playlists_into_sidebar()
self._load_settings()
self._setup_hotkeys()
self._wire_signals()
self.library_tab._scan_library()
self._resume_playback_if_saved()
self._restore_library_state()
self.ipod_tab._on_refresh_devices_clicked()
self._start_device_monitor()
def _setup_player(self):
self.player = QMediaPlayer()
self.audio_output = QAudioOutput()
self.audio_output.setVolume(self._saved_volume / 100.0)
self.player.setAudioOutput(self.audio_output)
self.player.positionChanged.connect(self._on_player_position)
self.player.durationChanged.connect(self._on_player_duration)
self.player.mediaStatusChanged.connect(self._on_media_status)
def _setup_ui(self):
from ui.sidebar_widget import SidebarWidget
main_widget = QWidget()
main_layout = QVBoxLayout(main_widget)
main_layout.setContentsMargins(0, 0, 0, 0)
main_layout.setSpacing(0)
self.player_header = PlayerHeader()
self.player_header.prev_requested.connect(self._on_prev)
self.player_header.play_pause_requested.connect(self._on_play_pause)
self.player_header.next_requested.connect(self._on_next)
self.player_header.position_changed_by_user.connect(self._on_user_seek_frac)
self.player_header.volume_changed.connect(self._on_volume_changed)
main_layout.addWidget(self.player_header)
self.stack = QStackedWidget()
self.library_tab = LibraryTab(self.config_loader)
self.library_tab.play_requested.connect(self._on_play_requested)
self.stack.addWidget(self.library_tab)
self.ipod_tab = iPodTab()
self.stack.addWidget(self.ipod_tab)
self.settings_tab = SettingsTab()
self.stack.addWidget(self.settings_tab)
self.stack.setCurrentIndex(0)
self.sidebar = SidebarWidget()
self.sidebar.item_selected.connect(self._on_sidebar_item_selected)
self.sidebar.playlist_selected.connect(self._on_playlist_selected)
self.splitter = QSplitter(Qt.Orientation.Horizontal)
self.splitter.addWidget(self.sidebar)
self.splitter.addWidget(self.stack)
self.splitter.setStretchFactor(0, 0)
self.splitter.setStretchFactor(1, 1)
self.splitter.setSizes([SIDEBAR_DEFAULT_WIDTH, 600])
self.splitter.setHandleWidth(1)
mid = self.palette().color(self.palette().ColorRole.Midlight)
self.splitter.setStyleSheet(f"QSplitter::handle {{ background: {mid.name()}; }}")
main_layout.addWidget(self.splitter, stretch=1)
self.statusBar().showMessage("Ready")
self.setCentralWidget(main_widget)
def _wire_signals(self):
self.ipod_tab.device_mounted.connect(self._on_device_mounted)
self.ipod_tab.device_unmounted.connect(self._on_device_unmounted)
self.settings_tab.hide_library_buttons_changed.connect(
self.library_tab.set_buttons_visible
)
self.settings_tab.auto_detect_changed.connect(self._on_auto_detect_toggled)
def _on_sidebar_item_selected(self, kind: str, value: str):
if kind == "tab":
tab_index = {"library": 0, "ipod": 1, "settings": 2}.get(value, 0)
self.stack.setCurrentIndex(tab_index)
if tab_index == 0:
self.library_tab.set_view_mode("table")
elif kind == "filter":
self.stack.setCurrentIndex(0)
self.library_tab.set_view_mode(value)
elif kind == "playlist_create":
self._create_playlist(value)
elif kind == "playlist_rename":
playlist_id, new_name = value.split("|", 1)
self._rename_playlist(playlist_id, new_name)
elif kind == "playlist_delete":
self._delete_playlist(playlist_id=value)
elif kind == "playlist_reorder":
ids = value.split("|") if value else []
self.playlist_manager.reorder(ids)
def _on_playlist_selected(self, playlist_id: str):
self.stack.setCurrentIndex(0)
playlist = self.playlist_manager.get_by_id(playlist_id)
if playlist:
self.library_tab.set_current_playlist_id(playlist_id)
self.library_tab.apply_filter("playlist", playlist.track_paths)
def _create_playlist(self, name: str):
playlist_id = self.playlist_manager.create(name)
self.sidebar.add_playlist_item(playlist_id, name)
def _rename_playlist(self, playlist_id: str, new_name: str):
self.playlist_manager.rename(playlist_id, new_name)
def _delete_playlist(self, playlist_id: str):
self.playlist_manager.delete(playlist_id)
self.sidebar.remove_playlist_item(playlist_id)
self.library_tab.clear_filter()
self.sidebar.select_library_music()
def _load_playlists_into_sidebar(self):
for pl in self.playlist_manager.get_all():
self.sidebar.add_playlist_item(pl.id, pl.name)
def _on_device_mounted(self, mount_point: str):
self.library_tab.on_device_mounted(mount_point)
self.statusBar().showMessage(f"iPod mounted at {mount_point}")
def _on_device_unmounted(self):
self.library_tab.on_device_unmounted()
self.statusBar().showMessage("iPod disconnected")
def _toggle_sidebar(self):
visible = not self.sidebar.isVisible()
self.sidebar.setVisible(visible)
if visible:
self.splitter.setSizes([SIDEBAR_DEFAULT_WIDTH, self.splitter.width() - SIDEBAR_DEFAULT_WIDTH])
else:
self.splitter.setSizes([0, self.splitter.width()])
self._sidebar_visible = visible
def _start_device_monitor(self):
auto_detect = self.config_loader.get_boolean("Device", "auto_detect", fallback=True)
self.device_monitor = DeviceMonitor()
if auto_detect:
self.ipod_tab.start_monitoring(self.device_monitor)
self.device_monitor.start()
def _on_auto_detect_toggled(self, enabled: bool):
if not self.device_monitor:
return
if enabled:
self.ipod_tab.start_monitoring(self.device_monitor)
self.device_monitor.start()
else:
self.device_monitor.stop()
self.ipod_tab.stop_monitoring(self.device_monitor)
def _load_settings(self):
config = self.config_loader
self.settings_tab.load_settings(config)
hide_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False)
self.library_tab.set_buttons_visible(not hide_buttons)
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)
self.audio_output.setVolume(self._saved_volume / 100.0)
self.player_header.set_volume(self._saved_volume)
def _save_settings(self):
config = self.config_loader
self.settings_tab.save_settings(config)
config.set("Playback", "last_volume", str(self.player_header.volume_slider.value()))
if self.current_track_path and os.path.exists(self.current_track_path):
config.set("Playback", "last_track_path", self.current_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_now).lower())
else:
config.set("Playback", "last_track_path", "")
config.set("Playback", "last_position_ms", "0")
state = self.library_tab.get_state()
config.set("Library", "view_mode", state.get("view_mode", "table"))
config.set("Library", "filtered_album", state.get("filtered_album", ""))
config.set("Library", "filtered_playlist_id", state.get("filtered_playlist_id", ""))
config.save()
@property
def current_track_path(self) -> str:
if self.player.source().isValid():
return self.player.source().toLocalFile()
return ""
@property
def _is_playing_now(self) -> bool:
return self.player.playbackState() == QMediaPlayer.PlaybackState.PlayingState
def _setup_hotkeys(self):
self.hotkey_manager = HotkeyManager(self, self.config_loader)
self.hotkey_manager.register_all({
"play_pause": self._on_play_pause,
"prev_track": self._on_prev,
"next_track": self._on_next,
"volume_up": lambda: self._adjust_volume(5),
"volume_down": lambda: self._adjust_volume(-5),
"seek_forward": self._seek_forward,
"seek_backward": self._seek_backward,
"toggle_sidebar": self._toggle_sidebar,
"tab_library": lambda: self.stack.setCurrentIndex(0),
"tab_ipod": lambda: self.stack.setCurrentIndex(self.ipod_tab_index),
"tab_settings": lambda: self.stack.setCurrentIndex(2),
"search_focus": self.library_tab._focus_search,
"select_all": self.library_tab._select_all_current,
"library_refresh": self.library_tab._on_refresh_library_clicked,
"ipod_refresh": self.ipod_tab._on_refresh_devices_clicked,
"delete_selected": self.library_tab._delete_selected_current,
"edit_metadata": self.library_tab._on_edit_metadata,
"library_add_files": self.library_tab._on_library_add_files,
})
self.settings_tab.set_hotkey_manager(self.hotkey_manager)
self.settings_tab.refresh_shortcuts_table()
# ------------------------------------------------------------------
# Playback
# ------------------------------------------------------------------
def _on_play_requested(self, track_data: dict, start_position_ms: int, auto_play: bool):
path = track_data.get("path", "")
if not path or not os.path.exists(path):
return
self._pending_seek_ms = start_position_ms
self.player.setSource(QUrl.fromLocalFile(path))
if auto_play:
self.player.play()
self.player_header.set_playing(True)
else:
self.player_header.set_playing(False)
self.player_header.set_track_info(
f"{track_data['artist']} \u2014 {track_data['title']}"
)
cover_path = track_data.get("cover_path")
if cover_path and os.path.exists(cover_path):
self.player_header.set_cover(QPixmap(cover_path), track_data.get("album", ""))
else:
artist = track_data.get("artist", "")
album = track_data.get("album", "")
cached = cover_cache.get(artist, album) if artist else None
if cached and os.path.exists(cached):
self.player_header.set_cover(QPixmap(cached), album)
else:
self.player_header.set_cover()
def _on_play_pause(self):
if self.player.playbackState() == QMediaPlayer.PlaybackState.PlayingState:
self.player.pause()
self.player_header.set_playing(False)
self.library_tab.on_playing_changed(False)
elif self.player.playbackState() == QMediaPlayer.PlaybackState.StoppedState:
self.library_tab.play_track_at_index(
max(0, self.library_tab.current_playback_index)
)
else:
self.player.play()
self.player_header.set_playing(True)
self.library_tab.on_playing_changed(True)
def _on_prev(self):
if self.player.position() > 3000:
self.player.setPosition(0)
return
idx = max(0, self.library_tab.current_playback_index - 1)
self.library_tab.play_track_at_index(idx)
def _on_next(self):
rows = self.library_tab.library_table.rowCount()
idx = min(rows - 1, self.library_tab.current_playback_index + 1)
self.library_tab.play_track_at_index(idx)
def _seek_forward(self):
if 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.duration() <= 0:
return
new_pos = max(0, self.player.position() - 5000)
self.player.setPosition(new_pos)
def _adjust_volume(self, delta: int):
slider = self.player_header.volume_slider
new_val = max(0, min(100, slider.value() + delta))
slider.setValue(new_val)
def _on_volume_changed(self, value: int):
if self.audio_output:
self.audio_output.setVolume(value / 100.0)
def _on_user_seek_frac(self, frac: int):
self.player.setPosition(int(frac * self.player.duration()))
def _on_player_position(self, position: int):
duration = self.player.duration()
self.player_header.set_position(position, duration)
self.library_tab.on_position_changed(position, duration)
def _on_player_duration(self, duration: int):
pos = self.player.position()
self.player_header.set_position(pos, duration)
def _on_media_status(self, status):
if status == QMediaPlayer.MediaStatus.EndOfMedia:
self.library_tab.on_track_ended()
self._on_next()
elif status == QMediaPlayer.MediaStatus.NoMedia:
self.library_tab.on_playing_changed(False)
self.player_header.reset_info()
elif status == QMediaPlayer.MediaStatus.LoadedMedia:
if self._pending_seek_ms > 0:
self.player.setPosition(self._pending_seek_ms)
self._pending_seek_ms = 0
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_tab.library_table.rowCount()):
item = self.library_tab.library_table.item(row, 0)
if item is None:
continue
_is_ready, data = item.data(Qt.ItemDataRole.UserRole)
if data and data.get("path", "") == self._saved_track_path:
self.library_tab.play_track_at_index(
row, start_position_ms=self._saved_position_ms,
auto_play=self._saved_playing
)
return
def _restore_library_state(self):
view_mode = self.config_loader.get("Library", "view_mode", fallback="table")
filtered_album = self.config_loader.get("Library", "filtered_album", fallback="")
filtered_playlist_id = self.config_loader.get("Library", "filtered_playlist_id", fallback="")
if filtered_album:
self.library_tab.filter_by_album(filtered_album)
elif filtered_playlist_id:
pl = self.playlist_manager.get_by_id(filtered_playlist_id)
if pl:
self.library_tab.set_current_playlist_id(filtered_playlist_id)
self.library_tab.apply_filter("playlist", pl.track_paths)
else:
self.library_tab.set_view_mode(view_mode)
self.sidebar.blockSignals(True)
if filtered_playlist_id:
self.sidebar.select_playlist_silent(filtered_playlist_id)
else:
self.sidebar.select_by_view_mode(view_mode)
self.sidebar.blockSignals(False)
def closeEvent(self, event):
if self.device_monitor:
self.device_monitor.stop()
self._save_settings()
event.accept()
# GTK3 platform theme paths for Linux desktop integration
_GTK3_PLUGIN_PATHS = [
"/usr/lib/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib/x86_64-linux-gnu/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib64/qt6/plugins/platformthemes/libqgtk3.so",
"/usr/lib/aarch64-linux-gnu/qt6/plugins/platformthemes/libqgtk3.so",
]
def _setup_linux_theming():
if sys.platform != "linux":
return
if "QT_QPA_PLATFORMTHEME" in os.environ:
return
for path in _GTK3_PLUGIN_PATHS:
if os.path.exists(path):
os.environ["QT_QPA_PLATFORMTHEME"] = "gtk3"
logger.debug(f"Using GTK3 platform theme: {path}")
return
if os.path.exists("/usr/bin/qt6ct"):
os.environ["QT_QPA_PLATFORMTHEME"] = "qt6ct"
logger.debug("Using qt6ct platform theme")
return
logger.debug("No GTK platform theme plugin found, using Qt default style")
def main():
_setup_linux_theming()
app = QApplication(sys.argv)
if not os.environ.get("QT_QPA_PLATFORMTHEME"):
app.setStyle("Fusion")
window = MainWindow()
window.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()

2
src/audio_converter.py
View File

@ -10,7 +10,7 @@ import subprocess
from typing import Optional, Dict, Any
from pathlib import Path
from track_info import TrackInfo
from youtube_downloader import TrackInfo
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

337
src/cli.py Normal file
View File

@ -0,0 +1,337 @@
#!/usr/bin/env python3
"""
Command Line Interface for neo-pod-desktop
Provides a CLI for downloading, converting, managing and transferring music to iPod Nano devices
"""
import os
import sys
import logging
import argparse
import time
from typing import List, Dict, Optional, Tuple
from pathlib import Path
from tqdm import tqdm
from youtube_downloader import YouTubeDownloader, TrackInfo
from audio_converter import AudioConverter
from metadata_handler import MetadataHandler
from ipod_device import IPodDevice
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class YouTubeToIPodCLI:
"""Command Line Interface for neo-pod-desktop"""
def __init__(self):
"""Initialize the CLI"""
self.downloader = None
self.converter = None
self.metadata_handler = None
self.ipod_device = None
self.downloaded_tracks = []
self.converted_tracks = []
self.temp_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "temp")
os.makedirs(self.temp_dir, exist_ok=True)
def parse_arguments(self):
"""Parse command line arguments"""
parser = argparse.ArgumentParser(
description="neo-pod-desktop",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
# URL argument
parser.add_argument(
"--url", "-u",
help="YouTube video or playlist URL",
required=True
)
# Output directory
parser.add_argument(
"--output-dir", "-o",
help="Output directory (iPod mount point)",
default=os.path.join(os.path.expanduser("~"), "Music", "iPod")
)
# Format
parser.add_argument(
"--format", "-f",
help="Audio format",
choices=["m4a", "mp3"],
default="m4a"
)
# Quality
parser.add_argument(
"--quality", "-q",
help="Audio quality in kbps",
type=int,
choices=[128, 192, 256, 320],
default=256
)
# Video support
parser.add_argument(
"--video",
help="Enable video download (for compatible iPod models)",
action="store_true"
)
# Video resolution
parser.add_argument(
"--resolution", "-r",
help="Video resolution (for video download)",
choices=["640x480", "480x360", "320x240"],
default="640x480"
)
# iPod device
parser.add_argument(
"--device", "-d",
help="iPod device mount point (if not specified, auto-detection will be attempted)",
default=None
)
# Skip steps
parser.add_argument(
"--skip-download",
help="Skip download step (use previously downloaded files)",
action="store_true"
)
parser.add_argument(
"--skip-convert",
help="Skip conversion step (use previously converted files)",
action="store_true"
)
parser.add_argument(
"--skip-transfer",
help="Skip transfer step (only download and convert)",
action="store_true"
)
# Clean temp files
parser.add_argument(
"--clean-temp",
help="Clean temporary files after transfer",
action="store_true"
)
return parser.parse_args()
def download(self, url: str, format: str = "m4a", quality: int = 256, video: bool = False) -> List[TrackInfo]:
"""
Download audio/video from YouTube
Args:
url: YouTube URL
format: Audio format
quality: Audio quality in kbps
video: Whether to download video
Returns:
List of TrackInfo objects
"""
logger.info(f"Downloading from URL: {url}")
# Initialize downloader
self.downloader = YouTubeDownloader(output_dir=self.temp_dir)
# Process URL
tracks = self.downloader.process_url(url, format, quality)
logger.info(f"Downloaded {len(tracks)} tracks")
return tracks
def convert(self,
tracks: List[TrackInfo],
output_dir: str,
format: str = "mp3",
quality: int = 256,
video: bool = False,
resolution: str = "640x480") -> List[Tuple[TrackInfo, str]]:
"""
Convert downloaded tracks to iPod-compatible format
Args:
tracks: List of TrackInfo objects
output_dir: Output directory
format: Audio format
quality: Audio quality in kbps
video: Whether to convert video
resolution: Video resolution
Returns:
List of (TrackInfo, output_path) tuples
"""
logger.info(f"Converting {len(tracks)} tracks to iPod-compatible format")
# Initialize converter and metadata handler
self.converter = AudioConverter(output_dir=output_dir)
self.metadata_handler = MetadataHandler(temp_dir=self.temp_dir)
# Convert each track
converted_tracks = []
for track in tqdm(tracks, desc="Converting", unit="track"):
if track.download_path and os.path.exists(track.download_path):
# Convert to iPod format
output_path = self.converter.convert_to_ipod_format(
track, track.download_path, format, quality
)
# Embed metadata
if output_path:
self.metadata_handler.process_file(output_path, track)
converted_tracks.append((track, output_path))
logger.info(f"Converted {len(converted_tracks)} tracks")
return converted_tracks
def transfer(self,
tracks: List[Tuple[TrackInfo, str]],
mount_point: Optional[str] = None) -> int:
"""
Transfer tracks to iPod
Args:
tracks: List of (TrackInfo, path) tuples
mount_point: iPod mount point (optional)
Returns:
Number of tracks transferred
"""
logger.info("Transferring tracks to iPod")
# Initialize iPod device
self.ipod_device = IPodDevice(mount_point=mount_point)
# If no mount point specified, try to detect and mount
if not mount_point:
logger.info("No mount point specified, attempting to detect iPod")
devices = self.ipod_device.detect_devices()
if not devices:
logger.error("No iPod devices found")
return 0
logger.info(f"Found iPod: {devices[0]['name']} ({devices[0]['model']})")
mount_point = self.ipod_device.mount_device(devices[0]['id'])
if not mount_point:
logger.error(f"Failed to mount iPod: {devices[0]['name']}")
return 0
logger.info(f"Mounted iPod at: {mount_point}")
# Get device info
device_info = self.ipod_device.get_device_info()
free_space = device_info.get("free_space", 0)
# Calculate total size of tracks
total_size = 0
for _, path in tracks:
if os.path.exists(path):
total_size += os.path.getsize(path)
# Check if there's enough space
if total_size > free_space:
logger.error(
f"Not enough space on iPod. Need {total_size / 1024**2:.1f} MB, " +
f"but only {free_space / 1024**2:.1f} MB available"
)
return 0
# Transfer each track
transferred_count = 0
for track, path in tqdm(tracks, desc="Transferring", unit="track"):
if os.path.exists(path):
# Transfer file
success = self.ipod_device.transfer_file(path)
if success:
transferred_count += 1
logger.info(f"Transferred {transferred_count} tracks to iPod")
return transferred_count
def clean_temp_files(self):
"""Clean temporary files"""
logger.info("Cleaning temporary files")
# Remove all files in temp directory
for file in os.listdir(self.temp_dir):
file_path = os.path.join(self.temp_dir, file)
try:
if os.path.isfile(file_path):
os.unlink(file_path)
except Exception as e:
logger.error(f"Error deleting {file_path}: {e}")
def run(self):
"""Run the CLI"""
# Parse arguments
args = self.parse_arguments()
try:
# Download step
if not args.skip_download:
self.downloaded_tracks = self.download(
args.url,
args.format,
args.quality,
args.video
)
else:
logger.info("Skipping download step")
# TODO: Load previously downloaded tracks
# Convert step
if not args.skip_convert:
self.converted_tracks = self.convert(
self.downloaded_tracks,
args.output_dir,
args.format,
args.quality,
args.video,
args.resolution
)
else:
logger.info("Skipping conversion step")
# TODO: Load previously converted tracks
# Transfer step
if not args.skip_transfer:
transferred_count = self.transfer(
self.converted_tracks,
args.device
)
if transferred_count > 0:
logger.info(f"Successfully transferred {transferred_count} tracks to iPod")
else:
logger.error("Failed to transfer tracks to iPod")
else:
logger.info("Skipping transfer step")
# Clean temp files
if args.clean_temp:
self.clean_temp_files()
return 0
except Exception as e:
logger.exception(f"Error: {e}")
return 1
if __name__ == "__main__":
cli = YouTubeToIPodCLI()
sys.exit(cli.run())

19
src/config_loader.py
View File

@ -116,6 +116,8 @@ class ConfigLoader:
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'):
@ -132,10 +134,14 @@ class ConfigLoader:
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', 'hide_library_buttons'):
self.config.set('Advanced', 'hide_library_buttons', 'false')
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'):
@ -149,15 +155,16 @@ class ConfigLoader:
'volume_down': 'Ctrl+Down',
'seek_forward': 'Right',
'seek_backward': 'Left',
'tab_library': 'Ctrl+1',
'tab_ipod': 'Ctrl+2',
'tab_settings': 'Ctrl+3',
'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',
'library_add_files': 'Ctrl+O',
}
for key, value in hotkey_defaults.items():
if not self.config.has_option('Hotkeys', key):

57
src/device_monitor.py
View File

@ -1,57 +0,0 @@
import logging
from typing import Set
from PyQt6.QtCore import QObject, QTimer, QThread, pyqtSignal
from ipod_device import IPodDevice
logger = logging.getLogger(__name__)
class _DetectWorker(QThread):
finished = pyqtSignal(object)
def run(self):
try:
devices = IPodDevice().detect_devices()
self.finished.emit(devices)
except Exception:
logger.exception("Device detection failed")
self.finished.emit([])
class DeviceMonitor(QObject):
state_changed = pyqtSignal(list)
def __init__(self, parent=None):
super().__init__(parent)
self.timer = QTimer(self)
self.timer.timeout.connect(self._poll)
self._device_ids: Set[str] = set()
self._poll_worker = None
def start(self, interval_ms: int = 3000):
self._device_ids.clear()
self._poll()
self.timer.start(interval_ms)
def stop(self):
self.timer.stop()
if self._poll_worker and self._poll_worker.isRunning():
self._poll_worker.quit()
self._poll_worker.wait(3000)
self._device_ids.clear()
def _poll(self):
if self._poll_worker and self._poll_worker.isRunning():
return
self._poll_worker = _DetectWorker()
self._poll_worker.finished.connect(self._on_devices_detected)
self._poll_worker.start()
def _on_devices_detected(self, devices):
current_ids = {d["id"] for d in devices}
if current_ids != self._device_ids:
self._device_ids = current_ids
self.state_changed.emit(devices)

10
src/hotkeys.py
View File

@ -84,17 +84,17 @@ DEFAULTS = [
("volume_down", "Ctrl+Down", "Volume Down", "global"),
("seek_forward", "Right", "Seek Forward 5s", "global"),
("seek_backward", "Left", "Seek Backward 5s", "global"),
("tab_library", "Ctrl+1", "Switch to Library", "global"),
("tab_ipod", "Ctrl+2", "Switch to iPod", "global"),
("tab_settings", "Ctrl+3", "Switch to Settings", "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"),
("toggle_sidebar", "Ctrl+B", "Toggle Sidebar", "global"),
("library_add_files", "Ctrl+O", "Library: Add Files", "global"),
]

18
src/ipod_device.py
View File

@ -294,24 +294,6 @@ class IPodDevice:
self.logger.error(f"Device {device_id} does not exist")
return None
# Check if already mounted via findmnt (avoids udisksctl AlreadyMounted error)
try:
fm_result = subprocess.run(
["findmnt", "-n", "-o", "TARGET", device_id],
capture_output=True, check=False, text=True
)
if fm_result.returncode == 0 and fm_result.stdout.strip():
mount_point = fm_result.stdout.strip()
if os.path.exists(mount_point):
self.mount_point = mount_point
self._block_device = device_id
name = device_id[5:]
self._parent_drive = f"/dev/{name.rstrip('0123456789')}"
self.logger.info(f"Device already mounted at: {mount_point}")
return mount_point
except Exception:
self.logger.warning("findmnt check failed, falling through to udisksctl")
try:
result = subprocess.run(
["udisksctl", "mount", "-b", device_id],

332
src/ipod_nano7_db.py
View File

@ -132,27 +132,12 @@ MASTER_CONTAINER_PID = 203092939621887772
MUSIC_CONTAINER_PID = 4872745547565090077
_REMOVE_ARTWORK = object()
"""Sentinel: used in artwork_overrides to remove artwork from a track."""
# ── Helpers ──────────────────────────────────────────────────────────
def _to_mac_epoch(ts: int) -> int:
return ts + MAC_EPOCH_OFFSET if ts > 0 else 0
def content_hash(filepath: str, max_bytes: int = 65536) -> str:
"""SHA1 of first max_bytes of audio data (fast content fingerprint)."""
h = hashlib.sha1()
try:
with open(filepath, "rb") as f:
h.update(f.read(max_bytes))
except OSError:
return ""
return h.hexdigest()
def _read_audio_tags(filepath: str) -> dict[str, Any]:
"""Extract title, artist, album, duration, etc. from an audio file."""
info: dict[str, Any] = {}
@ -329,11 +314,6 @@ def _scanned_tracks_to_iop(mountpoint: str,
sample_count=sample_count,
gapless_track_flag=1,
gapless_album_flag=1 if has_album else 0,
play_count=s.get("play_count", 0),
rating=s.get("rating", 0),
last_played=s.get("last_played", 0),
skip_count=s.get("skip_count", 0),
played_mark=-1,
)
tracks.append(t)
return tracks
@ -394,7 +374,6 @@ class Nano7Database:
tags["file_name"] = fname
tags["file_size"] = os.path.getsize(fpath)
tags["pid"] = self._hash_to_pid(fpath)
tags["content_hash"] = content_hash(fpath)
results.append(tags)
return results
@ -474,18 +453,10 @@ class Nano7Database:
return len(self._scan_ipod_files())
def get_all_tracks(self) -> list[dict]:
"""Return all tracks on iPod with full metadata including play stats."""
"""Return all tracks on iPod with full metadata."""
scanned = self._scan_ipod_files()
base_stats, delta_stats = self.read_play_stats()
tracks = []
for s in scanned:
rel = os.path.relpath(s["file_path"], self.music_base)
location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
bs = base_stats.get(location, {})
dd = delta_stats.get(location, {})
pc = bs.get("play_count", 0) + dd.get("play_count", 0)
sc = bs.get("skip_count", 0) + dd.get("skip_count", 0)
rating = dd.get("rating", 0) or bs.get("rating", 0)
tracks.append({
"pid": s["pid"],
"title": s.get("title") or os.path.splitext(s["file_name"])[0],
@ -500,11 +471,6 @@ class Nano7Database:
"year": s.get("year", 0),
"ipod_path": None,
"file_path": s["file_path"],
"content_hash": s.get("content_hash", ""),
"play_count": pc,
"rating": rating,
"last_played": bs.get("last_played", 0),
"skip_count": sc,
})
return tracks
@ -635,283 +601,16 @@ class Nano7Database:
except Exception as e:
logger.exception("iTunesDB sync failed: %s", e)
def _merge_play_stats(self, scanned: list[dict]) -> None:
"""Merge iPod + local play stats into *scanned* track dicts in-place.
Uses additive model (like iTunes):
``new_value = max(ipod_base, local_count) + ipod_delta``
"""
base_stats, delta_stats = self.read_play_stats()
local_play_stats = self._load_local_play_stats()
for s in scanned:
rel = os.path.relpath(s["file_path"], self.music_base)
location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
bs = base_stats.get(location, {})
dd = delta_stats.get(location, {})
ch = s.get("content_hash", "")
local_pc = local_play_stats.get(ch, {})
s["play_count"] = max(
bs.get("play_count", 0), local_pc.get("play_count", 0),
) + dd.get("play_count", 0)
s["skip_count"] = max(
bs.get("skip_count", 0), local_pc.get("skip_count", 0),
) + dd.get("skip_count", 0)
s["rating"] = dd.get("rating", 0) or bs.get("rating", 0) or local_pc.get("rating", 0)
s["last_played"] = max(
bs.get("last_played", 0), local_pc.get("last_played", 0),
)
def _cleanup_play_counts_file(self) -> None:
"""Delete the iPod 'Play Counts' delta file after a full sync."""
pc_file = os.path.join(
self.mountpoint, "iPod_Control", "iTunes", "Play Counts",
)
if os.path.exists(pc_file):
try:
os.remove(pc_file)
logger.debug("Deleted Play Counts file after sync")
except OSError as e:
logger.debug("Could not delete Play Counts file: %s", e)
def consume_play_deltas(self) -> None:
"""Public entry point: delete the Play Counts file after consuming
its data during mount. Prevents double-counting on subsequent
``read_play_stats()`` calls."""
self._cleanup_play_counts_file()
def _sync_itunescdb_impl(self) -> None:
# 1. Scan all audio files on iPod
scanned = self._scan_ipod_files()
if not scanned:
logger.warning("No audio files found on iPod — writing empty databases")
# Write minimal empty databases
self._write_empty_databases()
return
self._merge_play_stats(scanned)
self._write_databases_from_tracks(scanned)
self._cleanup_play_counts_file()
def find_track_by_content_hash(self, content_hash: str) -> dict | None:
"""Find an iPod track by its content hash."""
if not content_hash:
return None
for t in self.get_all_tracks():
if t.get("content_hash") == content_hash:
return t
return None
def find_track_by_metadata(
self, title: str, artist: str, album: str = "",
) -> dict | None:
"""Find an iPod track by title+artist+album (case-insensitive, fallback matching)."""
for t in self.get_all_tracks():
if (t["title"].lower() == title.lower() and
t["artist"].lower() == artist.lower() and
(t.get("album") or "").lower() == album.lower()):
return t
return None
def read_play_stats(self) -> tuple[dict[str, dict[str, int]], dict[str, dict[str, int]]]:
"""Read iPod play stats split into base (written) and delta (since last write).
Returns ``(base_stats, delta_stats)`` keyed by iPod ``location``.
**base_stats** — values from the binary iTunesDB / Dynamic.itdb
(the last written totals). Fields: ``play_count``, ``rating``,
``last_played``, ``skip_count``.
**delta_stats** — deltas from the ``Play Counts`` file accumulated
since the last database write. Fields: ``play_count``, ``skip_count``,
``rating`` (0 if no delta activity).
Both return empty dicts when the databases cannot be read (new or
corrupted iPod state).
"""
ipod_device = _import_iop("ipod_device")
try:
itdb_path = ipod_device.resolve_itdb_path(self.mountpoint)
except Exception:
itdb_path = os.path.join(
self.mountpoint, "iPod_Control", "iTunes", "iTunesDB",
)
base_stats: dict[str, dict[str, int]] = {}
delta_stats: dict[str, dict[str, int]] = {}
try:
ipl = _import_iop("iTunesDB_Parser.ipod_library")
data = ipl.load_ipod_library(itdb_path, merge_playcounts=False)
except Exception as e:
logger.debug("Cannot read iPod play base stats: %s", e)
return base_stats, delta_stats
if not data:
return base_stats, delta_stats
tracks = data.get("mhlt", [])
for track in tracks:
loc = track.get("location", "")
if not loc:
continue
base_stats[loc] = {
"play_count": track.get("play_count_1", 0) or 0,
"rating": track.get("rating", 0) or 0,
"last_played": track.get("last_played", 0) or 0,
"skip_count": track.get("skip_count", 0) or 0,
}
try:
pc_mod = _import_iop("iTunesDB_Parser.playcounts")
pc_dir = os.path.dirname(itdb_path)
pc_path = os.path.join(pc_dir, "Play Counts")
entries = pc_mod.parse_playcounts(pc_path)
if entries:
for i, entry in enumerate(entries):
if i >= len(tracks):
break
loc = tracks[i].get("location", "")
if not loc:
continue
delta_stats[loc] = {
"play_count": entry.play_count,
"skip_count": entry.skip_count,
"rating": entry.rating,
}
except Exception as e:
logger.debug("Cannot read iPod Play Counts deltas: %s", e)
logger.debug(
"Read play stats — base: %d tracks, delta: %d tracks",
len(base_stats), len(delta_stats),
)
return base_stats, delta_stats
@staticmethod
def _load_local_play_stats() -> dict[str, dict[str, int]]:
"""Read accumulated play stats from the local library cache.
Returns ``{content_hash: {play_count, rating, last_played,
skip_count}}`` for every cached entry that has play data.
"""
from library_cache import get_library_cache
cache = get_library_cache()
result: dict[str, dict[str, int]] = {}
for entry in cache._data.values():
ch = entry.get("_content_hash", "")
pc = entry.get("play_count", 0)
rating = entry.get("rating", 0)
lp = entry.get("last_played", 0)
sc = entry.get("skip_count", 0)
if pc or rating or lp or sc:
if ch:
result[ch] = {
"play_count": pc,
"rating": rating,
"last_played": lp,
"skip_count": sc,
}
return result
def update_track_metadata(
self,
pid: int,
new_metadata: dict,
local_source_path: str | None = None,
) -> bool:
"""Update metadata for a single iPod track without touching audio files.
Only updates the databases (iTunesDB + SQLite + ArtworkDB).
The audio file on the iPod is never modified — new metadata is
written directly into the database records.
Args:
pid: The iPod track pid (from get_all_tracks).
new_metadata: Dict with optional keys: title, artist, album,
album_artist, genre, composer, year, track_number,
disc_number, comment.
local_source_path: If provided, artwork is extracted from this
local file and encoded into iPod-native formats.
Returns:
True if the track was found and updated, False otherwise.
"""
scanned = self._scan_ipod_files()
if not scanned:
logger.warning("No tracks on iPod to update")
return False
target_idx = None
for idx, s in enumerate(scanned):
if s["pid"] == pid:
target_idx = idx
break
if target_idx is None:
logger.warning("Track with pid=%d not found on iPod", pid)
return False
target = scanned[target_idx]
text_fields = {
"title": "title",
"artist": "artist",
"album": "album",
"album_artist": "album_artist",
"genre": "genre",
"composer": "composer",
"comment": "comment",
}
for meta_key, scan_key in text_fields.items():
if meta_key in new_metadata:
target[scan_key] = str(new_metadata[meta_key]) if new_metadata[meta_key] else ""
if "year" in new_metadata:
target["year"] = int(new_metadata["year"] or 0)
if "track_number" in new_metadata:
target["track_number"] = int(new_metadata["track_number"] or 0)
if "disc_number" in new_metadata:
target["disc_number"] = int(new_metadata["disc_number"] or 0)
artwork_overrides: dict[str, object] = {}
if local_source_path and os.path.exists(local_source_path):
try:
art_result = prepare_artwork_for_track(local_source_path)
if art_result is not None:
artwork_overrides[target["file_path"]] = art_result
else:
artwork_overrides[target["file_path"]] = _REMOVE_ARTWORK
except Exception as e:
logger.warning("Failed to extract artwork from %s: %s",
local_source_path, e)
self._merge_play_stats(scanned)
self._write_databases_from_tracks(scanned, artwork_overrides=artwork_overrides)
self._cleanup_play_counts_file()
logger.info("Updated metadata for track: %s", target.get("title", "?"))
return True
def _write_databases_from_tracks(
self,
scanned: list[dict],
artwork_overrides: dict[str, object] | None = None,
) -> None:
"""Build artwork, convert to TrackInfo, write iTunesDB + SQLite.
Args:
scanned: List of track dicts from _scan_ipod_files().
artwork_overrides: Optional dict mapping iPod file_path →
(art_hash, {fmt_id: EncodedFormatPayload}) for tracks whose
artwork should come from a local file instead of the iPod file.
Use the ``_REMOVE_ARTWORK`` sentinel to explicitly remove
artwork for a track. Omit a key to keep existing iPod artwork.
"""
artwork_overrides = artwork_overrides or {}
# 1. Prepare artwork for all tracks
# 2. Prepare artwork for all tracks
artwork_map: dict[str, tuple[int, int, int]] = {}
art_entries: list[ArtworkEntry] = []
art_hash_to_img_id: dict[str, int] = {}
@ -919,16 +618,10 @@ class Nano7Database:
for idx, s in enumerate(scanned):
fpath = s["file_path"]
override = artwork_overrides.get(fpath)
if override is None:
try:
art_result = prepare_artwork_for_track(fpath)
except Exception:
art_result = None
elif override is _REMOVE_ARTWORK:
try:
art_result = prepare_artwork_for_track(fpath)
except Exception:
art_result = None
else:
art_result = override
if art_result is not None:
art_hash, formats = art_result
art_count = len(formats)
@ -964,11 +657,11 @@ class Nano7Database:
except Exception as e:
logger.warning("ArtworkDB write failed: %s", e)
# 2. Convert to iOpenPod TrackInfo
# 3. Convert to iOpenPod TrackInfo
tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map)
logger.info("Built %d iOpenPod TrackInfo objects", len(tracks))
# 3. Determine capabilities for this device
# 4. Determine capabilities for this device
ipod_device = _import_iop("ipod_device")
capabilities = None
try:
@ -982,7 +675,7 @@ class Nano7Database:
except Exception as e:
logger.debug("Capabilities detection failed: %s", e)
# 4. Generate and write binary iTunesDB first (to get db_pid)
# 5. Generate and write binary iTunesDB first (to get db_pid)
try:
firewire_id = ipod_device.get_firewire_id(self.mountpoint)
except Exception:
@ -1002,7 +695,7 @@ class Nano7Database:
return
logger.info("iTunesDB written (%d tracks)", len(tracks))
# 5. Extract db_pid from the binary DB for SQLite cross-reference
# 6. Extract db_pid from the binary DB for SQLite cross-reference
db_pid = 0
try:
ipod_dev = _import_iop("ipod_device")
@ -1016,7 +709,7 @@ class Nano7Database:
except Exception as e:
logger.warning("Could not extract db_pid: %s", e)
# 6. Write SQLite databases (Nano 6G/7G require this)
# 7. Write SQLite databases (Nano 6G/7G require this)
try:
write_sqlite_databases = _import_iop("SQLiteDB_Writer").write_sqlite_databases
sqlite_ok = write_sqlite_databases(
@ -1061,4 +754,3 @@ class Nano7Database:
db_pid=random.getrandbits(64),
capabilities=capabilities, firewire_id=fwid, backup=True,
)
self._cleanup_play_counts_file()

26
src/library_cache.py
View File

@ -5,7 +5,6 @@ Caches extracted tag data per file path with mtime-based invalidation.
Stored as JSON in XDG cache directory.
"""
import hashlib
import json
import logging
import os
@ -18,23 +17,11 @@ logger = logging.getLogger(__name__)
_CACHE_FILE = "library_cache.json"
def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
"""SHA1 of first max_bytes of audio data (fast content fingerprint)."""
h = hashlib.sha1()
try:
with open(filepath, "rb") as f:
h.update(f.read(max_bytes))
except OSError:
return ""
return h.hexdigest()
class LibraryCache:
"""JSON-based cache of audio file metadata.
Keys are absolute file paths. Values are dicts of metadata
plus internal ``_mtime`` and ``_content_hash`` fields for
change detection and track matching.
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.
@ -77,15 +64,7 @@ class LibraryCache:
actual_mtime = os.path.getmtime(file_path)
if actual_mtime != entry.get("_mtime", 0):
return None
result = {}
for k, v in entry.items():
if k == "_mtime":
continue
if k == "_content_hash":
result["content_hash"] = v
else:
result[k] = v
return result
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)
@ -93,7 +72,6 @@ class LibraryCache:
entry["_mtime"] = os.path.getmtime(file_path)
except OSError:
return
entry["_content_hash"] = _content_hash(file_path)
self._data[file_path] = entry
def remove(self, file_path: str) -> None:

2170
src/main.py Normal file
View File

File diff suppressed because it is too large Load Diff

86
src/ui/metadata_editor.py → src/metadata_editor.py
View File

@ -25,12 +25,11 @@ from PyQt6.QtWidgets import (
QPushButton, QLabel, QLineEdit, QSpinBox, QCheckBox,
QFileDialog, QMessageBox, QFrame, QDialogButtonBox,
)
from PyQt6.QtCore import Qt, QSize, QTimer
from PyQt6.QtCore import Qt, QSize
from PyQt6.QtGui import QPixmap, QDragEnterEvent, QDropEvent
from metadata_handler import MetadataHandler
from artwork.cache import cover_cache
from services.itunes_search import search_itunes
COVER_SIZE = 250
@ -344,19 +343,6 @@ class MetadataEditorDialog(QDialog):
self._artist_label = QLabel("Artist:")
form_layout.addRow(self._artist_label, self.artist_edit)
self._lookup_btn = QPushButton("\U0001F50D Lookup in iTunes")
self._lookup_btn.setToolTip("Search iTunes Store for metadata by artist and title")
self._lookup_btn.clicked.connect(self._on_lookup_clicked)
self._lookup_btn.setVisible(not self._batch_mode)
self._lookup_status = QLabel("")
self._lookup_status.setStyleSheet("color: #888; font-size: 11px; font-style: italic;")
self._lookup_status.setVisible(False)
lookup_row = QHBoxLayout()
lookup_row.addWidget(self._lookup_btn)
lookup_row.addWidget(self._lookup_status)
lookup_row.addStretch()
form_layout.addRow(QLabel(""), lookup_row)
self._album_label = QLabel("Album:")
form_layout.addRow(self._album_label, self.album_edit)
@ -423,9 +409,6 @@ class MetadataEditorDialog(QDialog):
self.cover_apply_check.setVisible(True)
self._lookup_btn.setVisible(False)
self._lookup_status.setVisible(False)
for key in self._mixed_fields:
self._set_mixed(key)
@ -534,73 +517,6 @@ class MetadataEditorDialog(QDialog):
self._cover_removed = True
self._update_cover_display()
def _on_lookup_clicked(self):
artist = self.artist_edit.text().strip()
title = self.title_edit.text().strip()
if not artist and not title:
self._show_lookup_status("Enter artist or title first", is_error=True)
return
self._lookup_btn.setEnabled(False)
self._show_lookup_status("Searching iTunes\u2026")
result = search_itunes(artist=artist, title=title)
self._lookup_btn.setEnabled(True)
if not result:
self._show_lookup_status("No results found", is_error=True)
return
self.title_edit.setText(result.get("title", ""))
self.artist_edit.setText(result.get("artist", ""))
self.album_edit.setText(result.get("album", ""))
self.album_artist_edit.setText(result.get("album_artist", ""))
self.genre_edit.setText(result.get("genre", ""))
year = result.get("year", 0)
if year and year > 0:
self.year_spin.setValue(year)
track_num = result.get("track_number", 0)
if track_num:
self.track_num_spin.setValue(track_num)
track_total = result.get("track_total", 0)
if track_total:
self.track_total_spin.setValue(track_total)
disc_num = result.get("disc_number", 0)
if disc_num:
self.disc_num_spin.setValue(disc_num)
disc_total = result.get("disc_total", 0)
if disc_total:
self.disc_total_spin.setValue(disc_total)
cover_url = result.get("cover_url", "")
if cover_url and not self._existing_cover_data and not self._new_cover_path:
try:
import tempfile
import requests
resp = requests.get(cover_url, timeout=10)
resp.raise_for_status()
fd, tmp_path = tempfile.mkstemp(suffix=".jpg")
os.close(fd)
with open(tmp_path, "wb") as f:
f.write(resp.content)
self._new_cover_path = tmp_path
self._update_cover_display()
except Exception:
pass
self._show_lookup_status("Metadata loaded from iTunes")
def _show_lookup_status(self, text: str, is_error: bool = False):
self._lookup_status.setText(text)
self._lookup_status.setStyleSheet(
"color: #c0392b; font-size: 11px; font-style: italic;"
if is_error else
"color: #27ae60; font-size: 11px; font-style: italic;"
)
self._lookup_status.setVisible(True)
QTimer.singleShot(4000, lambda: self._lookup_status.setVisible(False))
def dragEnterEvent(self, event: QDragEnterEvent):
if event.mimeData().hasUrls():
event.acceptProposedAction()

21
src/metadata_handler.py
View File

@ -20,7 +20,7 @@ from mutagen.mp4 import MP4, MP4Cover
from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK
from mutagen.easyid3 import EasyID3
from track_info import TrackInfo
from youtube_downloader import TrackInfo
from artwork.cache import cover_cache, xdg_cache_path
# Configure logging
@ -356,7 +356,7 @@ class MetadataHandler:
cover_path = cover_cache.put(artist or "Unknown Artist", album or "", cover_data)
return TrackInfo(
source_id=f"local_{hash(file_path)}",
video_id=f"local_{hash(file_path)}",
title=title or base_name,
artist=artist or "Unknown Artist",
album=album or "",
@ -366,7 +366,6 @@ class MetadataHandler:
genre=genre,
download_path=file_path,
cover_path=cover_path,
album_artist=album_artist,
)
except Exception as e:
@ -534,7 +533,7 @@ class MetadataHandler:
artist, title = filename.split(" - ", 1)
return TrackInfo(
source_id=f"local_{hash(file_path or filename)}",
video_id=f"local_{hash(file_path or filename)}",
title=title,
artist=artist,
album="",
@ -542,6 +541,20 @@ class MetadataHandler:
duration=0,
download_path=file_path,
)
def enhance_metadata_with_musicbrainz(self, track_info: TrackInfo) -> TrackInfo:
"""
Enhance track metadata using MusicBrainz API
Args:
track_info: TrackInfo object with track metadata
Returns:
Enhanced TrackInfo object
"""
# This is a placeholder for future implementation
# 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:

0
src/models/__init__.py
View File

130
src/playlist_manager.py
View File

@ -1,130 +0,0 @@
import os
import json
import uuid
import time
import logging
from typing import List, Dict, Optional
logger = logging.getLogger(__name__)
_PLAYLIST_FILE = os.path.join(
os.path.expanduser("~"), ".config", "neo-pod-desktop", "playlists.json"
)
class Playlist:
def __init__(self, name: str, playlist_id: Optional[str] = None,
track_paths: Optional[List[str]] = None,
created_at: Optional[float] = None,
updated_at: Optional[float] = None):
self.id = playlist_id or str(uuid.uuid4())
self.name = name
self.track_paths = track_paths or []
self.created_at = created_at or time.time()
self.updated_at = updated_at or time.time()
def to_dict(self) -> Dict:
return {
"id": self.id,
"name": self.name,
"track_paths": self.track_paths,
"created_at": self.created_at,
"updated_at": self.updated_at,
}
@classmethod
def from_dict(cls, data: Dict) -> "Playlist":
return cls(
name=data["name"],
playlist_id=data.get("id"),
track_paths=data.get("track_paths", []),
created_at=data.get("created_at"),
updated_at=data.get("updated_at"),
)
class PlaylistManager:
def __init__(self, filepath: str = _PLAYLIST_FILE):
self._filepath = filepath
self._playlists: List[Playlist] = []
self._load()
def _load(self):
try:
os.makedirs(os.path.dirname(self._filepath), exist_ok=True)
if os.path.exists(self._filepath):
with open(self._filepath, "r", encoding="utf-8") as f:
data = json.load(f)
self._playlists = [Playlist.from_dict(p) for p in data]
except Exception as e:
logger.error("Failed to load playlists: %s", e)
self._playlists = []
def save(self):
try:
os.makedirs(os.path.dirname(self._filepath), exist_ok=True)
with open(self._filepath, "w", encoding="utf-8") as f:
json.dump([p.to_dict() for p in self._playlists], f, indent=2)
except Exception as e:
logger.error("Failed to save playlists: %s", e)
def create(self, name: str) -> str:
playlist = Playlist(name=name)
self._playlists.append(playlist)
self.save()
return playlist.id
def rename(self, playlist_id: str, new_name: str) -> bool:
for p in self._playlists:
if p.id == playlist_id:
p.name = new_name
p.updated_at = time.time()
self.save()
return True
return False
def delete(self, playlist_id: str) -> bool:
for i, p in enumerate(self._playlists):
if p.id == playlist_id:
del self._playlists[i]
self.save()
return True
return False
def add_tracks(self, playlist_id: str, paths: List[str]) -> bool:
for p in self._playlists:
if p.id == playlist_id:
existing = set(p.track_paths)
for path in paths:
if path not in existing:
p.track_paths.append(path)
existing.add(path)
p.updated_at = time.time()
self.save()
return True
return False
def remove_tracks(self, playlist_id: str, paths: List[str]) -> bool:
for p in self._playlists:
if p.id == playlist_id:
path_set = set(paths)
p.track_paths = [t for t in p.track_paths if t not in path_set]
p.updated_at = time.time()
self.save()
return True
return False
def reorder(self, playlist_ids: List[str]) -> bool:
id_order = {pid: i for i, pid in enumerate(playlist_ids)}
self._playlists.sort(key=lambda p: id_order.get(p.id, 999))
self.save()
return True
def get_all(self) -> List[Playlist]:
return list(self._playlists)
def get_by_id(self, playlist_id: str) -> Optional[Playlist]:
for p in self._playlists:
if p.id == playlist_id:
return p
return None

0
src/services/__init__.py
View File

63
src/services/itunes_search.py
View File

@ -1,63 +0,0 @@
#!/usr/bin/env python3
"""
iTunes Search API service for neo-pod-desktop.
"""
import urllib.parse
from typing import Optional, Dict, Any
import requests
def search_itunes(artist: str, title: str, limit: int = 5) -> Optional[Dict[str, Any]]:
"""
Search iTunes Store for track metadata.
Args:
artist: Artist name
title: Track title
limit: Max results to fetch (default 5)
Returns:
Dict with keys: title, artist, album, album_artist, genre, year,
track_number, track_total, disc_number, disc_total, cover_url.
Returns None on failure or no results.
"""
try:
term = " ".join(p for p in (artist, title) if p)
if not term:
return None
quoted = urllib.parse.quote(term)
url = f"https://itunes.apple.com/search?term={quoted}&entity=song&limit={limit}&country=US"
resp = requests.get(url, timeout=10)
resp.raise_for_status()
data = resp.json()
results = data.get("results", [])
if not results:
return None
r = results[0]
cover_url = r.get("artworkUrl100", "")
if cover_url:
cover_url = cover_url.replace("100x100", "600x600")
year = 0
release_date = r.get("releaseDate", "")
if release_date and len(release_date) >= 4:
try:
year = int(release_date[:4])
except (ValueError, TypeError):
pass
return {
"title": r.get("trackName", ""),
"artist": r.get("artistName", ""),
"album": r.get("collectionName", ""),
"album_artist": r.get("collectionArtistName", ""),
"genre": r.get("primaryGenreName", ""),
"year": year,
"track_number": r.get("trackNumber", 0) or 0,
"track_total": r.get("trackCount", 0) or 0,
"disc_number": r.get("discNumber", 0) or 0,
"disc_total": r.get("discCount", 0) or 0,
"cover_url": cover_url,
}
except Exception:
return None

28
src/track_info.py
View File

@ -1,28 +0,0 @@
#!/usr/bin/env python3
"""
Track Info data class shared across the iPod transfer pipeline.
"""
from typing import Optional
from dataclasses import dataclass
@dataclass
class TrackInfo:
"""Data class to store track information."""
source_id: str
title: str
artist: str
album: str
thumbnail_url: str
duration: int
track_number: Optional[int] = None
playlist_title: Optional[str] = None
download_path: Optional[str] = None
genre: Optional[str] = None
album_artist: Optional[str] = None
cover_path: Optional[str] = None
play_count: int = 0
rating: int = 0
last_played: int = 0
skip_count: int = 0

0
src/ui/__init__.py
View File

517
src/ui/ipod_tab.py
View File

@ -1,517 +0,0 @@
import os
import logging
from typing import List, Dict, Optional
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout,
QPushButton, QLabel, QProgressBar,
QComboBox, QListWidget, QListWidgetItem, QMessageBox,
QFileDialog,
)
from PyQt6.QtCore import Qt, pyqtSignal
from ipod_device import IPodDevice
from device_monitor import DeviceMonitor
from worker import WorkerThread
logger = logging.getLogger(__name__)
class iPodTab(QWidget):
device_mounted = pyqtSignal(str)
device_unmounted = pyqtSignal()
def __init__(self, parent=None):
super().__init__(parent)
self.ipod_devices: List[Dict] = []
self.current_mount_point: Optional[str] = None
self.worker_thread: Optional[WorkerThread] = None
self._setup_ui()
def _setup_ui(self):
layout = QVBoxLayout(self)
device_layout = QHBoxLayout()
device_label = QLabel("iPod Device:")
self.device_combo = QComboBox()
self.device_combo.setPlaceholderText("No devices found")
refresh_button = QPushButton("Refresh")
refresh_button.clicked.connect(self._on_refresh_devices_clicked)
device_layout.addWidget(device_label)
device_layout.addWidget(self.device_combo)
device_layout.addWidget(refresh_button)
layout.addLayout(device_layout)
mount_layout = QHBoxLayout()
self.mount_button = QPushButton("Mount")
self.mount_button.setEnabled(False)
self.mount_button.clicked.connect(self._on_mount_clicked)
self.eject_button = QPushButton("Eject")
self.eject_button.setEnabled(False)
self.eject_button.clicked.connect(self._on_eject_clicked)
mount_layout.addWidget(self.mount_button)
mount_layout.addWidget(self.eject_button)
mount_layout.addStretch()
layout.addLayout(mount_layout)
self.device_info_label = QLabel("No device selected")
layout.addWidget(self.device_info_label)
self.device_status_label = QLabel("Status: Not connected")
layout.addWidget(self.device_status_label)
self.track_count_label = QLabel("Tracks on device: 0")
layout.addWidget(self.track_count_label)
on_device_label = QLabel("On Device:")
layout.addWidget(on_device_label)
self.transferred_list = QListWidget()
self.transferred_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
layout.addWidget(self.transferred_list)
remove_button = QPushButton("Remove Selected from iPod")
remove_button.clicked.connect(self._on_remove_tracks_clicked)
layout.addWidget(remove_button)
orphan_button = QPushButton("Scan for Orphaned Files")
orphan_button.clicked.connect(self._on_scan_orphans_clicked)
layout.addWidget(orphan_button)
dedup_button = QPushButton("Remove Duplicate Tracks")
dedup_button.clicked.connect(self._on_remove_duplicates_clicked)
layout.addWidget(dedup_button)
self.export_button = QPushButton("\u2B07 Download to Computer...")
self.export_button.setEnabled(False)
self.export_button.clicked.connect(self._on_export_ipod_clicked)
layout.addWidget(self.export_button)
self.export_progress = QProgressBar()
self.export_progress.setRange(0, 100)
self.export_progress.setVisible(False)
layout.addWidget(self.export_progress)
self.export_status = QLabel("")
self.export_status.setVisible(False)
layout.addWidget(self.export_status)
self.delete_progress = QProgressBar()
self.delete_progress.setRange(0, 100)
self.delete_progress.setVisible(False)
layout.addWidget(self.delete_progress)
self.delete_status = QLabel("")
self.delete_status.setVisible(False)
layout.addWidget(self.delete_status)
def _on_refresh_devices_clicked(self):
if self.worker_thread and self.worker_thread.isRunning():
return
self.device_combo.clear()
self.device_info_label.setText("Detecting devices...")
self.worker_thread = WorkerThread(task_type="detect_devices")
self.worker_thread.progress_signal.connect(self._on_device_detection_progress)
self.worker_thread.finished_signal.connect(self._on_device_detection_finished)
self.worker_thread.start()
def _on_device_detection_progress(self, progress, status):
self.device_info_label.setText(status)
def _on_device_detection_finished(self, success, message, result):
if success:
self.ipod_devices = result
self.device_combo.clear()
for device in self.ipod_devices:
mounted = device.get("mounted", False)
mount_text = " [mounted]" if mounted else " [not mounted]"
self.device_combo.addItem(device.get("name", "Unknown Device") + mount_text)
if self.ipod_devices:
device = self.ipod_devices[0]
detected_mount = device.get("mount_point")
already_mounted = device.get("mounted", False)
self._start_mount_and_load(device, detected_mount, already_mounted)
else:
self._set_device_not_found()
else:
self._set_device_not_found()
self._cleanup_worker()
def _start_mount_and_load(self, device: dict, mount_point=None, already_mounted=False):
self.device_status_label.setText("Mounting and loading tracks...")
self.mount_button.setEnabled(False)
self.worker_thread = WorkerThread(
task_type="mount_and_load",
device=device,
mount_point=mount_point,
already_mounted=already_mounted,
)
self.worker_thread.progress_signal.connect(self._on_mount_load_progress)
self.worker_thread.finished_signal.connect(self._on_mount_load_finished)
self.worker_thread.start()
def _on_mount_load_progress(self, progress, status):
self.device_info_label.setText(status)
def _on_mount_load_finished(self, success, message, result):
if success:
device = result["device"]
mount_point = result["mount_point"]
device_info = result["device_info"]
tracks = result["tracks"]
self.current_mount_point = mount_point
device["info"] = device_info
free_space = device_info.get("free_space", 0)
total_space = device_info.get("total_space", 0)
self.device_info_label.setText(
f"Device: {device['name']}\n"
f"Mount Point: {mount_point}\n"
f"Free Space: {free_space / 1024**2:.1f} MB / {total_space / 1024**2:.1f} MB"
)
self.device_status_label.setText("Status: Mounted and ready")
self.mount_button.setEnabled(False)
self.eject_button.setEnabled(True)
self.device_mounted.emit(mount_point)
self.transferred_list.clear()
for track in tracks:
display = f"{track['artist']} — {track['title']}"
if track.get("album"):
display += f" ({track['album']})"
item = QListWidgetItem(display)
item.setData(Qt.ItemDataRole.UserRole, track)
self.transferred_list.addItem(item)
self.track_count_label.setText(f"Tracks on device: {len(tracks)}")
self.export_button.setEnabled(len(tracks) > 0)
else:
self.device_status_label.setText(f"Status: {message}")
self.mount_button.setEnabled(True)
if self.ipod_devices:
self._set_device_unmounted(self.ipod_devices[0])
else:
self._set_device_not_found()
self._cleanup_worker()
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"
)
self.device_status_label.setText("Status: Connected but not mounted")
self.mount_button.setEnabled(True)
self.eject_button.setEnabled(False)
self.track_count_label.setText("Tracks on device: 0")
self.transferred_list.clear()
self.export_button.setEnabled(False)
self.export_progress.setVisible(False)
self.export_status.setVisible(False)
self.device_unmounted.emit()
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()
self.mount_button.setEnabled(False)
self.eject_button.setEnabled(False)
self.track_count_label.setText("Tracks on device: 0")
self.transferred_list.clear()
self.export_button.setEnabled(False)
self.export_progress.setVisible(False)
self.export_status.setVisible(False)
self.device_unmounted.emit()
def _on_mount_clicked(self):
if not self.ipod_devices:
return
device = self.ipod_devices[0]
detected_mount = device.get("mount_point")
self._start_mount_and_load(device, detected_mount, already_mounted=False)
def _on_eject_clicked(self):
reply = QMessageBox.question(
self, "Eject iPod",
"Eject iPod? Make sure no transfers are in progress.",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply != QMessageBox.StandardButton.Yes:
return
self.device_status_label.setText("Ejecting...")
self.eject_button.setEnabled(False)
try:
ipod = IPodDevice(mount_point=self.current_mount_point)
success = ipod.unmount_device()
if success or True:
self.current_mount_point = None
if self.ipod_devices:
self.ipod_devices[0]["mounted"] = False
self.ipod_devices[0]["mount_point"] = None
self.device_info_label.setText("iPod ejected. You can now disconnect it.")
self.device_status_label.setText("Status: Ejected - safe to disconnect")
self.mount_button.setEnabled(False)
self.eject_button.setEnabled(False)
self.track_count_label.setText("Tracks on device: 0")
self.transferred_list.clear()
self.device_unmounted.emit()
else:
self.device_status_label.setText("Status: Eject failed")
self.eject_button.setEnabled(True)
QMessageBox.warning(self, "Eject Error", "Failed to eject iPod safely.")
except Exception as e:
self.device_status_label.setText("Status: Eject error")
self.eject_button.setEnabled(True)
QMessageBox.warning(self, "Eject Error", f"Error ejecting iPod: {e}")
def _load_ipod_tracks(self):
if not self.current_mount_point:
return
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
tracks = db.get_all_tracks()
self.transferred_list.clear()
for track in tracks:
display = f"{track['artist']} — {track['title']}"
if track.get("album"):
display += f" ({track['album']})"
item = QListWidgetItem(display)
item.setData(Qt.ItemDataRole.UserRole, track)
self.transferred_list.addItem(item)
self.track_count_label.setText(f"Tracks on device: {len(tracks)}")
self.export_button.setEnabled(len(tracks) > 0)
except Exception as e:
logger.error(f"Failed to load iPod tracks: {e}")
self.track_count_label.setText("Tracks on device: unknown")
def _on_remove_tracks_clicked(self):
selected = self.transferred_list.selectedItems()
if not selected:
QMessageBox.information(self, "Info", "Select tracks to remove")
return
count = len(selected)
reply = QMessageBox.question(
self, "Confirm Delete",
f"Remove {count} track(s) from iPod?\nThis will permanently delete the files from the device.",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply != QMessageBox.StandardButton.Yes:
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
pids = []
for item in selected:
track_data = item.data(Qt.ItemDataRole.UserRole)
if track_data and "pid" in track_data:
pids.append(track_data["pid"])
if not pids:
QMessageBox.warning(self, "Error", "No valid tracks selected")
return
self._set_delete_buttons_enabled(False)
self.delete_progress.setVisible(True)
self.delete_progress.setValue(0)
self.delete_status.setVisible(True)
self.delete_status.setText(f"Removing {len(pids)} track(s)...")
self.worker_thread = WorkerThread(
task_type="delete_tracks",
pids=pids,
mount_point=self.current_mount_point,
)
self.worker_thread.progress_signal.connect(self._on_delete_progress)
self.worker_thread.finished_signal.connect(self._on_delete_finished)
self.worker_thread.start()
def _set_delete_buttons_enabled(self, enabled: bool):
for w in self.findChildren(QPushButton):
if w.text() in ("Remove Selected from iPod", "Remove Duplicate Tracks",
"Scan for Orphaned Files", "\u2B07 Download to Computer..."):
w.setEnabled(enabled)
def _on_delete_progress(self, progress, status):
self.delete_progress.setValue(progress)
self.delete_status.setText(status)
def _on_delete_finished(self, success, message, result):
self._set_delete_buttons_enabled(True)
if success:
self.delete_progress.setValue(100)
self._load_ipod_tracks()
QMessageBox.information(self, "Done" if success else "Error", message)
self._cleanup_worker()
def _on_export_ipod_clicked(self):
if not self.current_mount_point:
QMessageBox.warning(self, "Error", "No iPod device mounted.")
return
selected = self.transferred_list.selectedItems()
if not selected:
QMessageBox.information(self, "Info", "Select tracks to download")
return
dest_dir = QFileDialog.getExistingDirectory(self, "Select Download Destination")
if not dest_dir:
return
tracks = []
for item in selected:
data = item.data(Qt.ItemDataRole.UserRole)
if data and data.get("file_path") and os.path.exists(data["file_path"]):
tracks.append(data)
if not tracks:
QMessageBox.warning(self, "Error", "No valid track files found on device")
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
self.export_button.setEnabled(False)
self.export_progress.setVisible(True)
self.export_progress.setValue(0)
self.export_status.setVisible(True)
self.export_status.setText(f"Exporting {len(tracks)} track(s)...")
self.worker_thread = WorkerThread(
task_type="export_ipod",
tracks=tracks,
dest_dir=dest_dir,
mount_point=self.current_mount_point,
)
self.worker_thread.progress_signal.connect(self._on_export_ipod_progress)
self.worker_thread.finished_signal.connect(self._on_export_ipod_finished)
self.worker_thread.start()
def _on_export_ipod_progress(self, progress, status):
self.export_progress.setValue(progress)
self.export_status.setText(status)
def _on_export_ipod_finished(self, success, message, result):
self.export_button.setEnabled(True)
self.export_progress.setValue(100 if success else 0)
if success:
self.export_status.setText(message)
QMessageBox.information(self, "Export Complete", message)
else:
self.export_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Export Error", message)
self._cleanup_worker()
def _on_scan_orphans_clicked(self):
if not self.current_mount_point:
QMessageBox.warning(self, "Error", "No iPod device mounted.")
return
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
orphans = db.get_orphaned_files()
if not orphans:
QMessageBox.information(self, "Scan Complete", "No orphaned files found.")
return
total_size = sum(o["size"] for o in orphans)
total_mb = total_size / (1024 * 1024)
details = f"Found {len(orphans)} orphaned file(s) occupying {total_mb:.1f} MB:\n\n"
for o in orphans[:20]:
size_kb = o["size"] / 1024
details += f" {o['artist']} — {o['title']} ({size_kb:.0f} KB)\n"
if len(orphans) > 20:
details += f" ... and {len(orphans) - 20} more\n"
reply = QMessageBox.question(
self, "Orphaned Files Found",
f"{details}\nDelete these files to free up space on iPod?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.Yes:
deleted = db.delete_orphaned_files()
QMessageBox.information(self, "Cleanup Complete", f"Deleted {deleted} orphaned file(s).")
self._load_ipod_tracks()
except Exception as e:
logger.error(f"Failed to scan for orphaned files: {e}")
QMessageBox.warning(self, "Error", f"Failed to scan: {e}")
def _on_remove_duplicates_clicked(self):
if not self.current_mount_point:
QMessageBox.warning(self, "Error", "No iPod device mounted.")
return
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
self._set_delete_buttons_enabled(False)
self.delete_progress.setVisible(True)
self.delete_progress.setValue(0)
self.delete_status.setVisible(True)
self.delete_status.setText("Scanning for duplicates...")
self.worker_thread = WorkerThread(
task_type="remove_duplicates",
mount_point=self.current_mount_point,
)
self.worker_thread.progress_signal.connect(self._on_delete_progress)
self.worker_thread.finished_signal.connect(self._on_delete_duplicates_finished)
self.worker_thread.start()
def _on_delete_duplicates_finished(self, success, message, result):
self._set_delete_buttons_enabled(True)
self._load_ipod_tracks()
if success:
self.delete_progress.setValue(100)
QMessageBox.information(self, "Duplicates" if success else "Error", message)
self._cleanup_worker()
def start_monitoring(self, monitor: DeviceMonitor):
monitor.state_changed.connect(self._on_auto_state_changed)
def stop_monitoring(self, monitor: DeviceMonitor):
try:
monitor.state_changed.disconnect(self._on_auto_state_changed)
except TypeError:
pass
def _on_auto_state_changed(self, devices):
if self.worker_thread and self.worker_thread.isRunning():
return
self._on_device_detection_finished(True, f"Found {len(devices)} devices", devices)
def _cleanup_worker(self):
if self.worker_thread:
self.worker_thread.wait(3000)
self.worker_thread.deleteLater()
self.worker_thread = None

1326
src/ui/library_tab.py
View File

File diff suppressed because it is too large Load Diff

137
src/ui/player_header.py
View File

@ -1,137 +0,0 @@
from PyQt6.QtWidgets import (
QWidget, QHBoxLayout, QVBoxLayout, QPushButton, QLabel, QSlider, QGroupBox, QStyle,
)
from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QPixmap, QColor
def _fmt_time(ms: int) -> str:
total = ms // 1000
m, s = divmod(total, 60)
return f"{m}:{s:02d}"
class PlayerHeader(QGroupBox):
prev_requested = pyqtSignal()
play_pause_requested = pyqtSignal()
next_requested = pyqtSignal()
position_changed_by_user = pyqtSignal(int)
volume_changed = pyqtSignal(int)
def __init__(self, parent=None):
super().__init__(parent)
self.setFlat(True)
self._setup_ui()
def _setup_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(8, 6, 8, 6)
controls_row = QHBoxLayout()
controls_row.setSpacing(8)
self.prev_btn = QPushButton("\u23EE")
self.prev_btn.setFixedWidth(36)
self.prev_btn.setToolTip("Previous")
self.prev_btn.clicked.connect(self.prev_requested)
self.play_btn = QPushButton("\u25B6")
self.play_btn.setFixedWidth(44)
self.play_btn.setToolTip("Play / Pause")
self.play_btn.clicked.connect(self.play_pause_requested)
self.next_btn = QPushButton("\u23ED")
self.next_btn.setFixedWidth(36)
self.next_btn.setToolTip("Next")
self.next_btn.clicked.connect(self.next_requested)
controls_row.addWidget(self.prev_btn)
controls_row.addWidget(self.play_btn)
controls_row.addWidget(self.next_btn)
self.time_label_start = QLabel("0:00")
self.time_label_start.setFixedWidth(40)
self.time_label_start.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self.position_slider = QSlider(Qt.Orientation.Horizontal)
self.position_slider.setMinimum(0)
self.position_slider.setMaximum(1000)
self.position_slider.setValue(0)
self.position_slider.sliderPressed.connect(self._on_slider_user_action)
self.position_slider.sliderMoved.connect(self._on_slider_user_action)
self.position_slider.sliderReleased.connect(self._on_slider_user_action)
self.time_label_end = QLabel("0:00")
self.time_label_end.setFixedWidth(40)
self.time_label_end.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
controls_row.addWidget(self.time_label_start)
controls_row.addWidget(self.position_slider, stretch=1)
controls_row.addWidget(self.time_label_end)
vol_icon = QLabel("\U0001F50A")
self.volume_slider = QSlider(Qt.Orientation.Horizontal)
self.volume_slider.setFixedWidth(80)
self.volume_slider.setRange(0, 100)
self.volume_slider.setValue(80)
self.volume_slider.valueChanged.connect(self.volume_changed)
controls_row.addWidget(vol_icon)
controls_row.addWidget(self.volume_slider)
layout.addLayout(controls_row)
now_playing_layout = QHBoxLayout()
now_playing_layout.setContentsMargins(6, 0, 0, 0)
self.cover_label = QLabel()
self.cover_label.setFixedSize(60, 60)
self.cover_label.setScaledContents(True)
self.cover_label.setStyleSheet("background: palette(window); border-radius: 6px;")
now_playing_layout.addWidget(self.cover_label)
self.now_playing_label = QLabel("Select a track to play")
self.now_playing_label.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
self.now_playing_label.setStyleSheet("color: palette(mid); font-size: 12px;")
now_playing_layout.addWidget(self.now_playing_label, stretch=1)
layout.addLayout(now_playing_layout)
def _on_slider_user_action(self):
frac = self.position_slider.value() / 1000.0
self.position_changed_by_user.emit(int(frac))
def set_cover(self, pixmap: QPixmap = None, tooltip: str = ""):
if pixmap:
self.cover_label.setPixmap(pixmap)
else:
self.cover_label.clear()
self.cover_label.setToolTip(tooltip)
def set_track_info(self, text: str):
self.now_playing_label.setText(text)
self.now_playing_label.setStyleSheet(
"color: palette(text); font-weight: bold;" if text else "color: palette(mid); font-size: 12px;"
)
def set_playing(self, playing: bool):
self.play_btn.setText("\u23F8" if playing else "\u25B6")
def set_position(self, position_ms: int, duration_ms: int):
if not self.position_slider.isSliderDown():
if duration_ms > 0:
self.position_slider.setValue(int(position_ms / duration_ms * 1000))
else:
self.position_slider.setValue(0)
self.time_label_start.setText(_fmt_time(position_ms))
self.time_label_end.setText(_fmt_time(duration_ms))
def set_volume(self, value: int):
self.volume_slider.setValue(value)
def reset_info(self):
self.cover_label.clear()
self.cover_label.setToolTip("")
self.now_playing_label.setText("Select a track to play")
self.now_playing_label.setStyleSheet("color: palette(mid); font-size: 12px;")
self.time_label_start.setText("0:00")
self.time_label_end.setText("0:00")
self.position_slider.setValue(0)
self.play_btn.setText("\u25B6")

204
src/ui/settings_tab.py
View File

@ -1,204 +0,0 @@
#!/usr/bin/env python3
"""
Settings Tab for neo-pod-desktop.
"""
import os
from typing import Optional
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QLineEdit,
QComboBox, QCheckBox, QSpinBox, QGroupBox, QHeaderView,
QTableWidget, QTableWidgetItem, QFileDialog, QMessageBox,
)
from PyQt6.QtCore import Qt, pyqtSignal
from hotkeys import HotkeyManager, KeyCaptureDialog
class SettingsTab(QWidget):
"""Settings tab widget."""
hide_library_buttons_changed = pyqtSignal(bool)
auto_detect_changed = pyqtSignal(bool)
def __init__(self, parent=None):
super().__init__(parent)
self.hotkey_manager: Optional[HotkeyManager] = None
self._setup_ui()
def set_hotkey_manager(self, hm: HotkeyManager):
self.hotkey_manager = hm
def _setup_ui(self):
layout = QVBoxLayout(self)
library_group = QGroupBox("Library")
library_layout = QVBoxLayout(library_group)
output_layout = QHBoxLayout()
output_label = QLabel("Output Directory:")
self.output_dir_input = QLineEdit()
self.output_dir_input.setText(os.path.join(os.path.expanduser("~"), "Music", "iPod"))
browse_btn = QPushButton("Browse...")
browse_btn.clicked.connect(self._on_browse_output_clicked)
output_layout.addWidget(output_label)
output_layout.addWidget(self.output_dir_input)
output_layout.addWidget(browse_btn)
library_layout.addLayout(output_layout)
convert_layout = QHBoxLayout()
convert_format_label = QLabel("Convert Format:")
self.format_combo = QComboBox()
self.format_combo.addItems(["mp3", "m4a"])
convert_quality_label = QLabel("Quality (kbps):")
self.quality_spin = QSpinBox()
self.quality_spin.setRange(128, 320)
self.quality_spin.setValue(256)
self.quality_spin.setSingleStep(32)
convert_layout.addWidget(convert_format_label)
convert_layout.addWidget(self.format_combo)
convert_layout.addWidget(convert_quality_label)
convert_layout.addWidget(self.quality_spin)
convert_layout.addStretch()
library_layout.addLayout(convert_layout)
layout.addWidget(library_group)
metadata_group = QGroupBox("Metadata Options")
metadata_layout = QVBoxLayout(metadata_group)
self.embed_artwork_check = QCheckBox("Embed Album Artwork")
self.embed_artwork_check.setChecked(True)
metadata_layout.addWidget(self.embed_artwork_check)
layout.addWidget(metadata_group)
advanced_group = QGroupBox("Advanced Options")
advanced_layout = QVBoxLayout(advanced_group)
self.clean_temp_check = QCheckBox("Clean Temporary Files After Transfer")
self.clean_temp_check.setChecked(True)
advanced_layout.addWidget(self.clean_temp_check)
self.auto_detect_check = QCheckBox("Auto-Detect iPod")
self.auto_detect_check.setChecked(True)
self.auto_detect_check.toggled.connect(self.auto_detect_changed.emit)
advanced_layout.addWidget(self.auto_detect_check)
self.hide_library_buttons_check = QCheckBox("Hide Library Toolbar Buttons")
self.hide_library_buttons_check.setChecked(False)
self.hide_library_buttons_check.toggled.connect(self._on_hide_buttons_toggled)
advanced_layout.addWidget(self.hide_library_buttons_check)
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(
"neo-pod-desktop\n"
"Version 1.0.0\n\n"
"Desktop application for downloading, converting, managing and transferring music to iPod Nano devices."
)
about_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(about_label)
def _on_hide_buttons_toggled(self, checked: bool):
self.hide_library_buttons_changed.emit(checked)
def _on_browse_output_clicked(self):
directory = QFileDialog.getExistingDirectory(
self, "Select Output Directory",
self.output_dir_input.text()
)
if directory:
self.output_dir_input.setText(directory)
def load_settings(self, config):
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.embed_artwork_check.setChecked(
config.get_boolean("Metadata", "embed_artwork", fallback=True)
)
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)
)
hide_library_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False)
self.hide_library_buttons_check.setChecked(hide_library_buttons)
def save_settings(self, config):
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("Metadata", "embed_artwork", str(self.embed_artwork_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", "hide_library_buttons", str(self.hide_library_buttons_check.isChecked()).lower())
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() == KeyCaptureDialog.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.")

307
src/ui/sidebar_widget.py
View File

@ -1,307 +0,0 @@
import logging
from typing import List, Optional
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QLabel, QPushButton,
QListWidget, QListWidgetItem, QMenu, QInputDialog, QMessageBox,
QFrame,
)
from PyQt6.QtCore import Qt, pyqtSignal
from PyQt6.QtGui import QColor
logger = logging.getLogger(__name__)
SIDEBAR_MIN_WIDTH = 48
SIDEBAR_MAX_WIDTH = 240
SIDEBAR_DEFAULT_WIDTH = 200
class SidebarWidget(QWidget):
item_selected = pyqtSignal(str, str)
playlist_selected = pyqtSignal(str)
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("sidebar")
self.setMinimumWidth(SIDEBAR_MIN_WIDTH)
self.setMaximumWidth(SIDEBAR_MAX_WIDTH)
self.setFixedWidth(SIDEBAR_DEFAULT_WIDTH)
self.setAutoFillBackground(True)
p = self.palette()
self._bg = p.color(p.ColorRole.Window)
self._text = p.color(p.ColorRole.WindowText)
self._dim_text = p.color(p.ColorRole.PlaceholderText)
if not self._dim_text.isValid():
self._dim_text = self._text.darker(160)
self._border = p.color(p.ColorRole.Midlight)
if not self._border.isValid():
self._border = self._bg.darker(120)
hl = p.color(p.ColorRole.Highlight)
hl_light = QColor(hl)
hl_light.setAlpha(60)
hl_text = p.color(p.ColorRole.HighlightedText)
self._apply_palette_style(hl, hl_light, hl_text)
self._build_ui()
def _apply_palette_style(self, hl: QColor, hl_light: QColor, hl_text: QColor):
bg_name = self._bg.name()
border_name = self._border.name()
dim_name = self._dim_text.name()
text_name = self._text.name()
hl_name = hl.name()
hl_light_name = hl_light.name()
hl_text_name = hl_text.name()
self.setStyleSheet(f"""
QWidget#sidebar {{
background: {bg_name};
border-right: 1px solid {border_name};
}}
QListWidget {{
background: transparent;
border: none;
outline: none;
font-size: 13px;
color: {text_name};
}}
QListWidget::item {{
padding: 4px 12px;
border-radius: 4px;
}}
QListWidget::item:selected {{
background: {hl_name};
color: {hl_text_name};
}}
QListWidget::item:hover:!selected {{
background: {hl_light_name};
}}
""")
def _build_ui(self):
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 8, 0, 8)
layout.setSpacing(0)
dim_color = self._dim_text.name()
section_style = f"""
color: {dim_color};
font-size: 11px;
font-weight: bold;
letter-spacing: 1px;
padding: 8px 12px 4px 12px;
background: transparent;
"""
header = QLabel("LIBRARY")
header.setStyleSheet(section_style)
layout.addWidget(header)
self._library_list = QListWidget()
items = [
("\U0001F4BF Music", "tab", "library"),
("\u23F1 Recently Added", "filter", "recent"),
("\U0001F3A4 Artists", "filter", "artists"),
("\U0001F4BF Albums", "filter", "albums"),
("\U0001F3B5 Songs", "filter", "songs"),
]
for label, kind, value in items:
item = QListWidgetItem(label)
item.setData(Qt.ItemDataRole.UserRole + 1, kind)
item.setData(Qt.ItemDataRole.UserRole + 2, value)
self._library_list.addItem(item)
self._library_list.itemClicked.connect(self._on_library_item_clicked)
self._library_list.setCurrentRow(0)
layout.addWidget(self._library_list)
line = QFrame()
line.setFrameShape(QFrame.Shape.HLine)
line.setStyleSheet(f"color: {self._border.name()}; margin: 4px 12px;")
line.setFixedHeight(1)
layout.addWidget(line)
self._playlists_header = QLabel("PLAYLISTS")
self._playlists_header.setStyleSheet(section_style)
layout.addWidget(self._playlists_header)
self._playlist_list = QListWidget()
self._playlist_list.setDragDropMode(QListWidget.DragDropMode.InternalMove)
self._playlist_list.setDefaultDropAction(Qt.DropAction.MoveAction)
self._playlist_list.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self._playlist_list.customContextMenuRequested.connect(
self._on_playlist_context_menu
)
self._playlist_list.itemClicked.connect(self._on_playlist_clicked)
self._playlist_list.model().rowsMoved.connect(self._on_playlists_reordered)
layout.addWidget(self._playlist_list, stretch=1)
btn_style = f"""
QPushButton {{
text-align: left;
padding: 6px 12px;
border: none;
color: {dim_color};
font-size: 12px;
}}
QPushButton:hover {{
color: {self._text.name()};
background: {self._border.name()};
}}
"""
self._new_playlist_btn = QPushButton(" \u2795 New Playlist")
self._new_playlist_btn.setFlat(True)
self._new_playlist_btn.setCursor(Qt.CursorShape.PointingHandCursor)
self._new_playlist_btn.setStyleSheet(btn_style)
self._new_playlist_btn.clicked.connect(self._on_new_playlist)
layout.addWidget(self._new_playlist_btn)
line2 = QFrame()
line2.setFrameShape(QFrame.Shape.HLine)
line2.setStyleSheet(f"color: {self._border.name()}; margin: 4px 12px;")
line2.setFixedHeight(1)
layout.addWidget(line2)
self._nav_list = QListWidget()
items = [
("🎸 iPod", "ipod"),
("⚙️ Settings", "settings"),
]
for label, tab_id in items:
item = QListWidgetItem(label)
item.setData(Qt.ItemDataRole.UserRole + 1, "tab")
item.setData(Qt.ItemDataRole.UserRole + 2, tab_id)
self._nav_list.addItem(item)
self._nav_list.itemClicked.connect(self._on_nav_item_clicked)
layout.addWidget(self._nav_list)
self._playlist_items: dict = {}
def _on_library_item_clicked(self, item: QListWidgetItem):
kind = item.data(Qt.ItemDataRole.UserRole + 1)
value = item.data(Qt.ItemDataRole.UserRole + 2)
self._library_list.setCurrentItem(item)
self._playlist_list.clearSelection()
self._nav_list.clearSelection()
self.item_selected.emit(kind, value)
def _on_playlist_clicked(self, item: QListWidgetItem):
playlist_id = item.data(Qt.ItemDataRole.UserRole)
self._library_list.clearSelection()
self._playlist_list.setCurrentItem(item)
self._nav_list.clearSelection()
self.playlist_selected.emit(playlist_id)
def _on_playlist_context_menu(self, pos):
item = self._playlist_list.itemAt(pos)
if item is None:
return
menu = QMenu(self)
rename_action = menu.addAction("Rename")
delete_action = menu.addAction("Delete")
action = menu.exec(self._playlist_list.viewport().mapToGlobal(pos))
playlist_id = item.data(Qt.ItemDataRole.UserRole)
if action == rename_action:
self._rename_playlist(item, playlist_id)
elif action == delete_action:
self._delete_playlist(item, playlist_id)
def _rename_playlist(self, item: QListWidgetItem, playlist_id: str):
current = item.text()
name, ok = QInputDialog.getText(
self, "Rename Playlist", "New name:", text=current
)
if ok and name.strip():
item.setText(name.strip())
self.item_selected.emit("playlist_rename", playlist_id + "|" + name.strip())
def _delete_playlist(self, item: QListWidgetItem, playlist_id: str):
reply = QMessageBox.question(
self, "Delete Playlist",
f'Delete playlist "{item.text()}"?\nThis cannot be undone.',
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply == QMessageBox.StandardButton.Yes:
row = self._playlist_list.row(item)
self._playlist_list.takeItem(row)
self.item_selected.emit("playlist_delete", playlist_id)
def _on_new_playlist(self):
name, ok = QInputDialog.getText(self, "New Playlist", "Playlist name:")
if ok and name.strip():
self.item_selected.emit("playlist_create", name.strip())
def add_playlist_item(self, playlist_id: str, name: str):
item = QListWidgetItem(f"\U0001F4CB {name}")
item.setData(Qt.ItemDataRole.UserRole, playlist_id)
self._playlist_list.addItem(item)
self._playlist_items[playlist_id] = item
def remove_playlist_item(self, playlist_id: str):
item = self._playlist_items.pop(playlist_id, None)
if item:
row = self._playlist_list.row(item)
self._playlist_list.takeItem(row)
def rename_playlist_item(self, playlist_id: str, new_name: str):
item = self._playlist_items.get(playlist_id)
if item:
item.setText(f"\U0001F4CB {new_name}")
def clear_playlist_items(self):
self._playlist_list.clear()
self._playlist_items.clear()
def get_playlist_order_ids(self) -> List[str]:
ids = []
for i in range(self._playlist_list.count()):
item = self._playlist_list.item(i)
ids.append(item.data(Qt.ItemDataRole.UserRole))
return ids
def _on_playlists_reordered(self):
order = self.get_playlist_order_ids()
self.item_selected.emit("playlist_reorder", "|".join(order))
def _on_nav_item_clicked(self, item: QListWidgetItem):
kind = item.data(Qt.ItemDataRole.UserRole + 1)
value = item.data(Qt.ItemDataRole.UserRole + 2)
self._library_list.clearSelection()
self._playlist_list.clearSelection()
self._nav_list.setCurrentItem(item)
self.item_selected.emit(kind, value)
def select_library_music(self):
if self._library_list.count() > 0:
self._library_list.setCurrentRow(0)
self._on_library_item_clicked(self._library_list.item(0))
def select_playlist(self, playlist_id: str):
for i in range(self._playlist_list.count()):
item = self._playlist_list.item(i)
if item.data(Qt.ItemDataRole.UserRole) == playlist_id:
self._playlist_list.setCurrentItem(item)
self._on_playlist_clicked(item)
return
def select_playlist_silent(self, playlist_id: str):
for i in range(self._playlist_list.count()):
item = self._playlist_list.item(i)
if item.data(Qt.ItemDataRole.UserRole) == playlist_id:
self._playlist_list.setCurrentItem(item)
self._library_list.clearSelection()
self._nav_list.clearSelection()
return
def select_by_view_mode(self, mode: str):
mapping = {"songs": 4, "albums": 3, "artists": 2, "recent": 1}
row = mapping.get(mode, 0)
if row < self._library_list.count():
self._library_list.setCurrentRow(row)
self._playlist_list.clearSelection()
self._nav_list.clearSelection()
def width(self) -> int:
return self.geometry().width()

366
src/worker.py
View File

@ -1,366 +0,0 @@
#!/usr/bin/env python3
"""
Worker thread for background tasks.
"""
import os
import logging
from typing import Optional
from PyQt6.QtCore import QThread, pyqtSignal
from track_info import TrackInfo
from audio_converter import AudioConverter
from metadata_handler import MetadataHandler
from ipod_device import IPodDevice
from library_cache import invalidate_cache_for_path
logger = logging.getLogger(__name__)
AUDIO_EXTS = {'.m4a', '.mp3', '.aac'}
SOURCE_EXTS = {'.flac', '.wav', '.ogg', '.opus', '.wma', '.ape', '.aiff', '.alac'}
class WorkerThread(QThread):
"""Worker thread for background tasks"""
progress_signal = pyqtSignal(int, str)
finished_signal = pyqtSignal(bool, str, object)
def __init__(self, task_type: str, **kwargs):
super().__init__()
self.task_type = task_type
self.kwargs = kwargs
self.is_running = True
def run(self):
try:
handlers = {
"convert": self._run_convert,
"transfer": self._run_transfer,
"detect_devices": self._run_detect_devices,
"mount_and_load": self._run_mount_and_load,
"export_ipod": self._run_export_ipod,
"delete_tracks": self._run_delete_tracks,
"remove_duplicates": self._run_remove_duplicates,
"sync_metadata": self._run_sync_metadata,
}
handler = handlers.get(self.task_type)
if handler:
handler()
else:
raise ValueError(f"Unknown task type: {self.task_type}")
except Exception as e:
logger.exception(f"Error in worker thread: {e}")
self.finished_signal.emit(False, str(e), None)
def _run_convert(self):
tracks = self.kwargs.get("tracks", [])
local_files = self.kwargs.get("local_files", [])
output_dir = self.kwargs.get("output_dir", "converted")
fmt = self.kwargs.get("format", "mp3")
quality = self.kwargs.get("quality", 256)
valid_tracks = [
t for t in tracks
if hasattr(t, 'download_path') and t.download_path and os.path.exists(t.download_path)
]
metadata_handler = MetadataHandler()
for f in local_files:
if os.path.exists(f):
track = metadata_handler.extract_tags(f)
if track:
track.download_path = f
valid_tracks.append(track)
if not valid_tracks:
self.finished_signal.emit(False, "No valid tracks to convert", None)
return
self.progress_signal.emit(0, f"Initializing conversion of {len(valid_tracks)} tracks")
converter = AudioConverter(output_dir=output_dir)
metadata_handler = MetadataHandler()
converted_tracks = []
for i, track in enumerate(valid_tracks):
if not self.is_running:
break
progress = int(((i + 1) / len(valid_tracks)) * 100)
self.progress_signal.emit(progress, f"Converting {track.title}...")
try:
output_path = converter.convert_to_ipod_format(
track, track.download_path, fmt, quality
)
if output_path:
metadata_handler.process_file(output_path, track)
converted_tracks.append((track, output_path))
except Exception as e:
logger.error(f"Error converting track {track.title}: {e}")
continue
if not converted_tracks:
self.finished_signal.emit(False, "Failed to convert any tracks", None)
return
self.finished_signal.emit(True, f"Converted {len(converted_tracks)} tracks", converted_tracks)
def _run_transfer(self):
tracks = self.kwargs.get("tracks", [])
mount_point = self.kwargs.get("mount_point")
if not tracks:
self.finished_signal.emit(False, "No tracks to transfer", None)
return
if not mount_point:
self.finished_signal.emit(False, "No iPod device mounted", None)
return
self.progress_signal.emit(0, f"Initializing transfer of {len(tracks)} tracks")
ipod = IPodDevice(mount_point=mount_point)
device_info = ipod.get_device_info()
free_space = device_info.get("free_space", 0)
total_size = sum(os.path.getsize(path) for _, path in tracks if os.path.exists(path))
if total_size > free_space:
self.finished_signal.emit(
False,
f"Not enough space on iPod. Need {total_size / 1024**2:.1f} MB, "
f"but only {free_space / 1024**2:.1f} MB available",
None
)
return
transferred_tracks = []
for i, (track, path) in enumerate(tracks):
if not self.is_running:
break
progress = int(((i + 1) / len(tracks)) * 100)
self.progress_signal.emit(progress, f"Transferring {track.title}...")
if os.path.exists(path):
try:
success = ipod.transfer_file(path)
if success:
transferred_tracks.append((track, path))
except Exception as e:
logger.error(f"Error transferring track {track.title}: {e}")
continue
self.finished_signal.emit(
True, f"Transferred {len(transferred_tracks)} tracks to iPod", transferred_tracks
)
def _run_detect_devices(self):
self.progress_signal.emit(0, "Detecting iPod devices...")
ipod = IPodDevice()
devices = ipod.detect_devices()
if not devices:
self.finished_signal.emit(False, "No iPod devices found", [])
return
self.finished_signal.emit(True, f"Found {len(devices)} iPod devices", devices)
def _run_mount_and_load(self):
import shutil
device = self.kwargs.get("device", {})
mount_point = self.kwargs.get("mount_point")
already_mounted = self.kwargs.get("already_mounted", False)
if not already_mounted:
device_id = device.get("id")
detected_mount = device.get("mount_point")
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if not mount_point:
self.finished_signal.emit(False, "Failed to mount iPod", None)
return
device["mount_point"] = mount_point
device["mounted"] = True
from ipod_nano7_db import Nano7Database
usage = shutil.disk_usage(mount_point)
device_info = {
"mount_point": mount_point,
"total_space": usage.total,
"free_space": usage.free,
"used_space": usage.used,
}
db = Nano7Database(mount_point)
tracks = db.get_all_tracks()
device_info["track_count"] = len(tracks)
self.progress_signal.emit(60, "Syncing play counts from iPod...")
from library_cache import get_library_cache
cache = get_library_cache()
ipod_base, ipod_delta = db.read_play_stats()
hash_to_location: dict[str, str] = {}
for t in tracks:
ch = t.get("content_hash")
if not ch:
continue
fpath = t.get("file_path", "")
if not fpath:
continue
rel = os.path.relpath(fpath, os.path.join(db.mountpoint, "iPod_Control", "Music"))
loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}"
hash_to_location[ch] = loc
updated_cache = 0
for file_path, entry in list(cache._data.items()):
if not os.path.exists(file_path):
continue
ch = entry.get("_content_hash", "")
if not ch or ch not in hash_to_location:
continue
loc = hash_to_location[ch]
bs = ipod_base.get(loc, {})
dd = ipod_delta.get(loc, {})
entry["play_count"] = max(
entry.get("play_count", 0), bs.get("play_count", 0),
) + dd.get("play_count", 0)
entry["skip_count"] = max(
entry.get("skip_count", 0), bs.get("skip_count", 0),
) + dd.get("skip_count", 0)
entry["rating"] = dd.get("rating", 0) or bs.get("rating", 0) or entry.get("rating", 0)
entry["last_played"] = max(
entry.get("last_played", 0), bs.get("last_played", 0),
)
updated_cache += 1
if updated_cache:
cache.save()
logger.info("Updated play counts for %d local tracks from iPod", updated_cache)
db.consume_play_deltas()
self.progress_signal.emit(80, "Loading complete")
result = {
"device": device,
"mount_point": mount_point,
"device_info": device_info,
"tracks": tracks,
}
self.finished_signal.emit(
True, f"Loaded {len(tracks)} tracks from iPod", result
)
def _run_export_ipod(self):
tracks = self.kwargs.get("tracks", [])
dest_dir = self.kwargs.get("dest_dir", "")
mount_point = self.kwargs.get("mount_point")
if not tracks or not dest_dir or not mount_point:
self.finished_signal.emit(False, "Missing parameters for export", None)
return
self.progress_signal.emit(0, f"Exporting {len(tracks)} track(s)...")
ipod = IPodDevice(mount_point=mount_point)
def progress(p, s):
self.progress_signal.emit(p, s)
exported = ipod.export_tracks(tracks, dest_dir, progress_callback=progress)
self.finished_signal.emit(True, f"Exported {len(exported)} track(s)", exported)
def _run_delete_tracks(self):
pids = self.kwargs.get("pids", [])
mount_point = self.kwargs.get("mount_point")
if not pids or not mount_point:
self.finished_signal.emit(False, "No tracks to delete or device not mounted", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
def progress(p, s):
self.progress_signal.emit(p, s)
removed = db.delete_tracks(pids, progress_callback=progress)
self.finished_signal.emit(True, f"Removed {removed} track(s)", removed)
def _run_remove_duplicates(self):
mount_point = self.kwargs.get("mount_point")
if not mount_point:
self.finished_signal.emit(False, "No iPod device mounted", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
def progress(p, s):
self.progress_signal.emit(p, s)
self.progress_signal.emit(0, "Scanning for duplicates...")
removed = db.remove_duplicates(progress_callback=progress)
self.finished_signal.emit(True, f"Removed {len(removed)} duplicate(s)", removed)
def _run_sync_metadata(self):
"""Update metadata and artwork for tracks already on iPod (batch)."""
sync_batch = self.kwargs.get("sync_batch", [])
mount_point = self.kwargs.get("mount_point")
if not sync_batch or not mount_point:
self.finished_signal.emit(False, "Missing parameters for sync", None)
return
from ipod_nano7_db import Nano7Database
db = Nano7Database(mount_point)
total = len(sync_batch)
synced = 0
errors = []
for i, item in enumerate(sync_batch):
if not self.is_running:
break
self.progress_signal.emit(
int((i / total) * 100),
f"Syncing metadata: {item['new_metadata'].get('title', '?')}...",
)
try:
success = db.update_track_metadata(
pid=item["ipod_pid"],
new_metadata=item["new_metadata"],
local_source_path=item.get("local_path"),
)
if success:
synced += 1
else:
errors.append(item['new_metadata'].get('title', '?'))
except Exception as e:
logger.exception("Metadata sync failed for track: %s", e)
errors.append(f"{item['new_metadata'].get('title', '?')}: {e}")
msg = f"Synced metadata for {synced}/{total} track(s)"
if errors:
msg += f" ({len(errors)} error(s))"
self.progress_signal.emit(100, msg)
self.finished_signal.emit(synced > 0, msg, {"synced": synced, "errors": errors})
def stop(self):
self.is_running = False
self.wait()

417
src/youtube_downloader.py Normal file
View File

@ -0,0 +1,417 @@
#!/usr/bin/env python3
"""
YouTube Downloader Module for iPod Nano Transfer Tool
Handles downloading audio from YouTube videos and playlists
"""
import os
import re
import logging
import time
from typing import List, Dict, Optional, Tuple, Callable
from dataclasses import dataclass
import yt_dlp
from tqdm import tqdm
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class TrackInfo:
"""Data class to store track information"""
video_id: str
title: str
artist: str
album: str
thumbnail_url: str
duration: int
track_number: Optional[int] = None
playlist_id: Optional[str] = None
playlist_title: Optional[str] = None
playlist_index: Optional[int] = None
download_path: Optional[str] = None
genre: Optional[str] = None
cover_path: Optional[str] = None
class YouTubeDownloader:
"""Class to handle YouTube video and playlist downloads"""
def __init__(self, output_dir: str = "downloads", temp_dir: str = "temp", progress_callback: Optional[Callable] = None):
"""
Initialize the YouTube downloader
Args:
output_dir: Directory to save downloaded files
temp_dir: Directory for temporary files
progress_callback: Optional callback function for progress updates
"""
self.output_dir = output_dir
self.temp_dir = temp_dir
self.progress_callback = progress_callback
# Create directories if they don't exist
os.makedirs(output_dir, exist_ok=True)
os.makedirs(temp_dir, exist_ok=True)
# Playlist regex pattern
self.playlist_regex = r'(?:list=)([a-zA-Z0-9_-]+)'
# Track download progress
self.download_start_time = 0
self.current_track = None
def _parse_artist_title(self, video_title: str) -> Tuple[str, str]:
"""
Parse artist and title from video title
Args:
video_title: YouTube video title
Returns:
Tuple of (artist, title)
"""
# Common patterns: "Artist - Title", "Artist: Title", "Artist | Title"
patterns = [
r'^(.*?)\s*-\s*(.*?)$', # Artist - Title
r'^(.*?)\s*:\s*(.*?)$', # Artist: Title
r'^(.*?)\s*\|\s*(.*?)$' # Artist | Title
]
for pattern in patterns:
match = re.match(pattern, video_title)
if match:
artist, title = match.groups()
return artist.strip(), title.strip()
# If no pattern matches, return empty artist and full title
return "", video_title.strip()
def extract_video_info(self, video_url: str) -> Optional[TrackInfo]:
"""
Extract information from a YouTube video
Args:
video_url: YouTube video URL
Returns:
TrackInfo object with video metadata or None if video is unavailable
"""
logger.info(f"Extracting info from video: {video_url}")
ydl_opts = {
'quiet': True,
'no_warnings': True,
'skip_download': True,
'extract_flat': True,
'ignoreerrors': True, # Continue on download errors
}
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=False)
if not info:
logger.warning(f"Could not extract info from video: {video_url}")
return None
video_id = info.get('id')
video_title = info.get('title', '')
artist, title = self._parse_artist_title(video_title)
# Get best thumbnail
thumbnails = info.get('thumbnails', [])
thumbnail_url = next((t['url'] for t in reversed(thumbnails)
if 'url' in t), None) if thumbnails else None
return TrackInfo(
video_id=video_id,
title=title,
artist=artist,
album=artist, # Default album to artist name
thumbnail_url=thumbnail_url,
duration=info.get('duration', 0)
)
except Exception as e:
logger.warning(f"Error extracting video info: {e}")
return None
def extract_playlist_info(self, playlist_url: str) -> List[TrackInfo]:
"""
Extract information from a YouTube playlist
Args:
playlist_url: YouTube playlist URL
Returns:
List of TrackInfo objects for each video in the playlist
"""
logger.info(f"Extracting info from playlist: {playlist_url}")
# Extract playlist ID
playlist_id_match = re.search(self.playlist_regex, playlist_url)
if not playlist_id_match:
raise ValueError(f"Invalid playlist URL: {playlist_url}")
playlist_id = playlist_id_match.group(1)
ydl_opts = {
'quiet': True,
'no_warnings': True,
'extract_flat': True,
'skip_download': True,
'ignoreerrors': True, # Continue on download errors
}
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
playlist_info = ydl.extract_info(
f'https://www.youtube.com/playlist?list={playlist_id}',
download=False
)
if not playlist_info or 'entries' not in playlist_info:
logger.warning(f"Could not extract playlist info: {playlist_url}")
return []
playlist_title = playlist_info.get('title', 'Unknown Playlist')
tracks = []
track_number = 1 # Keep track of actual track numbers
# Update progress if callback is available
if self.progress_callback:
self.progress_callback(0, f"Analyzing playlist: {playlist_title}")
total_entries = len(playlist_info.get('entries', []))
for idx, entry in enumerate(playlist_info.get('entries', []), 1):
# Update progress for playlist analysis
if self.progress_callback:
progress = int((idx / total_entries) * 10) # Use first 10% for playlist analysis
self.progress_callback(progress, f"Analyzing playlist: {idx}/{total_entries} tracks")
if not entry or entry.get('_type') != 'url':
continue
try:
video_info = self.extract_video_info(f"https://www.youtube.com/watch?v={entry['id']}")
if video_info:
video_info.playlist_id = playlist_id
video_info.playlist_title = playlist_title
video_info.playlist_index = idx
video_info.track_number = track_number
video_info.album = playlist_title # Set album to playlist title
tracks.append(video_info)
track_number += 1
else:
logger.warning(f"Skipping unavailable video in playlist: {entry.get('id')}")
except Exception as e:
logger.warning(f"Error processing playlist entry {idx}: {e}")
return tracks
except Exception as e:
logger.error(f"Error extracting playlist info: {e}")
return []
def download_audio(self, track_info: TrackInfo, format: str = "m4a", quality: int = 256) -> Optional[str]:
"""
Download audio from YouTube video
Args:
track_info: TrackInfo object with video metadata
format: Audio format (m4a, mp3)
quality: Audio quality in kbps
Returns:
Path to downloaded file or None if download failed
"""
logger.info(f"Downloading audio for: {track_info.title}")
self.current_track = track_info
self.download_start_time = time.time()
# Create output filename
safe_title = re.sub(r'[^\w\-_\. ]', '_', track_info.title)
if track_info.track_number:
filename = f"{track_info.track_number:02d}. {safe_title}.{format}"
else:
filename = f"{safe_title}.{format}"
output_path = os.path.join(self.temp_dir, filename)
# Set up yt-dlp options
ydl_opts = {
'format': 'bestaudio/best',
'outtmpl': output_path,
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': format,
'preferredquality': str(quality),
}],
'quiet': False,
'progress_hooks': [self._download_progress_hook],
'ignoreerrors': True, # Continue on download errors
}
try:
# Download the file
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([f"https://www.youtube.com/watch?v={track_info.video_id}"])
# Check if file was actually downloaded
expected_path = f"{output_path}.{format}"
if os.path.exists(expected_path):
# Update track_info with download path
track_info.download_path = expected_path
return track_info.download_path
else:
logger.warning(f"Download failed for: {track_info.title}")
return None
except Exception as e:
logger.error(f"Error downloading audio: {e}")
return None
def _download_progress_hook(self, d: Dict):
"""Progress hook for yt-dlp"""
if d['status'] == 'downloading':
# Calculate download progress
if 'total_bytes' in d and d['total_bytes'] > 0:
percent = d['downloaded_bytes'] / d['total_bytes'] * 100
# Calculate download speed
elapsed_time = time.time() - self.download_start_time
if elapsed_time > 0:
download_speed = d['downloaded_bytes'] / elapsed_time / 1024 # KB/s
# Format download speed
speed_str = f"{download_speed:.1f} KB/s"
if download_speed > 1024:
speed_str = f"{download_speed/1024:.1f} MB/s"
# Format file size
total_size_mb = d['total_bytes'] / (1024 * 1024)
downloaded_mb = d['downloaded_bytes'] / (1024 * 1024)
# Create status message
status_msg = f"Downloading {self.current_track.title}: {downloaded_mb:.1f}/{total_size_mb:.1f} MB ({percent:.1f}%) at {speed_str}"
# Log progress
logger.info(status_msg)
# Call progress callback if available
if self.progress_callback:
# Scale to 10-95% range to leave room for playlist analysis and completion
scaled_percent = 10 + (percent * 0.85)
self.progress_callback(int(scaled_percent), status_msg)
# If we don't have total_bytes, show indeterminate progress
elif 'downloaded_bytes' in d:
elapsed_time = time.time() - self.download_start_time
if elapsed_time > 0:
download_speed = d['downloaded_bytes'] / elapsed_time / 1024 # KB/s
# Format download speed
speed_str = f"{download_speed:.1f} KB/s"
if download_speed > 1024:
speed_str = f"{download_speed/1024:.1f} MB/s"
# Format file size
downloaded_mb = d['downloaded_bytes'] / (1024 * 1024)
# Create status message
status_msg = f"Downloading {self.current_track.title}: {downloaded_mb:.1f} MB at {speed_str}"
# Log progress
logger.info(status_msg)
# Call progress callback if available
if self.progress_callback:
# Use a pulsing progress between 10-90% for indeterminate progress
pulse_progress = 10 + (int(time.time() * 10) % 80)
self.progress_callback(pulse_progress, status_msg)
elif d['status'] == 'finished':
logger.info(f"Download complete: {self.current_track.title}")
if self.progress_callback:
self.progress_callback(95, f"Download complete: {self.current_track.title}")
elif d['status'] == 'error':
logger.warning(f"Download error: {self.current_track.title}")
if self.progress_callback:
self.progress_callback(95, f"Download error: {self.current_track.title}")
def process_url(self, url: str, format: str = "m4a", quality: int = 256) -> List[TrackInfo]:
"""
Process a YouTube URL (video or playlist)
Args:
url: YouTube URL
format: Audio format
quality: Audio quality in kbps
Returns:
List of TrackInfo objects with download paths
"""
successful_tracks = []
try:
# Check if URL is a playlist
if 'list=' in url:
# Process as playlist
tracks = self.extract_playlist_info(url)
if not tracks:
logger.warning("No valid tracks found in playlist")
return []
# Download each track
for i, track in enumerate(tracks):
if self.progress_callback:
status_msg = f"Preparing to download {i+1}/{len(tracks)}: {track.title}"
self.progress_callback(10, status_msg)
try:
download_path = self.download_audio(track, format, quality)
if download_path:
successful_tracks.append(track)
except Exception as e:
logger.error(f"Error downloading track {track.title}: {e}")
else:
# Process as single video
track = self.extract_video_info(url)
if track:
try:
download_path = self.download_audio(track, format, quality)
if download_path:
successful_tracks.append(track)
except Exception as e:
logger.error(f"Error downloading track {track.title}: {e}")
else:
logger.warning("Could not extract video information")
# Final progress update
if self.progress_callback:
self.progress_callback(100, f"Downloaded {len(successful_tracks)} tracks successfully")
return successful_tracks
except Exception as e:
logger.error(f"Error processing URL: {e}")
if self.progress_callback:
self.progress_callback(100, f"Error: {str(e)}")
return successful_tracks
if __name__ == "__main__":
# Example usage
downloader = YouTubeDownloader()
# Test with a single video
video_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
video_info = downloader.extract_video_info(video_url)
print(f"Video: {video_info.artist} - {video_info.title}")
# Uncomment to test download
# downloader.download_audio(video_info)

52
tests/test_basic.py
View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Basic tests for neo-pod-desktop
Basic tests for YouTube Music to iPod Nano Transfer Tool
"""
import os
@ -8,22 +8,55 @@ import sys
import unittest
from unittest.mock import patch, MagicMock
# Add src directory to path
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src"))
from youtube_downloader import YouTubeDownloader, TrackInfo
from audio_converter import AudioConverter
from metadata_handler import MetadataHandler
from ipod_device import IPodDevice
class TestYouTubeDownloader(unittest.TestCase):
"""Test YouTube downloader functionality"""
def test_parse_artist_title(self):
"""Test parsing artist and title from video title"""
downloader = YouTubeDownloader()
# Test case 1: Artist - Title
artist, title = downloader._parse_artist_title("Rick Astley - Never Gonna Give You Up")
self.assertEqual(artist, "Rick Astley")
self.assertEqual(title, "Never Gonna Give You Up")
# Test case 2: Artist: Title
artist, title = downloader._parse_artist_title("Rick Astley: Never Gonna Give You Up")
self.assertEqual(artist, "Rick Astley")
self.assertEqual(title, "Never Gonna Give You Up")
# Test case 3: Artist | Title
artist, title = downloader._parse_artist_title("Rick Astley | Never Gonna Give You Up")
self.assertEqual(artist, "Rick Astley")
self.assertEqual(title, "Never Gonna Give You Up")
# Test case 4: No pattern match
artist, title = downloader._parse_artist_title("Never Gonna Give You Up")
self.assertEqual(artist, "")
self.assertEqual(title, "Never Gonna Give You Up")
class TestAudioConverter(unittest.TestCase):
"""Test audio converter functionality"""
def test_sanitize_filename(self):
"""Test filename sanitization"""
converter = AudioConverter()
# Test case 1: Invalid characters
sanitized = converter._sanitize_filename('File: with "invalid" chars?')
self.assertEqual(sanitized, 'File_ with _invalid_ chars_')
# Test case 2: Long filename
long_name = "A" * 150
sanitized = converter._sanitize_filename(long_name)
self.assertEqual(len(sanitized), 100)
@ -32,23 +65,26 @@ class TestAudioConverter(unittest.TestCase):
class TestMetadataHandler(unittest.TestCase):
"""Test metadata handler functionality"""
@patch('requests.get')
def test_download_thumbnail(self, mock_get):
"""Test thumbnail download"""
# Mock response
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.content = b'fake_image_data'
mock_get.return_value = mock_response
# Mock PIL Image
with patch('PIL.Image.open') as mock_image_open:
mock_image = MagicMock()
mock_image.save.return_value = None
mock_image_open.return_value = mock_image
handler = MetadataHandler()
result = handler.download_thumbnail("https://example.com/image.jpg")
# Check that the function returned a path
self.assertIsNotNone(result)
mock_get.assert_called_once_with("https://example.com/image.jpg", timeout=10)