diff --git a/src/ipod_device.py b/src/ipod_device.py index 90e3072..2affcd9 100644 --- a/src/ipod_device.py +++ b/src/ipod_device.py @@ -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,36 +285,61 @@ 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": - # Create mount point - mount_point = "/tmp/ipod_mount" - os.makedirs(mount_point, exist_ok=True) + # 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 - # Mount with ifuse - cmd = ["ifuse", mount_point] - if device_id: - cmd.extend(["-u", device_id]) + # 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) - subprocess.run(cmd, check=True) - - self.mount_point = mount_point - self.logger.info(f"Device mounted at: {mount_point}") - return mount_point + # Mount with ifuse + cmd = ["ifuse", mount_point] + if device_id: + cmd.extend(["-u", device_id]) + + subprocess.run(cmd, check=True) + + 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,21 +435,26 @@ class IPodDevice: info["free_space"] = usage.free info["used_space"] = usage.used - # Check for iPod directory structure - 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 + # 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 + 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 return info @@ -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) diff --git a/src/ipod_nano7_db.py b/src/ipod_nano7_db.py new file mode 100644 index 0000000..41ac161 --- /dev/null +++ b/src/ipod_nano7_db.py @@ -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 diff --git a/src/main.py b/src/main.py index c8f06b1..fdcff1a 100644 --- a/src/main.py +++ b/src/main.py @@ -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