initial app

This commit is contained in:
Antonio Lorusso 2025-03-12 01:01:43 +01:00
commit 9dcb1228ca
17 changed files with 3215 additions and 0 deletions

37
.gitignore vendored Normal file
View File

@ -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

63
README.md Normal file
View File

@ -0,0 +1,63 @@
<!-- # YouTube Music to iPod Nano -->
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

43
config.ini Normal file
View File

@ -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

48
install.bat Normal file
View File

@ -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

54
install.sh Executable file
View File

@ -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."

11
requirements.txt Normal file
View File

@ -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

34
run.py Executable file
View File

@ -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())

39
setup.py Normal file
View File

@ -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",
)

6
src/__init__.py Normal file
View File

@ -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"

212
src/audio_converter.py Normal file
View File

@ -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")

337
src/cli.py Normal file
View File

@ -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())

238
src/config_loader.py Normal file
View File

@ -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()

562
src/ipod_device.py Normal file
View File

@ -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()

762
src/main.py Normal file
View File

@ -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())

261
src/metadata_handler.py Normal file
View File

@ -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
# )
# )

415
src/youtube_downloader.py Normal file
View File

@ -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)

93
tests/test_basic.py Normal file
View File

@ -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()