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

Add iPod Nano 7 SQLite database support for proper track transfer

Browse Source 
- Create ipod_nano7_db.py module for Nano 7 SQLite DB management (Library.itdb)
- Fix case-insensitive iPod label detection in device detection
- Add /run/media to mount point search paths for Arch Linux
- Support already-mounted devices without requiring ifuse
- Add local file selection in Convert tab (FLAC, WAV, OGG, etc.)
- Transfer now properly registers tracks in iPod database so they appear in UI
This commit is contained in:
Maksim Totmin 2026-05-31 12:07:17 +07:00
parent 436898987b
commit 761f896ea8
3 changed files with 522 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
src/ipod_device.py
View File

@ -228,7 +228,7 @@ class IPodDevice:
for child in device["children"]:
label = child.get("label", "")
mountpoint = child.get("mountpoint")
if mountpoint and label and "iPod" in label:
if mountpoint and label and "ipod" in label.lower():
devices.append({
"id": child.get("name", "unknown"),
"name": label,
@ -242,13 +242,13 @@ class IPodDevice:
# Fallback to checking /media and /mnt directories
if not devices:
for mount_dir in ["/media", "/mnt"]:
for mount_dir in ["/media", "/mnt", "/run/media"]:
if os.path.exists(mount_dir):
for user_dir in os.listdir(mount_dir):
user_path = os.path.join(mount_dir, user_dir)
if os.path.isdir(user_path):
for device_dir in os.listdir(user_path):
if "iPod" in device_dir:
if "ipod" in device_dir.lower():
device_path = os.path.join(user_path, device_dir)
if os.path.isdir(device_path):
devices.append({
@ -285,22 +285,44 @@ class IPodDevice:
# You may need to implement it according to your requirements
pass
def mount_device(self, device_id: Optional[str] = None) -> Optional[str]:
def mount_device(self, device_id: Optional[str] = None, detected_mount: Optional[str] = None) -> Optional[str]:
"""
Mount iPod device
Args:
device_id: Device ID (optional)
detected_mount: Already detected mount point (optional)
Returns:
Mount point if successful, None otherwise
"""
# If device is already mounted at detected path, just use it
if detected_mount and os.path.exists(detected_mount):
# Verify it looks like an iPod
if os.path.exists(os.path.join(detected_mount, "iPod_Control")):
self.mount_point = detected_mount
self.logger.info(f"Using already mounted device at: {detected_mount}")
return detected_mount
if self.mount_point and os.path.exists(self.mount_point):
self.logger.info(f"Device already mounted at: {self.mount_point}")
return self.mount_point
try:
if self.system == "Linux":
# Check common mount points for already-mounted USB storage
for base in ["/run/media", "/media", "/mnt"]:
if os.path.exists(base):
for user_dir in os.listdir(base):
user_path = os.path.join(base, user_dir)
if os.path.isdir(user_path):
if os.path.exists(os.path.join(user_path, "iPod_Control")):
self.mount_point = user_path
self.logger.info(f"Found already mounted iPod at: {user_path}")
return user_path
# Try to mount with ifuse if available
if shutil.which("ifuse"):
# Create mount point
mount_point = "/tmp/ipod_mount"
os.makedirs(mount_point, exist_ok=True)
@ -315,6 +337,9 @@ class IPodDevice:
self.mount_point = mount_point
self.logger.info(f"Device mounted at: {mount_point}")
return mount_point
else:
self.logger.error("ifuse not installed and device not already mounted")
return None
elif self.system == "Darwin":
# On macOS, devices are automatically mounted
@ -410,18 +435,23 @@ class IPodDevice:
info["free_space"] = usage.free
info["used_space"] = usage.used
# Check for iPod directory structure
# Try to get track count from Nano7 database
try:
from ipod_nano7_db import Nano7Database
nano7_db = Nano7Database(self.mount_point)
info["track_count"] = nano7_db.get_track_count()
info["has_nano7_db"] = True
except Exception:
info["has_nano7_db"] = False
# Fallback: count files in Music directory
music_dir = os.path.join(self.mount_point, "Music")
if os.path.exists(music_dir):
info["has_music_dir"] = True
# Count music files
music_files = []
for root, _, files in os.walk(music_dir):
for file in files:
if file.lower().endswith(('.m4a', '.mp3', '.aac')):
music_files.append(os.path.join(root, file))
info["music_file_count"] = len(music_files)
else:
info["has_music_dir"] = False
@ -452,24 +482,30 @@ class IPodDevice:
return False
try:
# Determine destination path
# Try to use Nano7 database for proper track registration
try:
from ipod_nano7_db import Nano7Database
nano7_db = Nano7Database(self.mount_point)
result = nano7_db.add_track(source_file)
self.logger.info(f"Track added to iPod database: {result['title']}")
return True
except ImportError:
self.logger.warning("ipod_nano7_db not available, using simple file copy")
except Exception as e:
self.logger.error(f"Nano7 database transfer failed: {e}, falling back to simple copy")
# Fallback: simple file copy (won't show in iPod interface)
if relative_path:
dest_dir = os.path.join(self.mount_point, relative_path)
else:
# Default to Music directory
dest_dir = os.path.join(self.mount_point, "Music")
# Extract artist and album from source path
parts = Path(source_file).parts
if len(parts) >= 3:
artist = parts[-3]
album = parts[-2]
dest_dir = os.path.join(self.mount_point, "Music", artist, album)
# Create destination directory if it doesn't exist
os.makedirs(dest_dir, exist_ok=True)
# Copy file
filename = os.path.basename(source_file)
dest_file = os.path.join(dest_dir, filename)

370
src/ipod_nano7_db.py Normal file
View File

@ -0,0 +1,370 @@
#!/usr/bin/env python3
"""
iPod Nano 7 Database Manager
Handles SQLite database and iTunesCDB for iPod Nano 7 (firmware 2.x+)
"""
import os
import struct
import time
import random
import zlib
import sqlite3
import logging
import shutil
from typing import Optional, Dict, Any
from mutagen import File as MutagenFile
from mutagen.mp4 import MP4
from mutagen.mp3 import MP3
from mutagen.id3 import ID3
logger = logging.getLogger(__name__)
MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist
class Nano7Database:
"""Manages iPod Nano 7 SQLite database and iTunesCDB"""
def __init__(self, mountpoint: str):
self.mountpoint = mountpoint
self.sqlite_path = os.path.join(
mountpoint, "iPod_Control", "iTunes", "iTunes Library.itlp", "Library.itdb"
)
self.itunes_dir = os.path.join(mountpoint, "iPod_Control", "iTunes")
self.cdb_path = os.path.join(self.itunes_dir, "iTunesCDB")
self.cdb_ext_path = os.path.join(self.itunes_dir, "iTunesCDB.ext")
self.music_base = os.path.join(mountpoint, "iPod_Control", "Music")
def get_next_pid(self) -> int:
"""Generate a unique negative pid (like existing tracks)"""
return random.randint(-(2**63), -1)
def find_free_music_dir(self) -> str:
"""Find a FXX directory with the least files"""
dirs = sorted(os.listdir(self.music_base))
min_files = float("inf")
target = None
for d in dirs:
full = os.path.join(self.music_base, d)
if os.path.isdir(full):
count = len(os.listdir(full))
if count < min_files:
min_files = count
target = d
return target or "F00"
def copy_track_to_ipod(self, filepath: str) -> str:
"""Copy track to iPod_Control/Music/FXX/ and return iPod path"""
filename = os.path.basename(filepath)
base, ext = os.path.splitext(filename)
# Generate unique name like pygpod does
prefix = "pygpod" if "pygpod" in filename else ""
if not prefix:
prefix = "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ", k=4))
rand_num = random.randint(0, 999999)
new_name = f"{prefix}{rand_num:06d}{ext}"
target_dir = self.find_free_music_dir()
target_path = os.path.join(self.music_base, target_dir, new_name)
shutil.copy2(filepath, target_path)
# Return iPod path format (colon-separated)
ipod_path = f":iPod_Control:Music:{target_dir}:{new_name}"
logger.info(f"Copied track to {ipod_path}")
return ipod_path
def get_audio_info(self, filepath: str) -> Dict[str, Any]:
"""Extract audio information from file"""
info = {}
audio = MutagenFile(filepath)
if audio is None:
raise ValueError(f"Cannot read audio file: {filepath}")
# Duration in ms
if hasattr(audio, "info") and audio.info.length:
info["duration_ms"] = int(audio.info.length * 1000)
else:
info["duration_ms"] = 0
# Bitrate
if hasattr(audio.info, "bitrate"):
info["bitrate"] = audio.info.bitrate // 1000
else:
info["bitrate"] = 256
# Channels
if hasattr(audio.info, "channels"):
info["channels"] = audio.info.channels or 2
else:
info["channels"] = 2
# Sample rate
if hasattr(audio.info, "sample_rate"):
info["sample_rate"] = audio.info.sample_rate
else:
info["sample_rate"] = 44100
# Audio format code (502=AAC, 1=MP3)
ext = os.path.splitext(filepath)[1].lower()
if ext in (".m4a", ".aac", ".mp4"):
info["audio_format"] = 502 # AAC
elif ext == ".mp3":
info["audio_format"] = 1 # MP3
else:
info["audio_format"] = 1
# Extract tags
for tag_name, file_key in [
("title", "title"),
("artist", "artist"),
("album", "album"),
("albumartist", "album_artist"),
("composer", "composer"),
("genre", "genre"),
("comment", "comment"),
]:
if tag_name in audio:
val = audio[tag_name]
if isinstance(val, list):
val = val[0]
if isinstance(val, str):
info[file_key] = val
# Track/disc numbers
for tag_name, file_key in [
("tracknumber", "track_number"),
("discnumber", "disc_number"),
]:
if tag_name in audio:
val = audio[tag_name]
if isinstance(val, list):
val = val[0]
if isinstance(val, str):
parts = val.split("/")
info[file_key] = int(parts[0]) if parts[0].isdigit() else 0
if len(parts) > 1:
info[file_key + "_count"] = (
int(parts[1]) if parts[1].isdigit() else 0
)
elif isinstance(val, int):
info[file_key] = val
# Year
if "date" in audio:
val = audio["date"]
if isinstance(val, list):
val = val[0]
if isinstance(val, str) and len(val) >= 4:
info["year"] = int(val[:4])
elif isinstance(val, int):
info["year"] = val
return info
def add_track(self, filepath: str, **kwargs) -> Dict[str, Any]:
"""Add a track to the iPod Nano 7 database"""
if not os.path.exists(filepath):
raise FileNotFoundError(f"File not found: {filepath}")
# Copy file to iPod
ipod_path = self.copy_track_to_ipod(filepath)
# Get audio info
audio_info = self.get_audio_info(filepath)
# Apply overrides
for key, value in kwargs.items():
audio_info[key] = value
# Generate pid
pid = self.get_next_pid()
# Get current max physical_order
conn = sqlite3.connect(self.sqlite_path)
cursor = conn.cursor()
cursor.execute("SELECT MAX(physical_order) FROM item")
max_order = cursor.fetchone()[0] or 0
physical_order = max_order + 1
# Get current timestamp (mac epoch: seconds since 1904)
mac_epoch = int(time.time()) + 2082844800
# Get or create artist/album entries
artist = audio_info.get("artist", "Unknown Artist")
album = audio_info.get("album", "Unknown Album")
title = audio_info.get("title", os.path.splitext(os.path.basename(filepath))[0])
# Check if artist exists
cursor.execute("SELECT pid FROM artist WHERE name = ?", (artist,))
row = cursor.fetchone()
if row:
artist_pid = row[0]
else:
artist_pid = self.get_next_pid()
cursor.execute(
"""
INSERT INTO artist (
pid, kind, artwork_status, artwork_album_pid, name,
name_order, sort_name, is_unknown, has_songs,
has_music_videos, has_non_compilation_tracks, album_count
) VALUES (?, 2, 0, 0, ?, 100, ?, 0, 1, 0, 1, 1)
""",
(artist_pid, artist, artist),
)
# Check if album exists
cursor.execute("SELECT pid FROM album WHERE name = ?", (album,))
row = cursor.fetchone()
if row:
album_pid = row[0]
else:
album_pid = self.get_next_pid()
cursor.execute(
"""
INSERT INTO album (
pid, kind, artwork_status, artwork_item_pid, artist_pid,
user_rating, name, name_order, all_compilations, feed_url,
season_number, is_unknown, has_songs, has_music_videos,
sort_order, artist_order, has_any_compilations, sort_name,
artist_count_calc, has_movies, item_count, min_volume_normalization_energy
) VALUES (?, 2, 0, 0, ?, 0, ?, 100, 0, NULL, 0, 0, 1, 0, 100, 100, 0, ?, 1, 0, 1, 0)
""",
(album_pid, artist_pid, album, album),
)
# Insert into item table
cursor.execute(
"""
INSERT INTO item (
pid, media_kind, is_song, date_modified, year,
is_compilation, is_user_disabled, remember_bookmark,
exclude_from_shuffle, part_of_gapless_album, chosen_by_auto_fill,
artwork_status, artwork_cache_id,
total_time_ms, track_number, track_count,
disc_number, disc_count,
album_pid, artist_pid,
title, artist, album, album_artist,
sort_title, sort_artist, sort_album, sort_album_artist,
title_order, artist_order, album_order,
physical_order
) VALUES (
?, 1, 1, ?, ?, 0, 0, 0, 0, 0, 0, 1, 0,
?, ?, ?, ?, ?,
?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
1000, 1000, 1000,
?
)
""",
(
pid,
mac_epoch,
audio_info.get("year", 0),
audio_info["duration_ms"],
audio_info.get("track_number", 0),
audio_info.get("track_count", 0),
audio_info.get("disc_number", 0),
audio_info.get("disc_count", 0),
album_pid,
artist_pid,
title,
artist,
album,
audio_info.get("album_artist", artist),
title,
artist,
album,
audio_info.get("album_artist", artist),
physical_order,
),
)
# Insert into avformat_info
cursor.execute(
"""
INSERT INTO avformat_info (
item_pid, sub_id, audio_format, bit_rate, channels,
sample_rate, duration, gapless_heuristic_info,
gapless_encoding_delay, gapless_encoding_drain,
gapless_last_frame_resynch, analysis_inhibit_flags,
audio_fingerprint, volume_normalization_energy
) VALUES (?, 0, ?, ?, ?, ?, ?, 1, 2112, 316, 0, 0, 0, 0)
""",
(
pid,
audio_info["audio_format"],
audio_info["bitrate"],
audio_info["channels"],
audio_info["sample_rate"],
audio_info["duration_ms"],
),
)
# Insert into store_info
cursor.execute(
"""
INSERT INTO store_info (item_pid) VALUES (?)
""",
(pid,),
)
# Add to master playlist (iPod container)
cursor.execute(
"""
INSERT INTO item_to_container (item_pid, container_pid, physical_order, shuffle_order)
VALUES (?, ?, ?, NULL)
""",
(pid, MASTER_CONTAINER_PID, physical_order),
)
# Add to Music container (pid 4872745547565090077)
music_container_pid = 4872745547565090077
cursor.execute(
"""
INSERT INTO item_to_container (item_pid, container_pid, physical_order, shuffle_order)
VALUES (?, ?, ?, NULL)
""",
(pid, music_container_pid, physical_order),
)
# Add to artist album container (Music container)
conn.commit()
conn.close()
result = {
"pid": pid,
"title": title,
"artist": artist,
"album": album,
"ipod_path": ipod_path,
"physical_order": physical_order,
}
logger.info(f"Added track: {title} by {artist}")
return result
def get_track_count(self) -> int:
"""Get number of tracks in database"""
conn = sqlite3.connect(self.sqlite_path)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM item")
count = cursor.fetchone()[0]
conn.close()
return count
def get_tracks(self, limit: int = 10) -> list:
"""Get recent tracks from database"""
conn = sqlite3.connect(self.sqlite_path)
cursor = conn.cursor()
cursor.execute(
"SELECT pid, title, artist, album, physical_order FROM item ORDER BY physical_order DESC LIMIT ?",
(limit,),
)
tracks = cursor.fetchall()
conn.close()
return tracks

87
src/main.py
View File

@ -100,6 +100,7 @@ class WorkerThread(QThread):
def _run_convert(self):
"""Run conversion task"""
tracks = self.kwargs.get("tracks", [])
local_files = self.kwargs.get("local_files", [])
output_dir = self.kwargs.get("output_dir", "converted")
format = self.kwargs.get("format", "m4a")
quality = self.kwargs.get("quality", 256)
@ -107,6 +108,28 @@ class WorkerThread(QThread):
# Filter out tracks without download paths
valid_tracks = [track for track in tracks if hasattr(track, 'download_path') and track.download_path and os.path.exists(track.download_path)]
# Process local files into TrackInfo objects
for f in local_files:
if os.path.exists(f):
filename = os.path.basename(f)
title = os.path.splitext(filename)[0]
# Try to parse "Artist - Title" format
if " - " in title:
artist, track_title = title.split(" - ", 1)
else:
artist = "Unknown"
track_title = title
track = TrackInfo(
video_id=f"local_{hash(f)}",
title=track_title,
artist=artist,
album="",
thumbnail_url="",
duration=0,
download_path=f
)
valid_tracks.append(track)
if not valid_tracks:
self.finished_signal.emit(False, "No valid tracks to convert", None)
return
@ -357,8 +380,25 @@ class MainWindow(QMainWindow):
output_layout.addWidget(browse_button)
layout.addLayout(output_layout)
# Local file selection
file_layout = QHBoxLayout()
select_files_btn = QPushButton("Select Local Files...")
select_files_btn.clicked.connect(self._on_select_files_clicked)
remove_files_btn = QPushButton("Remove Selected")
remove_files_btn.clicked.connect(self._on_remove_files_clicked)
file_layout.addWidget(select_files_btn)
file_layout.addWidget(remove_files_btn)
layout.addLayout(file_layout)
# Source files list
source_label = QLabel("Files to Convert (FLAC, WAV, OGG, etc.):")
layout.addWidget(source_label)
self.source_files_list = QListWidget()
self.source_files_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
layout.addWidget(self.source_files_list)
# Convert button
self.convert_button = QPushButton("Convert Downloaded Tracks")
self.convert_button = QPushButton("Convert")
self.convert_button.clicked.connect(self._on_convert_clicked)
layout.addWidget(self.convert_button)
@ -494,6 +534,24 @@ class MainWindow(QMainWindow):
if directory:
self.output_dir_input.setText(directory)
def _on_select_files_clicked(self):
"""Handle select local files button click"""
audio_filter = "Audio Files (*.flac *.wav *.ogg *.opus *.wma *.ape *.aiff *.alac)"
files, _ = QFileDialog.getOpenFileNames(
self, "Select Audio Files", "", audio_filter
)
for f in files:
# Avoid duplicates
items = self.source_files_list.findItems(f, Qt.MatchFlag.MatchExactly)
if not items:
self.source_files_list.addItem(f)
def _on_remove_files_clicked(self):
"""Handle remove selected files button click"""
for item in self.source_files_list.selectedItems():
self.source_files_list.takeItem(self.source_files_list.row(item))
def _on_download_clicked(self):
"""Handle download button click"""
url = self.url_input.text().strip()
@ -564,8 +622,12 @@ class MainWindow(QMainWindow):
def _on_convert_clicked(self):
"""Handle convert button click"""
if not self.downloaded_tracks:
QMessageBox.warning(self, "Error", "No tracks to convert. Download tracks first.")
local_files = [self.source_files_list.item(i).text() for i in range(self.source_files_list.count())]
has_downloaded = bool(self.downloaded_tracks)
has_local = bool(local_files)
if not has_downloaded and not has_local:
QMessageBox.warning(self, "Error", "No tracks to convert. Download tracks first or select local audio files.")
return
output_dir = self.output_dir_input.text().strip()
@ -574,6 +636,15 @@ class MainWindow(QMainWindow):
QMessageBox.warning(self, "Error", "Please enter an output directory")
return
task_type = "convert"
kwargs = {
"tracks": self.downloaded_tracks if has_downloaded else [],
"local_files": local_files,
"output_dir": output_dir,
"format": self.format_combo.currentText(),
"quality": self.quality_spin.value()
}
# Disable convert button
self.convert_button.setEnabled(False)
self.convert_status.setText("Converting...")
@ -584,11 +655,8 @@ class MainWindow(QMainWindow):
# Start worker thread
self.worker_thread = WorkerThread(
task_type="convert",
tracks=self.downloaded_tracks,
output_dir=output_dir,
format=self.format_combo.currentText(),
quality=self.quality_spin.value()
task_type=task_type,
**kwargs
)
self.worker_thread.progress_signal.connect(self._on_convert_progress)
@ -650,8 +718,9 @@ class MainWindow(QMainWindow):
# Try to mount the first device
if self.ipod_devices:
device_id = self.ipod_devices[0]["id"]
detected_mount = self.ipod_devices[0].get("mount_point")
ipod = IPodDevice()
mount_point = ipod.mount_device(device_id)
mount_point = ipod.mount_device(device_id, detected_mount=detected_mount)
if mount_point:
# Get device info