Add track management: view and delete tracks on iPod from Transfer tab
- Add get_all_tracks() to load all tracks with file paths via tag matching - Add delete_track() to remove tracks from SQLite DB and delete physical files - Add _cleanup_orphaned_entries() to remove orphaned artist/album records - Add track count label and 'Remove Selected' button to Transfer tab - Load existing tracks on device detection - Confirmation dialog before deleting tracks - Reload track list after successful transfer
This commit is contained in:
Maksim Totmin
parent
1b4c092a04
commit
47bfdf0011
@ -435,3 +435,264 @@ class Nano7Database:
|
||||
tracks = cursor.fetchall()
|
||||
conn.close()
|
||||
return tracks
|
||||
|
||||
def get_all_tracks(self) -> list:
|
||||
"""Get all tracks with full metadata and file paths."""
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT pid, title, artist, album, album_artist,
|
||||
total_time_ms, track_number, disc_number,
|
||||
physical_order, date_modified, year
|
||||
FROM item
|
||||
ORDER BY physical_order
|
||||
"""
|
||||
)
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
|
||||
tracks = []
|
||||
for row in rows:
|
||||
track = {
|
||||
"pid": row["pid"],
|
||||
"title": row["title"] or "Unknown",
|
||||
"artist": row["artist"] or "Unknown Artist",
|
||||
"album": row["album"] or "",
|
||||
"album_artist": row["album_artist"] or "",
|
||||
"duration_ms": row["total_time_ms"] or 0,
|
||||
"track_number": row["track_number"] or 0,
|
||||
"disc_number": row["disc_number"] or 0,
|
||||
"physical_order": row["physical_order"] or 0,
|
||||
"date_modified": row["date_modified"] or 0,
|
||||
"year": row["year"] or 0,
|
||||
"ipod_path": None,
|
||||
"file_path": None,
|
||||
}
|
||||
tracks.append(track)
|
||||
|
||||
# Resolve physical file paths by scanning Music/FXX/ directories
|
||||
# and matching tags to tracks
|
||||
track_lookup = {}
|
||||
for t in tracks:
|
||||
key = (t["title"].lower(), t["artist"].lower(), t["duration_ms"])
|
||||
track_lookup[key] = t
|
||||
|
||||
music_base = os.path.join(self.mountpoint, "iPod_Control", "Music")
|
||||
if os.path.exists(music_base):
|
||||
for folder in sorted(os.listdir(music_base)):
|
||||
folder_path = os.path.join(music_base, folder)
|
||||
if not os.path.isdir(folder_path):
|
||||
continue
|
||||
for fname in os.listdir(folder_path):
|
||||
fpath = os.path.join(folder_path, fname)
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
ext = os.path.splitext(fname)[1].lower()
|
||||
if ext not in (".mp3", ".m4a", ".aac", ".mp4"):
|
||||
continue
|
||||
|
||||
# Try to match by tags
|
||||
try:
|
||||
audio = MutagenFile(fpath)
|
||||
if audio is None or audio.tags is None:
|
||||
continue
|
||||
|
||||
# Get title
|
||||
file_title = ""
|
||||
file_artist = ""
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
easy = EasyID3(fpath)
|
||||
if "title" in easy:
|
||||
file_title = str(easy["title"][0])
|
||||
if "artist" in easy:
|
||||
file_artist = str(easy["artist"][0])
|
||||
except Exception:
|
||||
pass
|
||||
elif ext in (".m4a", ".aac", ".mp4"):
|
||||
from mutagen.mp4 import MP4
|
||||
tags = audio.tags
|
||||
if "\xa9nam" in tags:
|
||||
file_title = str(tags["\xa9nam"][0])
|
||||
if "\xa9ART" in tags:
|
||||
file_artist = str(tags["\xa9ART"][0])
|
||||
else:
|
||||
tags = audio.tags
|
||||
try:
|
||||
if "title" in tags:
|
||||
val = tags["title"]
|
||||
file_title = str(val[0] if isinstance(val, list) else val)
|
||||
if "artist" in tags:
|
||||
val = tags["artist"]
|
||||
file_artist = str(val[0] if isinstance(val, list) else val)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
duration_ms = int(audio.info.length * 1000) if hasattr(audio, "info") and audio.info.length else 0
|
||||
|
||||
key = (file_title.lower(), file_artist.lower(), duration_ms)
|
||||
if key in track_lookup:
|
||||
track = track_lookup[key]
|
||||
track["ipod_path"] = f":iPod_Control:Music:{folder}:{fname}"
|
||||
track["file_path"] = fpath
|
||||
del track_lookup[key]
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# For unmatched tracks, try to find file by ipod_path pattern
|
||||
for t in tracks:
|
||||
if t["file_path"]:
|
||||
continue
|
||||
# Try to find by scanning for any audio file in FXX dirs
|
||||
for folder in sorted(os.listdir(music_base)):
|
||||
folder_path = os.path.join(music_base, folder)
|
||||
if not os.path.isdir(folder_path):
|
||||
continue
|
||||
for fname in os.listdir(folder_path):
|
||||
fpath = os.path.join(folder_path, fname)
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
ext = os.path.splitext(fname)[1].lower()
|
||||
if ext in (".mp3", ".m4a", ".aac", ".mp4"):
|
||||
t["ipod_path"] = f":iPod_Control:Music:{folder}:{fname}"
|
||||
t["file_path"] = fpath
|
||||
break
|
||||
if t["file_path"]:
|
||||
break
|
||||
|
||||
return tracks
|
||||
|
||||
def delete_track(self, pid: int) -> bool:
|
||||
"""Delete a track from the iPod database and remove its physical file."""
|
||||
try:
|
||||
conn = sqlite3.connect(self.sqlite_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Get the track info first (for file path)
|
||||
cursor.execute(
|
||||
"SELECT title, artist, album FROM item WHERE pid = ?",
|
||||
(pid,),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row is None:
|
||||
conn.close()
|
||||
logger.warning(f"Track pid={pid} not found in database")
|
||||
return False
|
||||
|
||||
title, artist, album = row
|
||||
|
||||
# Find the physical file by matching title+artist in tags
|
||||
file_path = self._find_track_file(pid, title, artist)
|
||||
|
||||
# Remove from item_to_container
|
||||
cursor.execute("DELETE FROM item_to_container WHERE item_pid = ?", (pid,))
|
||||
|
||||
# Remove from avformat_info
|
||||
cursor.execute("DELETE FROM avformat_info WHERE item_pid = ?", (pid,))
|
||||
|
||||
# Remove from store_info
|
||||
cursor.execute("DELETE FROM store_info WHERE item_pid = ?", (pid,))
|
||||
|
||||
# Remove from item
|
||||
cursor.execute("DELETE FROM item WHERE pid = ?", (pid,))
|
||||
|
||||
# Clean up orphaned artist/album entries
|
||||
self._cleanup_orphaned_entries(cursor)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# Remove physical file
|
||||
if file_path and os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
logger.info(f"Deleted physical file: {file_path}")
|
||||
|
||||
logger.info(f"Deleted track pid={pid}: {artist} - {title}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete track pid={pid}: {e}")
|
||||
return False
|
||||
|
||||
def _find_track_file(self, pid: int, title: str, artist: str) -> Optional[str]:
|
||||
"""Find the physical file for a track by scanning Music/FXX/ directories."""
|
||||
music_base = os.path.join(self.mountpoint, "iPod_Control", "Music")
|
||||
if not os.path.exists(music_base):
|
||||
return None
|
||||
|
||||
title_lower = (title or "").lower()
|
||||
artist_lower = (artist or "").lower()
|
||||
|
||||
for folder in sorted(os.listdir(music_base)):
|
||||
folder_path = os.path.join(music_base, folder)
|
||||
if not os.path.isdir(folder_path):
|
||||
continue
|
||||
for fname in os.listdir(folder_path):
|
||||
fpath = os.path.join(folder_path, fname)
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
ext = os.path.splitext(fname)[1].lower()
|
||||
if ext not in (".mp3", ".m4a", ".aac", ".mp4"):
|
||||
continue
|
||||
|
||||
try:
|
||||
audio = MutagenFile(fpath)
|
||||
if audio is None or audio.tags is None:
|
||||
continue
|
||||
|
||||
file_title = ""
|
||||
file_artist = ""
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
easy = EasyID3(fpath)
|
||||
if "title" in easy:
|
||||
file_title = str(easy["title"][0])
|
||||
if "artist" in easy:
|
||||
file_artist = str(easy["artist"][0])
|
||||
except Exception:
|
||||
pass
|
||||
elif ext in (".m4a", ".aac", ".mp4"):
|
||||
tags = audio.tags
|
||||
if "\xa9nam" in tags:
|
||||
file_title = str(tags["\xa9nam"][0])
|
||||
if "\xa9ART" in tags:
|
||||
file_artist = str(tags["\xa9ART"][0])
|
||||
else:
|
||||
tags = audio.tags
|
||||
try:
|
||||
if "title" in tags:
|
||||
val = tags["title"]
|
||||
file_title = str(val[0] if isinstance(val, list) else val)
|
||||
if "artist" in tags:
|
||||
val = tags["artist"]
|
||||
file_artist = str(val[0] if isinstance(val, list) else val)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if file_title.lower() == title_lower and file_artist.lower() == artist_lower:
|
||||
return fpath
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return None
|
||||
|
||||
def _cleanup_orphaned_entries(self, cursor):
|
||||
"""Remove artist/album records with no remaining tracks."""
|
||||
# Remove artists with no tracks referencing them
|
||||
cursor.execute(
|
||||
"""
|
||||
DELETE FROM artist
|
||||
WHERE pid NOT IN (SELECT DISTINCT artist_pid FROM item WHERE artist_pid IS NOT NULL)
|
||||
"""
|
||||
)
|
||||
|
||||
# Remove albums with no tracks referencing them
|
||||
cursor.execute(
|
||||
"""
|
||||
DELETE FROM album
|
||||
WHERE pid NOT IN (SELECT DISTINCT album_pid FROM item WHERE album_pid IS NOT NULL)
|
||||
"""
|
||||
)
|
||||
|
||||
78
src/main.py
78
src/main.py
@ -428,6 +428,10 @@ class MainWindow(QMainWindow):
|
||||
self.device_info_label = QLabel("No device selected")
|
||||
layout.addWidget(self.device_info_label)
|
||||
|
||||
# Track count
|
||||
self.track_count_label = QLabel("Tracks on device: 0")
|
||||
layout.addWidget(self.track_count_label)
|
||||
|
||||
# Transfer button
|
||||
self.transfer_button = QPushButton("Transfer to iPod")
|
||||
self.transfer_button.clicked.connect(self._on_transfer_clicked)
|
||||
@ -448,7 +452,13 @@ class MainWindow(QMainWindow):
|
||||
layout.addWidget(transferred_list_label)
|
||||
|
||||
self.transferred_list = QListWidget()
|
||||
self.transferred_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
||||
layout.addWidget(self.transferred_list)
|
||||
|
||||
# Remove selected button
|
||||
remove_button = QPushButton("Remove Selected from iPod")
|
||||
remove_button.clicked.connect(self._on_remove_tracks_clicked)
|
||||
layout.addWidget(remove_button)
|
||||
|
||||
def _setup_settings_tab(self, tab):
|
||||
"""Set up the settings tab"""
|
||||
@ -730,6 +740,9 @@ class MainWindow(QMainWindow):
|
||||
|
||||
# Enable transfer button
|
||||
self.transfer_button.setEnabled(True)
|
||||
|
||||
# Load existing tracks from iPod
|
||||
self._load_ipod_tracks()
|
||||
else:
|
||||
self.device_info_label.setText(
|
||||
f"Device: {self.ipod_devices[0]['name']}\n"
|
||||
@ -789,10 +802,8 @@ class MainWindow(QMainWindow):
|
||||
if success:
|
||||
self.transfer_status.setText(message)
|
||||
|
||||
# Update transferred list
|
||||
for track, path in result:
|
||||
item = QListWidgetItem(f"{track.artist} - {track.title}")
|
||||
self.transferred_list.addItem(item)
|
||||
# Reload track list from iPod
|
||||
self._load_ipod_tracks()
|
||||
|
||||
QMessageBox.information(self, "Transfer Complete", message)
|
||||
else:
|
||||
@ -802,6 +813,65 @@ class MainWindow(QMainWindow):
|
||||
# Clean up worker thread
|
||||
self.worker_thread = None
|
||||
|
||||
def _load_ipod_tracks(self):
|
||||
"""Load existing tracks from iPod database into transferred_list"""
|
||||
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["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)}")
|
||||
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):
|
||||
"""Remove selected tracks from iPod with confirmation"""
|
||||
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
|
||||
|
||||
try:
|
||||
from ipod_nano7_db import Nano7Database
|
||||
db = Nano7Database(self.current_mount_point)
|
||||
|
||||
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
|
||||
|
||||
# Reload list
|
||||
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}")
|
||||
|
||||
def closeEvent(self, event):
|
||||
"""Handle window close event"""
|
||||
# Stop worker thread if running
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user