import copy import logging import os import threading import traceback import webbrowser from pathlib import Path from typing import Callable from PySide6.QtWidgets import QMessageBox from yt_dlp import YoutubeDL from constant import YT_PLAYLISTS_URL from utils import check_url_scheme, parser_url_scheme, find_http_status logger = logging.getLogger("PLib") def sanitize_filename(value: str): invalid_chars = '<>:"/\\|?*' return "".join("_" if char in invalid_chars else char for char in value) def fetch_video_details(video_url: str, opts): opts = copy.deepcopy(opts) opts.update({"extract_flat": True, }) try: with YoutubeDL(opts) as ydl: info = ydl.extract_info(video_url, download=False) except Exception as e: logger.error(f"Unable to fetch video details: {e}\nURL:{video_url}") return None return info def resolve_video_page_url(video: dict | None=None, video_id: str=None) -> str: if not video and not video_id: raise Exception("No video id provided") if isinstance(video_id, str): return f"https://www.youtube.com/watch?v={video_id}" return ( video.get("webpage_url") or video.get("url") or (f"https://www.youtube.com/watch?v={video.get('id')}" if video.get("id") else None) ) def fetch_playlists_data(url_schema, cookie_file_path: Path, opts, callback: Callable, work_dir: Path, fetch_lock: threading.Lock, main_filepath: Path, secret_key: str | None = None): check_url_scheme(main_filepath.absolute().as_posix(), work_dir) error = None if url_schema is not None: try: parser_url_scheme(url_schema, cookie_path=cookie_file_path, secret_key=secret_key) except Exception as e: logger.error("Unable to parse URL scheme: %s", e) playlists_data = None with fetch_lock: opts = copy.deepcopy(opts) opts.update({ 'extract_flat': True, 'skip_download': True, }) try: with YoutubeDL(opts) as ydl: playlists_data = ydl.extract_info(YT_PLAYLISTS_URL, download=False) except Exception as e: status_code = find_http_status(e) if status_code == 401: error = "Cookie are expired. Reopen the tool again in browser!", None elif status_code == 403: error = "Server forbidden. Did you logged in?", None elif status_code == 429: error = "Too many requests. Try again later.", None elif status_code == 503: error = "Server unavailable. Try again later.", None elif status_code == 522: error = "Connection timed out.", None else: error = (f"Unexpected error: {e} (Type: {type(e)})\n" f"Traceback: {traceback.format_exc()}", None) if playlists_data is None: logger.error(f"Unable to fetch playlists: {error[0]}") playlists_data = {"entries": []} pf_exec = None if error is not None: reason = error[0] if isinstance(error, tuple) else error pf_exec = PlaylistFetchException( url=YT_PLAYLISTS_URL, title="Playlists Fetch Error", reason=reason, enable_option=True, options=[ {"label": "Reopen browser", "action": lambda: webbrowser.open(YT_PLAYLISTS_URL)}, ], ) callback(playlists_data, pf_exec) class PlaylistFetchException(Exception): def __init__(self, url, title, reason, enable_option=False, options=None, no_cancel=False): super().__init__() self.url = url self.title = title self.reason = reason self.qt_options = { "enable": enable_option, "options": options, "noCancelOption": no_cancel, } def create_messagebox(self, master): box = QMessageBox(master) box.setWindowTitle(self.title) box.setText(self.reason or self.title) box.setIcon(QMessageBox.Icon.Critical) is_custom_option_enabled = self.qt_options["enable"] is True and type(self.qt_options["options"]) is list if is_custom_option_enabled: for option in self.qt_options["options"]: label = option.get("label", None) action_func = option.get("action", None) role = option.get("role", QMessageBox.ButtonRole.AcceptRole) # Some type check if label is None or (action_func is None or not callable(action_func)): logger.warning(f"Skipping {option} ({action_func}) because its label or action is not set yet or not callable.") continue if not isinstance(role, QMessageBox.ButtonRole): logger.warning(f"Skipping {option} ({role}) because its role type is not QMessageBox.ButtonRole.") continue btn = box.addButton(label, role) option["button"] = btn if not self.qt_options["noCancelOption"]: box.addButton("Cancel", QMessageBox.ButtonRole.RejectRole) elif not is_custom_option_enabled: box.addButton(QMessageBox.StandardButton.Ok) box.exec() clicked = box.clickedButton() # If the option's button is clicked, Call the target action function if is_custom_option_enabled: for option in self.qt_options["options"]: btn = option.get("button") action_func = option.get("action") if clicked == btn: return action_func() return None