diff --git a/app/core/metadata_manager.py b/app/core/metadata_manager.py new file mode 100644 index 0000000..3c55ad5 --- /dev/null +++ b/app/core/metadata_manager.py @@ -0,0 +1,82 @@ +from pathlib import Path +import re +from mutagen.id3 import ID3, TALB, TIT2, TPE1, delete as id3_delete + +class MetadataManager: + def apply_mp3_tags(self, file_path: Path, metadata: dict, fallback_title: str) -> None: + track_title = str(metadata.get("mp3_title") or metadata.get("title") or fallback_title).strip() + track_artist = str(metadata.get("artist", "")).strip() + track_album = str(metadata.get("album", "")).strip() + track_cover = str(metadata.get("cover_path", "")).strip() + + try: + id3_delete(str(file_path)) + except Exception: + pass + + tags = ID3() + tags.delall("TIT2") + tags.add(TIT2(encoding=3, text=track_title or fallback_title)) + tags.delall("TPE1") + if track_artist: + tags.add(TPE1(encoding=3, text=track_artist)) + tags.delall("TALB") + if track_album: + tags.add(TALB(encoding=3, text=track_album)) + tags.save(str(file_path), v2_version=4) + + # Use eyeD3 for cover art (Windows Explorer compatibility) + if track_cover: + try: + import eyed3 + from PIL import Image + import io + audiofile = eyed3.load(str(file_path)) + if audiofile is not None: + if audiofile.tag is None: + audiofile.initTag() + cover_file = Path(track_cover) + if cover_file.exists() and cover_file.is_file(): + # Always convert to JPEG and resize to 999x999 + try: + img = Image.open(cover_file) + img = img.convert("RGB") + img = img.resize((999, 999), Image.LANCZOS) + buf = io.BytesIO() + img.save(buf, format="JPEG") + image_data = buf.getvalue() + mime_type = "image/jpeg" + except Exception as e: + print(f"[eyeD3] Failed to process image: {e}") + image_data = cover_file.read_bytes() + mime_type = "image/jpeg" + audiofile.tag.images.set(3, image_data, mime_type, u"Cover") + audiofile.tag.version = (2, 3, 0) + audiofile.tag.save(version=(2, 3, 0)) + except Exception as e: + print(f"[eyeD3] Failed to embed cover: {e}") + + def sanitize_filename_part(self, value: str) -> str: + clean_value = re.sub(r"[<>:\"/\\|?*]", "_", value).strip() + clean_value = re.sub(r"\s+", " ", clean_value) + return clean_value.strip(". ") + + def rename_output_file(self, file_path: Path, title: str, artist: str) -> Path: + safe_title = self.sanitize_filename_part(title) + safe_artist = self.sanitize_filename_part(artist) + if safe_artist and safe_title: + base_name = f"{safe_artist} - {safe_title}" + else: + base_name = safe_title or self.sanitize_filename_part(file_path.stem) + target_path = file_path.with_name(f"{base_name}{file_path.suffix}") + if target_path == file_path: + return file_path + counter = 2 + while target_path.exists(): + target_path = file_path.with_name(f"{base_name} ({counter}){file_path.suffix}") + counter += 1 + try: + file_path.rename(target_path) + return target_path + except Exception: + return file_path diff --git a/app/core/youtube_manager.py b/app/core/youtube_manager.py new file mode 100644 index 0000000..0915bb6 --- /dev/null +++ b/app/core/youtube_manager.py @@ -0,0 +1,575 @@ +from pathlib import Path +import re +from urllib.parse import quote_plus +from typing import Optional + +from PyQt6.QtCore import QObject, QThread, pyqtSignal +from mutagen.id3 import APIC, ID3, TALB, TIT2, TPE1, delete as id3_delete +from yt_dlp import YoutubeDL + +import app.utils.paths as paths +from app.core.settings_manager import SettingsManager + + +class AudioDownloadWorker(QThread): + progress_text = pyqtSignal(str) + progress_value = pyqtSignal(int) + completed = pyqtSignal(str) + failed = pyqtSignal(str) + + def __init__(self, url: str, output_dir: str, audio_format: str, audio_quality: str, ffmpeg_location: Optional[str] = None, track_metadata: Optional[dict] = None) -> None: + super().__init__() + self.url = url + self.output_dir = output_dir + self.audio_format = audio_format + self.audio_quality = audio_quality + self.ffmpeg_location = ffmpeg_location + self.track_metadata = track_metadata if isinstance(track_metadata, dict) else {} + + def _progress_hook(self, data: dict) -> None: + status = data.get("status") + + if status == "downloading": + percent_text = str(data.get("_percent_str", "")).strip() + percent_text = re.sub(r"\x1b\[[0-9;]*m", "", percent_text) + percent_match = re.search(r"(\d+(?:\.\d+)?)%", percent_text) + + if percent_match: + value = int(float(percent_match.group(1))) + self.progress_value.emit(max(0, min(100, value))) + + self.progress_text.emit(percent_text or "Downloading...") + + elif status == "finished": + self.progress_value.emit(100) + + def run(self) -> None: + try: + ydl_opts = { + "format": "bestaudio/best", + "noplaylist": True, + "outtmpl": str(Path(self.output_dir) / "%(title)s.%(ext)s"), + "progress_hooks": [self._progress_hook], + "quiet": True, + "no_warnings": True, + } + + if self.ffmpeg_location: + ydl_opts["ffmpeg_location"] = self.ffmpeg_location + + if self.audio_format == "mp3": + ydl_opts.update( + { + "postprocessors": [ + { + "key": "FFmpegExtractAudio", + "preferredcodec": "mp3", + "preferredquality": self.audio_quality, + } + ] + } + ) + + with YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(self.url, download=True) + title = info.get("title", "audio") + self.progress_text.emit(title) + + if self.audio_format == "mp3": + downloaded_file = self._resolve_downloaded_file(info) + if downloaded_file: + self._apply_mp3_tags(downloaded_file, title) + + self.completed.emit(title) + except Exception as error: + self.failed.emit(str(error)) + + def _resolve_downloaded_file(self, info: dict) -> Optional[Path]: + candidates: list[Path] = [] + + requested_downloads = info.get("requested_downloads") if isinstance(info, dict) else None + if isinstance(requested_downloads, list): + for download_data in requested_downloads: + if not isinstance(download_data, dict): + continue + for key in ("filepath", "_filename"): + candidate_path = download_data.get(key) + if isinstance(candidate_path, str) and candidate_path: + candidates.append(Path(candidate_path)) + + for key in ("filepath", "_filename"): + candidate_path = info.get(key) if isinstance(info, dict) else None + if isinstance(candidate_path, str) and candidate_path: + candidates.append(Path(candidate_path)) + + for candidate in candidates: + if candidate.exists() and candidate.is_file() and candidate.suffix.lower() == ".mp3": + return candidate + mp3_candidate = candidate.with_suffix(".mp3") + if mp3_candidate.exists() and mp3_candidate.is_file(): + return mp3_candidate + + output_dir = Path(self.output_dir) + if output_dir.exists(): + mp3_files = sorted(output_dir.glob("*.mp3"), key=lambda file_path: file_path.stat().st_mtime, reverse=True) + if mp3_files: + return mp3_files[0] + return None + + def _apply_mp3_tags(self, file_path: Path, fallback_title: str) -> None: + from app.core.metadata_manager import MetadataManager + metadata_manager = MetadataManager() + metadata_manager.apply_mp3_tags(file_path, self.track_metadata, fallback_title) + renamed_file = metadata_manager.rename_output_file(file_path, self.track_metadata.get("mp3_title") or fallback_title, self.track_metadata.get("artist", "")) + if renamed_file is not None: + self.progress_text.emit(renamed_file.stem) + + +class YoutubeSearchWorker(QThread): + finished_search = pyqtSignal(list) + failed_search = pyqtSignal(str) + + def __init__(self, query: str) -> None: + super().__init__() + self.query = query + + def run(self) -> None: + try: + results: list[dict] = [] + seen_urls: set[str] = set() + ydl_opts = {"quiet": True, "no_warnings": True, "extract_flat": True} + + with YoutubeDL(ydl_opts) as ydl: + video_result = ydl.extract_info(f"ytsearch10:{self.query}", download=False) + playlist_url = f"https://www.youtube.com/results?search_query={quote_plus(self.query)}&sp=EgIQAw%253D%253D" + playlist_result = ydl.extract_info(playlist_url, download=False) + + video_entries = video_result.get("entries", []) if isinstance(video_result, dict) else [] + for entry in video_entries: + item = self._normalize_video_entry(entry) + if item and item["url"] not in seen_urls: + seen_urls.add(item["url"]) + results.append(item) + + playlist_entries = playlist_result.get("entries", []) if isinstance(playlist_result, dict) else [] + for entry in playlist_entries: + item = self._normalize_playlist_entry(entry) + if item and item["url"] not in seen_urls: + seen_urls.add(item["url"]) + results.append(item) + + self.finished_search.emit(results) + except Exception: + self.failed_search.emit("search_failed") + + def _normalize_video_entry(self, entry: dict) -> Optional[dict]: + if not isinstance(entry, dict): + return None + + video_id = entry.get("id") + if not video_id: + return None + title = entry.get("title") or "Untitled" + channel = entry.get("uploader") or entry.get("channel") or "" + raw_duration = int(entry.get("duration")) + duration = self._format_duration(raw_duration) + thumbnail_url = f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg" + + return { + "kind": "video", + "title": title, + "url": f"https://www.youtube.com/watch?v={video_id}", + "channel": channel, + "duration": duration, + "item_count": "", + "thumbnail_url": thumbnail_url, + "thumbnail_bytes": b"", + } + + def _normalize_playlist_entry(self, entry: dict) -> Optional[dict]: + if not isinstance(entry, dict): + return None + + playlist_id = entry.get("id") + if not playlist_id: + return None + + title = entry.get("title") or "Untitled playlist" + channel = entry.get("uploader") or entry.get("channel") or "" + count = entry.get("playlist_count") or entry.get("n_entries") or 0 + thumbnails = entry.get("thumbnails") or [] + thumbnail_url = "" + if thumbnails and isinstance(thumbnails, list): + thumbnail_url = (thumbnails[-1] or {}).get("url", "") + if not thumbnail_url: + thumbnail_url = f"https://i.ytimg.com/vi/{playlist_id}/hqdefault.jpg" + + return { + "kind": "playlist", + "title": title, + "url": f"https://www.youtube.com/playlist?list={playlist_id}", + "channel": channel, + "duration": "", + "item_count": str(count) if count else "", + "thumbnail_url": thumbnail_url, + "thumbnail_bytes": b"", + } + + def _format_duration(self, seconds: Optional[int]) -> str: + if not isinstance(seconds, int) or seconds < 0: + return "" + minutes, sec = divmod(seconds, 60) + hours, minutes = divmod(minutes, 60) + if hours: + return f"{hours}:{minutes:02}:{sec:02}" + return f"{minutes}:{sec:02}" + + +class PreviewStreamWorker(QThread): + finished_preview = pyqtSignal(str, str) + failed_preview = pyqtSignal(str) + + def __init__(self, url: str, title: str) -> None: + super().__init__() + self.url = url + self.title = title + + def run(self) -> None: + try: + ydl_opts = { + "quiet": True, + "no_warnings": True, + "format": "best[height<=720][vcodec!=none][acodec!=none]/best[ext=mp4][height<=720]/best", + "noplaylist": True, + } + with YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(self.url, download=False) + stream_url = info.get("url") if isinstance(info, dict) else None + if stream_url: + self.finished_preview.emit(stream_url, self.title) + else: + self.failed_preview.emit("preview_unavailable") + except Exception: + self.failed_preview.emit("preview_unavailable") + + +class YoutubeManager(QObject): + def apply_album_cover(self, album_name: str, cover_path: str) -> None: + """Applique la cover à tous les morceaux de l'album (business logic)""" + for idx, track in enumerate(self.track_list): + if str(track.get("album", "")).strip() == album_name: + self.update_track_metadata( + idx, + str(track.get("mp3_title") or track.get("title", "")), + str(track.get("artist", "")), + album_name, + cover_path, + ) + started = pyqtSignal() + progress_text = pyqtSignal(str) + progress_value = pyqtSignal(int) + completed = pyqtSignal(str) + failed = pyqtSignal(str) + list_changed = pyqtSignal(list) + queue_finished = pyqtSignal() + downloading_state_changed = pyqtSignal(bool) + search_started = pyqtSignal() + search_results = pyqtSignal(list) + preview_started = pyqtSignal() + preview_ready = pyqtSignal(str, str) + preview_failed = pyqtSignal(str) + + def __init__(self, settings_manager: SettingsManager) -> None: + super().__init__() + self.settings_manager = settings_manager + self.download_worker: Optional[AudioDownloadWorker] = None + self.search_worker: Optional[YoutubeSearchWorker] = None + self.preview_worker: Optional[PreviewStreamWorker] = None + self.track_list: list[dict[str, str]] = [] + self.download_queue: list[dict[str, str]] = [] + self.queue_output_dir: str = "" + self.last_playlist_added_count: int = 0 + + def get_tracks(self) -> list[dict[str, str]]: + return list(self.track_list) + + def _build_track_payload(self, title: str, url: str, artist: str = "", album: str = "", cover_path: str = "") -> dict[str, str]: + clean_title = title.strip() + clean_url = url.strip() + return { + "title": clean_title, + "url": clean_url, + "mp3_title": clean_title, + "artist": artist.strip(), + "album": album.strip(), + "cover_path": cover_path.strip(), + } + + def add_track(self, url: str) -> bool: + clean_url = url.strip() + if not self.is_valid_youtube_url(clean_url): + self.failed.emit("invalid_youtube_url") + return False + + title = self._extract_title_from_url(clean_url) + if not title: + self.failed.emit("metadata_fetch_failed") + return False + + self.track_list.append(self._build_track_payload(title=title, url=clean_url)) + self.list_changed.emit(self.get_tracks()) + return True + + def add_search_result(self, search_result: dict) -> bool: + self.last_playlist_added_count = 0 + kind = search_result.get("kind", "video") + url = search_result.get("url", "") + title = search_result.get("title", "") + + if kind == "playlist": + tracks = self._extract_playlist_tracks(url) + if not tracks: + self.failed.emit("playlist_expand_failed") + return False + + self.track_list.extend(tracks) + self.last_playlist_added_count = len(tracks) + self.list_changed.emit(self.get_tracks()) + return True + + return self.add_track_with_title(url, title) + + def add_track_with_title(self, url: str, title: str) -> bool: + clean_url = url.strip() + clean_title = title.strip() + + if not self.is_valid_youtube_url(clean_url): + self.failed.emit("invalid_youtube_url") + return False + + if not clean_title: + self.failed.emit("metadata_fetch_failed") + return False + + self.track_list.append(self._build_track_payload(title=clean_title, url=clean_url)) + self.list_changed.emit(self.get_tracks()) + return True + + def update_track_metadata(self, index: int, mp3_title: str, artist: str, album: str, cover_path: str) -> bool: + if index < 0 or index >= len(self.track_list): + self.failed.emit("no_track_selected") + return False + + track = self.track_list[index] + track["mp3_title"] = mp3_title.strip() + track["artist"] = artist.strip() + track["album"] = album.strip() + track["cover_path"] = cover_path.strip() + self.list_changed.emit(self.get_tracks()) + return True + + def search_tracks(self, query: str) -> bool: + clean_query = query.strip() + if len(clean_query) < 2: + self.failed.emit("invalid_search_query") + return False + + if self.search_worker and self.search_worker.isRunning(): + self.failed.emit("search_in_progress") + return False + + self.search_worker = YoutubeSearchWorker(clean_query) + self.search_worker.finished_search.connect(self.search_results.emit) + self.search_worker.failed_search.connect(self.failed.emit) + + self.search_started.emit() + self.search_worker.start() + return True + + def start_preview(self, search_result: dict) -> bool: + url = search_result.get("url", "") + title = search_result.get("title", "") + kind = search_result.get("kind", "video") + + if kind != "video": + self.preview_failed.emit("preview_playlist_unavailable") + return False + + if not self.is_valid_youtube_url(url): + self.preview_failed.emit("invalid_youtube_url") + return False + + if self.preview_worker and self.preview_worker.isRunning(): + self.preview_failed.emit("preview_in_progress") + return False + + self.preview_worker = PreviewStreamWorker(url, title) + self.preview_worker.finished_preview.connect(self.preview_ready.emit) + self.preview_worker.failed_preview.connect(self.preview_failed.emit) + self.preview_started.emit() + self.preview_worker.start() + return True + + def remove_track(self, index: int) -> bool: + if index < 0 or index >= len(self.track_list): + self.failed.emit("no_track_selected") + return False + + self.track_list.pop(index) + self.list_changed.emit(self.get_tracks()) + return True + + def clear_tracks(self) -> None: + self.track_list.clear() + self.list_changed.emit(self.get_tracks()) + + def download_track_at(self, index: int, output_dir: str) -> bool: + if index < 0 or index >= len(self.track_list): + self.failed.emit("no_track_selected") + return False + return self.download_audio(self.track_list[index]["url"], output_dir, self.track_list[index]) + + def download_all_tracks(self, output_dir: str) -> bool: + if not self.track_list: + self.failed.emit("empty_download_list") + return False + + if not output_dir or not Path(output_dir).exists(): + self.failed.emit("invalid_output_folder") + return False + + if self.download_worker and self.download_worker.isRunning(): + self.failed.emit("download_in_progress") + return False + + self.download_queue = list(self.track_list) + self.queue_output_dir = output_dir + self._download_next_in_queue() + return True + + def download_audio(self, url: str, output_dir: str, track_metadata: Optional[dict] = None) -> bool: + if not self.is_valid_youtube_url(url): + self.failed.emit("invalid_youtube_url") + return False + + if not output_dir or not Path(output_dir).exists(): + self.failed.emit("invalid_output_folder") + return False + + if self.download_worker and self.download_worker.isRunning(): + self.failed.emit("download_in_progress") + return False + + audio_format = self.settings_manager.get_audio_format() + audio_quality = self.settings_manager.get_audio_quality() + ffmpeg_location = self.resolve_ffmpeg_location() + + self.download_worker = AudioDownloadWorker(url, output_dir, audio_format, audio_quality, ffmpeg_location, track_metadata) + self.download_worker.progress_text.connect(self.progress_text.emit) + self.download_worker.progress_value.connect(self.progress_value.emit) + self.download_worker.completed.connect(self._on_worker_completed) + self.download_worker.failed.connect(self._on_worker_failed) + + self.started.emit() + self.downloading_state_changed.emit(True) + self.download_worker.start() + return True + + def _download_next_in_queue(self) -> None: + if not self.download_queue: + self.queue_output_dir = "" + self.downloading_state_changed.emit(False) + self.queue_finished.emit() + return + + next_track = self.download_queue.pop(0) + self.download_audio(next_track["url"], self.queue_output_dir, next_track) + + def _on_worker_completed(self, title: str) -> None: + self.completed.emit(title) + + if self.download_queue: + self._download_next_in_queue() + return + + self.downloading_state_changed.emit(False) + + def _on_worker_failed(self, error_text: str) -> None: + self.download_queue = [] + self.queue_output_dir = "" + self.downloading_state_changed.emit(False) + if "ffmpeg" in error_text.lower(): + self.failed.emit("ffmpeg_required") + else: + self.failed.emit("audio_download_error") + + def is_valid_youtube_url(self, url: str) -> bool: + if not url.startswith(("http://", "https://")): + return False + return "youtube.com" in url or "youtu.be" in url + + def _extract_title_from_url(self, url: str) -> Optional[str]: + try: + ydl_opts = { + "quiet": True, + "no_warnings": True, + } + with YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(url, download=False) + return info.get("title") if isinstance(info, dict) else None + except Exception: + return None + + def _extract_playlist_tracks(self, playlist_url: str) -> list[dict[str, str]]: + # Try with extract_flat first, fallback to full extraction if it fails + for flat in (True, False): + try: + ydl_opts = { + "quiet": True, + "no_warnings": True, + } + if flat: + ydl_opts["extract_flat"] = True + with YoutubeDL(ydl_opts) as ydl: + info = ydl.extract_info(playlist_url, download=False) + + entries = info.get("entries", []) if isinstance(info, dict) else [] + tracks: list[dict[str, str]] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + + video_id = entry.get("id") + title = entry.get("title") or "Untitled" + url = entry.get("url") + + if video_id: + video_url = f"https://www.youtube.com/watch?v={video_id}" + elif isinstance(url, str) and url.startswith("http"): + video_url = url + elif isinstance(url, str): + video_url = f"https://www.youtube.com/watch?v={url}" + else: + continue + + tracks.append(self._build_track_payload(title=title, url=video_url)) + + if tracks: + return tracks + except Exception as e: + print(f"[yt-dlp playlist extraction error] flat={flat}: {e}") + return [] + + def resolve_ffmpeg_location(self) -> Optional[str]: + candidates = [ + Path("others") / "ffmpeg.exe", + Path("data") / "others" / "ffmpeg.exe", + Path(paths.resource_path("others/ffmpeg.exe")), + Path(paths.resource_path("data/others/ffmpeg.exe")), + ] + + for candidate in candidates: + if candidate.exists(): + return str(candidate.parent) + + return None \ No newline at end of file