mirror of
https://github.com/shadoll/moma.git
synced 2026-08-28 11:33:25 +00:00
added media catalog mode, impooved cache
This commit is contained in:
+74
-12
@@ -1,6 +1,7 @@
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.widgets import Tree, Static, Footer, LoadingIndicator
|
||||
from textual.containers import Horizontal, Container, ScrollableContainer, Vertical
|
||||
from textual.widget import Widget
|
||||
from rich.markup import escape
|
||||
from pathlib import Path
|
||||
import threading
|
||||
@@ -9,11 +10,14 @@ import logging
|
||||
import os
|
||||
|
||||
from .constants import MEDIA_TYPES
|
||||
from .screens import OpenScreen, HelpScreen, RenameConfirmScreen
|
||||
from .screens import OpenScreen, HelpScreen, RenameConfirmScreen, SettingsScreen
|
||||
from .extractors.extractor import MediaExtractor
|
||||
from .formatters.media_formatter import MediaFormatter
|
||||
from .formatters.proposed_name_formatter import ProposedNameFormatter
|
||||
from .formatters.text_formatter import TextFormatter
|
||||
from .formatters.catalog_formatter import CatalogFormatter
|
||||
from .settings import Settings
|
||||
from .cache import Cache
|
||||
|
||||
|
||||
# Set up logging conditionally
|
||||
@@ -43,13 +47,17 @@ class RenamerApp(App):
|
||||
("f", "refresh", "Refresh"),
|
||||
("r", "rename", "Rename"),
|
||||
("p", "expand", "Toggle Tree"),
|
||||
("m", "toggle_mode", "Toggle Mode"),
|
||||
("h", "help", "Help"),
|
||||
("ctrl+s", "settings", "Settings"),
|
||||
]
|
||||
|
||||
def __init__(self, scan_dir):
|
||||
super().__init__()
|
||||
self.scan_dir = Path(scan_dir) if scan_dir else None
|
||||
self.tree_expanded = False
|
||||
self.settings = Settings()
|
||||
self.cache = Cache()
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Horizontal():
|
||||
@@ -60,7 +68,10 @@ class RenamerApp(App):
|
||||
yield LoadingIndicator(id="loading")
|
||||
with ScrollableContainer(id="details_container"):
|
||||
yield Static(
|
||||
"Select a file to view details", id="details", markup=True
|
||||
"Select a file to view details", id="details_technical", markup=True
|
||||
)
|
||||
yield Static(
|
||||
"", id="details_catalog", markup=False
|
||||
)
|
||||
yield Static("", id="proposed", markup=True)
|
||||
yield Footer()
|
||||
@@ -73,7 +84,7 @@ class RenamerApp(App):
|
||||
def scan_files(self):
|
||||
logging.info("scan_files called")
|
||||
if not self.scan_dir or not self.scan_dir.exists() or not self.scan_dir.is_dir():
|
||||
details = self.query_one("#details", Static)
|
||||
details = self.query_one("#details_technical", Static)
|
||||
details.update("Error: Directory does not exist or is not a directory")
|
||||
return
|
||||
tree = self.query_one("#file_tree", Tree)
|
||||
@@ -105,7 +116,11 @@ class RenamerApp(App):
|
||||
def _start_loading_animation(self):
|
||||
loading = self.query_one("#loading", LoadingIndicator)
|
||||
loading.display = True
|
||||
details = self.query_one("#details", Static)
|
||||
mode = self.settings.get("mode")
|
||||
if mode == "technical":
|
||||
details = self.query_one("#details_technical", Static)
|
||||
else:
|
||||
details = self.query_one("#details_catalog", Static)
|
||||
details.update("Retrieving media data")
|
||||
proposed = self.query_one("#proposed", Static)
|
||||
proposed.update("")
|
||||
@@ -119,7 +134,10 @@ class RenamerApp(App):
|
||||
if node.data and isinstance(node.data, Path):
|
||||
if node.data.is_dir():
|
||||
self._stop_loading_animation()
|
||||
details = self.query_one("#details", Static)
|
||||
details = self.query_one("#details_technical", Static)
|
||||
details.display = True
|
||||
details_catalog = self.query_one("#details_catalog", Static)
|
||||
details_catalog.display = False
|
||||
details.update("Directory")
|
||||
proposed = self.query_one("#proposed", Static)
|
||||
proposed.update("")
|
||||
@@ -133,12 +151,20 @@ class RenamerApp(App):
|
||||
time.sleep(1) # Minimum delay to show loading
|
||||
try:
|
||||
# Initialize extractors and formatters
|
||||
extractor = MediaExtractor(file_path)
|
||||
|
||||
extractor = MediaExtractor.create(file_path, self.cache, self.settings.get("cache_ttl_extractors"))
|
||||
|
||||
mode = self.settings.get("mode")
|
||||
if mode == "technical":
|
||||
formatter = MediaFormatter(extractor)
|
||||
full_info = formatter.file_info_panel()
|
||||
else: # catalog
|
||||
formatter = CatalogFormatter(extractor)
|
||||
full_info = formatter.format_catalog_info()
|
||||
|
||||
# Update UI
|
||||
self.call_later(
|
||||
self._update_details,
|
||||
MediaFormatter(extractor).file_info_panel(),
|
||||
full_info,
|
||||
ProposedNameFormatter(extractor).rename_line_formatted(file_path),
|
||||
)
|
||||
except Exception as e:
|
||||
@@ -150,9 +176,18 @@ class RenamerApp(App):
|
||||
|
||||
def _update_details(self, full_info: str, display_string: str):
|
||||
self._stop_loading_animation()
|
||||
details = self.query_one("#details", Static)
|
||||
details.update(full_info)
|
||||
|
||||
details_technical = self.query_one("#details_technical", Static)
|
||||
details_catalog = self.query_one("#details_catalog", Static)
|
||||
mode = self.settings.get("mode")
|
||||
if mode == "technical":
|
||||
details_technical.display = True
|
||||
details_catalog.display = False
|
||||
details_technical.update(full_info)
|
||||
else:
|
||||
details_technical.display = False
|
||||
details_catalog.display = True
|
||||
details_catalog.update(full_info)
|
||||
|
||||
proposed = self.query_one("#proposed", Static)
|
||||
proposed.update(display_string)
|
||||
|
||||
@@ -170,6 +205,11 @@ class RenamerApp(App):
|
||||
tree = self.query_one("#file_tree", Tree)
|
||||
node = tree.cursor_node
|
||||
if node and node.data and isinstance(node.data, Path) and node.data.is_file():
|
||||
# Clear cache for this file
|
||||
cache_key_base = str(node.data)
|
||||
# Invalidate all keys for this file (we can improve this later)
|
||||
for key in ["title", "year", "source", "extension", "video_tracks", "audio_tracks", "subtitle_tracks"]:
|
||||
self.cache.invalidate(f"{cache_key_base}_{key}")
|
||||
self._start_loading_animation()
|
||||
threading.Thread(
|
||||
target=self._extract_and_show_details, args=(node.data,)
|
||||
@@ -178,12 +218,29 @@ class RenamerApp(App):
|
||||
async def action_help(self):
|
||||
self.push_screen(HelpScreen())
|
||||
|
||||
async def action_settings(self):
|
||||
self.push_screen(SettingsScreen())
|
||||
|
||||
async def action_toggle_mode(self):
|
||||
current_mode = self.settings.get("mode")
|
||||
new_mode = "catalog" if current_mode == "technical" else "technical"
|
||||
self.settings.set("mode", new_mode)
|
||||
self.notify(f"Switched to {new_mode} mode", severity="information", timeout=2)
|
||||
# Refresh current file display if any
|
||||
tree = self.query_one("#file_tree", Tree)
|
||||
node = tree.cursor_node
|
||||
if node and node.data and isinstance(node.data, Path) and node.data.is_file():
|
||||
self._start_loading_animation()
|
||||
threading.Thread(
|
||||
target=self._extract_and_show_details, args=(node.data,)
|
||||
).start()
|
||||
|
||||
async def action_rename(self):
|
||||
tree = self.query_one("#file_tree", Tree)
|
||||
node = tree.cursor_node
|
||||
if node and node.data and isinstance(node.data, Path) and node.data.is_file():
|
||||
# Get the proposed name from the extractor
|
||||
extractor = MediaExtractor(node.data)
|
||||
extractor = MediaExtractor.create(node.data, self.cache, self.settings.get("cache_ttl_extractors"))
|
||||
proposed_formatter = ProposedNameFormatter(extractor)
|
||||
new_name = str(proposed_formatter)
|
||||
logging.info(f"Proposed new name: {new_name!r} for file: {node.data}")
|
||||
@@ -216,6 +273,11 @@ class RenamerApp(App):
|
||||
"""Update the tree node for a renamed file."""
|
||||
logging.info(f"update_renamed_file called with old_path={old_path}, new_path={new_path}")
|
||||
|
||||
# Clear cache for old file
|
||||
cache_key_base = str(old_path)
|
||||
for key in ["title", "year", "source", "extension", "video_tracks", "audio_tracks", "subtitle_tracks"]:
|
||||
self.cache.invalidate(f"{cache_key_base}_{key}")
|
||||
|
||||
tree = self.query_one("#file_tree", Tree)
|
||||
logging.info(f"Before update: cursor_node.data = {tree.cursor_node.data if tree.cursor_node else None}")
|
||||
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import hashlib
|
||||
import pickle
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
class Cache:
|
||||
"""File-based cache with TTL support."""
|
||||
|
||||
def __init__(self, cache_dir: Optional[Path] = None):
|
||||
if cache_dir is None:
|
||||
cache_dir = Path.home() / ".cache" / "renamer"
|
||||
self.cache_dir = cache_dir
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _get_cache_file(self, key: str) -> Path:
|
||||
"""Get cache file path with hashed filename and subdirs."""
|
||||
# Parse key format: ClassName.method_name.param_hash
|
||||
if '.' in key:
|
||||
parts = key.split('.')
|
||||
if len(parts) >= 3:
|
||||
class_name = parts[0]
|
||||
method_name = parts[1]
|
||||
param_hash = parts[2]
|
||||
|
||||
# Use class name as subdir
|
||||
cache_subdir = self.cache_dir / class_name
|
||||
cache_subdir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Use method_name.param_hash as filename
|
||||
return cache_subdir / f"{method_name}.{param_hash}.pkl"
|
||||
|
||||
# Fallback for old keys (tmdb_, poster_, etc.)
|
||||
if key.startswith("tmdb_"):
|
||||
subdir = "tmdb"
|
||||
subkey = key[5:] # Remove "tmdb_" prefix
|
||||
elif key.startswith("poster_"):
|
||||
subdir = "posters"
|
||||
subkey = key[7:] # Remove "poster_" prefix
|
||||
else:
|
||||
subdir = "general"
|
||||
subkey = key
|
||||
|
||||
# Create subdir
|
||||
cache_subdir = self.cache_dir / subdir
|
||||
cache_subdir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Hash the subkey for filename
|
||||
key_hash = hashlib.md5(subkey.encode('utf-8')).hexdigest()
|
||||
return cache_subdir / f"{key_hash}.json"
|
||||
|
||||
def get(self, key: str) -> Optional[Any]:
|
||||
"""Get cached value if not expired."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
if not cache_file.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(cache_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
if time.time() > data.get('expires', 0):
|
||||
# Expired, remove file
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
return data.get('value')
|
||||
except (json.JSONDecodeError, IOError):
|
||||
# Corrupted, remove
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
def set(self, key: str, value: Any, ttl_seconds: int) -> None:
|
||||
"""Set cached value with TTL."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
data = {
|
||||
'value': value,
|
||||
'expires': time.time() + ttl_seconds
|
||||
}
|
||||
try:
|
||||
with open(cache_file, 'w') as f:
|
||||
json.dump(data, f)
|
||||
except IOError:
|
||||
pass # Silently fail
|
||||
|
||||
def invalidate(self, key: str) -> None:
|
||||
"""Remove cache entry."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
cache_file.unlink(missing_ok=True)
|
||||
|
||||
def get_image(self, key: str) -> Optional[Path]:
|
||||
"""Get cached image path if not expired."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
if not cache_file.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(cache_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
if time.time() > data.get('expires', 0):
|
||||
# Expired, remove file and image
|
||||
image_path = data.get('image_path')
|
||||
if image_path and Path(image_path).exists():
|
||||
Path(image_path).unlink(missing_ok=True)
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
image_path = data.get('image_path')
|
||||
if image_path and Path(image_path).exists():
|
||||
return Path(image_path)
|
||||
return None
|
||||
except (json.JSONDecodeError, IOError):
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
def set_image(self, key: str, image_data: bytes, ttl_seconds: int) -> Optional[Path]:
|
||||
"""Set cached image and return path."""
|
||||
# Determine subdir and subkey
|
||||
if key.startswith("poster_"):
|
||||
subdir = "posters"
|
||||
subkey = key[7:]
|
||||
else:
|
||||
subdir = "images"
|
||||
subkey = key
|
||||
|
||||
# Create subdir
|
||||
image_dir = self.cache_dir / subdir
|
||||
image_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Hash for filename
|
||||
key_hash = hashlib.md5(subkey.encode('utf-8')).hexdigest()
|
||||
image_path = image_dir / f"{key_hash}.jpg"
|
||||
|
||||
try:
|
||||
with open(image_path, 'wb') as f:
|
||||
f.write(image_data)
|
||||
|
||||
# Cache metadata
|
||||
data = {
|
||||
'image_path': str(image_path),
|
||||
'expires': time.time() + ttl_seconds
|
||||
}
|
||||
cache_file = self._get_cache_file(key)
|
||||
with open(cache_file, 'w') as f:
|
||||
json.dump(data, f)
|
||||
|
||||
return image_path
|
||||
except IOError:
|
||||
return None
|
||||
|
||||
def get_object(self, key: str) -> Optional[Any]:
|
||||
"""Get pickled object from cache if not expired."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
if not cache_file.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(cache_file, 'rb') as f:
|
||||
data = pickle.load(f)
|
||||
|
||||
if time.time() > data.get('expires', 0):
|
||||
# Expired, remove file
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
return data.get('value')
|
||||
except (pickle.PickleError, IOError):
|
||||
# Corrupted, remove
|
||||
cache_file.unlink(missing_ok=True)
|
||||
return None
|
||||
|
||||
def set_object(self, key: str, obj: Any, ttl_seconds: int) -> None:
|
||||
"""Pickle and cache object with TTL."""
|
||||
cache_file = self._get_cache_file(key)
|
||||
data = {
|
||||
'value': obj,
|
||||
'expires': time.time() + ttl_seconds
|
||||
}
|
||||
try:
|
||||
with open(cache_file, 'wb') as f:
|
||||
pickle.dump(data, f)
|
||||
except IOError:
|
||||
pass # Silently fail
|
||||
@@ -47,6 +47,7 @@ SOURCE_DICT = {
|
||||
"DVDRip": ["DVDRip", "DVD-Rip", "DVDRIP"],
|
||||
"HDTVRip": ["HDTVRip", "HDTV"],
|
||||
"BluRay": ["BluRay", "BLURAY", "Blu-ray"],
|
||||
"SATRip": ["SATRip", "SAT-Rip", "SATRIP"],
|
||||
"VHSRecord": [
|
||||
"VHSRecord",
|
||||
"VHS Record",
|
||||
@@ -69,6 +70,11 @@ FRAME_CLASSES = {
|
||||
"typical_widths": [640, 704, 720],
|
||||
"description": "Standard Definition (SD) interlaced - NTSC quality",
|
||||
},
|
||||
"360p": {
|
||||
"nominal_height": 360,
|
||||
"typical_widths": [480, 640],
|
||||
"description": "Low Definition (LD) - 360p",
|
||||
},
|
||||
"576p": {
|
||||
"nominal_height": 576,
|
||||
"typical_widths": [720, 768],
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
# Decorators package
|
||||
from .caching import cached_method
|
||||
|
||||
__all__ = ['cached_method']
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Caching decorators for extractors."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
from renamer.cache import Cache
|
||||
|
||||
|
||||
# Global cache instance
|
||||
_cache = Cache()
|
||||
|
||||
|
||||
def cached_method(ttl_seconds: int = 3600) -> Callable:
|
||||
"""Decorator to cache method results with TTL.
|
||||
|
||||
Caches the result of a method call using a global file-based cache.
|
||||
The cache key includes class name, method name, and parameters hash.
|
||||
|
||||
Args:
|
||||
ttl_seconds: Time to live for cached results in seconds (default 1 hour)
|
||||
|
||||
Returns:
|
||||
The decorated method with caching
|
||||
"""
|
||||
def decorator(func: Callable) -> Callable:
|
||||
def wrapper(self, *args, **kwargs) -> Any:
|
||||
# Generate cache key: class_name.method_name.param_hash
|
||||
class_name = self.__class__.__name__
|
||||
method_name = func.__name__
|
||||
|
||||
# Create hash from args and kwargs
|
||||
param_str = json.dumps((args, kwargs), sort_keys=True, default=str)
|
||||
param_hash = hashlib.md5(param_str.encode('utf-8')).hexdigest()
|
||||
|
||||
cache_key = f"{class_name}.{method_name}.{param_hash}"
|
||||
|
||||
# Try to get from cache
|
||||
cached_result = _cache.get_object(cache_key)
|
||||
if cached_result is not None:
|
||||
return cached_result
|
||||
|
||||
# Compute result and cache it
|
||||
result = func(self, *args, **kwargs)
|
||||
_cache.set_object(cache_key, result, ttl_seconds)
|
||||
return result
|
||||
|
||||
return wrapper
|
||||
return decorator
|
||||
@@ -10,14 +10,40 @@ from .default_extractor import DefaultExtractor
|
||||
class MediaExtractor:
|
||||
"""Class to extract various metadata from media files using specialized extractors"""
|
||||
|
||||
def __init__(self, file_path: Path):
|
||||
@classmethod
|
||||
def create(cls, file_path: Path, cache=None, ttl_seconds: int = 21600):
|
||||
"""Factory method that returns cached object if available, else creates new."""
|
||||
if cache:
|
||||
cache_key = f"extractor_{file_path}"
|
||||
cached_obj = cache.get_object(cache_key)
|
||||
if cached_obj:
|
||||
print(f"Loaded MediaExtractor object from cache for {file_path.name}")
|
||||
return cached_obj
|
||||
|
||||
# Create new instance
|
||||
instance = cls(file_path, cache, ttl_seconds)
|
||||
|
||||
# Cache the object
|
||||
if cache:
|
||||
cache_key = f"extractor_{file_path}"
|
||||
cache.set_object(cache_key, instance, ttl_seconds)
|
||||
print(f"Cached MediaExtractor object for {file_path.name}")
|
||||
|
||||
return instance
|
||||
|
||||
def __init__(self, file_path: Path, cache=None, ttl_seconds: int = 21600):
|
||||
self.file_path = file_path
|
||||
self.cache = cache
|
||||
self.ttl_seconds = ttl_seconds
|
||||
self.cache_key = f"file_data_{file_path}"
|
||||
|
||||
self.filename_extractor = FilenameExtractor(file_path)
|
||||
self.metadata_extractor = MetadataExtractor(file_path)
|
||||
self.mediainfo_extractor = MediaInfoExtractor(file_path)
|
||||
self.fileinfo_extractor = FileInfoExtractor(file_path)
|
||||
self.tmdb_extractor = TMDBExtractor(file_path)
|
||||
self.tmdb_extractor = TMDBExtractor(file_path, cache, ttl_seconds)
|
||||
self.default_extractor = DefaultExtractor()
|
||||
|
||||
|
||||
# Extractor mapping
|
||||
self._extractors = {
|
||||
"Metadata": self.metadata_extractor,
|
||||
@@ -164,9 +190,16 @@ class MediaExtractor:
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# No caching logic here - handled in create() method
|
||||
|
||||
def get(self, key: str, source: str | None = None):
|
||||
"""Get extracted data by key, optionally from specific source"""
|
||||
print(f"Extracting real data for key '{key}' in {self.file_path.name}")
|
||||
return self._get_uncached(key, source)
|
||||
|
||||
def _get_uncached(self, key: str, source: str | None = None):
|
||||
"""Original get logic without caching"""
|
||||
if source:
|
||||
# Specific source requested - find the extractor and call the method directly
|
||||
for extractor_name, extractor in self._extractors.items():
|
||||
@@ -174,27 +207,20 @@ class MediaExtractor:
|
||||
method = f"extract_{key}"
|
||||
if hasattr(extractor, method):
|
||||
val = getattr(extractor, method)()
|
||||
# Apply condition if specified
|
||||
if key in self._data and "condition" in self._data[key]:
|
||||
condition = self._data[key]["condition"]
|
||||
return val if condition(val) else None
|
||||
return val
|
||||
return val if val is not None else None
|
||||
return None
|
||||
|
||||
# Fallback mode - try sources in order
|
||||
if key in self._data:
|
||||
data = self._data[key]
|
||||
sources = data["sources"]
|
||||
condition = data.get("condition", lambda x: x is not None)
|
||||
sources = self._data[key]["sources"]
|
||||
else:
|
||||
# Try extractors in order for unconfigured keys
|
||||
sources = [(name, f"extract_{key}") for name in ["MediaInfo", "Metadata", "Filename", "FileInfo"]]
|
||||
condition = lambda x: x is not None
|
||||
|
||||
# Try each source in order until a valid value is found
|
||||
for src, method in sources:
|
||||
if src in self._extractors and hasattr(self._extractors[src], method):
|
||||
val = getattr(self._extractors[src], method)()
|
||||
if condition(val):
|
||||
if val is not None:
|
||||
return val
|
||||
return None
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from pathlib import Path
|
||||
import logging
|
||||
import os
|
||||
from ..decorators import cached_method
|
||||
|
||||
# Set up logging conditionally
|
||||
if os.getenv('FORMATTER_LOG', '0') == '1':
|
||||
@@ -19,24 +20,30 @@ class FileInfoExtractor:
|
||||
self._modification_time = file_path.stat().st_mtime
|
||||
self._file_name = file_path.name
|
||||
self._file_path = str(file_path)
|
||||
self._cache = {} # Internal cache for method results
|
||||
logging.info(f"FileInfoExtractor: file_name={self._file_name!r}, file_path={self._file_path!r}")
|
||||
|
||||
@cached_method()
|
||||
def extract_size(self) -> int:
|
||||
"""Extract file size in bytes"""
|
||||
return self._size
|
||||
|
||||
@cached_method()
|
||||
def extract_modification_time(self) -> float:
|
||||
"""Extract file modification time"""
|
||||
return self._modification_time
|
||||
|
||||
@cached_method()
|
||||
def extract_file_name(self) -> str:
|
||||
"""Extract file name"""
|
||||
return self._file_name
|
||||
|
||||
@cached_method()
|
||||
def extract_file_path(self) -> str:
|
||||
"""Extract full file path as string"""
|
||||
return self._file_path
|
||||
|
||||
@cached_method()
|
||||
def extract_extension(self) -> str:
|
||||
"""Extract file extension without the dot"""
|
||||
return self.file_path.suffix.lower().lstrip('.')
|
||||
@@ -2,6 +2,7 @@ import re
|
||||
from pathlib import Path
|
||||
from collections import Counter
|
||||
from ..constants import SOURCE_DICT, FRAME_CLASSES, MOVIE_DB_DICT, SPECIAL_EDITIONS
|
||||
from ..decorators import cached_method
|
||||
import langcodes
|
||||
|
||||
|
||||
@@ -34,6 +35,7 @@ class FilenameExtractor:
|
||||
return frame_class
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_title(self) -> str | None:
|
||||
"""Extract movie title from filename"""
|
||||
# Find positions of year, source, and quality brackets
|
||||
@@ -120,6 +122,7 @@ class FilenameExtractor:
|
||||
|
||||
return title if title else None
|
||||
|
||||
@cached_method()
|
||||
def extract_year(self) -> str | None:
|
||||
"""Extract year from filename"""
|
||||
# First try to find year in parentheses (most common and reliable)
|
||||
@@ -144,6 +147,7 @@ class FilenameExtractor:
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_source(self) -> str | None:
|
||||
"""Extract video source from filename"""
|
||||
temp_name = re.sub(r'\s*\(\d{4}\)\s*|\s*\d{4}\s*|\.\d{4}\.', ' ', self.file_name)
|
||||
@@ -154,6 +158,7 @@ class FilenameExtractor:
|
||||
return src
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_order(self) -> str | None:
|
||||
"""Extract collection order number from filename (at the beginning)"""
|
||||
# Look for order patterns at the start of filename
|
||||
@@ -176,6 +181,7 @@ class FilenameExtractor:
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_frame_class(self) -> str | None:
|
||||
"""Extract frame class from filename (480p, 720p, 1080p, 2160p, etc.)"""
|
||||
# Normalize Cyrillic characters for resolution parsing
|
||||
@@ -200,6 +206,7 @@ class FilenameExtractor:
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_hdr(self) -> str | None:
|
||||
"""Extract HDR information from filename"""
|
||||
# Check for SDR first - indicates no HDR
|
||||
@@ -212,6 +219,7 @@ class FilenameExtractor:
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_movie_db(self) -> list[str] | None:
|
||||
"""Extract movie database identifier from filename"""
|
||||
# Look for patterns at the end of filename in brackets or braces
|
||||
@@ -233,6 +241,7 @@ class FilenameExtractor:
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_special_info(self) -> list[str] | None:
|
||||
"""Extract special edition information from filename"""
|
||||
# Look for special edition indicators in brackets or as standalone text
|
||||
@@ -258,6 +267,7 @@ class FilenameExtractor:
|
||||
|
||||
return special_info if special_info else None
|
||||
|
||||
@cached_method()
|
||||
def extract_audio_langs(self) -> str:
|
||||
"""Extract audio languages from filename"""
|
||||
# Look for language patterns in brackets and outside brackets
|
||||
@@ -389,6 +399,7 @@ class FilenameExtractor:
|
||||
audio_langs = [f"{count}{lang}" if count > 1 else lang for lang, count in lang_counts.items()]
|
||||
return ','.join(audio_langs)
|
||||
|
||||
@cached_method()
|
||||
def extract_audio_tracks(self) -> list[dict]:
|
||||
"""Extract audio track data from filename (simplified version with only language)"""
|
||||
# Similar to extract_audio_langs but returns list of dicts
|
||||
|
||||
@@ -2,6 +2,7 @@ from pathlib import Path
|
||||
from pymediainfo import MediaInfo
|
||||
from collections import Counter
|
||||
from ..constants import FRAME_CLASSES, MEDIA_TYPES
|
||||
from ..decorators import cached_method
|
||||
import langcodes
|
||||
|
||||
|
||||
@@ -10,6 +11,7 @@ class MediaInfoExtractor:
|
||||
|
||||
def __init__(self, file_path: Path):
|
||||
self.file_path = file_path
|
||||
self._cache = {} # Internal cache for method results
|
||||
try:
|
||||
self.media_info = MediaInfo.parse(file_path)
|
||||
self.video_tracks = [t for t in self.media_info.tracks if t.track_type == 'Video']
|
||||
@@ -54,6 +56,7 @@ class MediaInfoExtractor:
|
||||
return closest
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_duration(self) -> float | None:
|
||||
"""Extract duration from media info in seconds"""
|
||||
if self.media_info:
|
||||
@@ -62,6 +65,7 @@ class MediaInfoExtractor:
|
||||
return getattr(track, 'duration', 0) / 1000 if getattr(track, 'duration', None) else None
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_frame_class(self) -> str | None:
|
||||
"""Extract frame class from media info (480p, 720p, 1080p, etc.)"""
|
||||
if not self.video_tracks:
|
||||
@@ -106,6 +110,7 @@ class MediaInfoExtractor:
|
||||
return f"{closest_height}{scan_type}"
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_resolution(self) -> tuple[int, int] | None:
|
||||
"""Extract actual video resolution as (width, height) tuple from media info"""
|
||||
if not self.video_tracks:
|
||||
@@ -116,6 +121,7 @@ class MediaInfoExtractor:
|
||||
return width, height
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_aspect_ratio(self) -> str | None:
|
||||
"""Extract video aspect ratio from media info"""
|
||||
if not self.video_tracks:
|
||||
@@ -125,6 +131,7 @@ class MediaInfoExtractor:
|
||||
return str(aspect_ratio)
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_hdr(self) -> str | None:
|
||||
"""Extract HDR info from media info"""
|
||||
if not self.video_tracks:
|
||||
@@ -134,6 +141,7 @@ class MediaInfoExtractor:
|
||||
return 'HDR'
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_audio_langs(self) -> str | None:
|
||||
"""Extract audio languages from media info"""
|
||||
if not self.audio_tracks:
|
||||
@@ -154,6 +162,7 @@ class MediaInfoExtractor:
|
||||
audio_langs = [f"{count}{lang}" if count > 1 else lang for lang, count in lang_counts.items()]
|
||||
return ','.join(audio_langs)
|
||||
|
||||
@cached_method()
|
||||
def extract_video_tracks(self) -> list[dict]:
|
||||
"""Extract video track data"""
|
||||
tracks = []
|
||||
@@ -169,6 +178,7 @@ class MediaInfoExtractor:
|
||||
tracks.append(track_data)
|
||||
return tracks
|
||||
|
||||
@cached_method()
|
||||
def extract_audio_tracks(self) -> list[dict]:
|
||||
"""Extract audio track data"""
|
||||
tracks = []
|
||||
@@ -182,6 +192,7 @@ class MediaInfoExtractor:
|
||||
tracks.append(track_data)
|
||||
return tracks
|
||||
|
||||
@cached_method()
|
||||
def extract_subtitle_tracks(self) -> list[dict]:
|
||||
"""Extract subtitle track data"""
|
||||
tracks = []
|
||||
@@ -193,6 +204,7 @@ class MediaInfoExtractor:
|
||||
tracks.append(track_data)
|
||||
return tracks
|
||||
|
||||
@cached_method()
|
||||
def is_3d(self) -> bool:
|
||||
"""Check if the video is 3D"""
|
||||
if not self.video_tracks:
|
||||
@@ -205,6 +217,7 @@ class MediaInfoExtractor:
|
||||
return True
|
||||
return False
|
||||
|
||||
@cached_method()
|
||||
def extract_anamorphic(self) -> str | None:
|
||||
"""Extract anamorphic info for 3D videos"""
|
||||
if not self.video_tracks:
|
||||
@@ -214,6 +227,7 @@ class MediaInfoExtractor:
|
||||
return 'Anamorphic:Yes'
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_extension(self) -> str | None:
|
||||
"""Extract file extension based on container format"""
|
||||
if not self.media_info:
|
||||
@@ -233,6 +247,7 @@ class MediaInfoExtractor:
|
||||
return exts[0] if exts else None
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_3d_layout(self) -> str | None:
|
||||
"""Extract 3D stereoscopic layout from MediaInfo"""
|
||||
if not self.is_3d():
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import mutagen
|
||||
from pathlib import Path
|
||||
from ..constants import MEDIA_TYPES
|
||||
from ..decorators import cached_method
|
||||
|
||||
|
||||
class MetadataExtractor:
|
||||
@@ -8,36 +9,40 @@ class MetadataExtractor:
|
||||
|
||||
def __init__(self, file_path: Path):
|
||||
self.file_path = file_path
|
||||
self._cache = {} # Internal cache for method results
|
||||
try:
|
||||
self.info = mutagen.File(file_path) # type: ignore
|
||||
except Exception:
|
||||
self.info = None
|
||||
|
||||
@cached_method()
|
||||
def extract_title(self) -> str | None:
|
||||
"""Extract title from metadata"""
|
||||
if self.info:
|
||||
return getattr(self.info, 'title', None) or getattr(self.info, 'get', lambda x, default=None: default)('title', [None])[0] # type: ignore
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_duration(self) -> float | None:
|
||||
"""Extract duration from metadata"""
|
||||
if self.info:
|
||||
return getattr(self.info, 'length', None)
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_artist(self) -> str | None:
|
||||
"""Extract artist from metadata"""
|
||||
if self.info:
|
||||
return getattr(self.info, 'artist', None) or getattr(self.info, 'get', lambda x, default=None: default)('artist', [None])[0] # type: ignore
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_meta_type(self) -> str:
|
||||
"""Extract meta type from metadata"""
|
||||
if self.info:
|
||||
return type(self.info).__name__
|
||||
return self._detect_by_mime()
|
||||
|
||||
|
||||
def _detect_by_mime(self) -> str:
|
||||
"""Detect meta type by MIME"""
|
||||
try:
|
||||
|
||||
@@ -11,53 +11,22 @@ from ..secrets import TMDB_API_KEY, TMDB_ACCESS_TOKEN
|
||||
class TMDBExtractor:
|
||||
"""Class to extract TMDB movie information"""
|
||||
|
||||
CACHE_DIR = Path.home() / ".cache" / "renamer" / "tmdb"
|
||||
CACHE_DURATION = 5 * 24 * 60 * 60 # 5 days in seconds
|
||||
|
||||
def __init__(self, file_path: Path):
|
||||
def __init__(self, file_path: Path, cache=None, ttl_seconds: int = 21600):
|
||||
self.file_path = file_path
|
||||
self.cache = cache
|
||||
self.ttl_seconds = ttl_seconds
|
||||
self._movie_db_info = None
|
||||
|
||||
def _get_cache_file_path(self, cache_key: str) -> Path:
|
||||
"""Get the cache file path for a given cache key"""
|
||||
# Create a hash of the cache key for the filename
|
||||
key_hash = hashlib.md5(cache_key.encode('utf-8')).hexdigest()
|
||||
return self.CACHE_DIR / f"{key_hash}.json"
|
||||
|
||||
def _is_cache_valid(self, cache_key: str) -> bool:
|
||||
"""Check if cache entry is still valid"""
|
||||
cache_file = self._get_cache_file_path(cache_key)
|
||||
if not cache_file.exists():
|
||||
return False
|
||||
|
||||
try:
|
||||
# Check file modification time
|
||||
stat = cache_file.stat()
|
||||
return time.time() - stat.st_mtime < self.CACHE_DURATION
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
def _get_cached_data(self, cache_key: str) -> Optional[Dict[str, Any]]:
|
||||
"""Get data from cache if valid"""
|
||||
if not self._is_cache_valid(cache_key):
|
||||
return None
|
||||
|
||||
cache_file = self._get_cache_file_path(cache_key)
|
||||
try:
|
||||
with open(cache_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return None
|
||||
if self.cache:
|
||||
return self.cache.get(f"tmdb_{cache_key}")
|
||||
return None
|
||||
|
||||
def _set_cached_data(self, cache_key: str, data: Dict[str, Any]):
|
||||
"""Store data in cache"""
|
||||
try:
|
||||
self.CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cache_file = self._get_cache_file_path(cache_key)
|
||||
with open(cache_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
except OSError:
|
||||
pass # Silently fail if we can't save cache
|
||||
if self.cache:
|
||||
self.cache.set(f"tmdb_{cache_key}", data, self.ttl_seconds)
|
||||
|
||||
def _make_tmdb_request(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
|
||||
"""Make a request to TMDB API"""
|
||||
@@ -230,9 +199,70 @@ class TMDBExtractor:
|
||||
return f"https://www.themoviedb.org/movie/{movie_id}"
|
||||
return None
|
||||
|
||||
def extract_movie_db(self) -> Optional[Tuple[str, str]]:
|
||||
"""Extract TMDB database info as (name, id) tuple"""
|
||||
movie_id = self.extract_tmdb_id()
|
||||
if movie_id:
|
||||
return ("tmdb", movie_id)
|
||||
def extract_duration(self) -> Optional[str]:
|
||||
"""Extract TMDB runtime in minutes"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info and movie_info.get('runtime'):
|
||||
return str(movie_info['runtime'])
|
||||
return None
|
||||
|
||||
def extract_popularity(self) -> Optional[str]:
|
||||
"""Extract TMDB popularity"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return str(movie_info.get('popularity', ''))
|
||||
return None
|
||||
|
||||
def extract_vote_average(self) -> Optional[str]:
|
||||
"""Extract TMDB vote average"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return str(movie_info.get('vote_average', ''))
|
||||
return None
|
||||
|
||||
def extract_overview(self) -> Optional[str]:
|
||||
"""Extract TMDB overview"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return movie_info.get('overview')
|
||||
return None
|
||||
|
||||
def extract_genres(self) -> Optional[str]:
|
||||
"""Extract TMDB genres as codes"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info and movie_info.get('genres'):
|
||||
return ', '.join(genre['name'] for genre in movie_info['genres'])
|
||||
return None
|
||||
|
||||
def extract_poster_path(self) -> Optional[str]:
|
||||
"""Extract TMDB poster path"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return movie_info.get('poster_path')
|
||||
return None
|
||||
|
||||
def extract_poster_image_path(self) -> Optional[str]:
|
||||
"""Download and cache poster image, return local path"""
|
||||
poster_path = self.extract_poster_path()
|
||||
if not poster_path or not self.cache:
|
||||
return None
|
||||
|
||||
cache_key = f"poster_{poster_path}"
|
||||
cached_path = self.cache.get_image(cache_key)
|
||||
if cached_path:
|
||||
return str(cached_path)
|
||||
|
||||
# Download poster
|
||||
base_url = "https://image.tmdb.org/t/p/w500" # Medium size
|
||||
url = f"{base_url}{poster_path}"
|
||||
|
||||
try:
|
||||
response = requests.get(url, timeout=10)
|
||||
response.raise_for_status()
|
||||
image_data = response.content
|
||||
|
||||
# Cache image
|
||||
local_path = self.cache.set_image(cache_key, image_data, self.ttl_seconds)
|
||||
return str(local_path) if local_path else None
|
||||
except requests.RequestException:
|
||||
return None
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
from .text_formatter import TextFormatter
|
||||
import os
|
||||
|
||||
|
||||
class CatalogFormatter:
|
||||
"""Formatter for catalog mode display"""
|
||||
|
||||
def __init__(self, extractor):
|
||||
self.extractor = extractor
|
||||
|
||||
def format_catalog_info(self) -> str:
|
||||
"""Format catalog information for display"""
|
||||
lines = []
|
||||
|
||||
# Title
|
||||
title = self.extractor.get("title", "TMDB")
|
||||
if title:
|
||||
lines.append(f"{TextFormatter.bold('Title:')} {title}")
|
||||
|
||||
# Year
|
||||
year = self.extractor.get("year", "TMDB")
|
||||
if year:
|
||||
lines.append(f"{TextFormatter.bold('Year:')} {year}")
|
||||
|
||||
# Duration
|
||||
duration = self.extractor.get("duration", "TMDB")
|
||||
if duration:
|
||||
lines.append(f"{TextFormatter.bold('Duration:')} {duration} minutes")
|
||||
|
||||
# Rates
|
||||
popularity = self.extractor.get("popularity", "TMDB")
|
||||
vote_average = self.extractor.get("vote_average", "TMDB")
|
||||
if popularity or vote_average:
|
||||
rates = []
|
||||
if popularity:
|
||||
rates.append(f"Popularity: {popularity}")
|
||||
if vote_average:
|
||||
rates.append(f"Rating: {vote_average}/10")
|
||||
lines.append(f"{TextFormatter.bold('Rates:')} {', '.join(rates)}")
|
||||
|
||||
# Overview
|
||||
overview = self.extractor.get("overview", "TMDB")
|
||||
if overview:
|
||||
lines.append(f"{TextFormatter.bold('Overview:')}")
|
||||
lines.append(overview)
|
||||
|
||||
# Genres
|
||||
genres = self.extractor.get("genres", "TMDB")
|
||||
if genres:
|
||||
lines.append(f"{TextFormatter.bold('Genres:')} {genres}")
|
||||
|
||||
# Poster
|
||||
poster_image_path = self.extractor.tmdb_extractor.extract_poster_image_path()
|
||||
if poster_image_path:
|
||||
lines.append(f"{TextFormatter.bold('Poster:')}")
|
||||
lines.append(self._display_poster(poster_image_path))
|
||||
else:
|
||||
poster_path = self.extractor.get("poster_path", "TMDB")
|
||||
if poster_path:
|
||||
lines.append(f"{TextFormatter.bold('Poster:')} {poster_path} (not cached yet)")
|
||||
|
||||
full_text = "\n\n".join(lines) if lines else "No catalog information available"
|
||||
|
||||
# Render markup to ANSI
|
||||
from rich.console import Console
|
||||
from io import StringIO
|
||||
console = Console(file=StringIO(), width=120, legacy_windows=False)
|
||||
console.print(full_text, markup=True)
|
||||
return console.file.getvalue()
|
||||
|
||||
def _display_poster(self, image_path: str) -> str:
|
||||
"""Display poster image in terminal using simple ASCII art"""
|
||||
try:
|
||||
from PIL import Image
|
||||
import os
|
||||
|
||||
if not os.path.exists(image_path):
|
||||
return f"Image file not found: {image_path}"
|
||||
|
||||
# Open and resize image
|
||||
img = Image.open(image_path).convert('L').resize((80, 40), Image.Resampling.LANCZOS)
|
||||
|
||||
# ASCII characters from dark to light
|
||||
ascii_chars = '@%#*+=-:. '
|
||||
|
||||
# Convert to ASCII
|
||||
pixels = img.getdata()
|
||||
width, height = img.size
|
||||
|
||||
ascii_art = []
|
||||
for y in range(0, height, 2): # Skip every other row for aspect ratio
|
||||
row = []
|
||||
for x in range(width):
|
||||
# Average of two rows for better aspect
|
||||
pixel1 = pixels[y * width + x] if y < height else 255
|
||||
pixel2 = pixels[(y + 1) * width + x] if y + 1 < height else 255
|
||||
avg = (pixel1 + pixel2) // 2
|
||||
char = ascii_chars[avg * len(ascii_chars) // 256]
|
||||
row.append(char)
|
||||
ascii_art.append(''.join(row))
|
||||
|
||||
return '\n'.join(ascii_art)
|
||||
|
||||
except ImportError:
|
||||
return f"Image at {image_path} (PIL not available)"
|
||||
except Exception as e:
|
||||
return f"Failed to display image at {image_path}: {e}"
|
||||
@@ -35,7 +35,6 @@ class FormatterApplier:
|
||||
DateFormatter.format_year,
|
||||
ExtensionFormatter.format_extension_info,
|
||||
ResolutionFormatter.get_frame_class_from_resolution,
|
||||
ResolutionFormatter.format_resolution_p,
|
||||
ResolutionFormatter.format_resolution_dimensions,
|
||||
TrackFormatter.format_video_track,
|
||||
TrackFormatter.format_audio_track,
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from renamer.constants import FRAME_CLASSES
|
||||
|
||||
class ResolutionFormatter:
|
||||
"""Class for formatting video resolutions and frame classes"""
|
||||
|
||||
@@ -20,39 +22,20 @@ class ResolutionFormatter:
|
||||
else:
|
||||
return 'Unclassified'
|
||||
|
||||
if height == 4320:
|
||||
return '4320p'
|
||||
elif height >= 2160:
|
||||
return '2160p'
|
||||
elif height >= 1440:
|
||||
return '1440p'
|
||||
elif height >= 1080:
|
||||
return '1080p'
|
||||
elif height >= 720:
|
||||
return '720p'
|
||||
elif height >= 576:
|
||||
return '576p'
|
||||
elif height >= 480:
|
||||
return '480p'
|
||||
else:
|
||||
return 'Unclassified'
|
||||
# Find the closest frame class based on nominal height
|
||||
closest_class = 'Unclassified'
|
||||
min_diff = float('inf')
|
||||
for frame_class, info in FRAME_CLASSES.items():
|
||||
nominal_height = info['nominal_height']
|
||||
diff = abs(height - nominal_height)
|
||||
if diff < min_diff:
|
||||
min_diff = diff
|
||||
closest_class = frame_class
|
||||
|
||||
return closest_class
|
||||
except (ValueError, IndexError):
|
||||
return 'Unclassified'
|
||||
|
||||
@staticmethod
|
||||
def format_resolution_p(height: int) -> str:
|
||||
"""Format resolution as 2160p, 1080p, etc."""
|
||||
if height >= 2160:
|
||||
return '2160p'
|
||||
elif height >= 1080:
|
||||
return '1080p'
|
||||
elif height >= 720:
|
||||
return '720p'
|
||||
elif height >= 480:
|
||||
return '480p'
|
||||
else:
|
||||
return f'{height}p'
|
||||
|
||||
@staticmethod
|
||||
def format_resolution_dimensions(resolution: tuple[int, int]) -> str:
|
||||
"""Format resolution as WIDTHxHEIGHT"""
|
||||
|
||||
+91
-1
@@ -58,6 +58,8 @@ ACTIONS:
|
||||
• f: Refresh - Reload metadata for selected file
|
||||
• r: Rename - Rename selected file with proposed name
|
||||
• p: Expand/Collapse - Toggle expansion of selected directory
|
||||
• m: Toggle Mode - Switch between technical and catalog display modes
|
||||
• ctrl+s: Settings - Open settings window
|
||||
• h: Help - Show this help screen
|
||||
• q: Quit - Exit the application
|
||||
|
||||
@@ -237,4 +239,92 @@ Do you want to proceed with renaming?
|
||||
content.update(f"Error renaming file: {str(e)}")
|
||||
elif event.key == "n":
|
||||
# Cancel
|
||||
self.app.pop_screen()
|
||||
self.app.pop_screen()
|
||||
|
||||
|
||||
class SettingsScreen(Screen):
|
||||
CSS = """
|
||||
#settings_content {
|
||||
text-align: center;
|
||||
}
|
||||
Button:focus {
|
||||
background: $primary;
|
||||
}
|
||||
#buttons {
|
||||
align: center middle;
|
||||
}
|
||||
.input_field {
|
||||
width: 100%;
|
||||
margin: 1 0;
|
||||
}
|
||||
.label {
|
||||
text-align: left;
|
||||
margin-bottom: 0;
|
||||
}
|
||||
"""
|
||||
|
||||
def compose(self):
|
||||
from .formatters.text_formatter import TextFormatter
|
||||
|
||||
settings = self.app.settings # type: ignore
|
||||
|
||||
content = f"""
|
||||
{TextFormatter.bold("SETTINGS")}
|
||||
|
||||
Configure application settings.
|
||||
""".strip()
|
||||
|
||||
with Center():
|
||||
with Vertical():
|
||||
yield Static(content, id="settings_content", markup=True)
|
||||
|
||||
# Mode selection
|
||||
yield Static("Display Mode:", classes="label")
|
||||
with Horizontal():
|
||||
yield Button("Technical", id="mode_technical", variant="primary" if settings.get("mode") == "technical" else "default")
|
||||
yield Button("Catalog", id="mode_catalog", variant="primary" if settings.get("mode") == "catalog" else "default")
|
||||
|
||||
# TTL inputs
|
||||
yield Static("Cache TTL - Extractors (hours):", classes="label")
|
||||
yield Input(value=str(settings.get("cache_ttl_extractors") // 3600), id="ttl_extractors", classes="input_field")
|
||||
|
||||
yield Static("Cache TTL - TMDB (hours):", classes="label")
|
||||
yield Input(value=str(settings.get("cache_ttl_tmdb") // 3600), id="ttl_tmdb", classes="input_field")
|
||||
|
||||
yield Static("Cache TTL - Posters (days):", classes="label")
|
||||
yield Input(value=str(settings.get("cache_ttl_posters") // 86400), id="ttl_posters", classes="input_field")
|
||||
|
||||
with Horizontal(id="buttons"):
|
||||
yield Button("Save", id="save")
|
||||
yield Button("Cancel", id="cancel")
|
||||
|
||||
def on_button_pressed(self, event):
|
||||
if event.button.id == "save":
|
||||
self.save_settings()
|
||||
self.app.pop_screen() # type: ignore
|
||||
elif event.button.id == "cancel":
|
||||
self.app.pop_screen() # type: ignore
|
||||
elif event.button.id.startswith("mode_"):
|
||||
# Toggle mode buttons
|
||||
mode = event.button.id.split("_")[1]
|
||||
self.app.settings.set("mode", mode) # type: ignore
|
||||
# Update button variants
|
||||
tech_btn = self.query_one("#mode_technical", Button)
|
||||
cat_btn = self.query_one("#mode_catalog", Button)
|
||||
tech_btn.variant = "primary" if mode == "technical" else "default"
|
||||
cat_btn.variant = "primary" if mode == "catalog" else "default"
|
||||
|
||||
def save_settings(self):
|
||||
try:
|
||||
# Get values and convert to seconds
|
||||
ttl_extractors = int(self.query_one("#ttl_extractors", Input).value) * 3600
|
||||
ttl_tmdb = int(self.query_one("#ttl_tmdb", Input).value) * 3600
|
||||
ttl_posters = int(self.query_one("#ttl_posters", Input).value) * 86400
|
||||
|
||||
self.app.settings.set("cache_ttl_extractors", ttl_extractors) # type: ignore
|
||||
self.app.settings.set("cache_ttl_tmdb", ttl_tmdb) # type: ignore
|
||||
self.app.settings.set("cache_ttl_posters", ttl_posters) # type: ignore
|
||||
|
||||
self.app.notify("Settings saved!", severity="information", timeout=2) # type: ignore
|
||||
except ValueError:
|
||||
self.app.notify("Invalid TTL values. Please enter numbers only.", severity="error", timeout=3) # type: ignore
|
||||
@@ -0,0 +1,72 @@
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
class Settings:
|
||||
"""Manages application settings stored in a JSON file."""
|
||||
|
||||
DEFAULTS = {
|
||||
"mode": "technical", # "technical" or "catalog"
|
||||
"cache_ttl_extractors": 21600, # 6 hours in seconds
|
||||
"cache_ttl_tmdb": 21600, # 6 hours in seconds
|
||||
"cache_ttl_posters": 2592000, # 30 days in seconds
|
||||
}
|
||||
|
||||
def __init__(self, config_dir: Path = None):
|
||||
if config_dir is None:
|
||||
config_dir = Path.home() / ".config" / "renamer"
|
||||
self.config_dir = config_dir
|
||||
self.config_file = self.config_dir / "config.json"
|
||||
self._settings = self.DEFAULTS.copy()
|
||||
self.load()
|
||||
|
||||
def load(self) -> None:
|
||||
"""Load settings from file, using defaults if file doesn't exist."""
|
||||
if self.config_file.exists():
|
||||
try:
|
||||
with open(self.config_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
# Validate and merge with defaults
|
||||
for key, default_value in self.DEFAULTS.items():
|
||||
if key in data:
|
||||
# Basic type checking
|
||||
if isinstance(data[key], type(default_value)):
|
||||
self._settings[key] = data[key]
|
||||
else:
|
||||
print(f"Warning: Invalid type for {key}, using default")
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
print(f"Warning: Could not load settings: {e}, using defaults")
|
||||
else:
|
||||
# Create config directory and file with defaults
|
||||
self.save()
|
||||
|
||||
def save(self) -> None:
|
||||
"""Save current settings to file."""
|
||||
try:
|
||||
self.config_dir.mkdir(parents=True, exist_ok=True)
|
||||
with open(self.config_file, 'w') as f:
|
||||
json.dump(self._settings, f, indent=2)
|
||||
except IOError as e:
|
||||
print(f"Error: Could not save settings: {e}")
|
||||
|
||||
def get(self, key: str) -> Any:
|
||||
"""Get a setting value."""
|
||||
return self._settings.get(key, self.DEFAULTS.get(key))
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
"""Set a setting value and save."""
|
||||
if key in self.DEFAULTS:
|
||||
# Basic type checking
|
||||
if isinstance(value, type(self.DEFAULTS[key])):
|
||||
self._settings[key] = value
|
||||
self.save()
|
||||
else:
|
||||
raise ValueError(f"Invalid type for setting {key}")
|
||||
else:
|
||||
raise KeyError(f"Unknown setting: {key}")
|
||||
|
||||
def get_all(self) -> Dict[str, Any]:
|
||||
"""Get all current settings."""
|
||||
return self._settings.copy()
|
||||
Reference in New Issue
Block a user