From 9dcb1228caa4dd99d7b5a6fc7b7059d4dae04011 Mon Sep 17 00:00:00 2001 From: Antonio Lorusso Date: Wed, 12 Mar 2025 01:01:43 +0100 Subject: [PATCH] initial app --- .gitignore | 37 ++ README.md | 63 ++++ config.ini | 43 +++ install.bat | 48 +++ install.sh | 54 +++ requirements.txt | 11 + run.py | 34 ++ setup.py | 39 ++ src/__init__.py | 6 + src/audio_converter.py | 212 +++++++++++ src/cli.py | 337 +++++++++++++++++ src/config_loader.py | 238 ++++++++++++ src/ipod_device.py | 562 ++++++++++++++++++++++++++++ src/main.py | 762 ++++++++++++++++++++++++++++++++++++++ src/metadata_handler.py | 261 +++++++++++++ src/youtube_downloader.py | 415 +++++++++++++++++++++ tests/test_basic.py | 93 +++++ 17 files changed, 3215 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 config.ini create mode 100644 install.bat create mode 100755 install.sh create mode 100644 requirements.txt create mode 100755 run.py create mode 100644 setup.py create mode 100644 src/__init__.py create mode 100644 src/audio_converter.py create mode 100644 src/cli.py create mode 100644 src/config_loader.py create mode 100644 src/ipod_device.py create mode 100644 src/main.py create mode 100644 src/metadata_handler.py create mode 100644 src/youtube_downloader.py create mode 100644 tests/test_basic.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ae9a8af --- /dev/null +++ b/.gitignore @@ -0,0 +1,37 @@ +# Python bytecode +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +dist/ +build/ +*.egg-info/ + +# Virtual environments +venv/ +env/ +ENV/ + +# Temporary files +temp/ +downloads/ +converted/ + +# IDE files +.idea/ +.vscode/ +*.swp +*.swo + +# macOS files +.DS_Store + +# iPod test files +*.m4a +*.mp3 +*.mp4 +*.aac + +# Logs +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..ba8cb49 --- /dev/null +++ b/README.md @@ -0,0 +1,63 @@ + + +An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano, bypassing iTunes. This application combines YouTube audio extraction, format conversion, metadata management, and direct device transfer while adhering to iPod Nano's technical specifications. + +## Features + +- Download audio from YouTube videos/playlists +- Convert to iPod-compatible formats (AAC/M4A) +- Automatic metadata handling (title, artist, album art) +- Direct transfer to iPod Nano +- Cross-platform support (Windows, macOS, Linux) + +## Installation + +1. Clone this repository: + +```bash +git clone https://github.com/yourusername/youtube-to-ipod-nano.git +cd youtube-to-ipod-nano +``` + +1. Install dependencies: + +```bash +pip install -r requirements.txt +``` + +1. Install FFmpeg (required for audio conversion): + - **macOS**: `brew install ffmpeg` + - **Linux**: `sudo apt-get install ffmpeg` + - **Windows**: Download from [FFmpeg website](https://ffmpeg.org/download.html) + +## Usage + +### GUI Application + +Run the application: + +```bash +python src/main.py +``` + +### Command Line Interface + +```bash +python src/cli.py --url "https://www.youtube.com/watch?v=VIDEO_ID" --output-dir "/path/to/ipod" +``` + +Options: + +- `--url`: YouTube video or playlist URL +- `--output-dir`: Output directory (iPod mount point) +- `--format`: Audio format (default: m4a) +- `--quality`: Audio quality in kbps (default: 256) +- `--video`: Enable video download (for compatible iPod models) + +## Legal Disclaimer + +This tool is intended for personal use only. Users are responsible for complying with YouTube's Terms of Service and copyright laws. Please only download content that you have the right to access and use. + +## License + +MIT License diff --git a/config.ini b/config.ini new file mode 100644 index 0000000..c4f30f0 --- /dev/null +++ b/config.ini @@ -0,0 +1,43 @@ +[General] +# Default output directory +output_dir = ~/Music/iPod + +# Default audio format (m4a or mp3) +format = m4a + +# Default audio quality in kbps (128, 192, 256, 320) +quality = 256 + +# Clean temporary files after transfer (true/false) +clean_temp = true + +[Video] +# Enable video download (true/false) +enable_video = false + +# Default video resolution (640x480, 480x360, 320x240) +resolution = 640x480 + +[Metadata] +# Embed album artwork (true/false) +embed_artwork = true + +# Use MusicBrainz for enhanced metadata (true/false) +use_musicbrainz = false + +[Device] +# Auto-detect iPod on startup (true/false) +auto_detect = true + +# Custom mount point (leave empty for auto-detection) +mount_point = + +[Advanced] +# Temporary directory +temp_dir = temp + +# Number of simultaneous downloads (1-4) +concurrent_downloads = 2 + +# Debug mode (true/false) +debug = false diff --git a/install.bat b/install.bat new file mode 100644 index 0000000..df574b8 --- /dev/null +++ b/install.bat @@ -0,0 +1,48 @@ +@echo off +REM Installation script for YouTube Music to iPod Nano Transfer Tool (Windows) + +echo Installing YouTube Music to iPod Nano Transfer Tool... + +REM Check if Python 3 is installed +python --version 2>NUL +if %ERRORLEVEL% NEQ 0 ( + echo Python 3 is required but not installed. Please install Python 3 and try again. + exit /b 1 +) + +REM Check if pip is installed +pip --version 2>NUL +if %ERRORLEVEL% NEQ 0 ( + echo pip is required but not installed. Please install pip and try again. + exit /b 1 +) + +REM Create virtual environment +echo Creating virtual environment... +python -m venv venv + +REM Activate virtual environment +echo Activating virtual environment... +call venv\Scripts\activate.bat + +REM Install dependencies +echo Installing dependencies... +pip install -r requirements.txt + +REM Create necessary directories +echo Creating necessary directories... +if not exist temp mkdir temp +if not exist downloads mkdir downloads +if not exist converted mkdir converted + +echo. +echo Installation complete! +echo. +echo To run the application: +echo 1. Activate the virtual environment: venv\Scripts\activate.bat +echo 2. Run the GUI application: python run.py +echo 3. Or run the CLI application: python run.py --cli --url "https://www.youtube.com/watch?v=VIDEO_ID" +echo. +echo For more information, see the README.md file. + +pause diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..000e363 --- /dev/null +++ b/install.sh @@ -0,0 +1,54 @@ +#!/bin/bash +# Installation script for YouTube Music to iPod Nano Transfer Tool + +set -e + +echo "Installing YouTube Music to iPod Nano Transfer Tool..." + +# Check if Python 3 is installed +if ! command -v python3 &> /dev/null; then + echo "Python 3 is required but not installed. Please install Python 3 and try again." + exit 1 +fi + +# Check if pip is installed +if ! command -v pip3 &> /dev/null; then + echo "pip3 is required but not installed. Please install pip3 and try again." + exit 1 +fi + +# Check if FFmpeg is installed +if ! command -v ffmpeg &> /dev/null; then + echo "FFmpeg is required but not installed." + echo "Please install FFmpeg:" + echo " - macOS: brew install ffmpeg" + echo " - Linux: sudo apt-get install ffmpeg" + echo " - Windows: Download from https://ffmpeg.org/download.html" + exit 1 +fi + +# Create virtual environment +echo "Creating virtual environment..." +python3 -m venv venv + +# Activate virtual environment +echo "Activating virtual environment..." +source venv/bin/activate + +# Install dependencies +echo "Installing dependencies..." +pip3 install -r requirements.txt + +# Create necessary directories +echo "Creating necessary directories..." +mkdir -p temp downloads converted + +echo "" +echo "Installation complete!" +echo "" +echo "To run the application:" +echo " 1. Activate the virtual environment: source venv/bin/activate" +echo " 2. Run the GUI application: python run.py" +echo " 3. Or run the CLI application: python run.py --cli --url \"https://www.youtube.com/watch?v=VIDEO_ID\"" +echo "" +echo "For more information, see the README.md file." diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1441308 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +yt-dlp>=2023.3.4 +pytube>=12.1.3 +ffmpeg-python>=0.2.0 +mutagen>=1.46.0 +eyeD3>=0.9.7 +pyusb>=1.2.1 +pillow>=9.5.0 +requests>=2.28.2 +tqdm>=4.65.0 +musicbrainzngs>=0.7.1 +PyQt6>=6.5.0 diff --git a/run.py b/run.py new file mode 100755 index 0000000..a4fdec5 --- /dev/null +++ b/run.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +""" +Launcher script for YouTube Music to iPod Nano Transfer Tool +""" + +import os +import sys +import argparse + +def main(): + """Main entry point""" + parser = argparse.ArgumentParser(description="YouTube to iPod Nano Transfer Tool") + parser.add_argument("--cli", action="store_true", help="Run in command-line mode") + args = parser.parse_args() + + # Add src directory to path + src_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "src") + sys.path.insert(0, src_dir) + + if args.cli: + # Run CLI version + from src.cli import YouTubeToIPodCLI + cli = YouTubeToIPodCLI() + return cli.run() + else: + # Run GUI version + from src.main import QApplication, MainWindow + app = QApplication(sys.argv) + window = MainWindow() + window.show() + return app.exec() + +if __name__ == "__main__": + sys.exit(main()) diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..9c8acf6 --- /dev/null +++ b/setup.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +""" +Setup script for YouTube Music to iPod Nano Transfer Tool +""" + +from setuptools import setup, find_packages +import os + +# Read requirements from requirements.txt +with open('requirements.txt') as f: + requirements = f.read().splitlines() + +# Read long description from README.md +with open('README.md', encoding='utf-8') as f: + long_description = f.read() + +setup( + name="youtube-to-ipod-nano", + version="1.0.0", + description="Transfer YouTube music/playlists directly to iPod Nano", + long_description=long_description, + long_description_content_type="text/markdown", + author="Your Name", + author_email="your.email@example.com", + url="https://github.com/yourusername/youtube-to-ipod-nano", + packages=find_packages(), + install_requires=requirements, + entry_points={ + 'console_scripts': [ + 'youtube-to-ipod=src.cli:main', + ], + }, + classifiers=[ + "Programming Language :: Python :: 3", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + ], + python_requires=">=3.7", +) diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..33c699d --- /dev/null +++ b/src/__init__.py @@ -0,0 +1,6 @@ +""" +YouTube Music to iPod Nano Transfer Tool +An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano +""" + +__version__ = "1.0.0" diff --git a/src/audio_converter.py b/src/audio_converter.py new file mode 100644 index 0000000..d92fb80 --- /dev/null +++ b/src/audio_converter.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +""" +Audio Converter Module for iPod Nano Transfer Tool +Handles converting audio files to iPod-compatible formats +""" + +import os +import logging +import subprocess +from typing import Optional, Dict, Any +from pathlib import Path + +import ffmpeg + +from youtube_downloader import TrackInfo + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class AudioConverter: + """Class to handle audio conversion to iPod-compatible formats""" + + def __init__(self, output_dir: str = "converted"): + """ + Initialize the audio converter + + Args: + output_dir: Directory to save converted files + """ + self.output_dir = output_dir + os.makedirs(output_dir, exist_ok=True) + + # Check if ffmpeg is installed + try: + subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True) + except (subprocess.SubprocessError, FileNotFoundError): + logger.error("FFmpeg is not installed or not in PATH. Please install FFmpeg.") + raise RuntimeError("FFmpeg is required but not found") + + def convert_to_ipod_format(self, + track_info: TrackInfo, + input_file: str, + format: str = "m4a", + audio_bitrate: int = 256, + overwrite: bool = False) -> str: + """ + Convert audio file to iPod-compatible format + + Args: + track_info: TrackInfo object with track metadata + input_file: Path to input audio file + format: Output format (m4a, mp3) + audio_bitrate: Audio bitrate in kbps + overwrite: Whether to overwrite existing files + + Returns: + Path to converted file + """ + logger.info(f"Converting {input_file} to iPod-compatible {format}") + + # Create output directory structure + if track_info.playlist_title: + # For playlist tracks: Artist/Playlist/01 - Track.m4a + artist_dir = os.path.join(self.output_dir, self._sanitize_filename(track_info.artist)) + album_dir = os.path.join(artist_dir, self._sanitize_filename(track_info.playlist_title)) + os.makedirs(album_dir, exist_ok=True) + + if track_info.track_number: + filename = f"{track_info.track_number:02d} - {self._sanitize_filename(track_info.title)}.{format}" + else: + filename = f"{self._sanitize_filename(track_info.title)}.{format}" + + output_file = os.path.join(album_dir, filename) + else: + # For single tracks: Artist/Artist/Track.m4a + artist_dir = os.path.join(self.output_dir, self._sanitize_filename(track_info.artist)) + album_dir = os.path.join(artist_dir, self._sanitize_filename(track_info.album or track_info.artist)) + os.makedirs(album_dir, exist_ok=True) + + filename = f"{self._sanitize_filename(track_info.title)}.{format}" + output_file = os.path.join(album_dir, filename) + + # Check if file already exists + if os.path.exists(output_file) and not overwrite: + logger.info(f"File already exists: {output_file}") + return output_file + + # Set up FFmpeg conversion + try: + # Input file + stream = ffmpeg.input(input_file) + + # Output options + audio_options: Dict[str, Any] = { + 'acodec': 'aac' if format == 'm4a' else 'libmp3lame', + 'b:a': f'{audio_bitrate}k', + 'map_metadata': 0, + } + + # Run conversion + stream = ffmpeg.output(stream, output_file, **audio_options) + ffmpeg.run(stream, overwrite_output=overwrite, quiet=True) + + logger.info(f"Conversion complete: {output_file}") + return output_file + + except ffmpeg.Error as e: + logger.error(f"FFmpeg error: {e.stderr.decode() if e.stderr else str(e)}") + raise + + def convert_video_for_ipod(self, + input_file: str, + output_file: Optional[str] = None, + resolution: str = "640x480", + video_bitrate: int = 1500, + audio_bitrate: int = 256, + overwrite: bool = False) -> str: + """ + Convert video file to iPod-compatible format + + Args: + input_file: Path to input video file + output_file: Path to output file (optional) + resolution: Video resolution (max 640x480 for iPod Nano) + video_bitrate: Video bitrate in kbps + audio_bitrate: Audio bitrate in kbps + overwrite: Whether to overwrite existing files + + Returns: + Path to converted file + """ + logger.info(f"Converting video {input_file} to iPod-compatible format") + + if not output_file: + input_path = Path(input_file) + output_file = os.path.join(self.output_dir, f"{input_path.stem}_ipod.mp4") + + # Check if file already exists + if os.path.exists(output_file) and not overwrite: + logger.info(f"File already exists: {output_file}") + return output_file + + # Set up FFmpeg conversion + try: + # Input file + stream = ffmpeg.input(input_file) + + # Output options + video_options = { + 'vcodec': 'h264', + 'b:v': f'{video_bitrate}k', + 'acodec': 'aac', + 'b:a': f'{audio_bitrate}k', + 'vf': f'scale={resolution}', + 'pix_fmt': 'yuv420p', # Required for compatibility + 'map_metadata': 0, + } + + # Run conversion + stream = ffmpeg.output(stream, output_file, **video_options) + ffmpeg.run(stream, overwrite_output=overwrite, quiet=True) + + logger.info(f"Video conversion complete: {output_file}") + return output_file + + except ffmpeg.Error as e: + logger.error(f"FFmpeg error: {e.stderr.decode() if e.stderr else str(e)}") + raise + + def _sanitize_filename(self, filename: str) -> str: + """ + Sanitize filename to be compatible with file systems + + Args: + filename: Input filename + + Returns: + Sanitized filename + """ + # Replace invalid characters with underscore + invalid_chars = '<>:"/\\|?*' + for char in invalid_chars: + filename = filename.replace(char, '_') + + # Limit length + if len(filename) > 100: + filename = filename[:97] + '...' + + return filename.strip() + + +if __name__ == "__main__": + # Example usage + converter = AudioConverter() + + # Test audio conversion + # converter.convert_to_ipod_format( + # TrackInfo( + # video_id="dQw4w9WgXcQ", + # title="Never Gonna Give You Up", + # artist="Rick Astley", + # album="Whenever You Need Somebody", + # thumbnail_url="", + # duration=213 + # ), + # "input.mp3" + # ) + + # Test video conversion + # converter.convert_video_for_ipod("input.mp4") diff --git a/src/cli.py b/src/cli.py new file mode 100644 index 0000000..f4b1bac --- /dev/null +++ b/src/cli.py @@ -0,0 +1,337 @@ +#!/usr/bin/env python3 +""" +Command Line Interface for iPod Nano Transfer Tool +Provides a CLI for the YouTube Music to iPod Nano transfer tool +""" + +import os +import sys +import logging +import argparse +import time +from typing import List, Dict, Optional, Tuple +from pathlib import Path + +from tqdm import tqdm + +from youtube_downloader import YouTubeDownloader, TrackInfo +from audio_converter import AudioConverter +from metadata_handler import MetadataHandler +from ipod_device import IPodDevice + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class YouTubeToIPodCLI: + """Command Line Interface for YouTube to iPod Nano Transfer Tool""" + + def __init__(self): + """Initialize the CLI""" + self.downloader = None + self.converter = None + self.metadata_handler = None + self.ipod_device = None + + self.downloaded_tracks = [] + self.converted_tracks = [] + + self.temp_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "temp") + os.makedirs(self.temp_dir, exist_ok=True) + + def parse_arguments(self): + """Parse command line arguments""" + parser = argparse.ArgumentParser( + description="YouTube to iPod Nano Transfer Tool", + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + + # URL argument + parser.add_argument( + "--url", "-u", + help="YouTube video or playlist URL", + required=True + ) + + # Output directory + parser.add_argument( + "--output-dir", "-o", + help="Output directory (iPod mount point)", + default=os.path.join(os.path.expanduser("~"), "Music", "iPod") + ) + + # Format + parser.add_argument( + "--format", "-f", + help="Audio format", + choices=["m4a", "mp3"], + default="m4a" + ) + + # Quality + parser.add_argument( + "--quality", "-q", + help="Audio quality in kbps", + type=int, + choices=[128, 192, 256, 320], + default=256 + ) + + # Video support + parser.add_argument( + "--video", + help="Enable video download (for compatible iPod models)", + action="store_true" + ) + + # Video resolution + parser.add_argument( + "--resolution", "-r", + help="Video resolution (for video download)", + choices=["640x480", "480x360", "320x240"], + default="640x480" + ) + + # iPod device + parser.add_argument( + "--device", "-d", + help="iPod device mount point (if not specified, auto-detection will be attempted)", + default=None + ) + + # Skip steps + parser.add_argument( + "--skip-download", + help="Skip download step (use previously downloaded files)", + action="store_true" + ) + + parser.add_argument( + "--skip-convert", + help="Skip conversion step (use previously converted files)", + action="store_true" + ) + + parser.add_argument( + "--skip-transfer", + help="Skip transfer step (only download and convert)", + action="store_true" + ) + + # Clean temp files + parser.add_argument( + "--clean-temp", + help="Clean temporary files after transfer", + action="store_true" + ) + + return parser.parse_args() + + def download(self, url: str, format: str = "m4a", quality: int = 256, video: bool = False) -> List[TrackInfo]: + """ + Download audio/video from YouTube + + Args: + url: YouTube URL + format: Audio format + quality: Audio quality in kbps + video: Whether to download video + + Returns: + List of TrackInfo objects + """ + logger.info(f"Downloading from URL: {url}") + + # Initialize downloader + self.downloader = YouTubeDownloader(output_dir=self.temp_dir) + + # Process URL + tracks = self.downloader.process_url(url, format, quality) + + logger.info(f"Downloaded {len(tracks)} tracks") + return tracks + + 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]]: + """ + Convert downloaded tracks to iPod-compatible format + + Args: + tracks: List of TrackInfo objects + output_dir: Output directory + format: Audio format + quality: Audio quality in kbps + video: Whether to convert video + resolution: Video resolution + + Returns: + List of (TrackInfo, output_path) tuples + """ + logger.info(f"Converting {len(tracks)} tracks to iPod-compatible format") + + # Initialize converter and metadata handler + self.converter = AudioConverter(output_dir=output_dir) + self.metadata_handler = MetadataHandler(temp_dir=self.temp_dir) + + # Convert each track + converted_tracks = [] + for track in tqdm(tracks, desc="Converting", unit="track"): + if track.download_path and os.path.exists(track.download_path): + # Convert to iPod format + output_path = self.converter.convert_to_ipod_format( + track, track.download_path, format, quality + ) + + # Embed metadata + if output_path: + self.metadata_handler.process_file(output_path, track) + converted_tracks.append((track, output_path)) + + logger.info(f"Converted {len(converted_tracks)} tracks") + return converted_tracks + + def transfer(self, + tracks: List[Tuple[TrackInfo, str]], + mount_point: Optional[str] = None) -> int: + """ + Transfer tracks to iPod + + Args: + tracks: List of (TrackInfo, path) tuples + mount_point: iPod mount point (optional) + + Returns: + Number of tracks transferred + """ + logger.info("Transferring tracks to iPod") + + # Initialize iPod device + self.ipod_device = IPodDevice(mount_point=mount_point) + + # If no mount point specified, try to detect and mount + if not mount_point: + logger.info("No mount point specified, attempting to detect iPod") + + devices = self.ipod_device.detect_devices() + if not devices: + logger.error("No iPod devices found") + return 0 + + logger.info(f"Found iPod: {devices[0]['name']} ({devices[0]['model']})") + + mount_point = self.ipod_device.mount_device(devices[0]['id']) + if not mount_point: + logger.error(f"Failed to mount iPod: {devices[0]['name']}") + return 0 + + logger.info(f"Mounted iPod at: {mount_point}") + + # Get device info + device_info = self.ipod_device.get_device_info() + free_space = device_info.get("free_space", 0) + + # Calculate total size of tracks + total_size = 0 + for _, path in tracks: + if os.path.exists(path): + total_size += os.path.getsize(path) + + # Check if there's enough space + if total_size > free_space: + logger.error( + f"Not enough space on iPod. Need {total_size / 1024**2:.1f} MB, " + + f"but only {free_space / 1024**2:.1f} MB available" + ) + return 0 + + # Transfer each track + transferred_count = 0 + for track, path in tqdm(tracks, desc="Transferring", unit="track"): + if os.path.exists(path): + # Transfer file + success = self.ipod_device.transfer_file(path) + if success: + transferred_count += 1 + + logger.info(f"Transferred {transferred_count} tracks to iPod") + return transferred_count + + def clean_temp_files(self): + """Clean temporary files""" + logger.info("Cleaning temporary files") + + # Remove all files in temp directory + for file in os.listdir(self.temp_dir): + file_path = os.path.join(self.temp_dir, file) + try: + if os.path.isfile(file_path): + os.unlink(file_path) + except Exception as e: + logger.error(f"Error deleting {file_path}: {e}") + + def run(self): + """Run the CLI""" + # Parse arguments + args = self.parse_arguments() + + try: + # Download step + if not args.skip_download: + self.downloaded_tracks = self.download( + args.url, + args.format, + args.quality, + args.video + ) + else: + logger.info("Skipping download step") + # TODO: Load previously downloaded tracks + + # Convert step + if not args.skip_convert: + self.converted_tracks = self.convert( + self.downloaded_tracks, + args.output_dir, + args.format, + args.quality, + args.video, + args.resolution + ) + else: + logger.info("Skipping conversion step") + # TODO: Load previously converted tracks + + # Transfer step + if not args.skip_transfer: + transferred_count = self.transfer( + self.converted_tracks, + args.device + ) + + if transferred_count > 0: + logger.info(f"Successfully transferred {transferred_count} tracks to iPod") + else: + logger.error("Failed to transfer tracks to iPod") + else: + logger.info("Skipping transfer step") + + # Clean temp files + if args.clean_temp: + self.clean_temp_files() + + return 0 + + except Exception as e: + logger.exception(f"Error: {e}") + return 1 + + +if __name__ == "__main__": + cli = YouTubeToIPodCLI() + sys.exit(cli.run()) diff --git a/src/config_loader.py b/src/config_loader.py new file mode 100644 index 0000000..f9be2e2 --- /dev/null +++ b/src/config_loader.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +""" +Configuration Loader Module for iPod Nano Transfer Tool +Handles reading and writing configuration settings +""" + +import os +import logging +import configparser +from pathlib import Path +from typing import Any, Dict, Optional + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class ConfigLoader: + """Class to handle configuration loading and saving""" + + def __init__(self, config_file: Optional[str] = None): + """ + Initialize the configuration loader + + Args: + config_file: Path to configuration file (optional) + """ + # Default config file is in the project root directory + if not config_file: + self.config_file = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config.ini" + ) + else: + self.config_file = config_file + + self.config = configparser.ConfigParser() + + # Load configuration + self.load() + + def load(self) -> bool: + """ + Load configuration from file + + Returns: + True if successful, False otherwise + """ + try: + if os.path.exists(self.config_file): + logger.info(f"Loading configuration from {self.config_file}") + self.config.read(self.config_file) + return True + else: + logger.warning(f"Configuration file {self.config_file} not found, using defaults") + self._set_defaults() + return False + except Exception as e: + logger.error(f"Failed to load configuration: {e}") + self._set_defaults() + return False + + def save(self) -> bool: + """ + Save configuration to file + + Returns: + True if successful, False otherwise + """ + try: + logger.info(f"Saving configuration to {self.config_file}") + + with open(self.config_file, 'w') as f: + self.config.write(f) + + return True + except Exception as e: + logger.error(f"Failed to save configuration: {e}") + return False + + def _set_defaults(self): + """Set default configuration values""" + # General section + if not self.config.has_section('General'): + self.config.add_section('General') + + self.config.set('General', 'output_dir', os.path.join(os.path.expanduser("~"), "Music", "iPod")) + self.config.set('General', 'format', 'm4a') + self.config.set('General', 'quality', '256') + self.config.set('General', 'clean_temp', 'true') + + # Video section + if not self.config.has_section('Video'): + self.config.add_section('Video') + + self.config.set('Video', 'enable_video', 'false') + self.config.set('Video', 'resolution', '640x480') + + # Metadata section + if not self.config.has_section('Metadata'): + self.config.add_section('Metadata') + + self.config.set('Metadata', 'embed_artwork', 'true') + self.config.set('Metadata', 'use_musicbrainz', 'false') + + # Device section + if not self.config.has_section('Device'): + self.config.add_section('Device') + + self.config.set('Device', 'auto_detect', 'true') + self.config.set('Device', 'mount_point', '') + + # Advanced section + if not self.config.has_section('Advanced'): + self.config.add_section('Advanced') + + self.config.set('Advanced', 'temp_dir', 'temp') + self.config.set('Advanced', 'concurrent_downloads', '2') + self.config.set('Advanced', 'debug', 'false') + + def get(self, section: str, option: str, fallback: Any = None) -> Any: + """ + Get configuration value + + Args: + section: Configuration section + option: Configuration option + fallback: Fallback value if option not found + + Returns: + Configuration value + """ + try: + return self.config.get(section, option, fallback=fallback) + except (configparser.NoSectionError, configparser.NoOptionError): + return fallback + + def get_boolean(self, section: str, option: str, fallback: bool = False) -> bool: + """ + Get boolean configuration value + + Args: + section: Configuration section + option: Configuration option + fallback: Fallback value if option not found + + Returns: + Boolean configuration value + """ + try: + return self.config.getboolean(section, option, fallback=fallback) + except (configparser.NoSectionError, configparser.NoOptionError): + return fallback + + def get_int(self, section: str, option: str, fallback: int = 0) -> int: + """ + Get integer configuration value + + Args: + section: Configuration section + option: Configuration option + fallback: Fallback value if option not found + + Returns: + Integer configuration value + """ + try: + return self.config.getint(section, option, fallback=fallback) + except (configparser.NoSectionError, configparser.NoOptionError): + return fallback + + def get_float(self, section: str, option: str, fallback: float = 0.0) -> float: + """ + Get float configuration value + + Args: + section: Configuration section + option: Configuration option + fallback: Fallback value if option not found + + Returns: + Float configuration value + """ + try: + return self.config.getfloat(section, option, fallback=fallback) + except (configparser.NoSectionError, configparser.NoOptionError): + return fallback + + def set(self, section: str, option: str, value: Any) -> None: + """ + Set configuration value + + Args: + section: Configuration section + option: Configuration option + value: Configuration value + """ + if not self.config.has_section(section): + self.config.add_section(section) + + self.config.set(section, option, str(value)) + + def get_all(self) -> Dict[str, Dict[str, str]]: + """ + Get all configuration values + + Returns: + Dictionary of all configuration values + """ + result = {} + + for section in self.config.sections(): + result[section] = {} + + for option in self.config.options(section): + result[section][option] = self.config.get(section, option) + + return result + + +if __name__ == "__main__": + # Example usage + config = ConfigLoader() + + # Get configuration values + output_dir = config.get('General', 'output_dir') + format = config.get('General', 'format') + quality = config.get_int('General', 'quality') + + print(f"Output directory: {output_dir}") + print(f"Format: {format}") + print(f"Quality: {quality} kbps") + + # Set configuration value + config.set('General', 'format', 'mp3') + + # Save configuration + config.save() diff --git a/src/ipod_device.py b/src/ipod_device.py new file mode 100644 index 0000000..90e3072 --- /dev/null +++ b/src/ipod_device.py @@ -0,0 +1,562 @@ +#!/usr/bin/env python3 +""" +iPod Device Handler Module for iPod Nano Transfer Tool +Handles device detection and file transfer to iPod Nano +""" + +import os +import sys +import logging +import platform +import shutil +import subprocess +from typing import List, Dict, Optional, Tuple, Any +from pathlib import Path + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class IPodDevice: + """Class to handle iPod device detection and file transfer""" + + def __init__(self, mount_point: Optional[str] = None): + """ + Initialize the iPod device handler + + Args: + mount_point: Path to iPod mount point (optional) + """ + 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 + + def detect_devices(self) -> List[Dict[str, str]]: + """ + Detect connected iPod devices + + Returns: + 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 + }) + + 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: + 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"]: + 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: + 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 + }) + 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 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 + }) + + return devices + + def _get_ipod_mount_point(self, device_id: str) -> Optional[str]: + """ + Get the mount point for an iPod device + + Args: + device_id: Device ID + + 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 + + def mount_device(self, device_id: Optional[str] = None) -> Optional[str]: + """ + Mount iPod device + + Args: + device_id: Device ID (optional) + + Returns: + Mount point if successful, None otherwise + """ + 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) + + # 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 + + 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 + + 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 + + except subprocess.SubprocessError as e: + self.logger.error(f"Failed to mount device: {e}") + return None + + def unmount_device(self) -> bool: + """ + Unmount iPod device + + Returns: + True if successful, False otherwise + """ + if not self.mount_point: + 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 + + 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 + + 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 + + except subprocess.SubprocessError as e: + self.logger.error(f"Failed to unmount device: {e}") + return False + + def get_device_info(self) -> Dict[str, Any]: + """ + Get information about the mounted device + + Returns: + Dictionary with device information + """ + if not self.mount_point or not os.path.exists(self.mount_point): + self.logger.warning("No device mounted") + return {} + + info = { + "mount_point": self.mount_point, + "total_space": 0, + "free_space": 0, + "used_space": 0 + } + + try: + # Get disk usage + usage = shutil.disk_usage(self.mount_point) + info["total_space"] = usage.total + 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 + + return info + + except Exception as e: + self.logger.error(f"Failed to get device info: {e}") + return info + + def transfer_file(self, source_file: str, relative_path: Optional[str] = None) -> bool: + """ + Transfer file to iPod device + + Args: + source_file: Path to source file + relative_path: Relative path on iPod (optional) + + Returns: + True if successful, False otherwise + """ + if not self.mount_point or not os.path.exists(self.mount_point): + self.logger.warning("No device mounted") + return False + + if not os.path.exists(source_file): + self.logger.warning(f"Source file does not exist: {source_file}") + return False + + try: + # Determine destination path + 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) + + shutil.copy2(source_file, dest_file) + self.logger.info(f"File transferred: {dest_file}") + return True + + except Exception as e: + self.logger.error(f"Failed to transfer file: {e}") + return False + + def transfer_directory(self, source_dir: str, dest_dir: Optional[str] = None) -> bool: + """ + Transfer directory to iPod device + + Args: + source_dir: Path to source directory + dest_dir: Destination directory on iPod (optional) + + Returns: + True if successful, False otherwise + """ + if not self.mount_point or not os.path.exists(self.mount_point): + self.logger.warning("No device mounted") + return False + + if not os.path.exists(source_dir) or not os.path.isdir(source_dir): + self.logger.warning(f"Source directory does not exist: {source_dir}") + return False + + try: + # Determine destination path + 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) + dst_file = os.path.join(dest_subdir, file) + shutil.copy2(src_file, dst_file) + self.logger.info(f"Transferred: {dst_file}") + + self.logger.info(f"Directory transferred: {source_dir} -> {full_dest_dir}") + return True + + except Exception as e: + self.logger.error(f"Failed to transfer directory: {e}") + return False + + +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']})") + + # Mount device + 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() diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..c8f06b1 --- /dev/null +++ b/src/main.py @@ -0,0 +1,762 @@ +#!/usr/bin/env python3 +""" +Main Application Module for iPod Nano Transfer Tool +Provides a GUI for the YouTube Music to iPod Nano transfer tool +""" + +import os +import sys +import logging +import threading +import time +from typing import List, Dict, Optional, Tuple +from pathlib import Path + +from PyQt6.QtWidgets import ( + QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, + QPushButton, QLabel, QLineEdit, QProgressBar, QFileDialog, + QComboBox, QCheckBox, QTabWidget, QListWidget, QListWidgetItem, + QMessageBox, QSpinBox, QGroupBox, QRadioButton, QSplitter +) +from PyQt6.QtCore import Qt, QThread, pyqtSignal, pyqtSlot, QSize +from PyQt6.QtGui import QIcon, QPixmap + +from youtube_downloader import YouTubeDownloader, TrackInfo +from audio_converter import AudioConverter +from metadata_handler import MetadataHandler +from ipod_device import IPodDevice + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class WorkerThread(QThread): + """Worker thread for background tasks""" + + progress_signal = pyqtSignal(int, str) + finished_signal = pyqtSignal(bool, str, object) + + def __init__(self, task_type: str, **kwargs): + """ + Initialize worker thread + + Args: + task_type: Type of task to perform + **kwargs: Additional arguments for the task + """ + super().__init__() + self.task_type = task_type + self.kwargs = kwargs + self.is_running = True + + def run(self): + """Run the worker thread""" + try: + if self.task_type == "download": + self._run_download() + elif self.task_type == "convert": + self._run_convert() + elif self.task_type == "transfer": + self._run_transfer() + elif self.task_type == "detect_devices": + self._run_detect_devices() + else: + raise ValueError(f"Unknown task type: {self.task_type}") + + except Exception as e: + logger.exception(f"Error in worker thread: {e}") + self.finished_signal.emit(False, str(e), None) + + def _run_download(self): + """Run download task""" + url = self.kwargs.get("url") + output_dir = self.kwargs.get("output_dir", "downloads") + format = self.kwargs.get("format", "m4a") + quality = self.kwargs.get("quality", 256) + + self.progress_signal.emit(0, f"Initializing download from {url}") + + # Initialize downloader with progress callback + downloader = YouTubeDownloader( + output_dir=output_dir, + progress_callback=self._download_progress_callback + ) + + # Process URL + tracks = downloader.process_url(url, format, quality) + + if not tracks: + self.finished_signal.emit(False, "No tracks were successfully downloaded", []) + return + + # Report success + self.finished_signal.emit(True, f"Downloaded {len(tracks)} tracks", tracks) + + def _download_progress_callback(self, progress: int, status: str): + """Callback for download progress updates""" + self.progress_signal.emit(progress, status) + + def _run_convert(self): + """Run conversion task""" + tracks = self.kwargs.get("tracks", []) + output_dir = self.kwargs.get("output_dir", "converted") + format = self.kwargs.get("format", "m4a") + quality = self.kwargs.get("quality", 256) + + # 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)] + + if not valid_tracks: + self.finished_signal.emit(False, "No valid tracks to convert", None) + return + + self.progress_signal.emit(0, f"Initializing conversion of {len(valid_tracks)} tracks") + + # Initialize converter and metadata handler + converter = AudioConverter(output_dir=output_dir) + metadata_handler = MetadataHandler() + + # Convert each track + converted_tracks = [] + for i, track in enumerate(valid_tracks): + if not self.is_running: + break + + progress = int((i / len(valid_tracks)) * 100) + self.progress_signal.emit(progress, f"Converting {track.title}...") + + try: + # Convert to iPod format + output_path = converter.convert_to_ipod_format( + track, track.download_path, format, quality + ) + + # Embed metadata + if output_path: + metadata_handler.process_file(output_path, track) + converted_tracks.append((track, output_path)) + except Exception as e: + logger.error(f"Error converting track {track.title}: {e}") + continue + + if not converted_tracks: + self.finished_signal.emit(False, "Failed to convert any tracks", None) + return + + # Report success + self.finished_signal.emit( + True, + f"Converted {len(converted_tracks)} tracks", + converted_tracks + ) + + def _run_transfer(self): + """Run transfer task""" + tracks = self.kwargs.get("tracks", []) + mount_point = self.kwargs.get("mount_point") + + if not tracks: + self.finished_signal.emit(False, "No tracks to transfer", None) + return + + if not mount_point: + self.finished_signal.emit(False, "No iPod device mounted", None) + return + + self.progress_signal.emit(0, f"Initializing transfer of {len(tracks)} tracks") + + # Initialize iPod device + ipod = IPodDevice(mount_point=mount_point) + + # Get device info + device_info = ipod.get_device_info() + free_space = device_info.get("free_space", 0) + + # Calculate total size of tracks + total_size = 0 + for track, path in tracks: + if os.path.exists(path): + total_size += os.path.getsize(path) + + # Check if there's enough space + if total_size > free_space: + self.finished_signal.emit( + False, + f"Not enough space on iPod. Need {total_size / 1024**2:.1f} MB, " + + f"but only {free_space / 1024**2:.1f} MB available", + None + ) + return + + # Transfer each track + transferred_tracks = [] + for i, (track, path) in enumerate(tracks): + if not self.is_running: + break + + progress = int((i / len(tracks)) * 100) + self.progress_signal.emit(progress, f"Transferring {track.title}...") + + if os.path.exists(path): + # Transfer file + success = ipod.transfer_file(path) + if success: + transferred_tracks.append((track, path)) + + # Report success + self.finished_signal.emit( + True, + f"Transferred {len(transferred_tracks)} tracks to iPod", + transferred_tracks + ) + + def _run_detect_devices(self): + """Run device detection task""" + self.progress_signal.emit(0, "Detecting iPod devices...") + + # Initialize iPod device + ipod = IPodDevice() + + # Detect devices + devices = ipod.detect_devices() + + if not devices: + self.finished_signal.emit(False, "No iPod devices found", []) + return + + # Report success + self.finished_signal.emit(True, f"Found {len(devices)} iPod devices", devices) + + def stop(self): + """Stop the worker thread""" + self.is_running = False + + +class MainWindow(QMainWindow): + """Main application window""" + + def __init__(self): + """Initialize the main window""" + super().__init__() + + self.setWindowTitle("YouTube to iPod Nano") + self.setMinimumSize(800, 600) + + # Initialize variables + self.worker_thread = None + self.downloaded_tracks = [] + self.converted_tracks = [] + self.ipod_devices = [] + self.current_mount_point = None + + # Set up UI + self._setup_ui() + + def _setup_ui(self): + """Set up the user interface""" + # Main widget and layout + main_widget = QWidget() + main_layout = QVBoxLayout(main_widget) + + # Tab widget + tab_widget = QTabWidget() + main_layout.addWidget(tab_widget) + + # Download tab + download_tab = QWidget() + tab_widget.addTab(download_tab, "Download") + self._setup_download_tab(download_tab) + + # Convert tab + convert_tab = QWidget() + tab_widget.addTab(convert_tab, "Convert") + self._setup_convert_tab(convert_tab) + + # Transfer tab + transfer_tab = QWidget() + tab_widget.addTab(transfer_tab, "Transfer") + self._setup_transfer_tab(transfer_tab) + + # Settings tab + settings_tab = QWidget() + tab_widget.addTab(settings_tab, "Settings") + self._setup_settings_tab(settings_tab) + + # Status bar + self.statusBar().showMessage("Ready") + + # Set central widget + self.setCentralWidget(main_widget) + + def _setup_download_tab(self, tab): + """Set up the download tab""" + layout = QVBoxLayout(tab) + + # URL input + url_layout = QHBoxLayout() + url_label = QLabel("YouTube URL:") + self.url_input = QLineEdit() + self.url_input.setPlaceholderText("https://www.youtube.com/watch?v=...") + url_layout.addWidget(url_label) + url_layout.addWidget(self.url_input) + layout.addLayout(url_layout) + + # Format selection + format_layout = QHBoxLayout() + format_label = QLabel("Format:") + self.format_combo = QComboBox() + self.format_combo.addItems(["m4a", "mp3"]) + quality_label = QLabel("Quality (kbps):") + self.quality_spin = QSpinBox() + self.quality_spin.setRange(128, 320) + self.quality_spin.setValue(256) + self.quality_spin.setSingleStep(32) + format_layout.addWidget(format_label) + format_layout.addWidget(self.format_combo) + format_layout.addWidget(quality_label) + format_layout.addWidget(self.quality_spin) + format_layout.addStretch() + layout.addLayout(format_layout) + + # Download button + self.download_button = QPushButton("Download") + self.download_button.clicked.connect(self._on_download_clicked) + layout.addWidget(self.download_button) + + # Progress bar + self.download_progress = QProgressBar() + self.download_progress.setRange(0, 100) + self.download_progress.setValue(0) + layout.addWidget(self.download_progress) + + # Status label + self.download_status = QLabel("Ready to download") + layout.addWidget(self.download_status) + + # Track list + track_list_label = QLabel("Downloaded Tracks:") + layout.addWidget(track_list_label) + + self.track_list = QListWidget() + layout.addWidget(self.track_list) + + def _setup_convert_tab(self, tab): + """Set up the convert tab""" + layout = QVBoxLayout(tab) + + # Output directory + output_layout = QHBoxLayout() + output_label = QLabel("Output Directory:") + self.output_dir_input = QLineEdit() + self.output_dir_input.setText(os.path.join(os.path.expanduser("~"), "Music", "iPod")) + browse_button = QPushButton("Browse...") + browse_button.clicked.connect(self._on_browse_output_clicked) + output_layout.addWidget(output_label) + output_layout.addWidget(self.output_dir_input) + output_layout.addWidget(browse_button) + layout.addLayout(output_layout) + + # Convert button + self.convert_button = QPushButton("Convert Downloaded Tracks") + self.convert_button.clicked.connect(self._on_convert_clicked) + layout.addWidget(self.convert_button) + + # Progress bar + self.convert_progress = QProgressBar() + self.convert_progress.setRange(0, 100) + self.convert_progress.setValue(0) + layout.addWidget(self.convert_progress) + + # Status label + self.convert_status = QLabel("Ready to convert") + layout.addWidget(self.convert_status) + + # Converted track list + converted_list_label = QLabel("Converted Tracks:") + layout.addWidget(converted_list_label) + + self.converted_list = QListWidget() + layout.addWidget(self.converted_list) + + def _setup_transfer_tab(self, tab): + """Set up the transfer tab""" + layout = QVBoxLayout(tab) + + # Device selection + device_layout = QHBoxLayout() + device_label = QLabel("iPod Device:") + self.device_combo = QComboBox() + self.device_combo.setPlaceholderText("No devices found") + refresh_button = QPushButton("Refresh") + refresh_button.clicked.connect(self._on_refresh_devices_clicked) + device_layout.addWidget(device_label) + device_layout.addWidget(self.device_combo) + device_layout.addWidget(refresh_button) + layout.addLayout(device_layout) + + # Device info + self.device_info_label = QLabel("No device selected") + layout.addWidget(self.device_info_label) + + # Transfer button + self.transfer_button = QPushButton("Transfer to iPod") + self.transfer_button.clicked.connect(self._on_transfer_clicked) + layout.addWidget(self.transfer_button) + + # Progress bar + self.transfer_progress = QProgressBar() + self.transfer_progress.setRange(0, 100) + self.transfer_progress.setValue(0) + layout.addWidget(self.transfer_progress) + + # Status label + self.transfer_status = QLabel("Ready to transfer") + layout.addWidget(self.transfer_status) + + # Transferred track list + transferred_list_label = QLabel("Transferred Tracks:") + layout.addWidget(transferred_list_label) + + self.transferred_list = QListWidget() + layout.addWidget(self.transferred_list) + + def _setup_settings_tab(self, tab): + """Set up the settings tab""" + layout = QVBoxLayout(tab) + + # Video support + video_group = QGroupBox("Video Support") + video_layout = QVBoxLayout(video_group) + + self.enable_video_check = QCheckBox("Enable Video Download (iPod Nano 5th-7th gen)") + video_layout.addWidget(self.enable_video_check) + + video_resolution_layout = QHBoxLayout() + video_resolution_label = QLabel("Video Resolution:") + self.video_resolution_combo = QComboBox() + self.video_resolution_combo.addItems(["640x480", "480x360", "320x240"]) + self.video_resolution_combo.setEnabled(False) + self.enable_video_check.toggled.connect(self.video_resolution_combo.setEnabled) + + video_resolution_layout.addWidget(video_resolution_label) + video_resolution_layout.addWidget(self.video_resolution_combo) + video_resolution_layout.addStretch() + video_layout.addLayout(video_resolution_layout) + + layout.addWidget(video_group) + + # Metadata options + metadata_group = QGroupBox("Metadata Options") + metadata_layout = QVBoxLayout(metadata_group) + + self.embed_artwork_check = QCheckBox("Embed Album Artwork") + self.embed_artwork_check.setChecked(True) + metadata_layout.addWidget(self.embed_artwork_check) + + self.use_musicbrainz_check = QCheckBox("Use MusicBrainz for Enhanced Metadata (Experimental)") + metadata_layout.addWidget(self.use_musicbrainz_check) + + layout.addWidget(metadata_group) + + # Advanced options + advanced_group = QGroupBox("Advanced Options") + advanced_layout = QVBoxLayout(advanced_group) + + self.clean_temp_check = QCheckBox("Clean Temporary Files After Transfer") + self.clean_temp_check.setChecked(True) + advanced_layout.addWidget(self.clean_temp_check) + + self.auto_detect_check = QCheckBox("Auto-Detect iPod on Startup") + self.auto_detect_check.setChecked(True) + advanced_layout.addWidget(self.auto_detect_check) + + layout.addWidget(advanced_group) + + # Spacer + layout.addStretch() + + # About section + about_label = QLabel( + "YouTube to iPod Nano Transfer Tool\n" + "Version 1.0.0\n\n" + "An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano." + ) + about_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + layout.addWidget(about_label) + + def _on_browse_output_clicked(self): + """Handle browse output directory button click""" + directory = QFileDialog.getExistingDirectory( + self, "Select Output Directory", self.output_dir_input.text() + ) + + if directory: + self.output_dir_input.setText(directory) + + def _on_download_clicked(self): + """Handle download button click""" + url = self.url_input.text().strip() + + if not url: + QMessageBox.warning(self, "Error", "Please enter a YouTube URL") + return + + if not (url.startswith("http://") or url.startswith("https://")): + QMessageBox.warning(self, "Error", "Please enter a valid URL") + return + + # Disable download button + self.download_button.setEnabled(False) + self.download_status.setText("Downloading...") + self.download_progress.setValue(0) + + # Clear track list + self.track_list.clear() + + # Start worker thread + self.worker_thread = WorkerThread( + task_type="download", + url=url, + format=self.format_combo.currentText(), + quality=self.quality_spin.value() + ) + + self.worker_thread.progress_signal.connect(self._on_download_progress) + self.worker_thread.finished_signal.connect(self._on_download_finished) + self.worker_thread.start() + + def _on_download_progress(self, progress, status): + """Handle download progress updates""" + self.download_progress.setValue(progress) + self.download_status.setText(status) + + def _on_download_finished(self, success, message, result): + """Handle download completion""" + # Enable download button + self.download_button.setEnabled(True) + + if success and result: + self.download_status.setText(message) + self.downloaded_tracks = result + + # Update track list + self.track_list.clear() + for track in self.downloaded_tracks: + if hasattr(track, 'download_path') and track.download_path: + item = QListWidgetItem(f"{track.artist} - {track.title}") + self.track_list.addItem(item) + + # Enable convert button if we have tracks + if self.track_list.count() > 0: + self.convert_button.setEnabled(True) + else: + self.convert_button.setEnabled(False) + self.download_status.setText("No valid tracks were downloaded") + else: + self.download_status.setText(f"Error: {message}") + QMessageBox.warning(self, "Download Error", message) + self.downloaded_tracks = [] + self.track_list.clear() + + # Clean up worker thread + self.worker_thread = None + + 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.") + return + + output_dir = self.output_dir_input.text().strip() + + if not output_dir: + QMessageBox.warning(self, "Error", "Please enter an output directory") + return + + # Disable convert button + self.convert_button.setEnabled(False) + self.convert_status.setText("Converting...") + self.convert_progress.setValue(0) + + # Clear converted list + self.converted_list.clear() + + # 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() + ) + + self.worker_thread.progress_signal.connect(self._on_convert_progress) + self.worker_thread.finished_signal.connect(self._on_convert_finished) + self.worker_thread.start() + + def _on_convert_progress(self, progress, status): + """Handle convert progress updates""" + self.convert_progress.setValue(progress) + self.convert_status.setText(status) + + def _on_convert_finished(self, success, message, result): + """Handle convert completion""" + # Enable convert button + self.convert_button.setEnabled(True) + + if success: + self.convert_status.setText(message) + self.converted_tracks = result + + # Update converted list + for track, path in self.converted_tracks: + item = QListWidgetItem(f"{track.artist} - {track.title}") + self.converted_list.addItem(item) + else: + self.convert_status.setText(f"Error: {message}") + QMessageBox.warning(self, "Conversion Error", message) + + # Clean up worker thread + self.worker_thread = None + + def _on_refresh_devices_clicked(self): + """Handle refresh devices button click""" + # Disable refresh button + self.device_combo.clear() + self.device_info_label.setText("Detecting devices...") + + # Start worker thread + self.worker_thread = WorkerThread(task_type="detect_devices") + + self.worker_thread.progress_signal.connect(self._on_device_detection_progress) + self.worker_thread.finished_signal.connect(self._on_device_detection_finished) + self.worker_thread.start() + + def _on_device_detection_progress(self, progress, status): + """Handle device detection progress updates""" + self.device_info_label.setText(status) + + def _on_device_detection_finished(self, success, message, result): + """Handle device detection completion""" + if success: + self.ipod_devices = result + + # Update device combo box + self.device_combo.clear() + for device in self.ipod_devices: + self.device_combo.addItem(device.get("name", "Unknown Device")) + + # Try to mount the first device + if self.ipod_devices: + device_id = self.ipod_devices[0]["id"] + ipod = IPodDevice() + mount_point = ipod.mount_device(device_id) + + if mount_point: + # Get device info + self.current_mount_point = mount_point + device_info = ipod.get_device_info() + self.ipod_devices[0]["mount_point"] = mount_point + self.ipod_devices[0]["info"] = device_info + + # Update device info label + free_space = device_info.get("free_space", 0) + total_space = device_info.get("total_space", 0) + + self.device_info_label.setText( + f"Device: {self.ipod_devices[0]['name']}\n" + f"Mount Point: {mount_point}\n" + f"Free Space: {free_space / 1024**2:.1f} MB / {total_space / 1024**2:.1f} MB" + ) + + # Enable transfer button + self.transfer_button.setEnabled(True) + else: + self.device_info_label.setText( + f"Device: {self.ipod_devices[0]['name']}\n" + f"Failed to mount device. Please check permissions." + ) + self.transfer_button.setEnabled(False) + + self.statusBar().showMessage(message) + else: + self.device_info_label.setText("No iPod devices found") + self.device_combo.clear() + self.transfer_button.setEnabled(False) + self.statusBar().showMessage(message) + + # Clean up worker thread + self.worker_thread = None + + def _on_transfer_clicked(self): + """Handle transfer button click""" + if not self.converted_tracks: + QMessageBox.warning(self, "Error", "No tracks to transfer. Convert tracks first.") + return + + if not self.current_mount_point: + QMessageBox.warning(self, "Error", "No iPod device mounted. Refresh devices first.") + return + + # Disable transfer button + self.transfer_button.setEnabled(False) + self.transfer_status.setText("Transferring...") + self.transfer_progress.setValue(0) + + # Clear transferred list + self.transferred_list.clear() + + # Start worker thread + self.worker_thread = WorkerThread( + task_type="transfer", + tracks=self.converted_tracks, + mount_point=self.current_mount_point + ) + + self.worker_thread.progress_signal.connect(self._on_transfer_progress) + self.worker_thread.finished_signal.connect(self._on_transfer_finished) + self.worker_thread.start() + + def _on_transfer_progress(self, progress, status): + """Handle transfer progress updates""" + self.transfer_progress.setValue(progress) + self.transfer_status.setText(status) + + def _on_transfer_finished(self, success, message, result): + """Handle transfer completion""" + # Enable transfer button + self.transfer_button.setEnabled(True) + + 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) + + QMessageBox.information(self, "Transfer Complete", message) + else: + self.transfer_status.setText(f"Error: {message}") + QMessageBox.warning(self, "Transfer Error", message) + + # Clean up worker thread + self.worker_thread = None + + def closeEvent(self, event): + """Handle window close event""" + # Stop worker thread if running + if self.worker_thread and self.worker_thread.isRunning(): + self.worker_thread.stop() + self.worker_thread.wait() + + # Accept close event + event.accept() + + +if __name__ == "__main__": + app = QApplication(sys.argv) + window = MainWindow() + window.show() + sys.exit(app.exec()) diff --git a/src/metadata_handler.py b/src/metadata_handler.py new file mode 100644 index 0000000..5dc4299 --- /dev/null +++ b/src/metadata_handler.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +""" +Metadata Handler Module for iPod Nano Transfer Tool +Handles metadata injection and album art embedding +""" + +import os +import logging +import tempfile +from typing import Optional, Dict, Any +from io import BytesIO +from pathlib import Path + +import requests +from PIL import Image +from mutagen.mp4 import MP4, MP4Cover +from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK +from mutagen.easyid3 import EasyID3 + +from youtube_downloader import TrackInfo + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + + +class MetadataHandler: + """Class to handle metadata injection and album art embedding""" + + def __init__(self, temp_dir: Optional[str] = None): + """ + Initialize the metadata handler + + Args: + temp_dir: Directory for temporary files (optional) + """ + self.temp_dir = temp_dir or tempfile.gettempdir() + os.makedirs(self.temp_dir, exist_ok=True) + + def download_thumbnail(self, thumbnail_url: str) -> Optional[str]: + """ + Download thumbnail image from URL + + Args: + thumbnail_url: URL of the thumbnail image + + Returns: + Path to downloaded thumbnail file, or None if download failed + """ + if not thumbnail_url: + logger.warning("No thumbnail URL provided") + return None + + try: + logger.info(f"Downloading thumbnail: {thumbnail_url}") + response = requests.get(thumbnail_url, timeout=10) + response.raise_for_status() + + # Save thumbnail to temporary file + thumbnail_path = os.path.join(self.temp_dir, f"thumbnail_{hash(thumbnail_url)}.jpg") + + # Process image with PIL to ensure it's a valid format + img = Image.open(BytesIO(response.content)) + img.save(thumbnail_path, "JPEG") + + logger.info(f"Thumbnail saved to: {thumbnail_path}") + return thumbnail_path + + except (requests.RequestException, IOError) as e: + logger.error(f"Failed to download thumbnail: {e}") + return None + + def embed_metadata_m4a(self, + file_path: str, + track_info: TrackInfo, + thumbnail_path: Optional[str] = None) -> bool: + """ + Embed metadata and album art in M4A/AAC file + + Args: + file_path: Path to M4A/AAC file + track_info: TrackInfo object with track metadata + thumbnail_path: Path to thumbnail image (optional) + + Returns: + True if successful, False otherwise + """ + logger.info(f"Embedding metadata in M4A file: {file_path}") + + try: + audio = MP4(file_path) + + # Clear existing tags + audio.clear() + + # Add metadata tags + audio['\xa9nam'] = [track_info.title] # Title + audio['\xa9ART'] = [track_info.artist] # Artist + audio['\xa9alb'] = [track_info.album or track_info.artist] # Album + + # Add track number if available + if track_info.track_number is not None: + audio['trkn'] = [(track_info.track_number, 0)] + + # Add album art if available + if thumbnail_path: + with open(thumbnail_path, 'rb') as f: + album_art_data = f.read() + + # Determine image format + if album_art_data.startswith(b'\xff\xd8'): + # JPEG + cover = MP4Cover(album_art_data, imageformat=MP4Cover.FORMAT_JPEG) + else: + # PNG + cover = MP4Cover(album_art_data, imageformat=MP4Cover.FORMAT_PNG) + + audio['covr'] = [cover] + + # Save changes + audio.save() + logger.info(f"Metadata embedded successfully: {file_path}") + return True + + except Exception as e: + logger.error(f"Failed to embed metadata in M4A file: {e}") + return False + + def embed_metadata_mp3(self, + file_path: str, + track_info: TrackInfo, + thumbnail_path: Optional[str] = None) -> bool: + """ + Embed metadata and album art in MP3 file + + Args: + file_path: Path to MP3 file + track_info: TrackInfo object with track metadata + thumbnail_path: Path to thumbnail image (optional) + + Returns: + True if successful, False otherwise + """ + logger.info(f"Embedding metadata in MP3 file: {file_path}") + + try: + # First try with EasyID3 for basic tags + try: + audio = EasyID3(file_path) + except: + # If the file doesn't have an ID3 tag, add one + audio = ID3() + audio.save(file_path) + audio = EasyID3(file_path) + + # Clear existing tags + audio.clear() + + # Add metadata tags + audio['title'] = track_info.title + audio['artist'] = track_info.artist + audio['album'] = track_info.album or track_info.artist + + # Add track number if available + if track_info.track_number is not None: + audio['tracknumber'] = str(track_info.track_number) + + # Save basic tags + audio.save() + + # Now add album art using ID3 + if thumbnail_path: + audio = ID3(file_path) + + with open(thumbnail_path, 'rb') as f: + album_art_data = f.read() + + # Add album art + audio.add( + APIC( + encoding=3, # UTF-8 + mime='image/jpeg', + type=3, # Cover (front) + desc='Cover', + data=album_art_data + ) + ) + + # Save with album art + audio.save() + + logger.info(f"Metadata embedded successfully: {file_path}") + return True + + except Exception as e: + logger.error(f"Failed to embed metadata in MP3 file: {e}") + return False + + def process_file(self, + file_path: str, + track_info: TrackInfo) -> bool: + """ + Process audio file to embed metadata and album art + + Args: + file_path: Path to audio file + track_info: TrackInfo object with track metadata + + Returns: + True if successful, False otherwise + """ + logger.info(f"Processing metadata for file: {file_path}") + + # Download thumbnail if available + thumbnail_path = None + if track_info.thumbnail_url: + thumbnail_path = self.download_thumbnail(track_info.thumbnail_url) + + # Determine file type and process accordingly + file_ext = Path(file_path).suffix.lower() + + if file_ext in ['.m4a', '.aac', '.mp4']: + return self.embed_metadata_m4a(file_path, track_info, thumbnail_path) + elif file_ext == '.mp3': + return self.embed_metadata_mp3(file_path, track_info, thumbnail_path) + else: + logger.warning(f"Unsupported file format: {file_ext}") + return False + + def enhance_metadata_with_musicbrainz(self, track_info: TrackInfo) -> TrackInfo: + """ + Enhance track metadata using MusicBrainz API + + Args: + track_info: TrackInfo object with track metadata + + Returns: + Enhanced TrackInfo object + """ + # This is a placeholder for future implementation + # MusicBrainz integration would go here + return track_info + + +if __name__ == "__main__": + # Example usage + handler = MetadataHandler() + + # Test metadata embedding + # handler.process_file( + # "test.m4a", + # TrackInfo( + # video_id="dQw4w9WgXcQ", + # title="Never Gonna Give You Up", + # artist="Rick Astley", + # album="Whenever You Need Somebody", + # thumbnail_url="https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg", + # duration=213, + # track_number=1 + # ) + # ) diff --git a/src/youtube_downloader.py b/src/youtube_downloader.py new file mode 100644 index 0000000..1b200b9 --- /dev/null +++ b/src/youtube_downloader.py @@ -0,0 +1,415 @@ +#!/usr/bin/env python3 +""" +YouTube Downloader Module for iPod Nano Transfer Tool +Handles downloading audio from YouTube videos and playlists +""" + +import os +import re +import logging +import time +from typing import List, Dict, Optional, Tuple, Callable +from dataclasses import dataclass + +import yt_dlp +from tqdm import tqdm + +# Configure logging +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) + +@dataclass +class TrackInfo: + """Data class to store track information""" + video_id: str + title: str + artist: str + album: str + thumbnail_url: str + duration: int + track_number: Optional[int] = None + playlist_id: Optional[str] = None + playlist_title: Optional[str] = None + playlist_index: Optional[int] = None + download_path: Optional[str] = None + + +class YouTubeDownloader: + """Class to handle YouTube video and playlist downloads""" + + def __init__(self, output_dir: str = "downloads", temp_dir: str = "temp", progress_callback: Optional[Callable] = None): + """ + Initialize the YouTube downloader + + Args: + output_dir: Directory to save downloaded files + temp_dir: Directory for temporary files + progress_callback: Optional callback function for progress updates + """ + self.output_dir = output_dir + self.temp_dir = temp_dir + self.progress_callback = progress_callback + + # Create directories if they don't exist + os.makedirs(output_dir, exist_ok=True) + os.makedirs(temp_dir, exist_ok=True) + + # Playlist regex pattern + self.playlist_regex = r'(?:list=)([a-zA-Z0-9_-]+)' + + # Track download progress + self.download_start_time = 0 + self.current_track = None + + def _parse_artist_title(self, video_title: str) -> Tuple[str, str]: + """ + Parse artist and title from video title + + Args: + video_title: YouTube video title + + Returns: + Tuple of (artist, title) + """ + # Common patterns: "Artist - Title", "Artist: Title", "Artist | Title" + patterns = [ + r'^(.*?)\s*-\s*(.*?)$', # Artist - Title + r'^(.*?)\s*:\s*(.*?)$', # Artist: Title + r'^(.*?)\s*\|\s*(.*?)$' # Artist | Title + ] + + for pattern in patterns: + match = re.match(pattern, video_title) + if match: + artist, title = match.groups() + return artist.strip(), title.strip() + + # If no pattern matches, return empty artist and full title + return "", video_title.strip() + + def extract_video_info(self, video_url: str) -> Optional[TrackInfo]: + """ + Extract information from a YouTube video + + Args: + video_url: YouTube video URL + + Returns: + TrackInfo object with video metadata or None if video is unavailable + """ + logger.info(f"Extracting info from video: {video_url}") + + ydl_opts = { + 'quiet': True, + 'no_warnings': True, + 'skip_download': True, + 'extract_flat': True, + 'ignoreerrors': True, # Continue on download errors + } + + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(video_url, download=False) + + if not info: + logger.warning(f"Could not extract info from video: {video_url}") + return None + + video_id = info.get('id') + video_title = info.get('title', '') + artist, title = self._parse_artist_title(video_title) + + # Get best thumbnail + thumbnails = info.get('thumbnails', []) + thumbnail_url = next((t['url'] for t in reversed(thumbnails) + if 'url' in t), None) if thumbnails else None + + return TrackInfo( + video_id=video_id, + title=title, + artist=artist, + album=artist, # Default album to artist name + thumbnail_url=thumbnail_url, + duration=info.get('duration', 0) + ) + except Exception as e: + logger.warning(f"Error extracting video info: {e}") + return None + + def extract_playlist_info(self, playlist_url: str) -> List[TrackInfo]: + """ + Extract information from a YouTube playlist + + Args: + playlist_url: YouTube playlist URL + + Returns: + List of TrackInfo objects for each video in the playlist + """ + logger.info(f"Extracting info from playlist: {playlist_url}") + + # Extract playlist ID + playlist_id_match = re.search(self.playlist_regex, playlist_url) + if not playlist_id_match: + raise ValueError(f"Invalid playlist URL: {playlist_url}") + + playlist_id = playlist_id_match.group(1) + + ydl_opts = { + 'quiet': True, + 'no_warnings': True, + 'extract_flat': True, + 'skip_download': True, + 'ignoreerrors': True, # Continue on download errors + } + + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + playlist_info = ydl.extract_info( + f'https://www.youtube.com/playlist?list={playlist_id}', + download=False + ) + + if not playlist_info or 'entries' not in playlist_info: + logger.warning(f"Could not extract playlist info: {playlist_url}") + return [] + + playlist_title = playlist_info.get('title', 'Unknown Playlist') + tracks = [] + track_number = 1 # Keep track of actual track numbers + + # Update progress if callback is available + if self.progress_callback: + self.progress_callback(0, f"Analyzing playlist: {playlist_title}") + + total_entries = len(playlist_info.get('entries', [])) + + for idx, entry in enumerate(playlist_info.get('entries', []), 1): + # Update progress for playlist analysis + if self.progress_callback: + progress = int((idx / total_entries) * 10) # Use first 10% for playlist analysis + self.progress_callback(progress, f"Analyzing playlist: {idx}/{total_entries} tracks") + + if not entry or entry.get('_type') != 'url': + continue + + try: + video_info = self.extract_video_info(f"https://www.youtube.com/watch?v={entry['id']}") + if video_info: + video_info.playlist_id = playlist_id + video_info.playlist_title = playlist_title + video_info.playlist_index = idx + video_info.track_number = track_number + video_info.album = playlist_title # Set album to playlist title + tracks.append(video_info) + track_number += 1 + else: + logger.warning(f"Skipping unavailable video in playlist: {entry.get('id')}") + except Exception as e: + logger.warning(f"Error processing playlist entry {idx}: {e}") + + return tracks + except Exception as e: + logger.error(f"Error extracting playlist info: {e}") + return [] + + def download_audio(self, track_info: TrackInfo, format: str = "m4a", quality: int = 256) -> Optional[str]: + """ + Download audio from YouTube video + + Args: + track_info: TrackInfo object with video metadata + format: Audio format (m4a, mp3) + quality: Audio quality in kbps + + Returns: + Path to downloaded file or None if download failed + """ + logger.info(f"Downloading audio for: {track_info.title}") + self.current_track = track_info + self.download_start_time = time.time() + + # Create output filename + safe_title = re.sub(r'[^\w\-_\. ]', '_', track_info.title) + if track_info.track_number: + filename = f"{track_info.track_number:02d}. {safe_title}.{format}" + else: + filename = f"{safe_title}.{format}" + + output_path = os.path.join(self.temp_dir, filename) + + # Set up yt-dlp options + ydl_opts = { + 'format': 'bestaudio/best', + 'outtmpl': output_path, + 'postprocessors': [{ + 'key': 'FFmpegExtractAudio', + 'preferredcodec': format, + 'preferredquality': str(quality), + }], + 'quiet': False, + 'progress_hooks': [self._download_progress_hook], + 'ignoreerrors': True, # Continue on download errors + } + + try: + # Download the file + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + ydl.download([f"https://www.youtube.com/watch?v={track_info.video_id}"]) + + # Check if file was actually downloaded + expected_path = f"{output_path}.{format}" + if os.path.exists(expected_path): + # Update track_info with download path + track_info.download_path = expected_path + return track_info.download_path + else: + logger.warning(f"Download failed for: {track_info.title}") + return None + except Exception as e: + logger.error(f"Error downloading audio: {e}") + return None + + def _download_progress_hook(self, d: Dict): + """Progress hook for yt-dlp""" + if d['status'] == 'downloading': + # Calculate download progress + if 'total_bytes' in d and d['total_bytes'] > 0: + percent = d['downloaded_bytes'] / d['total_bytes'] * 100 + + # Calculate download speed + elapsed_time = time.time() - self.download_start_time + if elapsed_time > 0: + download_speed = d['downloaded_bytes'] / elapsed_time / 1024 # KB/s + + # Format download speed + speed_str = f"{download_speed:.1f} KB/s" + if download_speed > 1024: + speed_str = f"{download_speed/1024:.1f} MB/s" + + # Format file size + total_size_mb = d['total_bytes'] / (1024 * 1024) + downloaded_mb = d['downloaded_bytes'] / (1024 * 1024) + + # Create status message + status_msg = f"Downloading {self.current_track.title}: {downloaded_mb:.1f}/{total_size_mb:.1f} MB ({percent:.1f}%) at {speed_str}" + + # Log progress + logger.info(status_msg) + + # Call progress callback if available + if self.progress_callback: + # Scale to 10-95% range to leave room for playlist analysis and completion + scaled_percent = 10 + (percent * 0.85) + self.progress_callback(int(scaled_percent), status_msg) + + # If we don't have total_bytes, show indeterminate progress + elif 'downloaded_bytes' in d: + elapsed_time = time.time() - self.download_start_time + if elapsed_time > 0: + download_speed = d['downloaded_bytes'] / elapsed_time / 1024 # KB/s + + # Format download speed + speed_str = f"{download_speed:.1f} KB/s" + if download_speed > 1024: + speed_str = f"{download_speed/1024:.1f} MB/s" + + # Format file size + downloaded_mb = d['downloaded_bytes'] / (1024 * 1024) + + # Create status message + status_msg = f"Downloading {self.current_track.title}: {downloaded_mb:.1f} MB at {speed_str}" + + # Log progress + logger.info(status_msg) + + # Call progress callback if available + if self.progress_callback: + # Use a pulsing progress between 10-90% for indeterminate progress + pulse_progress = 10 + (int(time.time() * 10) % 80) + self.progress_callback(pulse_progress, status_msg) + + elif d['status'] == 'finished': + logger.info(f"Download complete: {self.current_track.title}") + if self.progress_callback: + self.progress_callback(95, f"Download complete: {self.current_track.title}") + + elif d['status'] == 'error': + logger.warning(f"Download error: {self.current_track.title}") + if self.progress_callback: + self.progress_callback(95, f"Download error: {self.current_track.title}") + + def process_url(self, url: str, format: str = "m4a", quality: int = 256) -> List[TrackInfo]: + """ + Process a YouTube URL (video or playlist) + + Args: + url: YouTube URL + format: Audio format + quality: Audio quality in kbps + + Returns: + List of TrackInfo objects with download paths + """ + successful_tracks = [] + + try: + # Check if URL is a playlist + if 'list=' in url: + # Process as playlist + tracks = self.extract_playlist_info(url) + + if not tracks: + logger.warning("No valid tracks found in playlist") + return [] + + # Download each track + for i, track in enumerate(tracks): + if self.progress_callback: + status_msg = f"Preparing to download {i+1}/{len(tracks)}: {track.title}" + self.progress_callback(10, status_msg) + + try: + download_path = self.download_audio(track, format, quality) + if download_path: + successful_tracks.append(track) + except Exception as e: + logger.error(f"Error downloading track {track.title}: {e}") + + else: + # Process as single video + track = self.extract_video_info(url) + if track: + try: + download_path = self.download_audio(track, format, quality) + if download_path: + successful_tracks.append(track) + except Exception as e: + logger.error(f"Error downloading track {track.title}: {e}") + else: + logger.warning("Could not extract video information") + + # Final progress update + if self.progress_callback: + self.progress_callback(100, f"Downloaded {len(successful_tracks)} tracks successfully") + + return successful_tracks + + except Exception as e: + logger.error(f"Error processing URL: {e}") + if self.progress_callback: + self.progress_callback(100, f"Error: {str(e)}") + return successful_tracks + + +if __name__ == "__main__": + # Example usage + downloader = YouTubeDownloader() + + # Test with a single video + video_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" + video_info = downloader.extract_video_info(video_url) + print(f"Video: {video_info.artist} - {video_info.title}") + + # Uncomment to test download + # downloader.download_audio(video_info) diff --git a/tests/test_basic.py b/tests/test_basic.py new file mode 100644 index 0000000..2313cfb --- /dev/null +++ b/tests/test_basic.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +""" +Basic tests for YouTube Music to iPod Nano Transfer Tool +""" + +import os +import sys +import unittest +from unittest.mock import patch, MagicMock + +# Add src directory to path +sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src")) + +from youtube_downloader import YouTubeDownloader, TrackInfo +from audio_converter import AudioConverter +from metadata_handler import MetadataHandler +from ipod_device import IPodDevice + + +class TestYouTubeDownloader(unittest.TestCase): + """Test YouTube downloader functionality""" + + def test_parse_artist_title(self): + """Test parsing artist and title from video title""" + downloader = YouTubeDownloader() + + # Test case 1: Artist - Title + artist, title = downloader._parse_artist_title("Rick Astley - Never Gonna Give You Up") + self.assertEqual(artist, "Rick Astley") + self.assertEqual(title, "Never Gonna Give You Up") + + # Test case 2: Artist: Title + artist, title = downloader._parse_artist_title("Rick Astley: Never Gonna Give You Up") + self.assertEqual(artist, "Rick Astley") + self.assertEqual(title, "Never Gonna Give You Up") + + # Test case 3: Artist | Title + artist, title = downloader._parse_artist_title("Rick Astley | Never Gonna Give You Up") + self.assertEqual(artist, "Rick Astley") + self.assertEqual(title, "Never Gonna Give You Up") + + # Test case 4: No pattern match + artist, title = downloader._parse_artist_title("Never Gonna Give You Up") + self.assertEqual(artist, "") + self.assertEqual(title, "Never Gonna Give You Up") + + +class TestAudioConverter(unittest.TestCase): + """Test audio converter functionality""" + + def test_sanitize_filename(self): + """Test filename sanitization""" + converter = AudioConverter() + + # Test case 1: Invalid characters + sanitized = converter._sanitize_filename('File: with "invalid" chars?') + self.assertEqual(sanitized, 'File_ with _invalid_ chars_') + + # Test case 2: Long filename + long_name = "A" * 150 + sanitized = converter._sanitize_filename(long_name) + self.assertEqual(len(sanitized), 100) + self.assertTrue(sanitized.endswith('...')) + + +class TestMetadataHandler(unittest.TestCase): + """Test metadata handler functionality""" + + @patch('requests.get') + def test_download_thumbnail(self, mock_get): + """Test thumbnail download""" + # Mock response + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_response.content = b'fake_image_data' + mock_get.return_value = mock_response + + # Mock PIL Image + with patch('PIL.Image.open') as mock_image_open: + mock_image = MagicMock() + mock_image.save.return_value = None + mock_image_open.return_value = mock_image + + handler = MetadataHandler() + result = handler.download_thumbnail("https://example.com/image.jpg") + + # Check that the function returned a path + self.assertIsNotNone(result) + mock_get.assert_called_once_with("https://example.com/image.jpg", timeout=10) + + +if __name__ == "__main__": + unittest.main()