From d5c3b4799b13a2977ecf9d6a6ae63af8e3035289 Mon Sep 17 00:00:00 2001 From: oop7 <110548351+oop7@users.noreply.github.com> Date: Sat, 15 Nov 2025 18:31:16 +0200 Subject: [PATCH] Add explicit type annotations across core modules Introduced explicit type annotations for variables, function arguments, and return types in ytsage_downloader.py, ytsage_utils.py, and ytsage_config_manager.py. This improves code clarity, maintainability, and static analysis support. --- src/core/ytsage_downloader.py | 45 ++++++++++++------------ src/core/ytsage_utils.py | 55 +++++++++++++++--------------- src/utils/ytsage_config_manager.py | 27 ++++++++------- 3 files changed, 65 insertions(+), 62 deletions(-) diff --git a/src/core/ytsage_downloader.py b/src/core/ytsage_downloader.py index 6dcfea0..0d218e0 100644 --- a/src/core/ytsage_downloader.py +++ b/src/core/ytsage_downloader.py @@ -4,6 +4,7 @@ import shlex # For safely parsing command arguments import subprocess # For direct CLI command execution import time from pathlib import Path +from typing import Optional, List, Set from PySide6.QtCore import QObject, QThread, Signal @@ -85,13 +86,13 @@ class DownloadThread(QThread): self.geo_proxy_url = geo_proxy_url self.force_output_format = force_output_format self.preferred_output_format = preferred_output_format - self.paused = False - self.cancelled = False - self.process = None - self.current_filename = None # Initialize filename storage - self.last_file_path = None # Initialize full file path storage - self.subtitle_files = [] # Track subtitle files that are created - self.initial_subtitle_files = set() # Track initial subtitle files before download + self.paused: bool = False + self.cancelled: bool = False + self.process: Optional[subprocess.Popen] = None + self.current_filename: Optional[str] = None # Initialize filename storage + self.last_file_path: Optional[str] = None # Initialize full file path storage + self.subtitle_files: List[str] = [] # Track subtitle files that are created + self.initial_subtitle_files: Set[Path] = set() # Track initial subtitle files before download def cleanup_partial_files(self) -> None: """Delete any partial files including .part and unmerged format-specific files""" @@ -127,7 +128,7 @@ class DownloadThread(QThread): def cleanup_subtitle_files(self) -> None: """Delete subtitle files after they have been merged into the video file""" - deleted_count = [0, 0] + deleted_count: List[int] = [0, 0] def safe_delete(path: Path) -> bool: try: @@ -146,7 +147,7 @@ class DownloadThread(QThread): logger.debug(f"Deleted {deleted_count[0]} of {len(self.subtitle_files)} tracked subtitle files") # --- Method 2: Delete new subtitle files not in initial set --- - new_subtitle_files = { + new_subtitle_files: Set[Path] = { f for f in Path(self.path).rglob("*") if f.suffix in [".vtt", ".srt"] and f not in self.initial_subtitle_files } for subtitle_file in new_subtitle_files: @@ -156,16 +157,16 @@ class DownloadThread(QThread): except Exception as e: logger.exception(f"Error cleaning subtitle files: {e}") - def _build_yt_dlp_command(self) -> list: + def _build_yt_dlp_command(self) -> List[str]: """Build the yt-dlp command line with all options for direct execution.""" # Use the new yt-dlp path function from ytsage_yt_dlp module - yt_dlp_path = get_yt_dlp_path() - cmd: list = [yt_dlp_path] + yt_dlp_path: str = get_yt_dlp_path() + cmd: List[str] = [yt_dlp_path] logger.debug(f"Using yt-dlp from: {yt_dlp_path}") # Format selection strategy - use format ID if provided or fallback to resolution if self.format_id: - clean_format_id = self.format_id.split("-drc")[0] if "-drc" in self.format_id else self.format_id + clean_format_id: str = self.format_id.split("-drc")[0] if "-drc" in self.format_id else self.format_id # If the selected format is audio-only, pass it directly. if self.is_audio_only: @@ -180,7 +181,7 @@ class DownloadThread(QThread): logger.debug(f"Using video-only format merged with best audio: {clean_format_id}+bestaudio/best") else: # If no specific format ID, use resolution-based sorting (-S) - res_value = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified + res_value: str = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified cmd.extend(["-S", f"res:{res_value}"]) # Force output format if enabled and merging is needed @@ -196,13 +197,13 @@ class DownloadThread(QThread): # Output template with resolution in filename # Use string concatenation instead of Path.joinpath to avoid Path object issues - base_path = self.path.as_posix() + base_path: str = self.path.as_posix() if self.is_playlist: # Create output template with playlist subfolder - output_template = f"{base_path}/%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s" + output_template: str = f"{base_path}/%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s" else: - output_template = f"{base_path}/%(title)s_%(resolution)s.%(ext)s" + output_template: str = f"{base_path}/%(title)s_%(resolution)s.%(ext)s" cmd.extend(["-o", str(output_template)]) @@ -220,7 +221,7 @@ class DownloadThread(QThread): cmd.append("--write-subs") # Get language codes from subtitle selections - lang_codes = [] + lang_codes: List[str] = [] for sub_selection in self.subtitle_langs: try: # Extract just the language code (e.g., 'en' from 'en - Manual') @@ -309,8 +310,8 @@ class DownloadThread(QThread): def _run_direct_command(self) -> None: """Run yt-dlp as a direct command line process instead of using Python API.""" try: - cmd = self._build_yt_dlp_command() - cmd_str = " ".join(shlex.quote(str(arg)) for arg in cmd) + cmd: List[str] = self._build_yt_dlp_command() + cmd_str: str = " ".join(shlex.quote(str(arg)) for arg in cmd) logger.debug(f"Executing command: {cmd_str}") self.status_signal.emit(_("download.starting")) @@ -355,7 +356,7 @@ class DownloadThread(QThread): self._parse_output_line(line) # Wait for process to complete - return_code = self.process.wait() + return_code: int = self.process.wait() # Special handling for specific errors # return code 127 typically means command not found @@ -454,7 +455,7 @@ class DownloadThread(QThread): time.sleep(1) self.cleanup_partial_files() - def _parse_output_line(self, line) -> None: + def _parse_output_line(self, line: str) -> None: """Parse yt-dlp command output to update progress and status.""" line = line.strip() # logger.info(f"yt-dlp: {line}") # Log all output - OPTIONALLY UNCOMMENT FOR VERBOSE DEBUG diff --git a/src/core/ytsage_utils.py b/src/core/ytsage_utils.py index 3cfe5e5..158acae 100644 --- a/src/core/ytsage_utils.py +++ b/src/core/ytsage_utils.py @@ -6,6 +6,7 @@ import tempfile import time from importlib.metadata import PackageNotFoundError from pathlib import Path +from typing import Any, Dict, Optional, Union import requests from packaging import version @@ -42,30 +43,30 @@ except ImportError: # Cache for version information to avoid delays -_version_cache = { +_version_cache: Dict[str, Dict[str, Any]] = { "ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0}, "ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0}, "deno": {"version": None, "path": None, "last_check": 0, "path_mtime": 0}, } # Cache expiry time in seconds (5 minutes) -CACHE_EXPIRY = 300 +CACHE_EXPIRY: int = 300 -def get_file_mtime(filepath) -> float: +def get_file_mtime(filepath: Optional[Union[str, Path]]) -> float: """Get file modification time safely.""" try: if filepath and Path(filepath).exists(): return Path(filepath).stat().st_mtime except Exception: pass - return 0 + return 0.0 -def should_refresh_cache(tool_name, current_path) -> bool: +def should_refresh_cache(tool_name: str, current_path: Optional[str]) -> bool: """Determine if cache should be refreshed for a tool.""" - cache = _version_cache.get(tool_name, {}) - current_time = time.time() + cache: Dict[str, Any] = _version_cache.get(tool_name, {}) + current_time: float = time.time() # Always refresh if no cached data if not cache.get("version"): @@ -87,10 +88,10 @@ def should_refresh_cache(tool_name, current_path) -> bool: return False -def update_version_cache(tool_name, version_info, path, force_save=False) -> None: +def update_version_cache(tool_name: str, version_info: str, path: Optional[str], force_save: bool = False) -> None: """Update the version cache and optionally save to config.""" - current_time = time.time() - current_mtime = get_file_mtime(path) + current_time: float = time.time() + current_mtime: float = get_file_mtime(path) _version_cache[tool_name] = { "version": version_info, @@ -239,7 +240,7 @@ def get_deno_version() -> str: return get_deno_version_cached() -def get_ytdlp_version_direct(yt_dlp_path=None) -> str: +def get_ytdlp_version_direct(yt_dlp_path: Optional[str] = None) -> str: """Get yt-dlp version directly without caching.""" try: if yt_dlp_path is None: @@ -324,9 +325,9 @@ def get_ffmpeg_version_direct() -> str: # ensure_app_data_dir() moved to src\utils\ytsage_constants.py -def load_config() -> dict: +def load_config() -> Dict[str, Any]: """Load the application configuration from file.""" - default_config = { + default_config: Dict[str, Any] = { "download_path": str(USER_HOME_DIR / "Downloads"), "speed_limit_value": None, "speed_limit_unit_index": 0, @@ -358,7 +359,7 @@ def load_config() -> dict: return default_config -def save_config(config) -> bool: +def save_config(config: Dict[str, Any]) -> bool: """Save the application configuration to file.""" try: with open(APP_CONFIG_FILE, "w", encoding="utf-8") as f: @@ -412,7 +413,7 @@ def check_ffmpeg() -> bool: return False -def load_saved_path(main_window_instance) -> None: +def load_saved_path(main_window_instance: Any) -> None: """Load saved download path with enhanced error handling.""" try: if APP_CONFIG_FILE.exists(): @@ -444,7 +445,7 @@ def load_saved_path(main_window_instance) -> None: main_window_instance.last_path = tempfile.gettempdir() -def save_path(main_window_instance, path) -> bool: +def save_path(main_window_instance: Any, path: Union[str, Path]) -> bool: """Save download path with enhanced error handling.""" try: # Verify the path is valid and writable @@ -474,13 +475,13 @@ def update_yt_dlp() -> bool: """Check for yt-dlp updates and update if a newer version is available.""" try: # Get the yt-dlp path - yt_dlp_path = get_yt_dlp_path() + yt_dlp_path: Path = get_yt_dlp_path() # Extra logic moved to src\utils\ytsage_constants.py # For binaries downloaded with our app, use direct binary update approach # Check if this is an app-managed binary by comparing paths safely - is_app_managed = False + is_app_managed: bool = False try: # Only compare if both files exist if yt_dlp_path.exists() and YTDLP_APP_BIN_PATH.exists(): @@ -599,12 +600,12 @@ def should_check_for_auto_update() -> bool: if not config.get("auto_update_ytdlp", False): return False - frequency = config.get("auto_update_frequency", "daily") - last_check = config.get("last_update_check", 0) - current_time = time.time() + frequency: str = config.get("auto_update_frequency", "daily") + last_check: float = config.get("last_update_check", 0) + current_time: float = time.time() # Calculate time since last check - time_diff = current_time - last_check + time_diff: float = current_time - last_check if frequency == "startup": # Always check on startup if we haven't checked in the last hour @@ -679,13 +680,13 @@ def check_and_update_ytdlp_auto() -> bool: return False -def get_auto_update_settings() -> dict: +def get_auto_update_settings() -> Dict[str, Any]: """Get current auto-update settings from config.""" from src.utils.ytsage_config_manager import ConfigManager - enabled = ConfigManager.get("auto_update_ytdlp") - frequency = ConfigManager.get("auto_update_frequency") - last_check = ConfigManager.get("last_update_check") + enabled: Optional[bool] = ConfigManager.get("auto_update_ytdlp") + frequency: Optional[str] = ConfigManager.get("auto_update_frequency") + last_check: Optional[float] = ConfigManager.get("last_update_check") return { "enabled": enabled if enabled is not None else True, @@ -694,7 +695,7 @@ def get_auto_update_settings() -> dict: } -def update_auto_update_settings(enabled, frequency) -> bool: +def update_auto_update_settings(enabled: bool, frequency: str) -> bool: """Update auto-update settings in config.""" try: from src.utils.ytsage_config_manager import ConfigManager diff --git a/src/utils/ytsage_config_manager.py b/src/utils/ytsage_config_manager.py index e024078..b65e280 100644 --- a/src/utils/ytsage_config_manager.py +++ b/src/utils/ytsage_config_manager.py @@ -51,7 +51,8 @@ Exceptions import json import threading -from typing import Any +from pathlib import Path +from typing import Any, Dict, Optional from src.utils.ytsage_constants import APP_CONFIG_FILE, USER_HOME_DIR from src.utils.ytsage_logger import logger @@ -65,10 +66,10 @@ class ConfigManager: Supports nested keys via dot notation and automatically persists changes. """ - _lock = threading.RLock() - _config_file = APP_CONFIG_FILE - _settings: dict[str, Any] = {} - _default_config = { + _lock: threading.RLock = threading.RLock() + _config_file: Path = APP_CONFIG_FILE + _settings: Dict[str, Any] = {} + _default_config: Dict[str, Any] = { "download_path": str(USER_HOME_DIR / "Downloads"), "speed_limit_value": None, "speed_limit_unit_index": 0, @@ -130,12 +131,12 @@ class ConfigManager: logger.exception(f"Unexpected error while saving config: {e}") @classmethod - def get(cls, key: str) -> Any: + def get(cls, key: str) -> Optional[Any]: """ Retrieve a configuration value using a dotted key notation. Args: key (str): The dotted key string representing the path to the desired setting (e.g., "database.host"). - Any: The value associated with the given key, or None if the key does not exist. + Optional[Any]: The value associated with the given key, or None if the key does not exist. Notes: - If the configuration settings are not loaded, this method will load them before attempting retrieval. - If any part of the dotted key path is missing, None is returned and a debug message is logged. @@ -143,8 +144,8 @@ class ConfigManager: with cls._lock: if not cls._settings: cls._load() - parts = key.split(".") - value = cls._settings + parts: list[str] = key.split(".") + value: Any = cls._settings for part in parts: if isinstance(value, dict) and part in value: value = value[part] @@ -170,8 +171,8 @@ class ConfigManager: with cls._lock: if not cls._settings: cls._load() - parts = key.split(".") - d = cls._settings + parts: list[str] = key.split(".") + d: Dict[str, Any] = cls._settings for part in parts[:-1]: d = d.setdefault(part, {}) d[parts[-1]] = value @@ -193,8 +194,8 @@ class ConfigManager: with cls._lock: if not cls._settings: cls._load() - parts = key.split(".") - d = cls._settings + parts: list[str] = key.split(".") + d: Any = cls._settings for part in parts[:-1]: if part not in d: logger.debug(f"Config key '{key}' not found for deletion.")