refactor: remove YouTube download feature, extract TrackInfo dataclass, clean up config and hotkeys
This commit is contained in:
Maksim Totmin
@@ -10,7 +10,7 @@ import subprocess
|
||||
from typing import Optional, Dict, Any
|
||||
from pathlib import Path
|
||||
|
||||
from youtube_downloader import TrackInfo
|
||||
from track_info import TrackInfo
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
|
||||
-337
@@ -1,337 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Command Line Interface for neo-pod-desktop
|
||||
Provides a CLI for downloading, converting, managing and transferring music to iPod Nano devices
|
||||
"""
|
||||
|
||||
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 neo-pod-desktop"""
|
||||
|
||||
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="neo-pod-desktop",
|
||||
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 = "mp3",
|
||||
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())
|
||||
+3
-11
@@ -134,14 +134,8 @@ class ConfigLoader:
|
||||
|
||||
if not self.config.has_option('Advanced', 'temp_dir'):
|
||||
self.config.set('Advanced', 'temp_dir', 'temp')
|
||||
if not self.config.has_option('Advanced', 'concurrent_downloads'):
|
||||
self.config.set('Advanced', 'concurrent_downloads', '2')
|
||||
if not self.config.has_option('Advanced', 'debug'):
|
||||
self.config.set('Advanced', 'debug', 'false')
|
||||
if not self.config.has_option('Advanced', 'clean_temp'):
|
||||
self.config.set('Advanced', 'clean_temp', 'true')
|
||||
if not self.config.has_option('Advanced', 'show_download'):
|
||||
self.config.set('Advanced', 'show_download', 'false')
|
||||
if not self.config.has_option('Advanced', 'hide_library_buttons'):
|
||||
self.config.set('Advanced', 'hide_library_buttons', 'false')
|
||||
|
||||
@@ -157,15 +151,13 @@ class ConfigLoader:
|
||||
'volume_down': 'Ctrl+Down',
|
||||
'seek_forward': 'Right',
|
||||
'seek_backward': 'Left',
|
||||
'tab_download': 'Ctrl+1',
|
||||
'tab_library': 'Ctrl+2',
|
||||
'tab_ipod': 'Ctrl+3',
|
||||
'tab_settings': 'Ctrl+4',
|
||||
'tab_library': 'Ctrl+1',
|
||||
'tab_ipod': 'Ctrl+2',
|
||||
'tab_settings': 'Ctrl+3',
|
||||
'search_focus': 'Ctrl+F',
|
||||
'select_all': 'Ctrl+A',
|
||||
'library_refresh': 'F5',
|
||||
'ipod_refresh': 'Ctrl+R',
|
||||
'download': 'Ctrl+Enter',
|
||||
'delete_selected': 'Delete',
|
||||
'library_add_files': 'Ctrl+O',
|
||||
}
|
||||
|
||||
+3
-5
@@ -84,15 +84,13 @@ DEFAULTS = [
|
||||
("volume_down", "Ctrl+Down", "Volume Down", "global"),
|
||||
("seek_forward", "Right", "Seek Forward 5s", "global"),
|
||||
("seek_backward", "Left", "Seek Backward 5s", "global"),
|
||||
("tab_download", "Ctrl+1", "Switch to Download", "global"),
|
||||
("tab_library", "Ctrl+2", "Switch to Library", "global"),
|
||||
("tab_ipod", "Ctrl+3", "Switch to iPod", "global"),
|
||||
("tab_settings", "Ctrl+4", "Switch to Settings", "global"),
|
||||
("tab_library", "Ctrl+1", "Switch to Library", "global"),
|
||||
("tab_ipod", "Ctrl+2", "Switch to iPod", "global"),
|
||||
("tab_settings", "Ctrl+3", "Switch to Settings", "global"),
|
||||
("search_focus", "Ctrl+F", "Focus Search", "global"),
|
||||
("select_all", "Ctrl+A", "Select All", "global"),
|
||||
("library_refresh", "F5", "Refresh Library", "global"),
|
||||
("ipod_refresh", "Ctrl+R", "Refresh Devices", "global"),
|
||||
("download", "Ctrl+Enter", "Start Download", "global"),
|
||||
("delete_selected", "Delete", "Delete Selected", "global"),
|
||||
("edit_metadata", "F2", "Edit Metadata", "global"),
|
||||
("library_add_files", "Ctrl+O", "Library: Add Files", "global"),
|
||||
|
||||
+30
-184
@@ -22,7 +22,7 @@ from PyQt6.QtCore import Qt, QThread, pyqtSignal, QUrl, QTimer
|
||||
from PyQt6.QtGui import QPixmap, QColor
|
||||
from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
|
||||
|
||||
from youtube_downloader import YouTubeDownloader, TrackInfo
|
||||
from track_info import TrackInfo
|
||||
from audio_converter import AudioConverter
|
||||
from metadata_handler import MetadataHandler
|
||||
from metadata_editor import MetadataEditorDialog
|
||||
@@ -72,7 +72,6 @@ class WorkerThread(QThread):
|
||||
def run(self):
|
||||
try:
|
||||
handlers = {
|
||||
"download": self._run_download,
|
||||
"convert": self._run_convert,
|
||||
"transfer": self._run_transfer,
|
||||
"detect_devices": self._run_detect_devices,
|
||||
@@ -89,29 +88,6 @@ class WorkerThread(QThread):
|
||||
logger.exception(f"Error in worker thread: {e}")
|
||||
self.finished_signal.emit(False, str(e), None)
|
||||
|
||||
def _run_download(self):
|
||||
url = self.kwargs.get("url")
|
||||
output_dir = self.kwargs.get("output_dir", "downloads")
|
||||
fmt = self.kwargs.get("format", "m4a")
|
||||
quality = self.kwargs.get("quality", 256)
|
||||
|
||||
self.progress_signal.emit(0, f"Initializing download from {url}")
|
||||
|
||||
downloader = YouTubeDownloader(
|
||||
output_dir=output_dir,
|
||||
progress_callback=self._download_progress_callback
|
||||
)
|
||||
tracks = downloader.process_url(url, fmt, quality)
|
||||
|
||||
if not tracks:
|
||||
self.finished_signal.emit(False, "No tracks were successfully downloaded", [])
|
||||
return
|
||||
|
||||
self.finished_signal.emit(True, f"Downloaded {len(tracks)} tracks", tracks)
|
||||
|
||||
def _download_progress_callback(self, progress: int, status: str):
|
||||
self.progress_signal.emit(progress, status)
|
||||
|
||||
def _run_convert(self):
|
||||
tracks = self.kwargs.get("tracks", [])
|
||||
local_files = self.kwargs.get("local_files", [])
|
||||
@@ -303,16 +279,14 @@ class MainWindow(QMainWindow):
|
||||
self.setMinimumSize(800, 600)
|
||||
|
||||
self.worker_thread = None
|
||||
self.downloaded_tracks: List[TrackInfo] = []
|
||||
self.library_ready: List[Dict] = []
|
||||
self.library_source_files: List[str] = []
|
||||
self.library_scanned_sources: List[Dict] = []
|
||||
self.ipod_devices: List[Dict] = []
|
||||
self.current_mount_point: Optional[str] = None
|
||||
self.show_download_tab = False
|
||||
self.hide_library_buttons = False
|
||||
self._library_toolbar_buttons: list = []
|
||||
self.ipod_tab_index = 2
|
||||
self.ipod_tab_index = 1
|
||||
|
||||
self.config_loader = ConfigLoader()
|
||||
self.hotkey_manager: Optional[HotkeyManager] = None
|
||||
@@ -358,12 +332,6 @@ class MainWindow(QMainWindow):
|
||||
)
|
||||
self.format_combo.setCurrentText(config.get("General", "format", fallback="m4a"))
|
||||
self.quality_spin.setValue(config.get_int("General", "quality", fallback=256))
|
||||
self.enable_video_check.setChecked(
|
||||
config.get_boolean("Video", "enable_video", fallback=False)
|
||||
)
|
||||
self.video_resolution_combo.setCurrentText(
|
||||
config.get("Video", "resolution", fallback="640x480")
|
||||
)
|
||||
self.embed_artwork_check.setChecked(
|
||||
config.get_boolean("Metadata", "embed_artwork", fallback=True)
|
||||
)
|
||||
@@ -376,11 +344,6 @@ class MainWindow(QMainWindow):
|
||||
self.auto_detect_check.setChecked(
|
||||
config.get_boolean("Device", "auto_detect", fallback=True)
|
||||
)
|
||||
show_download = config.get_boolean("Advanced", "show_download", fallback=False)
|
||||
self.show_download_check.setChecked(show_download)
|
||||
self.show_download_tab = show_download
|
||||
self.tab_widget.setTabVisible(self.download_tab_index, show_download)
|
||||
|
||||
hide_library_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False)
|
||||
self.hide_library_buttons_check.setChecked(hide_library_buttons)
|
||||
self._on_hide_library_buttons_toggled(hide_library_buttons)
|
||||
@@ -395,13 +358,10 @@ class MainWindow(QMainWindow):
|
||||
config.set("General", "output_dir", self.output_dir_input.text())
|
||||
config.set("General", "format", self.format_combo.currentText())
|
||||
config.set("General", "quality", str(self.quality_spin.value()))
|
||||
config.set("Video", "enable_video", str(self.enable_video_check.isChecked()).lower())
|
||||
config.set("Video", "resolution", self.video_resolution_combo.currentText())
|
||||
config.set("Metadata", "embed_artwork", str(self.embed_artwork_check.isChecked()).lower())
|
||||
config.set("Metadata", "use_musicbrainz", str(self.use_musicbrainz_check.isChecked()).lower())
|
||||
config.set("Advanced", "clean_temp", str(self.clean_temp_check.isChecked()).lower())
|
||||
config.set("Device", "auto_detect", str(self.auto_detect_check.isChecked()).lower())
|
||||
config.set("Advanced", "show_download", str(self.show_download_check.isChecked()).lower())
|
||||
config.set("Advanced", "hide_library_buttons", str(self.hide_library_buttons_check.isChecked()).lower())
|
||||
|
||||
config.set("Playback", "last_volume", str(self.volume_slider.value()))
|
||||
@@ -433,15 +393,13 @@ class MainWindow(QMainWindow):
|
||||
"volume_down": lambda: self._adjust_volume(-5),
|
||||
"seek_forward": self._seek_forward,
|
||||
"seek_backward": self._seek_backward,
|
||||
"tab_download": lambda: self.tab_widget.setCurrentIndex(self.download_tab_index),
|
||||
"tab_library": lambda: self.tab_widget.setCurrentIndex(1),
|
||||
"tab_library": lambda: self.tab_widget.setCurrentIndex(0),
|
||||
"tab_ipod": lambda: self.tab_widget.setCurrentIndex(self.ipod_tab_index),
|
||||
"tab_settings": lambda: self.tab_widget.setCurrentIndex(3),
|
||||
"tab_settings": lambda: self.tab_widget.setCurrentIndex(2),
|
||||
"search_focus": self._focus_search,
|
||||
"select_all": self._select_all_current,
|
||||
"library_refresh": self._on_refresh_library_clicked,
|
||||
"ipod_refresh": self._on_refresh_devices_clicked,
|
||||
"download": self._on_download_clicked,
|
||||
"delete_selected": self._delete_selected_current,
|
||||
"edit_metadata": self._on_edit_metadata,
|
||||
"library_add_files": self._on_library_add_files,
|
||||
@@ -459,11 +417,6 @@ class MainWindow(QMainWindow):
|
||||
self.tab_widget = QTabWidget()
|
||||
main_layout.addWidget(self.tab_widget)
|
||||
|
||||
download_tab = QWidget()
|
||||
self.tab_widget.addTab(download_tab, "Download")
|
||||
self._setup_download_tab(download_tab)
|
||||
self.download_tab_index = 0
|
||||
|
||||
library_tab = QWidget()
|
||||
self.tab_widget.addTab(library_tab, "Library")
|
||||
self._setup_library_tab(library_tab)
|
||||
@@ -476,56 +429,11 @@ class MainWindow(QMainWindow):
|
||||
self.tab_widget.addTab(settings_tab, "Settings")
|
||||
self._setup_settings_tab(settings_tab)
|
||||
|
||||
self.tab_widget.setTabVisible(self.download_tab_index, self.show_download_tab)
|
||||
self.tab_widget.setTabVisible(self.ipod_tab_index, False)
|
||||
|
||||
self.statusBar().showMessage("Ready")
|
||||
self.setCentralWidget(main_widget)
|
||||
|
||||
def _setup_download_tab(self, tab):
|
||||
layout = QVBoxLayout(tab)
|
||||
|
||||
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_layout = QHBoxLayout()
|
||||
format_label = QLabel("Format:")
|
||||
self.format_combo = QComboBox()
|
||||
self.format_combo.addItems(["mp3", "m4a"])
|
||||
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)
|
||||
|
||||
self.download_button = QPushButton("Download")
|
||||
self.download_button.clicked.connect(self._on_download_clicked)
|
||||
layout.addWidget(self.download_button)
|
||||
|
||||
self.download_progress = QProgressBar()
|
||||
self.download_progress.setRange(0, 100)
|
||||
layout.addWidget(self.download_progress)
|
||||
|
||||
self.download_status = QLabel("Ready to download")
|
||||
layout.addWidget(self.download_status)
|
||||
|
||||
track_list_label = QLabel("Downloaded Tracks:")
|
||||
layout.addWidget(track_list_label)
|
||||
|
||||
self.track_list = QListWidget()
|
||||
layout.addWidget(self.track_list)
|
||||
|
||||
def _setup_library_tab(self, tab):
|
||||
layout = QVBoxLayout(tab)
|
||||
|
||||
@@ -783,28 +691,25 @@ class MainWindow(QMainWindow):
|
||||
output_layout.addWidget(self.output_dir_input)
|
||||
output_layout.addWidget(browse_btn)
|
||||
library_layout.addLayout(output_layout)
|
||||
|
||||
convert_layout = QHBoxLayout()
|
||||
convert_format_label = QLabel("Convert Format:")
|
||||
self.format_combo = QComboBox()
|
||||
self.format_combo.addItems(["mp3", "m4a"])
|
||||
convert_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)
|
||||
convert_layout.addWidget(convert_format_label)
|
||||
convert_layout.addWidget(self.format_combo)
|
||||
convert_layout.addWidget(convert_quality_label)
|
||||
convert_layout.addWidget(self.quality_spin)
|
||||
convert_layout.addStretch()
|
||||
library_layout.addLayout(convert_layout)
|
||||
|
||||
layout.addWidget(library_group)
|
||||
|
||||
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_group = QGroupBox("Metadata Options")
|
||||
metadata_layout = QVBoxLayout(metadata_group)
|
||||
|
||||
@@ -828,11 +733,6 @@ class MainWindow(QMainWindow):
|
||||
self.auto_detect_check.setChecked(True)
|
||||
advanced_layout.addWidget(self.auto_detect_check)
|
||||
|
||||
self.show_download_check = QCheckBox("Show YouTube Download Tab")
|
||||
self.show_download_check.setChecked(False)
|
||||
self.show_download_check.toggled.connect(self._on_show_download_toggled)
|
||||
advanced_layout.addWidget(self.show_download_check)
|
||||
|
||||
self.hide_library_buttons_check = QCheckBox("Hide Library Toolbar Buttons")
|
||||
self.hide_library_buttons_check.setChecked(False)
|
||||
self.hide_library_buttons_check.toggled.connect(self._on_hide_library_buttons_toggled)
|
||||
@@ -872,10 +772,6 @@ class MainWindow(QMainWindow):
|
||||
about_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.addWidget(about_label)
|
||||
|
||||
def _on_show_download_toggled(self, checked: bool):
|
||||
self.show_download_tab = checked
|
||||
self.tab_widget.setTabVisible(self.download_tab_index, checked)
|
||||
|
||||
def _on_hide_library_buttons_toggled(self, checked: bool):
|
||||
self.hide_library_buttons = checked
|
||||
for btn in self._library_toolbar_buttons:
|
||||
@@ -1433,7 +1329,7 @@ class MainWindow(QMainWindow):
|
||||
if not is_ready:
|
||||
continue
|
||||
track_info = TrackInfo(
|
||||
video_id=f"local_{hash(data['path'])}",
|
||||
source_id=f"local_{hash(data['path'])}",
|
||||
title=data['title'],
|
||||
artist=data['artist'],
|
||||
album=data.get("album", ""),
|
||||
@@ -1680,66 +1576,13 @@ class MainWindow(QMainWindow):
|
||||
else:
|
||||
subprocess.Popen(["xdg-open", os.path.dirname(path)])
|
||||
|
||||
def _on_download_clicked(self):
|
||||
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
|
||||
|
||||
if self.worker_thread and self.worker_thread.isRunning():
|
||||
QMessageBox.information(self, "Info", "Please wait for the current task to finish")
|
||||
return
|
||||
|
||||
self.download_button.setEnabled(False)
|
||||
self.download_status.setText("Downloading...")
|
||||
self.download_progress.setValue(0)
|
||||
self.track_list.clear()
|
||||
|
||||
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):
|
||||
self.download_progress.setValue(progress)
|
||||
self.download_status.setText(status)
|
||||
|
||||
def _on_download_finished(self, success, message, result):
|
||||
self.download_button.setEnabled(True)
|
||||
|
||||
if success and result:
|
||||
self.download_status.setText(message)
|
||||
self.downloaded_tracks = result
|
||||
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)
|
||||
self._scan_library()
|
||||
else:
|
||||
self.download_status.setText(f"Error: {message}")
|
||||
QMessageBox.warning(self, "Download Error", message)
|
||||
self.downloaded_tracks = []
|
||||
self.track_list.clear()
|
||||
|
||||
self._cleanup_worker()
|
||||
|
||||
def _on_library_convert_clicked(self):
|
||||
source_files = self._get_selected_source_files()
|
||||
has_downloaded = bool(self.downloaded_tracks)
|
||||
|
||||
if not source_files and not has_downloaded:
|
||||
if not source_files:
|
||||
QMessageBox.warning(
|
||||
self, "Error",
|
||||
"No tracks to convert. Add local source files or download tracks first."
|
||||
"No tracks to convert. Add local source files first."
|
||||
)
|
||||
return
|
||||
|
||||
@@ -1757,7 +1600,7 @@ class MainWindow(QMainWindow):
|
||||
self.library_progress.setValue(0)
|
||||
|
||||
kwargs = {
|
||||
"tracks": self.downloaded_tracks if has_downloaded else [],
|
||||
"tracks": [],
|
||||
"local_files": source_files,
|
||||
"output_dir": output_dir,
|
||||
"format": self.format_combo.currentText(),
|
||||
@@ -2332,15 +2175,18 @@ def _setup_linux_theming():
|
||||
logger.debug("No GTK platform theme plugin found, using Qt default style")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
def main():
|
||||
_setup_linux_theming()
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
|
||||
# Apply Fusion as fallback style if no platform theme was applied
|
||||
if not os.environ.get("QT_QPA_PLATFORMTHEME"):
|
||||
app.setStyle("Fusion")
|
||||
|
||||
window = MainWindow()
|
||||
window.show()
|
||||
sys.exit(app.exec())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -20,7 +20,7 @@ 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
|
||||
from track_info import TrackInfo
|
||||
from artwork.cache import cover_cache, xdg_cache_path
|
||||
|
||||
# Configure logging
|
||||
@@ -356,7 +356,7 @@ class MetadataHandler:
|
||||
cover_path = cover_cache.put(artist or "Unknown Artist", album or "", cover_data)
|
||||
|
||||
return TrackInfo(
|
||||
video_id=f"local_{hash(file_path)}",
|
||||
source_id=f"local_{hash(file_path)}",
|
||||
title=title or base_name,
|
||||
artist=artist or "Unknown Artist",
|
||||
album=album or "",
|
||||
@@ -533,7 +533,7 @@ class MetadataHandler:
|
||||
artist, title = filename.split(" - ", 1)
|
||||
|
||||
return TrackInfo(
|
||||
video_id=f"local_{hash(file_path or filename)}",
|
||||
source_id=f"local_{hash(file_path or filename)}",
|
||||
title=title,
|
||||
artist=artist,
|
||||
album="",
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Track Info data class shared across the iPod transfer pipeline.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class TrackInfo:
|
||||
"""Data class to store track information."""
|
||||
source_id: str
|
||||
title: str
|
||||
artist: str
|
||||
album: str
|
||||
thumbnail_url: str
|
||||
duration: int
|
||||
track_number: Optional[int] = None
|
||||
playlist_title: Optional[str] = None
|
||||
download_path: Optional[str] = None
|
||||
genre: Optional[str] = None
|
||||
cover_path: Optional[str] = None
|
||||
@@ -1,417 +0,0 @@
|
||||
#!/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
|
||||
genre: Optional[str] = None
|
||||
cover_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)
|
||||
Reference in New Issue
Block a user