Compare commits

...

5 Commits

Author SHA1 Message Date
Maksim Totmin
340b0dea9e Switch default output to MP3, fix AAC for iPod Nano 7G, auto-remove gray source rows after convert
- Separate AAC and MP3 ffmpeg paths: remove +global_header and +genpts from MP3
- Clamp AAC bitrate to 160 kbps (iPod Nano 7G max); warn on override
- Make mp3 the default format in UI, converter, and CLI
- Auto-remove converted source tracks from library UI (keep files on disk)
- Bulk delete_tracks with single scan+sync (progress support)
- Async duplicate removal with progress via WorkerThread
2026-05-31 22:44:02 +07:00
Maksim Totmin
63d72adae6 fix: move iPod track deletion to background thread with progress bar, batch DB sync 2026-05-31 22:43:59 +07:00
Maksim Totmin
3da5e15fe0 fix: add +global_header +genpts ffmpeg flags, preserve cover art streams, gapless metadata for iPod Nano 7 2026-05-31 22:13:47 +07:00
Maksim Totmin
71928d3478 feat: integrate artwork module into sync pipeline for Nano 7G 2026-05-31 21:45:51 +07:00
Maksim Totmin
9da689782a fix: clamp default audio bitrate to 160 kbps for iPod Nano 7 compatibility 2026-05-31 21:45:49 +07:00
4 changed files with 334 additions and 90 deletions

View File

@ -42,7 +42,7 @@ class AudioConverter:
def convert_to_ipod_format(self,
track_info: TrackInfo,
input_file: str,
format: str = "m4a",
format: str = "mp3",
audio_bitrate: int = 256,
overwrite: bool = False) -> str:
"""
@ -90,31 +90,38 @@ class AudioConverter:
logger.warning(f"Existing file is too small ({file_size} bytes), re-converting")
os.remove(output_file)
# Build ffmpeg command — iPod Nano 7 is strict about AAC parameters:
# - AAC-LC profile only (HE-AAC causes playback issues/crashes)
# - 44100 Hz sample rate (firmware expects this exactly)
# - Max 160 kbps for stereo (higher bitrates may truncate playback)
codec = "aac" if format == "m4a" else "libmp3lame"
cmd = [
"ffmpeg", "-y", "-i", input_file,
"-c:a", codec,
"-b:a", f"{audio_bitrate}k",
"-ar", "44100",
"-ac", "2",
]
# For M4A/AAC: force LC profile and use libfdk_aac if available (best quality),
# otherwise fall back to built-in aac with profile flag
if format == "m4a":
# iPod Nano 7 AAC requirements:
# - AAC-LC profile only (HE-AAC causes playback issues/crashes)
# - 44100 Hz sample rate (firmware expects this exactly)
# - Max 160 kbps for stereo (higher may cause truncation/crackle)
# - +global_header: required for iPod AAC decoder init
# - +genpts: fix timestamp gaps from web sources
m4a_bitrate = min(audio_bitrate, 160)
if audio_bitrate > 160:
logger.warning(
"Bitrate %d kbps exceeds iPod Nano 7 max (160 kbps), clamping to 160",
audio_bitrate,
)
cmd = [
"ffmpeg", "-y", "-fflags", "+genpts", "-i", input_file,
"-c:a", "aac",
"-b:a", f"{m4a_bitrate}k",
"-ar", "44100",
"-ac", "2",
"-flags", "+global_header",
]
# Try libfdk_aac first (best compatibility)
fdkaac_cmd = cmd + [
"-map", "0:a",
"-map", "0:v?",
"-c:v", "copy",
"-map", "0:t?",
"-c:a", "libfdk_aac",
"-profile:a", "aac_low",
"-movflags", "+faststart",
"-map_metadata", "0",
"-loglevel", "error",
"-loglevel", "warning",
output_file,
]
try:
@ -125,33 +132,44 @@ class AudioConverter:
if "libfdk_aac" in test.stdout:
cmd = fdkaac_cmd
else:
# Fallback: built-in aac with LC profile
cmd += [
"-map", "0:a",
"-map", "0:v?",
"-c:v", "copy",
"-map", "0:t?",
"-profile:a", "aac_low",
"-movflags", "+faststart",
"-map_metadata", "0",
"-loglevel", "error",
"-loglevel", "warning",
output_file,
]
except Exception:
cmd += [
"-map", "0:a",
"-map", "0:v?",
"-c:v", "copy",
"-map", "0:t?",
"-profile:a", "aac_low",
"-movflags", "+faststart",
"-map_metadata", "0",
"-loglevel", "error",
"-loglevel", "warning",
output_file,
]
else:
cmd += [
# MP3 for iPod Nano 7G: reliable, no special decoder flags needed.
# libmp3lame is mature and produces streams safe for hardware decoders.
# No +global_header — that flag is AAC-only and confuses MP3 playback.
cmd = [
"ffmpeg", "-y", "-i", input_file,
"-c:a", "libmp3lame",
"-b:a", f"{audio_bitrate}k",
"-ar", "44100",
"-ac", "2",
"-map", "0:a",
"-map", "0:t?",
"-id3v2_version", "3",
"-map_metadata", "0",
"-loglevel", "error",
"-loglevel", "warning",
output_file,
]

View File

@ -155,10 +155,10 @@ class YouTubeToIPodCLI:
def convert(self,
tracks: List[TrackInfo],
output_dir: str,
format: str = "m4a",
quality: int = 256,
video: bool = False,
resolution: str = "640x480") -> List[Tuple[TrackInfo, str]]:
format: str = "mp3",
quality: int = 256,
video: bool = False,
resolution: str = "640x480") -> List[Tuple[TrackInfo, str]]:
"""
Convert downloaded tracks to iPod-compatible format

View File

@ -29,12 +29,15 @@ import random
import shutil
import sqlite3
import struct
import subprocess
import time
from typing import Any, Optional
from mutagen import File as MutagenFile
from mutagen.easyid3 import EasyID3
from artwork import prepare_artwork_for_track, ArtworkEntry, write_artworkdb
_IOP_ROOT = "/tmp/iOpenPod"
_IOP_CACHE: dict[str, Any] = {}
_IOP_READY = False
@ -187,8 +190,41 @@ def _read_audio_tags(filepath: str) -> dict[str, Any]:
return info
def _probe_gapless(filepath: str, sample_rate: int, duration_ms: int) -> tuple[int, int, int]:
"""Probe AAC/MP3 file for encoder delay and padding samples.
Returns (pregap, postgap, sample_count).
"""
pregap = 0
postgap = 0
try:
result = subprocess.run(
["ffprobe", "-v", "error", "-select_streams", "a:0",
"-show_entries", "stream_tags=encoder_delay,padding",
"-of", "csv=p=0", filepath],
capture_output=True, text=True, timeout=10,
)
if result.returncode == 0 and result.stdout.strip():
parts = result.stdout.strip().split(",")
pregap = int(parts[0]) if parts[0] else 0
postgap = int(parts[1]) if len(parts) > 1 and parts[1] else 0
except Exception:
pass
# Fallback: typical AAC-LC encoding delay (1 frame = 1024 samples at 44100 Hz)
if pregap == 0 and sample_rate == 44100:
ext = os.path.splitext(filepath)[1].lower()
if ext in (".m4a", ".aac", ".mp4"):
pregap = 1024
total_samples = int(duration_ms * sample_rate / 1000) if duration_ms else 0
sample_count = max(0, total_samples - pregap - postgap)
return pregap, postgap, sample_count
def _scanned_tracks_to_iop(mountpoint: str,
scanned: list[dict]) -> list:
scanned: list[dict],
artwork_map: dict[str, tuple[int, int, int]] | None = None) -> list:
"""Convert our scanned track dicts to iOpenPod TrackInfo objects."""
IopTrackInfo = _import_iop("iTunesDB_Writer.mhit_writer").TrackInfo
tracks: list = []
@ -201,14 +237,26 @@ def _scanned_tracks_to_iop(mountpoint: str,
location = f":iPod_Control:Music:{rel.replace('/', ':')}"
size = os.path.getsize(fpath)
art_count = 0
art_size = 0
mhii_link = 0
if artwork_map and fpath in artwork_map:
mhii_link, art_count, art_size = artwork_map[fpath]
sr = s.get("sample_rate", 44100)
dur_ms = s.get("duration_ms", 0)
pregap, postgap, sample_count = _probe_gapless(fpath, sr, dur_ms)
has_album = bool(s.get("album"))
t = IopTrackInfo(
title=s.get("title") or base,
location=location,
size=size,
length=s.get("duration_ms", 0),
length=dur_ms,
filetype=ft,
bitrate=s.get("bitrate", 0) // 1000,
sample_rate=s.get("sample_rate", 44100),
sample_rate=sr,
artist=s.get("artist") or "",
album=s.get("album") or "",
album_artist=s.get("album_artist") or s.get("artist") or "",
@ -219,9 +267,14 @@ def _scanned_tracks_to_iop(mountpoint: str,
total_tracks=s.get("track_count", 0),
disc_number=s.get("disc_number", 0),
total_discs=s.get("disc_count", 0),
artwork_count=s.get("art_count", 0),
artwork_size=s.get("art_size", 0),
mhii_link=s.get("mhii_link", 0),
artwork_count=art_count,
artwork_size=art_size,
mhii_link=mhii_link,
pregap=pregap,
postgap=postgap,
sample_count=sample_count,
gapless_track_flag=1,
gapless_album_flag=1 if has_album else 0,
)
tracks.append(t)
return tracks
@ -383,24 +436,46 @@ class Nano7Database:
return tracks
def delete_track(self, pid: int) -> bool:
"""Delete track by pid. Scans for matching file, deletes, re-syncs."""
deleted = False
for s in self._scan_ipod_files():
if s["pid"] == pid:
if os.path.exists(s["file_path"]):
os.remove(s["file_path"])
logger.info("Deleted: %s", s["file_path"])
deleted = True
break
if deleted:
self.sync_itunescdb()
return True
# Try deleting by file path if pid lookup fails
# pid might be a negative hash; try all files
logger.warning("No file found for pid=%d", pid)
return False
"""Delete a single track by pid (scans, deletes, syncs)."""
return self.delete_tracks([pid]) > 0
def remove_duplicates(self) -> list[dict]:
def delete_tracks(self, pids: list[int], progress_callback=None) -> int:
"""Delete tracks by pid. Scans once, deletes all matching files,
then regenerates the database exactly once.
Args:
pids: List of track pids to delete.
progress_callback: Optional callable(percent, status) for progress.
Returns:
Number of files actually deleted.
"""
pid_set = set(pids)
scanned = self._scan_ipod_files()
deleted = 0
for i, s in enumerate(scanned):
if s["pid"] in pid_set and os.path.exists(s["file_path"]):
os.remove(s["file_path"])
logger.info("Deleted: %s", s["file_path"])
deleted += 1
pid_set.discard(s["pid"])
if progress_callback and i % 5 == 0:
progress_callback(
min(50, int((i + 1) / max(len(scanned), 1) * 50)),
f"Deleting files... {deleted} removed",
)
if deleted:
if progress_callback:
progress_callback(55, "Regenerating iPod database...")
self.sync_itunescdb()
if progress_callback:
progress_callback(100, f"Removed {deleted} track(s)")
return deleted
def remove_duplicates(self, progress_callback=None) -> list[dict]:
"""Find and remove duplicate tracks (same title+artist+album)."""
all_tracks = self._scan_ipod_files()
seen: dict[tuple, list[dict]] = {}
@ -414,7 +489,6 @@ class Nano7Database:
for key, items in seen.items():
if len(items) <= 1:
continue
# Keep first, delete rest
for item in items[1:]:
fpath = item["file_path"]
if os.path.exists(fpath):
@ -427,9 +501,19 @@ class Nano7Database:
"file_path": fpath,
})
logger.info("Removed duplicate: %s", fpath)
if progress_callback:
progress_callback(
min(50, int(len(removed) / max(len(all_tracks), 1) * 50)),
f"Removing duplicates... {len(removed)} found",
)
if removed:
if progress_callback:
progress_callback(55, "Regenerating iPod database...")
self.sync_itunescdb()
if progress_callback:
progress_callback(100, f"Removed {len(removed)} duplicate(s)")
logger.info("Removed %d duplicate(s)", len(removed))
return removed
@ -487,11 +571,58 @@ class Nano7Database:
self._write_empty_databases()
return
# 2. Convert to iOpenPod TrackInfo
tracks = _scanned_tracks_to_iop(self.mountpoint, scanned)
# 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] = {}
next_img_id = 100
for idx, s in enumerate(scanned):
fpath = s["file_path"]
try:
art_result = prepare_artwork_for_track(fpath)
except Exception:
art_result = None
if art_result is not None:
art_hash, formats = art_result
art_count = len(formats)
total_size = sum(p.size for p in formats.values())
if art_hash not in art_hash_to_img_id:
img_id = next_img_id
next_img_id += 1
art_hash_to_img_id[art_hash] = img_id
art_entries.append(ArtworkEntry(
img_id=img_id,
db_track_id=idx,
art_hash=art_hash,
src_img_size=total_size,
formats=formats,
db_track_ids=[idx],
))
else:
img_id = art_hash_to_img_id[art_hash]
for entry in art_entries:
if entry.img_id == img_id:
if idx not in entry.db_track_ids:
entry.db_track_ids.append(idx)
break
artwork_map[fpath] = (img_id, art_count, total_size)
if art_entries:
try:
write_artworkdb(self.mountpoint, art_entries)
logger.info("ArtworkDB written: %d unique covers, %d total tracks",
len(art_entries), len(artwork_map))
except Exception as e:
logger.warning("ArtworkDB write failed: %s", e)
# 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:
@ -505,7 +636,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:
@ -525,7 +656,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")
@ -539,7 +670,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(

View File

@ -55,6 +55,8 @@ class WorkerThread(QThread):
"transfer": self._run_transfer,
"detect_devices": self._run_detect_devices,
"export_ipod": self._run_export_ipod,
"delete_tracks": self._run_delete_tracks,
"remove_duplicates": self._run_remove_duplicates,
}
handler = handlers.get(self.task_type)
if handler:
@ -92,7 +94,7 @@ class WorkerThread(QThread):
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", "m4a")
fmt = self.kwargs.get("format", "mp3")
quality = self.kwargs.get("quality", 256)
valid_tracks = [
@ -223,6 +225,40 @@ class WorkerThread(QThread):
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 stop(self):
self.is_running = False
self.wait()
@ -311,7 +347,7 @@ class MainWindow(QMainWindow):
format_layout = QHBoxLayout()
format_label = QLabel("Format:")
self.format_combo = QComboBox()
self.format_combo.addItems(["m4a", "mp3"])
self.format_combo.addItems(["mp3", "m4a"])
quality_label = QLabel("Quality (kbps):")
self.quality_spin = QSpinBox()
self.quality_spin.setRange(128, 320)
@ -555,6 +591,15 @@ class MainWindow(QMainWindow):
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 _setup_settings_tab(self, tab):
layout = QVBoxLayout(tab)
@ -1256,6 +1301,22 @@ class MainWindow(QMainWindow):
if success:
self.library_status.setText(message)
self._scan_library()
if result:
converted_source_paths = set()
for track, _output_path in result:
src = getattr(track, "download_path", None)
if src:
converted_source_paths.add(os.path.normpath(src))
if converted_source_paths:
self.library_scanned_sources = [
s for s in self.library_scanned_sources
if os.path.normpath(s.get("path", "")) not in converted_source_paths
]
self.library_source_files = [
p for p in self.library_source_files
if os.path.normpath(p) not in converted_source_paths
]
self._refresh_library_ui()
else:
self.library_status.setText(f"Error: {message}")
QMessageBox.warning(self, "Conversion Error", message)
@ -1508,23 +1569,52 @@ class MainWindow(QMainWindow):
if reply != QMessageBox.StandardButton.Yes:
return
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
removed = 0
for item in selected:
track_data = item.data(Qt.ItemDataRole.UserRole)
if track_data and "pid" in track_data:
success = db.delete_track(track_data["pid"])
if success:
removed += 1
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"])
self._load_ipod_tracks()
QMessageBox.information(self, "Done", f"Successfully removed {removed} track(s) from iPod")
except Exception as e:
logger.error(f"Failed to remove tracks: {e}")
QMessageBox.warning(self, "Error", f"Failed to remove tracks: {e}")
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:
@ -1629,26 +1719,31 @@ class MainWindow(QMainWindow):
QMessageBox.warning(self, "Error", "No iPod device mounted.")
return
try:
from ipod_nano7_db import Nano7Database
db = Nano7Database(self.current_mount_point)
removed = db.remove_duplicates()
if self.worker_thread and self.worker_thread.isRunning():
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
return
if not removed:
QMessageBox.information(self, "Scan Complete", "No duplicate tracks found.")
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...")
details = f"Found and removed {len(removed)} duplicate entry(ies):\n\n"
for r in removed[:15]:
details += f" {r['artist']} \u2014 {r['title']} ({r['album']})\n"
if len(removed) > 15:
details += f" ... and {len(removed) - 15} more\n"
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()
QMessageBox.information(self, "Duplicates Removed", details)
self._load_ipod_tracks()
except Exception as e:
logger.error(f"Failed to remove duplicates: {e}")
QMessageBox.warning(self, "Error", f"Failed to remove duplicates: {e}")
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 closeEvent(self, event):
if self.worker_thread and self.worker_thread.isRunning():