Houmeres/SageTube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Add explicit type annotations across core modules

Browse Source 
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.
This commit is contained in:
oop7
2025-11-15 18:31:16 +02:00
parent 15bf5673e3
commit d5c3b4799b
3 changed files with 65 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/ytsage_downloader.py
+23 -22
View File
@@ -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
src/core/ytsage_utils.py
+28 -27
View File
@@ -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
src/utils/ytsage_config_manager.py
+14 -13
View File
@@ -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.")