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

Refactor utils module structure and update imports

Browse Source 
Moved utility modules from src/utils/ to ytsage/utils/ and updated all relative imports accordingly. Adjusted docstring usage examples and fixed a path reference in LocalizationManager to reflect the new directory structure.
This commit is contained in:
oop7
2026-01-25 14:06:41 +02:00
parent ad5f304d4e
commit ca6b53715e
6 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ytsage/utils/__init__.py
View File
ytsage/utils/ytsage_config_manager.py
+215
View File
@@ -0,0 +1,215 @@
"""
Config Manager Module
=====================
This module provides **thread-safe** centralized management for application
configuration in YTSage. It handles reading, writing, and managing settings
stored in a JSON file, with support for nested keys via dot notation.
Thread safety is ensured using a reentrant lock (`RLock`), so multiple threads
can safely access or modify settings concurrently.
Features
--------
- Thread-safe operations for getting, setting, and deleting configuration values.
- Loads settings from a JSON config file (`APP_CONFIG_FILE`).
- Creates the config file with default values if missing or corrupt.
- Retrieves, sets, and deletes settings using dot-separated keys.
- Provides safe error handling with logging instead of raising exceptions.
- Persists updates back to disk automatically.
Usage
-----
from .ytsage_config_manager import ConfigManager
# Load settings (auto-loads if not already loaded)
download_path = ConfigManager.get("download_path")
# Update a value
ConfigManager.set("download_path", "D:/Downloads")
# Retrieve nested value
last_check = ConfigManager.get("cached_versions.ytdlp.last_check")
# Delete a key
ConfigManager.delete("cached_versions.ffmpeg.path")
Design Notes
------------
- Settings are stored in `ConfigManager.settings` (a dict).
- Default values are defined in `ConfigManager.default_config`.
- All modifications trigger a save (`_save`) to keep JSON in sync.
- Logs actions and errors using the app's central logger.
- Uses `RLock` to allow safe concurrent access from multiple threads.
Exceptions
----------
- Any issues during file I/O (permissions, disk errors, JSON corruption)
are caught and logged. The application continues running with defaults
when possible.
"""
import json
import threading
from pathlib import Path
from typing import Any, Dict, Optional
from .ytsage_constants import APP_CONFIG_FILE, USER_HOME_DIR
from .ytsage_logger import logger
class ConfigManager:
"""
Thread-safe configuration manager for YTSage.
Provides methods to load, save, get, set, and delete settings stored in a JSON file.
Supports nested keys via dot notation and automatically persists changes.
"""
_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,
"cookie_source": "browser", # "browser" or "file"
"cookie_browser": "chrome",
"cookie_browser_profile": "",
"cookie_file_path": None,
"cookie_active": False, # True only if user explicitly applied cookies
"last_used_cookie_file": None,
"proxy_url": None,
"geo_proxy_url": None,
"auto_update_ytdlp": True,
"auto_update_frequency": "daily",
"last_update_check": 0,
"language": "en",
"ytdlp_channel": "stable",
"force_output_format": False,
"preferred_output_format": "mp4",
"force_audio_format": False,
"preferred_audio_format": "best",
"cached_versions": {
"ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
"ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
},
}
@classmethod
def _load(cls) -> None:
"""
Loads configuration settings from a JSON file if it exists and is valid.
If the file is missing or corrupt, loads default settings and creates or overwrites the config file as needed.
Logs actions and errors during the process.
"""
with cls._lock:
if cls._config_file.exists():
try:
with open(cls._config_file, "r", encoding="utf-8") as f:
cls._settings = json.load(f)
logger.info("Config loaded from file.")
except json.JSONDecodeError:
cls._settings = cls._default_config.copy()
logger.warning("Config file corrupt, loaded defaults.")
else:
cls._settings = cls._default_config.copy()
cls._save()
logger.info("Config file not found, created default config.")
@classmethod
def _save(cls) -> None:
"""
Save current settings to JSON file.
Note:
May raise exceptions if the file cannot be written due to permission issues,
disk errors, or other I/O problems.
"""
with cls._lock:
try:
with open(cls._config_file, "w", encoding="utf-8") as f:
json.dump(cls._settings, f, indent=4)
logger.debug("Config saved to file.")
except (OSError, PermissionError) as e:
logger.exception(f"Failed to save config: {e}")
except Exception as e:
logger.exception(f"Unexpected error while saving config: {e}")
@classmethod
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").
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.
"""
with cls._lock:
if not cls._settings:
cls._load()
parts: list[str] = key.split(".")
value: Any = cls._settings
for part in parts:
if isinstance(value, dict) and part in value:
value = value[part]
else:
logger.debug(f"Config key '{key}' not found.")
return None
return value
@classmethod
def set(cls, key: str, value: Any) -> None:
"""
Sets a configuration value for a given key.
If the configuration settings are not loaded, loads them first.
Supports nested keys using dot notation (e.g., "database.host").
Updates the configuration dictionary with the provided value,
saves the updated settings, and logs the change.
Args:
key (str): The configuration key, possibly nested using dots.
value (Any): The value to set for the specified key.
Returns:
None
"""
with cls._lock:
if not cls._settings:
cls._load()
parts: list[str] = key.split(".")
d: Dict[str, Any] = cls._settings
for part in parts[:-1]:
d = d.setdefault(part, {})
d[parts[-1]] = value
cls._save()
logger.info(f"Config key '{key}' set to '{value}'.")
@classmethod
def delete(cls, key: str) -> None:
"""
Deletes a configuration key from the settings.
If the key is nested (dot-separated), traverses the settings dictionary accordingly.
If the key exists, removes it and saves the updated settings.
Logs the deletion or if the key was not found.
Args:
key (str): The dot-separated configuration key to delete.
Returns:
None
"""
with cls._lock:
if not cls._settings:
cls._load()
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.")
return
d = d[part]
if parts[-1] in d:
d.pop(parts[-1], None)
cls._save()
logger.info(f"Config key '{key}' deleted.")
else:
logger.debug(f"Config key '{key}' not found for deletion.")
ytsage/utils/ytsage_constants.py
+240
View File
@@ -0,0 +1,240 @@
"""
This module defines centralized constants used across the YTSage application.
By storing shared values in one place, it improves consistency, readability,
and maintainability of the codebase.
Constants include:
- Asset paths for icons and notification sounds.
- OS detection and platform-specific directory paths for application data, binaries, logs, and configuration.
- Download URLs for yt-dlp and ffmpeg binaries.
- SUBPROCESS_CREATIONFLAGS: Used to specify subprocess creation flags (e.g., subprocess.CREATE_NO_WINDOW on Windows to hide the console window).
Directories are automatically created when the module is imported, ensuring the required structure exists for the application.
YTSage application constants.
"""
import os
import platform
import subprocess
import sys
from pathlib import Path
# Handle resource paths for both development and installed package
def get_asset_path(asset_relative_path: str) -> Path:
"""
Get the absolute path to an asset file, works both in development and installed package.
Args:
asset_relative_path: Relative path to the asset (e.g., "assets/Icon/icon.png")
Returns:
Path: Absolute path to the asset file
"""
# Check if running as a frozen executable (PyInstaller, cx_Freeze, etc.)
if getattr(sys, "frozen", False):
# Running as a frozen executable
# Try sys._MEIPASS first (PyInstaller)
if hasattr(sys, "_MEIPASS"):
asset_path = Path(sys._MEIPASS) / asset_relative_path
if asset_path.exists():
return asset_path
# For cx_Freeze, assets are typically in lib/ directory next to the executable
executable_dir = Path(sys.executable).parent
# Try with lib/ prefix (cx_Freeze standard structure)
asset_path = executable_dir / "lib" / asset_relative_path
if asset_path.exists():
return asset_path
# Try directly in executable directory
asset_path = executable_dir / asset_relative_path
if asset_path.exists():
return asset_path
# Not frozen - try importlib.resources for installed packages
try:
# Use importlib.resources (standard in Python 3.9+)
import importlib.resources as resources
try:
# Navigate to the package root and then to the asset
package_path = resources.files('ytsage')
asset_path = package_path / asset_relative_path
if asset_path.is_file():
return Path(str(asset_path))
except (ImportError, AttributeError, FileNotFoundError):
pass
except Exception:
pass
# Fallback to relative path (for development environment)
current_file = Path(__file__)
# Go up from utils to ytsage root, then to asset
ytsage_root = current_file.parent.parent.parent
asset_path = ytsage_root / asset_relative_path
return asset_path
# Assets Constants
ICON_PATH: Path = get_asset_path("assets/Icon/icon.png")
SOUND_PATH: Path = get_asset_path("assets/sound/notification.mp3")
OS_NAME: str = platform.system() # Windows ; Darwin ; Linux
IS_FROZEN = getattr(sys, "frozen", False)
USER_HOME_DIR: Path = Path.home()
# OS Specific Constants
if OS_NAME == "Windows":
OS_FULL_NAME: str = f"{OS_NAME} {platform.release()}"
# Always use user data directory for app data, logs, config, and binaries
# Even when frozen, we don't want to create these folders next to the executable
APP_DIR: Path = Path(os.environ.get("LOCALAPPDATA", USER_HOME_DIR / "AppData" / "Local")) / "YTSage"
APP_BIN_DIR: Path = APP_DIR / "bin"
APP_DATA_DIR: Path = APP_DIR / "data"
APP_LOG_DIR: Path = APP_DIR / "logs"
APP_CONFIG_FILE: Path = APP_DATA_DIR / "ytsage_config.json"
APP_HISTORY_FILE: Path = APP_DATA_DIR / "ytsage_history.json"
APP_THUMBNAILS_DIR: Path = APP_DATA_DIR / "thumbnails"
YTDLP_DOWNLOAD_URL: str = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp.exe"
YTDLP_APP_BIN_PATH: Path = APP_BIN_DIR / "yt-dlp.exe"
SUBPROCESS_CREATIONFLAGS: int = subprocess.CREATE_NO_WINDOW
elif OS_NAME == "Darwin": # macOS
_mac_version = platform.mac_ver()[0]
OS_FULL_NAME: str = f"macOS {_mac_version}" if _mac_version else "macOS"
# Always use user data directory for app data, logs, config, and binaries
# Even when frozen, we don't want to create these folders next to the executable
APP_DIR: Path = USER_HOME_DIR / "Library" / "Application Support" / "YTSage"
APP_BIN_DIR: Path = APP_DIR / "bin"
APP_DATA_DIR: Path = APP_DIR / "data"
APP_LOG_DIR: Path = APP_DIR / "logs"
APP_CONFIG_FILE: Path = APP_DATA_DIR / "ytsage_config.json"
APP_HISTORY_FILE: Path = APP_DATA_DIR / "ytsage_history.json"
APP_THUMBNAILS_DIR: Path = APP_DATA_DIR / "thumbnails"
YTDLP_DOWNLOAD_URL: str = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp_macos"
YTDLP_APP_BIN_PATH: Path = APP_BIN_DIR / "yt-dlp"
SUBPROCESS_CREATIONFLAGS: int = 0
else: # Linux and other UNIX-like
OS_FULL_NAME: str = f"{OS_NAME} {platform.release()}"
# Always use user data directory for app data, logs, config, and binaries
# Even when frozen, we don't want to create these folders next to the executable
APP_DIR: Path = USER_HOME_DIR / ".local" / "share" / "YTSage"
APP_BIN_DIR: Path = APP_DIR / "bin"
APP_DATA_DIR: Path = APP_DIR / "data"
APP_LOG_DIR: Path = APP_DIR / "logs"
APP_CONFIG_FILE: Path = APP_DATA_DIR / "ytsage_config.json"
APP_HISTORY_FILE: Path = APP_DATA_DIR / "ytsage_history.json"
APP_THUMBNAILS_DIR: Path = APP_DATA_DIR / "thumbnails"
YTDLP_DOWNLOAD_URL: str = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp"
# Check for environment variable override (critical for Flatpak support)
_ytdlp_env_path = os.environ.get("YTDLP_APP_BIN_PATH")
if _ytdlp_env_path:
YTDLP_APP_BIN_PATH: Path = Path(_ytdlp_env_path)
else:
YTDLP_APP_BIN_PATH: Path = APP_BIN_DIR / "yt-dlp"
SUBPROCESS_CREATIONFLAGS: int = 0
# Documentation URLs
YTDLP_DOCS_URL: str = "https://github.com/yt-dlp/yt-dlp?tab=readme-ov-file#usage-and-options"
# yt-dlp SHA256 checksums URL
YTDLP_SHA256_URL: str = "https://github.com/yt-dlp/yt-dlp/releases/latest/download/SHA2-256SUMS"
# Deno download URLs and paths
if OS_NAME == "Windows":
DENO_DOWNLOAD_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-pc-windows-msvc.zip"
DENO_SHA256_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-pc-windows-msvc.zip.sha256sum"
DENO_APP_BIN_PATH: Path = APP_BIN_DIR / "deno.exe"
elif OS_NAME == "Darwin": # macOS
DENO_DOWNLOAD_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-apple-darwin.zip"
DENO_SHA256_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-apple-darwin.zip.sha256sum"
DENO_APP_BIN_PATH: Path = APP_BIN_DIR / "deno"
else: # Linux
DENO_DOWNLOAD_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-unknown-linux-gnu.zip"
DENO_SHA256_URL: str = "https://github.com/denoland/deno/releases/latest/download/deno-x86_64-unknown-linux-gnu.zip.sha256sum"
# Check for environment variable override (critical for Flatpak support)
_deno_env_path = os.environ.get("DENO_APP_BIN_PATH")
if _deno_env_path:
DENO_APP_BIN_PATH: Path = Path(_deno_env_path)
else:
DENO_APP_BIN_PATH: Path = APP_BIN_DIR / "deno"
# FFmpeg download links (Essentials build - always latest version)
FFMPEG_7Z_DOWNLOAD_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.7z"
FFMPEG_ZIP_DOWNLOAD_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip"
FFMPEG_7Z_SHA256_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.7z.sha256"
FFMPEG_ZIP_SHA256_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip.sha256"
FFMPEG_7Z_VERSION_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.7z.ver"
FFMPEG_ZIP_VERSION_URL = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip.ver"
# =============================================================================
# File Extension Constants
# =============================================================================
# Centralized file extension definitions to avoid duplication across modules
# Use these constants for file type detection throughout the application
# Video file extensions (container formats that typically contain video)
VIDEO_EXTENSIONS: frozenset[str] = frozenset({
".mp4", ".webm", ".mkv", ".avi", ".mov", ".flv"
})
# Audio file extensions (audio-only formats)
AUDIO_EXTENSIONS: frozenset[str] = frozenset({
".mp3", ".m4a", ".aac", ".wav", ".ogg", ".opus", ".flac"
})
# Subtitle file extensions
SUBTITLE_EXTENSIONS: frozenset[str] = frozenset({
".vtt", ".srt", ".ass", ".ssa"
})
# Combined video and audio extensions (for file search operations)
MEDIA_EXTENSIONS: frozenset[str] = VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
if __name__ == "__main__":
# If this file is run directly, print directory information; if imported, create the necessary directories for the application.
# for debug, to check os specific variable which can be different based on os.
info = {
"OS_NAME": OS_NAME,
"OS_FULL_NAME": OS_FULL_NAME,
"USER_HOME_DIR": str(USER_HOME_DIR),
"APP_DIR": str(APP_DIR),
"APP_BIN_DIR": str(APP_BIN_DIR),
"APP_DATA_DIR": str(APP_DATA_DIR),
"APP_LOG_DIR": str(APP_LOG_DIR),
"APP_CONFIG_FILE": str(APP_CONFIG_FILE),
"YTDLP_DOWNLOAD_URL": YTDLP_DOWNLOAD_URL,
"YTDLP_APP_BIN_PATH": YTDLP_APP_BIN_PATH,
"SUBPROCESS_CREATIONFLAGS": SUBPROCESS_CREATIONFLAGS,
}
for key, value in info.items():
print(f"{key}: {value}")
else:
APP_DIR.mkdir(parents=True, exist_ok=True)
APP_BIN_DIR.mkdir(parents=True, exist_ok=True)
APP_DATA_DIR.mkdir(parents=True, exist_ok=True)
APP_LOG_DIR.mkdir(parents=True, exist_ok=True)
APP_THUMBNAILS_DIR.mkdir(parents=True, exist_ok=True)
# Ensure custom yt-dlp directory exists if set
if OS_NAME not in ["Windows", "Darwin"]:
# Ensure parent directories exist for custom paths
if "YTDLP_APP_BIN_PATH" in globals():
YTDLP_APP_BIN_PATH.parent.mkdir(parents=True, exist_ok=True)
if "DENO_APP_BIN_PATH" in globals():
DENO_APP_BIN_PATH.parent.mkdir(parents=True, exist_ok=True)
ytsage/utils/ytsage_history_manager.py
+428
View File
@@ -0,0 +1,428 @@
"""
History Manager Module
======================
This module provides **thread-safe** centralized management for download
history in YTSage using SQLite for high performance and scalability.
Features
--------
- Scalable: Uses SQLite instead of parsing potentially large JSON files.
- Thread-safe: Handles database connections safely.
- Migration: Automatically migrates legacy JSON history to SQLite.
- CRUD: Create, Read, Delete, Clear operations for history entries.
Usage
-----
from .ytsage_history_manager import HistoryManager
# Add a download to history
HistoryManager.add_entry(...)
# Get all history entries
history = HistoryManager.get_all_entries()
# Get recent entries (limit + offset support planned)
# Remove an entry
HistoryManager.remove_entry(entry_id)
# Clear all history
HistoryManager.clear_history()
"""
import json
import sqlite3
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from .ytsage_constants import APP_HISTORY_FILE, APP_DATA_DIR
from .ytsage_logger import logger
class HistoryManager:
"""
Thread-safe history manager for YTSage using SQLite.
"""
_lock = threading.RLock()
# Define DB file next to the old JSON file
_db_file = APP_DATA_DIR / "ytsage_history.db"
_connection = None
_initialized = False
@classmethod
def _init_db(cls):
"""Initialize the database: create table and migrate if needed."""
if cls._initialized:
return
with cls._lock:
# Check if file exists to know if we need to migrate or just create schema
db_exists = cls._db_file.exists()
legacy_json_exists = APP_HISTORY_FILE.exists()
try:
# Ensure directory exists
cls._db_file.parent.mkdir(parents=True, exist_ok=True)
# We use a persistent connection to avoid churn
if cls._connection is None:
cls._connection = sqlite3.connect(cls._db_file, check_same_thread=False)
cls._connection.row_factory = sqlite3.Row
cursor = cls._connection.cursor()
# Create table
cursor.execute("""
CREATE TABLE IF NOT EXISTS history (
id TEXT PRIMARY KEY,
title TEXT,
url TEXT,
channel TEXT,
file_path TEXT,
download_date TEXT,
file_size INTEGER,
thumbnail_url TEXT,
format_id TEXT,
resolution TEXT,
is_audio_only INTEGER,
duration TEXT,
options TEXT,
timestamp REAL
)
""")
# Index for faster sorting by date
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON history (timestamp DESC)
""")
# Indexes for faster search (title, channel, url)
# This prevents full table scans during search
cursor.execute("CREATE INDEX IF NOT EXISTS idx_title ON history (title)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_channel ON history (channel)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_url ON history (url)")
cls._connection.commit()
# If we just created the DB and have a JSON file, migrate
if not db_exists and legacy_json_exists:
cls._migrate_legacy_json()
cls._initialized = True
except sqlite3.Error as e:
logger.error(f"Failed to initialize history database: {e}")
@classmethod
def _migrate_legacy_json(cls):
"""Migrate legacy JSON history to SQLite."""
logger.info("Migrating legacy history JSON to SQLite...")
try:
with open(APP_HISTORY_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, list):
count = 0
# Use the persistent connection
conn = cls._get_connection()
try:
with conn: # Transaction
cursor = conn.cursor()
for entry in data:
try:
# Safely extract download_options logic if complex
options_json = json.dumps(entry.get("download_options", {}))
# Construct timestamp from isoformat if missing
ts = entry.get("timestamp")
if not ts and "download_date" in entry:
try:
dt = datetime.fromisoformat(entry["download_date"])
ts = dt.timestamp()
except Exception:
ts = time.time()
cursor.execute("""
INSERT OR IGNORE INTO history (
id, title, url, channel, file_path, download_date,
file_size, thumbnail_url, format_id, resolution,
is_audio_only, duration, options, timestamp
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
entry.get("id", str(int(time.time()*1000))),
entry.get("title", ""),
entry.get("url", ""),
entry.get("channel", "Unknown"),
entry.get("file_path", ""),
entry.get("download_date", ""),
entry.get("file_size", 0),
entry.get("thumbnail_url", ""),
entry.get("format_id", ""),
entry.get("resolution", ""),
1 if entry.get("is_audio_only") else 0,
entry.get("duration", ""),
options_json,
ts or time.time()
))
count += 1
except Exception as e:
logger.error(f"Skipped invalid entry during migration: {e}")
logger.info(f"Successfully migrated {count} history entries.")
# Rename old JSON to .bak to avoid re-migration, or keep as backup
try:
APP_HISTORY_FILE.rename(APP_HISTORY_FILE.with_suffix(".json.bak"))
except Exception as e:
logger.warning(f"Could not rename legacy history file: {e}")
except Exception as e:
logger.error(f"Migration transaction failed: {e}")
except Exception as e:
logger.error(f"Migration failed: {e}")
@classmethod
def _get_connection(cls):
"""Get the persistent database connection."""
cls._init_db()
if cls._connection is None:
# Should be created in _init_db, but just in case
cls._connection = sqlite3.connect(cls._db_file, check_same_thread=False)
cls._connection.row_factory = sqlite3.Row
return cls._connection
@classmethod
def get_all_entries(cls, limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""
Get all history entries, sorted by most recent first.
Args:
limit: Optional limit on number of entries to return (most recent first)
Returns:
List of dictionary entries.
"""
entries = []
try:
with cls._lock: # Lock for simple concurrency safety
# Use persistent connection
conn = cls._get_connection()
# conn.row_factory is already set in _init_db/_get_connection
cursor = conn.cursor()
query = "SELECT * FROM history ORDER BY timestamp DESC"
params = ()
if limit is not None:
query += " LIMIT ?"
params = (limit,)
cursor.execute(query, params)
rows = cursor.fetchall()
for row in rows:
entry = dict(row)
# Convert boolean back
entry["is_audio_only"] = bool(entry["is_audio_only"])
# Parse options JSON
try:
entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {}
except json.JSONDecodeError:
entry["download_options"] = {}
del entry["options"] # Remove internal column
entries.append(entry)
except Exception as e:
logger.error(f"Error fetching history: {e}")
return entries
@classmethod
def get_entry(cls, entry_id: str) -> Optional[Dict[str, Any]]:
"""
Get a specific history entry by ID.
Args:
entry_id: The unique entry ID
Returns:
Entry dictionary or None if not found
"""
try:
with cls._lock:
conn = cls._get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM history WHERE id = ?", (entry_id,))
row = cursor.fetchone()
if row:
entry = dict(row)
entry["is_audio_only"] = bool(entry["is_audio_only"])
try:
entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {}
except json.JSONDecodeError:
entry["download_options"] = {}
del entry["options"]
return entry
return None
except Exception as e:
logger.error(f"Error fetching entry {entry_id}: {e}")
return None
@classmethod
def add_entry(
cls,
title: str,
url: str,
thumbnail_url: Optional[str],
file_path: str,
format_id: str,
is_audio_only: bool,
resolution: str,
file_size: Optional[int] = None,
channel: Optional[str] = None,
duration: Optional[str] = None,
download_options: Optional[Dict[str, Any]] = None,
) -> str:
"""
Add a new entry to the download history.
"""
if download_options is None:
download_options = {}
timestamp = time.time()
# Ensure unique ID
unique_id = f"{int(timestamp * 1000)}"
download_date = datetime.fromtimestamp(timestamp).isoformat()
# Determine file size if not provided
if file_size is None:
try:
p = Path(file_path)
if p.exists():
file_size = p.stat().st_size
except Exception:
file_size = 0
# Allow None for optional strings
channel = channel or "Unknown"
duration = duration or ""
thumbnail_url = thumbnail_url or ""
try:
with cls._lock:
conn = cls._get_connection()
cursor = conn.cursor()
cursor.execute("""
INSERT INTO history (
id, title, url, channel, file_path, download_date,
file_size, thumbnail_url, format_id, resolution,
is_audio_only, duration, options, timestamp
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
unique_id,
title,
url,
channel,
str(file_path),
download_date,
file_size,
thumbnail_url,
format_id,
resolution,
1 if is_audio_only else 0,
duration,
json.dumps(download_options),
timestamp
))
conn.commit()
logger.info(f"Added history entry: {title}")
return unique_id
except Exception as e:
logger.error(f"Error adding history entry: {e}")
return ""
@classmethod
def remove_entry(cls, entry_id: str) -> bool:
"""Remove an entry from history by ID."""
try:
with cls._lock:
conn = cls._get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM history WHERE id = ?", (entry_id,))
if cursor.rowcount > 0:
conn.commit()
logger.info(f"Removed history entry: {entry_id}")
return True
return False
except Exception as e:
logger.error(f"Error removing history entry: {e}")
return False
@classmethod
def clear_history(cls) -> int:
"""Clear all history entries."""
try:
with cls._lock:
conn = cls._get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM history")
count = cursor.rowcount
conn.commit()
logger.info("History cleared")
return count
except Exception as e:
logger.error(f"Error clearing history: {e}")
return 0
@classmethod
def search_entries(cls, query: str) -> List[Dict[str, Any]]:
"""
Search history entries by title, channel, or URL.
Args:
query: Search query string
Returns:
List of matching history entries
"""
if not query:
return cls.get_all_entries()
entries = []
try:
search_pattern = f"%{query}%"
with cls._lock:
conn = cls._get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM history
WHERE title LIKE ? OR channel LIKE ? OR url LIKE ?
ORDER BY timestamp DESC
""", (search_pattern, search_pattern, search_pattern))
rows = cursor.fetchall()
for row in rows:
entry = dict(row)
entry["is_audio_only"] = bool(entry["is_audio_only"])
try:
entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {}
except json.JSONDecodeError:
entry["download_options"] = {}
del entry["options"]
entries.append(entry)
except Exception as e:
logger.error(f"Error searching history: {e}")
return entries
ytsage/utils/ytsage_localization.py
+283
View File
@@ -0,0 +1,283 @@
"""
Localization Manager Module
==========================
This module provides centralized localization support for YTSage application.
It handles loading language files, switching languages, and retrieving localized strings.
Features
--------
- Thread-safe operations for getting localized text
- Fallback to English when translation is missing
- Support for multiple languages via JSON files
- Dynamic language switching without restart
- Nested key support with dot notation
Usage
-----
from .ytsage_localization import LocalizationManager
# Get localized text
text = LocalizationManager.get_text("download.ready")
button_text = LocalizationManager.get_text("buttons.download")
# Change language
LocalizationManager.set_language("es")
# Get available languages
languages = LocalizationManager.get_available_languages()
"""
import json
import threading
from pathlib import Path
from typing import Any, Dict
from .ytsage_logger import logger
class LocalizationManager:
"""
Thread-safe localization manager for YTSage.
Handles loading, caching, and retrieving localized strings from JSON language files.
"""
_lock = threading.RLock()
_current_language = "en"
_languages: Dict[str, Dict[str, Any]] = {}
_languages_dir = Path(__file__).parent.parent / "languages"
# Fallback English strings embedded in code
_fallback_strings = {
"app": {
"title": "YTSage",
"version": "v{version}",
"ready": "Ready"
},
"buttons": {
"download": "Download",
"pause": "Pause",
"resume": "Resume",
"cancel": "Cancel",
"browse": "Browse",
"clear": "Clear",
"ok": "OK",
"apply": "Apply",
"close": "Close"
},
"dialogs": {
"custom_options": "Custom Options",
"settings": "Settings"
},
"tabs": {
"cookies": "Login with Cookies",
"custom_command": "Custom Command",
"proxy": "Proxy",
"language": "Language"
},
"language": {
"select_language": "Select Language:",
"current_language": "Current language: {language}",
"restart_required": "Language changes will take effect after restarting the application.",
"english": "English",
"spanish": "Español (Spanish)",
"portuguese": "Português (Portuguese)",
"russian": "Русский (Russian)",
"chinese": "中文 (简体) (Chinese Simplified)",
"german": "Deutsch (German)",
"french": "Français (French)",
"hindi": "हिन्दी (Hindi)",
"indonesian": "Bahasa Indonesia (Indonesian)",
"turkish": "Türkçe (Turkish)",
"polish": "Polski (Polish)",
"italian": "Italiano (Italian)",
"arabic": "العربية (Arabic)",
"japanese": "日本語 (Japanese)"
},
"download": {
"preparing": "🚀 Preparing your download...",
"completed": "✅ Download completed!",
"video_completed": "✅ Video download completed!",
"audio_completed": "✅ Audio download completed!",
"subtitle_completed": "✅ Subtitle download completed!",
"please_set_path": "Please set a download path using 'Change Path'",
"please_enter_url": "Please enter a URL",
"please_enter_url_and_path": "Please enter URL and set download path",
"please_select_format": "Please select a format"
},
"formats": {
"show_formats": "Show formats:"
},
"errors": {
"download_failed_return_code_conflict": "Download failed with return code {return_code}. This may be due to a conflict with multiple yt-dlp installations. Try uninstalling any system-installed yt-dlp (e.g. through snap or apt) and restart the application.",
"download_failed_return_code": "Download failed with return code {return_code}",
"direct_command_error": "Error in direct command: {error}"
}
}
@classmethod
def _ensure_languages_dir(cls) -> None:
"""Ensure the languages directory exists."""
cls._languages_dir.mkdir(exist_ok=True)
@classmethod
def _load_language(cls, language_code: str) -> Dict[str, Any]:
"""
Load a language file from disk.
Args:
language_code: The language code (e.g., 'en', 'es')
Returns:
Dictionary containing the language strings, or empty dict if not found
"""
language_file = cls._languages_dir / f"{language_code}.json"
if not language_file.exists():
logger.warning(f"Language file not found: {language_file}")
return {}
try:
with open(language_file, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.error(f"Failed to load language file {language_file}: {e}")
return {}
@classmethod
def _get_nested_value(cls, data: Dict[str, Any], key: str) -> Any:
"""
Get a nested value from dictionary using dot notation.
Args:
data: Dictionary to search in
key: Dot-separated key (e.g., "app.title")
Returns:
The value if found, None otherwise
"""
parts = key.split(".")
value = data
for part in parts:
if isinstance(value, dict) and part in value:
value = value[part]
else:
return None
return value
@classmethod
def get_text(cls, key: str, **kwargs) -> str:
"""
Get localized text for the given key.
Args:
key: Dot-separated key for the text (e.g., "app.title")
**kwargs: Format parameters for the text
Returns:
Localized text, with fallback to English if not found
"""
with cls._lock:
# Load current language if not cached
if cls._current_language not in cls._languages:
cls._languages[cls._current_language] = cls._load_language(cls._current_language)
# Try to get from current language
current_lang_data = cls._languages.get(cls._current_language, {})
text = cls._get_nested_value(current_lang_data, key)
# Fallback to embedded English strings
if text is None:
text = cls._get_nested_value(cls._fallback_strings, key)
# Final fallback to key itself
if text is None:
logger.warning(f"Localization key not found: {key}")
text = key
# Format the text with provided parameters
if kwargs and isinstance(text, str):
try:
text = text.format(**kwargs)
except (KeyError, ValueError) as e:
logger.warning(f"Failed to format localized text '{key}': {e}")
return str(text)
@classmethod
def set_language(cls, language_code: str) -> None:
"""
Set the current language.
Args:
language_code: The language code to set (e.g., 'en', 'es')
"""
with cls._lock:
if language_code != cls._current_language:
cls._current_language = language_code
# Clear cache to force reload
cls._languages.clear()
logger.info(f"Language set to: {language_code}")
@classmethod
def get_current_language(cls) -> str:
"""Get the current language code."""
return cls._current_language
@classmethod
def get_available_languages(cls) -> Dict[str, str]:
"""
Get available languages from the languages directory.
Returns:
Dictionary mapping language codes to display names
"""
cls._ensure_languages_dir()
available_languages = {"en": cls.get_text("language.english")}
# Scan for language files
for language_file in cls._languages_dir.glob("*.json"):
lang_code = language_file.stem
if lang_code != "en": # Skip English as it's already added
# Try to get language display name from the file
lang_data = cls._load_language(lang_code)
display_name = cls._get_nested_value(lang_data, "language.display_name")
if display_name:
available_languages[lang_code] = display_name
else:
# Fallback display name
available_languages[lang_code] = lang_code.upper()
return available_languages
@classmethod
def initialize(cls, language_code: str = "en") -> None:
"""
Initialize the localization system.
Args:
language_code: Initial language code to use
"""
with cls._lock:
cls._ensure_languages_dir()
cls.set_language(language_code)
logger.info(f"Localization system initialized with language: {language_code}")
# Convenience function for getting localized text
def _(key: str, **kwargs) -> str:
"""
Convenience function to get localized text.
Args:
key: Dot-separated key for the text
**kwargs: Format parameters
Returns:
Localized text
"""
return LocalizationManager.get_text(key, **kwargs)
ytsage/utils/ytsage_logger.py
+56
View File
@@ -0,0 +1,56 @@
"""
YTSage centralized logging with loguru.
- This module provides centralized logging configuration for the entire YTSage application.
- Two log files: ytsage.log (all logs) & ytsage_error.log (errors only).
"""
import sys
from loguru import logger
from .ytsage_constants import APP_LOG_DIR, IS_FROZEN
# Separate configs for each handler
CONSOLE_CONFIG = {
"sink": sys.stdout if sys.stdout else sys.stderr,
"level": "INFO",
"colorize": True,
"enqueue": True,
}
ALL_LOGS_CONFIG = {
"sink": APP_LOG_DIR / "ytsage.log",
"level": "DEBUG",
"rotation": "10 MB",
"retention": "14 days",
"compression": "zip",
"enqueue": True,
}
ERROR_LOGS_CONFIG = {
"sink": APP_LOG_DIR / "ytsage_error.log",
"level": "ERROR",
"rotation": "5 MB",
"retention": "30 days",
"compression": "zip",
"enqueue": True,
}
# Logger initialization
def init_logger() -> None:
"""Configure loguru logger using separate configs for each handler."""
logger.remove() # Remove default loguru handler
if not IS_FROZEN:
logger.add(**CONSOLE_CONFIG)
logger.add(**ALL_LOGS_CONFIG)
logger.add(**ERROR_LOGS_CONFIG)
logger.info("YTSage logger initialized")
init_logger()
__all__ = ["logger"]