diff --git a/src/ipod_device.py b/src/ipod_device.py index 2affcd9..8cbaef9 100644 --- a/src/ipod_device.py +++ b/src/ipod_device.py @@ -1,11 +1,12 @@ #!/usr/bin/env python3 """ iPod Device Handler Module for iPod Nano Transfer Tool -Handles device detection and file transfer to iPod Nano +Handles device detection, mounting, and file transfer to iPod Nano """ import os import sys +import json import logging import platform import shutil @@ -31,37 +32,9 @@ class IPodDevice: self.mount_point = mount_point self.system = platform.system() self.device_info = {} - self.logger = logger # Add this line - - # Check if libimobiledevice is installed - if not self._check_dependencies(): - self.logger.warning("Some dependencies are missing. Device detection may not work properly.") - - def _check_dependencies(self) -> bool: - """ - Check if required dependencies are installed - - Returns: - True if all dependencies are installed, False otherwise - """ - try: - if self.system == "Linux": - # Check for ifuse and libimobiledevice - subprocess.run(["which", "ifuse"], check=True, capture_output=True) - subprocess.run(["which", "idevice_id"], check=True, capture_output=True) - return True - elif self.system == "Darwin": # macOS - # Check for libimobiledevice - subprocess.run(["which", "idevice_id"], check=True, capture_output=True) - return True - elif self.system == "Windows": - # Check for libimobiledevice-win32 (simplified check) - # In reality, this would need a more complex check - return os.path.exists("C:\\Program Files\\libimobiledevice") - - return False - except subprocess.SubprocessError: - return False + self._block_device = None # /dev/sdd1 etc. + self._parent_drive = None # /dev/sdd (for power-off) + self.logger = logger def detect_devices(self) -> List[Dict[str, str]]: """ @@ -71,234 +44,223 @@ class IPodDevice: List of dictionaries with device information """ devices = [] - - # Check platform system = platform.system() try: - if system == "Darwin": # macOS - # Try using libimobiledevice if available - try: - result = subprocess.run( - ["idevice_id", "-l"], - capture_output=True, - check=False, - text=True - ) - - if result.returncode == 0 and result.stdout.strip(): - # Parse device IDs - device_ids = [line.strip() for line in result.stdout.splitlines() if line.strip()] - - for device_id in device_ids: - # Get device info - info_result = subprocess.run( - ["ideviceinfo", "-u", device_id], - capture_output=True, - check=False, - text=True - ) - - if info_result.returncode == 0: - # Parse device info - device_info = {} - for line in info_result.stdout.splitlines(): - if ":" in line: - key, value = line.split(":", 1) - device_info[key.strip()] = value.strip() - - if "ProductType" in device_info and "iPod" in device_info["ProductType"]: - devices.append({ - "id": device_id, - "name": device_info.get("DeviceName", "iPod"), - "model": device_info.get("ProductType", "Unknown"), - "mount_point": self._get_ipod_mount_point(device_id) - }) - except FileNotFoundError: - # libimobiledevice not installed, fall back to disk utility - self.logger.warning("libimobiledevice not found, falling back to disk utility") - - # Fall back to diskutil list - result = subprocess.run( - ["diskutil", "list"], - capture_output=True, - check=False, - text=True - ) - - if result.returncode == 0: - # Look for iPod in the output - for line in result.stdout.splitlines(): - if "iPod" in line: - # Extract disk identifier - parts = line.split() - for i, part in enumerate(parts): - if part.startswith("/dev/"): - disk_id = part - - # Get mount point - mount_result = subprocess.run( - ["diskutil", "info", disk_id], - capture_output=True, - check=False, - text=True - ) - - if mount_result.returncode == 0: - mount_point = None - for info_line in mount_result.stdout.splitlines(): - if "Mount Point" in info_line: - mount_point = info_line.split(":", 1)[1].strip() - break - - if mount_point and os.path.exists(mount_point): - devices.append({ - "id": disk_id, - "name": "iPod", - "model": "Unknown", - "mount_point": mount_point - }) - - # If no devices found, check for mounted volumes with iPod in the name - if not devices: - volumes_dir = "/Volumes" - if os.path.exists(volumes_dir): - for volume in os.listdir(volumes_dir): - volume_path = os.path.join(volumes_dir, volume) - if "iPod" in volume and os.path.isdir(volume_path): - devices.append({ - "id": "unknown", - "name": volume, - "model": "Unknown", - "mount_point": volume_path - }) - + if system == "Linux": + devices = self._detect_linux() + elif system == "Darwin": + devices = self._detect_macos() elif system == "Windows": - # On Windows, check for mounted drives with iPod in the label - import win32api - import win32con - - try: - drives = win32api.GetLogicalDriveStrings().split('\0')[:-1] - for drive in drives: - try: - volume_name = win32api.GetVolumeInformation(drive)[0] - if volume_name and "iPod" in volume_name: - devices.append({ - "id": drive[0], - "name": volume_name, - "model": "Unknown", - "mount_point": drive - }) - except: - pass - except ImportError: - self.logger.warning("win32api module not available, cannot detect iPod on Windows") - - # Fallback to checking drive letters - import string - for letter in string.ascii_uppercase: - drive = f"{letter}:\\" - if os.path.exists(drive): - # Check if this looks like an iPod - if os.path.exists(os.path.join(drive, "iPod_Control")): - devices.append({ - "id": letter, - "name": "iPod", - "model": "Unknown", - "mount_point": drive - }) - - elif system == "Linux": - # On Linux, check mounted devices - try: - result = subprocess.run( - ["lsblk", "-o", "NAME,LABEL,MOUNTPOINT", "-J"], - capture_output=True, - check=False, - text=True - ) - - if result.returncode == 0: - import json - try: - data = json.loads(result.stdout) - for device in data.get("blockdevices", []): - if "children" in device: - for child in device["children"]: - label = child.get("label", "") - mountpoint = child.get("mountpoint") - if mountpoint and label and "ipod" in label.lower(): - devices.append({ - "id": child.get("name", "unknown"), - "name": label, - "model": "Unknown", - "mount_point": mountpoint - }) - except json.JSONDecodeError: - self.logger.warning("Failed to parse lsblk output as JSON") - except FileNotFoundError: - self.logger.warning("lsblk command not found") - - # Fallback to checking /media and /mnt directories - if not devices: - 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.lower(): - device_path = os.path.join(user_path, device_dir) - if os.path.isdir(device_path): - devices.append({ - "id": "unknown", - "name": device_dir, - "model": "Unknown", - "mount_point": device_path - }) + devices = self._detect_windows() except Exception as e: self.logger.error(f"Error detecting devices: {e}") - # If no devices found through system methods, check for manual mount point + # If no devices found, check for manual mount point if not devices and self.mount_point and os.path.exists(self.mount_point): devices.append({ "id": "manual", "name": os.path.basename(self.mount_point), "model": "Unknown", - "mount_point": self.mount_point + "mount_point": self.mount_point, + "mounted": True, }) return devices - def _get_ipod_mount_point(self, device_id: str) -> Optional[str]: - """ - Get the mount point for an iPod device + def _detect_linux(self) -> List[Dict[str, str]]: + """Detect iPod on Linux using lsblk and udisksctl""" + devices = [] - Args: - device_id: Device ID + # Method 1: lsblk JSON output - find devices with IPOD label + try: + result = subprocess.run( + ["lsblk", "-o", "NAME,LABEL,MOUNTPOINT,TYPE", "-J"], + capture_output=True, check=False, text=True + ) + + if result.returncode == 0: + data = json.loads(result.stdout) + for device in data.get("blockdevices", []): + if device.get("type") != "disk": + continue + parent_name = device.get("name", "") + for child in device.get("children", []): + label = child.get("label", "") + mountpoint = child.get("mountpoint") + child_name = child.get("name", "") + + if label and "ipod" in label.lower(): + dev_path = f"/dev/{child_name}" + parent_dev = f"/dev/{parent_name}" + + device_entry = { + "id": dev_path, + "name": label, + "model": "iPod Nano 7", + "mount_point": mountpoint, + "mounted": mountpoint is not None, + "_parent_drive": parent_dev, + } + devices.append(device_entry) + except (FileNotFoundError, json.JSONDecodeError) as e: + self.logger.warning(f"lsblk JSON failed: {e}") - Returns: - Mount point if found, None otherwise - """ - # This method is not implemented in the provided code - # You may need to implement it according to your requirements - pass + # Method 2: udisksctl dump - find iPod drives + if not devices: + try: + result = subprocess.run( + ["udisksctl", "dump"], + capture_output=True, check=False, text=True + ) + + if result.returncode == 0: + current_block = None + current_drive = None + current_label = None + current_mount = None + + for line in result.stdout.splitlines(): + if "/org/freedesktop/UDisks2/block_devices/" in line: + # Save previous if it was an iPod + if current_block and current_label and "ipod" in current_label.lower(): + devices.append({ + "id": current_block, + "name": current_label, + "model": "iPod", + "mount_point": current_mount if current_mount else None, + "mounted": current_mount is not None, + "_parent_drive": self._get_parent_drive_from_udisks(current_block), + }) + current_block = line.split("/block_devices/")[-1].strip().rstrip(":") + current_drive = None + current_label = None + current_mount = None + + if "IdLabel:" in line and current_block: + val = line.split(":", 1)[1].strip() + if val and val != "''": + current_label = val.strip("'") + if "MountPoints:" in line and current_block: + val = line.split(":", 1)[1].strip() + if val and val != "''" and val != "b''": + # MountPoints is a byte array like b'/run/media/mat/IPOD' + import ast + try: + decoded = ast.literal_eval(val) + if isinstance(decoded, bytes): + current_mount = decoded.decode("utf-8", errors="replace") + elif isinstance(decoded, str): + current_mount = decoded + except (ValueError, SyntaxError): + current_mount = val.strip("'\"") + + # Don't forget the last entry + if current_block and current_label and "ipod" in current_label.lower(): + devices.append({ + "id": current_block, + "name": current_label, + "model": "iPod", + "mount_point": current_mount if current_mount else None, + "mounted": current_mount is not None, + "_parent_drive": self._get_parent_drive_from_udisks(current_block), + }) + except FileNotFoundError: + self.logger.warning("udisksctl not found") + + # Method 3: Check common mount directories for already-mounted iPods + for device in devices: + if not device["mounted"]: + for base in ["/run/media", "/media", "/mnt"]: + if os.path.exists(base): + for user_or_device in os.listdir(base): + user_path = os.path.join(base, user_or_device) + if os.path.isdir(user_path): + # Could be /run/media/mat/IPOD or /media/IPOD + check_paths = [user_path] + if os.path.isdir(user_path): + for sub in os.listdir(user_path): + sub_path = os.path.join(user_path, sub) + if "ipod" in sub.lower() and os.path.isdir(sub_path): + check_paths.append(sub_path) + + for check_path in check_paths: + if os.path.exists(os.path.join(check_path, "iPod_Control")): + device["mount_point"] = check_path + device["mounted"] = True + break + if device["mounted"]: + break + if device["mounted"]: + break + + return devices + + def _get_parent_drive_from_udisks(self, block_name: str) -> Optional[str]: + """Get parent drive device from udisksctl for a partition""" + try: + # /dev/sdd1 -> /dev/sdd + if block_name and block_name.startswith("/dev/"): + name = block_name[5:] # sdd1 + # Strip trailing partition number + base = name.rstrip("0123456789") + if base: + return f"/dev/{base}" + except Exception: + pass + return None + + def _detect_macos(self) -> List[Dict[str, str]]: + """Detect iPod on macOS""" + devices = [] + + # Check /Volumes for iPod + volumes_dir = "/Volumes" + if os.path.exists(volumes_dir): + for volume in os.listdir(volumes_dir): + volume_path = os.path.join(volumes_dir, volume) + if "iPod" in volume and os.path.isdir(volume_path): + if os.path.exists(os.path.join(volume_path, "iPod_Control")): + devices.append({ + "id": "unknown", + "name": volume, + "model": "Unknown", + "mount_point": volume_path, + "mounted": True, + }) + + return devices + + def _detect_windows(self) -> List[Dict[str, str]]: + """Detect iPod on Windows""" + devices = [] + import string + for letter in string.ascii_uppercase: + drive = f"{letter}:\\" + if os.path.exists(drive): + if os.path.exists(os.path.join(drive, "iPod_Control")): + devices.append({ + "id": letter, + "name": "iPod", + "model": "Unknown", + "mount_point": drive, + "mounted": True, + }) + return devices def mount_device(self, device_id: Optional[str] = None, detected_mount: Optional[str] = None) -> Optional[str]: """ - Mount iPod device + Mount iPod device using udisksctl (Linux) or verify existing mount Args: - device_id: Device ID (optional) - detected_mount: Already detected mount point (optional) + device_id: Device path like /dev/sdd1 + detected_mount: Already detected mount point 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}") @@ -307,108 +269,205 @@ class IPodDevice: 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) - - # 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 - + return self._mount_linux(device_id) elif self.system == "Darwin": - # On macOS, devices are automatically mounted - # Find the iPod mount point - result = subprocess.run( - ["df", "-h"], - check=True, - capture_output=True, - text=True - ) - - for line in result.stdout.strip().split('\n'): - if "iPod" in line: - # Extract disk identifier - parts = line.split() - if len(parts) >= 9: # macOS df output format - mount_point = parts[8] - self.mount_point = mount_point - self.logger.info(f"Device found at: {mount_point}") - return mount_point - - self.logger.warning("iPod mount point not found") - return None - + return self._mount_macos() elif self.system == "Windows": - # Windows implementation would go here - # This is a placeholder for future implementation - self.logger.warning("Windows mounting not fully implemented") - return None - + return self._mount_windows() except subprocess.SubprocessError as e: self.logger.error(f"Failed to mount device: {e}") return None + + return None + + def _mount_linux(self, device_id: Optional[str]) -> Optional[str]: + """Mount on Linux using udisksctl""" + # device_id should be like /dev/sdd1 + if not device_id or not device_id.startswith("/dev/"): + self.logger.error("No valid device path provided for mounting") + return None + + if not os.path.exists(device_id): + self.logger.error(f"Device {device_id} does not exist") + return None + + try: + result = subprocess.run( + ["udisksctl", "mount", "-b", device_id], + capture_output=True, check=False, text=True + ) + + if result.returncode != 0: + self.logger.error(f"udisksctl mount failed: {result.stderr}") + return None + + # Parse mount point from output: "Mounted /dev/sdd1 at /run/media/mat/IPOD.\n" + output = result.stdout.strip() + mount_point = None + for line in output.splitlines(): + if "Mounted" in line and "at" in line: + # "Mounted /dev/sdd1 at /run/media/mat/IPOD." + parts = line.split(" at ") + if len(parts) >= 2: + mp = parts[1].strip().rstrip(".") + if os.path.exists(mp): + mount_point = mp + + if not mount_point: + # Fallback: check /run/media for the device + for user_dir in os.listdir("/run/media"): + user_path = os.path.join("/run/media", user_dir) + if os.path.isdir(user_path): + for d in os.listdir(user_path): + d_path = os.path.join(user_path, d) + if os.path.exists(os.path.join(d_path, "iPod_Control")): + mount_point = d_path + break + if mount_point: + break + + if mount_point: + self.mount_point = mount_point + self._block_device = device_id + # Extract parent drive: /dev/sdd1 -> /dev/sdd + name = device_id[5:] + self._parent_drive = f"/dev/{name.rstrip('0123456789')}" + self.logger.info(f"Device mounted at: {mount_point}") + return mount_point + + self.logger.error("Could not determine mount point") + return None + + except FileNotFoundError: + self.logger.error("udisksctl not found. Please install udisks2.") + return None + + def _mount_macos(self) -> Optional[str]: + """Mount on macOS - devices auto-mount, just find the mount point""" + result = subprocess.run( + ["diskutil", "list"], + capture_output=True, check=False, text=True + ) + + if result.returncode == 0: + for line in result.stdout.splitlines(): + if "iPod" in line: + df_result = subprocess.run( + ["df", "-h"], + capture_output=True, check=False, text=True + ) + for df_line in df_result.stdout.splitlines(): + if "iPod" in df_line: + parts = df_line.split() + if len(parts) >= 9: + mount_point = parts[8] + self.mount_point = mount_point + return mount_point + + return None + + def _mount_windows(self) -> Optional[str]: + """Mount on Windows""" + self.logger.warning("Windows mounting not fully implemented") + return None def unmount_device(self) -> bool: """ - Unmount iPod device + Unmount and safely eject iPod device Returns: True if successful, False otherwise """ - if not self.mount_point: + if not self.mount_point and not self._block_device: self.logger.warning("No device mounted") return False - + try: if self.system == "Linux": - # Unmount with fusermount - subprocess.run(["fusermount", "-u", self.mount_point], check=True) - self.logger.info(f"Device unmounted: {self.mount_point}") - self.mount_point = None - return True - + return self._unmount_linux() elif self.system == "Darwin": - # On macOS, use diskutil - subprocess.run(["diskutil", "eject", self.mount_point], check=True) - self.logger.info(f"Device ejected: {self.mount_point}") - self.mount_point = None - return True - + return self._unmount_macos() elif self.system == "Windows": - # Windows implementation would go here - # This is a placeholder for future implementation - self.logger.warning("Windows unmounting not fully implemented") - return False - + return self._unmount_windows() except subprocess.SubprocessError as e: self.logger.error(f"Failed to unmount device: {e}") return False + + return False + + def _unmount_linux(self) -> bool: + """Unmount and power-off on Linux using udisksctl""" + success = True + + # Step 1: Unmount + if self.mount_point or self._block_device: + device = self._block_device + if not device: + # Try to find the device from mount point + result = subprocess.run( + ["findmnt", "-n", "-o", "SOURCE", self.mount_point], + capture_output=True, check=False, text=True + ) + if result.returncode == 0 and result.stdout.strip(): + device = result.stdout.strip() + + if device: + result = subprocess.run( + ["udisksctl", "unmount", "-b", device], + capture_output=True, check=False, text=True + ) + + if result.returncode == 0: + self.logger.info(f"Device unmounted: {device}") + else: + self.logger.warning(f"Unmount warning: {result.stderr}") + # Not a fatal error - device might already be unmounted + + # Step 2: Power off the drive (safely remove) + drive = self._parent_drive + if not drive and self._block_device: + name = self._block_device[5:] + drive = f"/dev/{name.rstrip('0123456789')}" + + if drive: + result = subprocess.run( + ["udisksctl", "power-off", "-b", drive], + capture_output=True, check=False, text=True + ) + + if result.returncode == 0: + self.logger.info(f"Device powered off: {drive}") + else: + self.logger.warning(f"Power-off warning: {result.stderr}") + # Not fatal - USB may have been unplugged already + + self.mount_point = None + self._block_device = None + self._parent_drive = None + return success + + def _unmount_macos(self) -> bool: + """Eject on macOS""" + if self.mount_point: + subprocess.run(["diskutil", "eject", self.mount_point], check=True) + self.logger.info(f"Device ejected: {self.mount_point}") + self.mount_point = None + return True + return False + + def _unmount_windows(self) -> bool: + """Unmount on Windows""" + self.logger.warning("Windows unmounting not fully implemented") + return False + + def is_mounted(self) -> bool: + """Check if iPod is currently mounted""" + if self.mount_point and os.path.exists(self.mount_point): + return os.path.exists(os.path.join(self.mount_point, "iPod_Control")) + return False def get_device_info(self) -> Dict[str, Any]: """ @@ -443,18 +502,6 @@ class IPodDevice: 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 @@ -487,6 +534,10 @@ class IPodDevice: from ipod_nano7_db import Nano7Database nano7_db = Nano7Database(self.mount_point) result = nano7_db.add_track(source_file) + if result is None: + # Track already exists (duplicate) + self.logger.info(f"Track already on iPod, skipping: {os.path.basename(source_file)}") + return True # Not an error - track is already there self.logger.info(f"Track added to iPod database: {result['title']}") return True except ImportError: @@ -541,24 +592,18 @@ class IPodDevice: if dest_dir: full_dest_dir = os.path.join(self.mount_point, dest_dir) else: - # Default to Music directory full_dest_dir = os.path.join(self.mount_point, "Music") - # Create destination directory if it doesn't exist os.makedirs(full_dest_dir, exist_ok=True) - # Copy directory contents for root, dirs, files in os.walk(source_dir): - # Create relative path rel_path = os.path.relpath(root, source_dir) if rel_path == '.': rel_path = '' - # Create destination subdirectory dest_subdir = os.path.join(full_dest_dir, rel_path) os.makedirs(dest_subdir, exist_ok=True) - # Copy files for file in files: if file.lower().endswith(('.m4a', '.mp3', '.aac', '.mp4')): src_file = os.path.join(root, file) @@ -575,24 +620,30 @@ class IPodDevice: if __name__ == "__main__": - # Example usage ipod = IPodDevice() # Detect devices devices = ipod.detect_devices() for device in devices: - print(f"Found device: {device['name']} ({device['model']})") + status = "mounted" if device.get("mounted") else "NOT mounted" + print(f"Found device: {device['name']} ({device['model']}) - {status}") + if device.get("mount_point"): + print(f" Mount point: {device['mount_point']}") + print(f" Device: {device['id']}") - # Mount device + # Mount and eject if device found if devices: - mount_point = ipod.mount_device(devices[0]['id']) - if mount_point: - print(f"Device mounted at: {mount_point}") - - # Get device info - info = ipod.get_device_info() - print(f"Total space: {info.get('total_space', 0) / (1024**3):.2f} GB") - print(f"Free space: {info.get('free_space', 0) / (1024**3):.2f} GB") - - # Unmount device - ipod.unmount_device() + device = devices[0] + if not device.get("mounted"): + mount_point = ipod.mount_device(device["id"]) + if mount_point: + print(f"Mounted at: {mount_point}") + + info = ipod.get_device_info() + print(f"Free space: {info.get('free_space', 0) / 1024**2:.1f} MB") + + # Unmount + ipod.unmount_device() + print("Ejected.") + else: + print(f"Already mounted at: {device['mount_point']}")