mirror of
https://github.com/shadoll/moma.git
synced 2026-08-28 11:33:25 +00:00
Add rename service and utility modules for file renaming operations
- Implemented RenameService for handling file renaming with features like name validation, proposed name generation, conflict detection, and atomic rename operations. - Created utility modules for language code extraction, regex pattern matching, and frame class matching to centralize common functionalities. - Added comprehensive logging for error handling and debugging across all new modules.
This commit is contained in:
@@ -0,0 +1,21 @@
|
||||
"""Services package - business logic layer for the Renamer application.
|
||||
|
||||
This package contains service classes that encapsulate business logic and
|
||||
coordinate between different components. Services provide a clean separation
|
||||
of concerns and make the application more testable and maintainable.
|
||||
|
||||
Services:
|
||||
- FileTreeService: Manages file tree operations (scanning, building, filtering)
|
||||
- MetadataService: Coordinates metadata extraction with caching and threading
|
||||
- RenameService: Handles file rename operations with validation
|
||||
"""
|
||||
|
||||
from .file_tree_service import FileTreeService
|
||||
from .metadata_service import MetadataService
|
||||
from .rename_service import RenameService
|
||||
|
||||
__all__ = [
|
||||
'FileTreeService',
|
||||
'MetadataService',
|
||||
'RenameService',
|
||||
]
|
||||
@@ -0,0 +1,280 @@
|
||||
"""File tree service for managing directory scanning and tree building.
|
||||
|
||||
This service encapsulates all file system operations related to building
|
||||
and managing the file tree display.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable
|
||||
from rich.markup import escape
|
||||
|
||||
from renamer.constants import MEDIA_TYPES
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FileTreeService:
|
||||
"""Service for managing file tree operations.
|
||||
|
||||
This service handles:
|
||||
- Directory scanning and validation
|
||||
- File tree construction with filtering
|
||||
- File type filtering based on media types
|
||||
- Permission error handling
|
||||
|
||||
Example:
|
||||
service = FileTreeService()
|
||||
files = service.scan_directory(Path("/media/movies"))
|
||||
service.build_tree(Path("/media/movies"), tree_node)
|
||||
"""
|
||||
|
||||
def __init__(self, media_types: Optional[set[str]] = None):
|
||||
"""Initialize the file tree service.
|
||||
|
||||
Args:
|
||||
media_types: Set of file extensions to include (without dot).
|
||||
If None, uses MEDIA_TYPES from constants.
|
||||
"""
|
||||
self.media_types = media_types or MEDIA_TYPES
|
||||
logger.debug(f"FileTreeService initialized with {len(self.media_types)} media types")
|
||||
|
||||
def validate_directory(self, path: Path) -> tuple[bool, Optional[str]]:
|
||||
"""Validate that a path is a valid directory.
|
||||
|
||||
Args:
|
||||
path: The path to validate
|
||||
|
||||
Returns:
|
||||
Tuple of (is_valid, error_message). If valid, error_message is None.
|
||||
|
||||
Example:
|
||||
>>> service = FileTreeService()
|
||||
>>> is_valid, error = service.validate_directory(Path("/tmp"))
|
||||
>>> if is_valid:
|
||||
... print("Directory is valid")
|
||||
"""
|
||||
if not path:
|
||||
return False, "No directory specified"
|
||||
|
||||
if not path.exists():
|
||||
return False, f"Directory does not exist: {path}"
|
||||
|
||||
if not path.is_dir():
|
||||
return False, f"Path is not a directory: {path}"
|
||||
|
||||
try:
|
||||
# Test if we can read the directory
|
||||
list(path.iterdir())
|
||||
return True, None
|
||||
except PermissionError:
|
||||
return False, f"Permission denied: {path}"
|
||||
except Exception as e:
|
||||
return False, f"Error accessing directory: {e}"
|
||||
|
||||
def scan_directory(self, path: Path, recursive: bool = True) -> list[Path]:
|
||||
"""Scan a directory and return all media files.
|
||||
|
||||
Args:
|
||||
path: The directory to scan
|
||||
recursive: If True, scan subdirectories recursively
|
||||
|
||||
Returns:
|
||||
List of Path objects for all media files found
|
||||
|
||||
Example:
|
||||
>>> service = FileTreeService()
|
||||
>>> files = service.scan_directory(Path("/media/movies"))
|
||||
>>> print(f"Found {len(files)} media files")
|
||||
"""
|
||||
is_valid, error = self.validate_directory(path)
|
||||
if not is_valid:
|
||||
logger.warning(f"Cannot scan directory: {error}")
|
||||
return []
|
||||
|
||||
media_files = []
|
||||
try:
|
||||
for item in sorted(path.iterdir()):
|
||||
try:
|
||||
if item.is_dir():
|
||||
# Skip hidden directories and system directories
|
||||
if item.name.startswith(".") or item.name == "lost+found":
|
||||
continue
|
||||
|
||||
if recursive:
|
||||
# Recursively scan subdirectories
|
||||
media_files.extend(self.scan_directory(item, recursive=True))
|
||||
elif item.is_file():
|
||||
# Check if file has a media extension
|
||||
if self._is_media_file(item):
|
||||
media_files.append(item)
|
||||
logger.debug(f"Found media file: {item}")
|
||||
except PermissionError:
|
||||
logger.debug(f"Permission denied: {item}")
|
||||
continue
|
||||
except PermissionError:
|
||||
logger.warning(f"Permission denied scanning directory: {path}")
|
||||
|
||||
return media_files
|
||||
|
||||
def build_tree(
|
||||
self,
|
||||
path: Path,
|
||||
node,
|
||||
add_node_callback: Optional[Callable] = None
|
||||
):
|
||||
"""Build a tree structure from a directory.
|
||||
|
||||
This method recursively builds a tree by adding directories and media files
|
||||
to the provided node. Uses a callback to add nodes to maintain compatibility
|
||||
with different tree implementations.
|
||||
|
||||
Args:
|
||||
path: The directory path to build tree from
|
||||
node: The tree node to add children to
|
||||
add_node_callback: Optional callback(node, label, data) to add a child node.
|
||||
If None, uses node.add(label, data=data)
|
||||
|
||||
Example:
|
||||
>>> from textual.widgets import Tree
|
||||
>>> tree = Tree("Files")
|
||||
>>> service = FileTreeService()
|
||||
>>> service.build_tree(Path("/media"), tree.root)
|
||||
"""
|
||||
if add_node_callback is None:
|
||||
# Default implementation for Textual Tree
|
||||
add_node_callback = lambda parent, label, data: parent.add(label, data=data)
|
||||
|
||||
try:
|
||||
for item in sorted(path.iterdir()):
|
||||
try:
|
||||
if item.is_dir():
|
||||
# Skip hidden and system directories
|
||||
if item.name.startswith(".") or item.name == "lost+found":
|
||||
continue
|
||||
|
||||
# Add directory node
|
||||
subnode = add_node_callback(node, escape(item.name), item)
|
||||
# Recursively build tree for subdirectory
|
||||
self.build_tree(item, subnode, add_node_callback)
|
||||
|
||||
elif item.is_file() and self._is_media_file(item):
|
||||
# Add media file node
|
||||
logger.debug(f"Adding file to tree: {item.name!r} (full path: {item})")
|
||||
add_node_callback(node, escape(item.name), item)
|
||||
|
||||
except PermissionError:
|
||||
logger.debug(f"Permission denied: {item}")
|
||||
continue
|
||||
except PermissionError:
|
||||
logger.warning(f"Permission denied building tree: {path}")
|
||||
|
||||
def find_node_by_path(self, root_node, target_path: Path):
|
||||
"""Find a tree node by file path.
|
||||
|
||||
Recursively searches the tree for a node with matching data path.
|
||||
|
||||
Args:
|
||||
root_node: The root node to start searching from
|
||||
target_path: The Path to search for
|
||||
|
||||
Returns:
|
||||
The matching node or None if not found
|
||||
|
||||
Example:
|
||||
>>> node = service.find_node_by_path(tree.root, Path("/media/movie.mkv"))
|
||||
>>> if node:
|
||||
... node.label = "New Name.mkv"
|
||||
"""
|
||||
# Check if this node matches
|
||||
if hasattr(root_node, 'data') and root_node.data == target_path:
|
||||
return root_node
|
||||
|
||||
# Recursively search children
|
||||
if hasattr(root_node, 'children'):
|
||||
for child in root_node.children:
|
||||
result = self.find_node_by_path(child, target_path)
|
||||
if result:
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
def count_media_files(self, path: Path) -> int:
|
||||
"""Count the number of media files in a directory.
|
||||
|
||||
Args:
|
||||
path: The directory to count files in
|
||||
|
||||
Returns:
|
||||
Number of media files found (including subdirectories)
|
||||
|
||||
Example:
|
||||
>>> count = service.count_media_files(Path("/media/movies"))
|
||||
>>> print(f"Found {count} media files")
|
||||
"""
|
||||
return len(self.scan_directory(path, recursive=True))
|
||||
|
||||
def _is_media_file(self, path: Path) -> bool:
|
||||
"""Check if a file is a media file based on extension.
|
||||
|
||||
Args:
|
||||
path: The file path to check
|
||||
|
||||
Returns:
|
||||
True if the file has a media extension
|
||||
|
||||
Example:
|
||||
>>> service._is_media_file(Path("movie.mkv"))
|
||||
True
|
||||
>>> service._is_media_file(Path("readme.txt"))
|
||||
False
|
||||
"""
|
||||
extension = path.suffix.lower()
|
||||
# Remove the leading dot and check against media types
|
||||
return extension.lstrip('.') in {ext.lower() for ext in self.media_types}
|
||||
|
||||
def get_directory_stats(self, path: Path) -> dict[str, int]:
|
||||
"""Get statistics about a directory.
|
||||
|
||||
Args:
|
||||
path: The directory to analyze
|
||||
|
||||
Returns:
|
||||
Dictionary with stats: total_files, total_dirs, media_files
|
||||
|
||||
Example:
|
||||
>>> stats = service.get_directory_stats(Path("/media"))
|
||||
>>> print(f"Media files: {stats['media_files']}")
|
||||
"""
|
||||
stats = {
|
||||
'total_files': 0,
|
||||
'total_dirs': 0,
|
||||
'media_files': 0,
|
||||
}
|
||||
|
||||
is_valid, _ = self.validate_directory(path)
|
||||
if not is_valid:
|
||||
return stats
|
||||
|
||||
try:
|
||||
for item in path.iterdir():
|
||||
try:
|
||||
if item.is_dir():
|
||||
if not item.name.startswith(".") and item.name != "lost+found":
|
||||
stats['total_dirs'] += 1
|
||||
# Recursively count subdirectories
|
||||
sub_stats = self.get_directory_stats(item)
|
||||
stats['total_files'] += sub_stats['total_files']
|
||||
stats['total_dirs'] += sub_stats['total_dirs']
|
||||
stats['media_files'] += sub_stats['media_files']
|
||||
elif item.is_file():
|
||||
stats['total_files'] += 1
|
||||
if self._is_media_file(item):
|
||||
stats['media_files'] += 1
|
||||
except PermissionError:
|
||||
continue
|
||||
except PermissionError:
|
||||
pass
|
||||
|
||||
return stats
|
||||
@@ -0,0 +1,325 @@
|
||||
"""Metadata service for coordinating metadata extraction and caching.
|
||||
|
||||
This service manages the extraction of metadata from media files with:
|
||||
- Thread pool for concurrent extraction
|
||||
- Cache integration for performance
|
||||
- Formatter coordination for display
|
||||
- Error handling and recovery
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable
|
||||
from concurrent.futures import ThreadPoolExecutor, Future
|
||||
from threading import Lock
|
||||
|
||||
from renamer.cache import Cache
|
||||
from renamer.settings import Settings
|
||||
from renamer.extractors.extractor import MediaExtractor
|
||||
from renamer.formatters.media_formatter import MediaFormatter
|
||||
from renamer.formatters.catalog_formatter import CatalogFormatter
|
||||
from renamer.formatters.proposed_name_formatter import ProposedNameFormatter
|
||||
from renamer.formatters.text_formatter import TextFormatter
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MetadataService:
|
||||
"""Service for managing metadata extraction and formatting.
|
||||
|
||||
This service coordinates:
|
||||
- Metadata extraction from media files
|
||||
- Caching of extracted metadata
|
||||
- Thread pool management for concurrent operations
|
||||
- Formatting for different display modes (technical/catalog)
|
||||
- Proposed name generation
|
||||
|
||||
The service uses a thread pool to extract metadata concurrently while
|
||||
maintaining thread safety with proper locking mechanisms.
|
||||
|
||||
Example:
|
||||
cache = Cache()
|
||||
settings = Settings()
|
||||
service = MetadataService(cache, settings, max_workers=3)
|
||||
|
||||
# Extract metadata
|
||||
result = service.extract_metadata(Path("/media/movie.mkv"))
|
||||
if result:
|
||||
print(result['formatted_info'])
|
||||
|
||||
# Cleanup when done
|
||||
service.shutdown()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cache: Cache,
|
||||
settings: Settings,
|
||||
max_workers: int = 3
|
||||
):
|
||||
"""Initialize the metadata service.
|
||||
|
||||
Args:
|
||||
cache: Cache instance for storing extracted metadata
|
||||
settings: Settings instance for user preferences
|
||||
max_workers: Maximum number of concurrent extraction threads
|
||||
"""
|
||||
self.cache = cache
|
||||
self.settings = settings
|
||||
self.max_workers = max_workers
|
||||
|
||||
# Thread pool for concurrent extraction
|
||||
self.executor = ThreadPoolExecutor(
|
||||
max_workers=max_workers,
|
||||
thread_name_prefix="metadata_"
|
||||
)
|
||||
|
||||
# Lock for thread-safe operations
|
||||
self._lock = Lock()
|
||||
|
||||
# Track active futures for cancellation
|
||||
self._active_futures: dict[Path, Future] = {}
|
||||
|
||||
logger.info(f"MetadataService initialized with {max_workers} workers")
|
||||
|
||||
def extract_metadata(
|
||||
self,
|
||||
file_path: Path,
|
||||
callback: Optional[Callable] = None,
|
||||
error_callback: Optional[Callable] = None
|
||||
) -> Optional[dict]:
|
||||
"""Extract metadata from a media file.
|
||||
|
||||
This method can be called synchronously (returns result immediately) or
|
||||
asynchronously (uses callbacks when complete).
|
||||
|
||||
Args:
|
||||
file_path: Path to the media file
|
||||
callback: Optional callback(result_dict) called when extraction completes
|
||||
error_callback: Optional callback(error_message) called on error
|
||||
|
||||
Returns:
|
||||
Dictionary with 'formatted_info' and 'proposed_name' if synchronous,
|
||||
None if using callbacks (async mode)
|
||||
|
||||
Example:
|
||||
# Synchronous
|
||||
result = service.extract_metadata(path)
|
||||
print(result['formatted_info'])
|
||||
|
||||
# Asynchronous
|
||||
service.extract_metadata(
|
||||
path,
|
||||
callback=lambda r: print(r['formatted_info']),
|
||||
error_callback=lambda e: print(f"Error: {e}")
|
||||
)
|
||||
"""
|
||||
if callback or error_callback:
|
||||
# Asynchronous mode - submit to thread pool
|
||||
future = self.executor.submit(
|
||||
self._extract_metadata_internal,
|
||||
file_path
|
||||
)
|
||||
|
||||
# Track the future
|
||||
with self._lock:
|
||||
# Cancel any existing extraction for this file
|
||||
if file_path in self._active_futures:
|
||||
self._active_futures[file_path].cancel()
|
||||
self._active_futures[file_path] = future
|
||||
|
||||
# Add callback handlers
|
||||
def done_callback(f: Future):
|
||||
with self._lock:
|
||||
# Remove from active futures
|
||||
self._active_futures.pop(file_path, None)
|
||||
|
||||
try:
|
||||
result = f.result()
|
||||
if callback:
|
||||
callback(result)
|
||||
except Exception as e:
|
||||
logger.error(f"Error extracting metadata for {file_path}: {e}")
|
||||
if error_callback:
|
||||
error_callback(str(e))
|
||||
|
||||
future.add_done_callback(done_callback)
|
||||
return None
|
||||
else:
|
||||
# Synchronous mode - extract directly
|
||||
return self._extract_metadata_internal(file_path)
|
||||
|
||||
def _extract_metadata_internal(self, file_path: Path) -> dict:
|
||||
"""Internal method to extract and format metadata.
|
||||
|
||||
Args:
|
||||
file_path: Path to the media file
|
||||
|
||||
Returns:
|
||||
Dictionary with 'formatted_info' and 'proposed_name'
|
||||
|
||||
Raises:
|
||||
Exception: If extraction fails
|
||||
"""
|
||||
try:
|
||||
# Initialize extractor (uses cache internally via decorators)
|
||||
extractor = MediaExtractor(file_path)
|
||||
|
||||
# Get current mode from settings
|
||||
mode = self.settings.get("mode")
|
||||
|
||||
# Format based on mode
|
||||
if mode == "technical":
|
||||
formatter = MediaFormatter(extractor)
|
||||
formatted_info = formatter.file_info_panel()
|
||||
else: # catalog
|
||||
formatter = CatalogFormatter(extractor)
|
||||
formatted_info = formatter.format_catalog_info()
|
||||
|
||||
# Generate proposed name
|
||||
proposed_formatter = ProposedNameFormatter(extractor)
|
||||
proposed_name = proposed_formatter.rename_line_formatted(file_path)
|
||||
|
||||
return {
|
||||
'formatted_info': formatted_info,
|
||||
'proposed_name': proposed_name,
|
||||
'mode': mode,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to extract metadata for {file_path}: {e}")
|
||||
return {
|
||||
'formatted_info': TextFormatter.red(f"Error extracting details: {str(e)}"),
|
||||
'proposed_name': "",
|
||||
'mode': self.settings.get("mode"),
|
||||
}
|
||||
|
||||
def extract_for_display(
|
||||
self,
|
||||
file_path: Path,
|
||||
display_callback: Callable[[str, str], None],
|
||||
error_callback: Optional[Callable[[str], None]] = None
|
||||
):
|
||||
"""Extract metadata and update display via callback.
|
||||
|
||||
Convenience method that extracts metadata and calls the display callback
|
||||
with the formatted info and proposed name.
|
||||
|
||||
Args:
|
||||
file_path: Path to the media file
|
||||
display_callback: Callback(formatted_info, proposed_name) to update UI
|
||||
error_callback: Optional callback(error_message) for errors
|
||||
|
||||
Example:
|
||||
def update_ui(info, proposed):
|
||||
details_widget.update(info)
|
||||
proposed_widget.update(proposed)
|
||||
|
||||
service.extract_for_display(path, update_ui)
|
||||
"""
|
||||
def on_success(result: dict):
|
||||
display_callback(result['formatted_info'], result['proposed_name'])
|
||||
|
||||
def on_error(error_message: str):
|
||||
if error_callback:
|
||||
error_callback(error_message)
|
||||
else:
|
||||
display_callback(
|
||||
TextFormatter.red(f"Error: {error_message}"),
|
||||
""
|
||||
)
|
||||
|
||||
self.extract_metadata(file_path, callback=on_success, error_callback=on_error)
|
||||
|
||||
def cancel_extraction(self, file_path: Path) -> bool:
|
||||
"""Cancel an ongoing extraction for a file.
|
||||
|
||||
Args:
|
||||
file_path: Path to the file whose extraction should be canceled
|
||||
|
||||
Returns:
|
||||
True if an extraction was canceled, False if none was active
|
||||
|
||||
Example:
|
||||
# User selected a different file
|
||||
service.cancel_extraction(old_path)
|
||||
service.extract_metadata(new_path, callback=update_ui)
|
||||
"""
|
||||
with self._lock:
|
||||
future = self._active_futures.get(file_path)
|
||||
if future and not future.done():
|
||||
future.cancel()
|
||||
self._active_futures.pop(file_path, None)
|
||||
logger.debug(f"Canceled extraction for {file_path}")
|
||||
return True
|
||||
return False
|
||||
|
||||
def cancel_all_extractions(self):
|
||||
"""Cancel all ongoing extractions.
|
||||
|
||||
Useful when closing the application or switching directories.
|
||||
|
||||
Example:
|
||||
# User closing app
|
||||
service.cancel_all_extractions()
|
||||
service.shutdown()
|
||||
"""
|
||||
with self._lock:
|
||||
canceled_count = 0
|
||||
for file_path, future in list(self._active_futures.items()):
|
||||
if not future.done():
|
||||
future.cancel()
|
||||
canceled_count += 1
|
||||
self._active_futures.clear()
|
||||
|
||||
if canceled_count > 0:
|
||||
logger.info(f"Canceled {canceled_count} active extractions")
|
||||
|
||||
def get_active_extraction_count(self) -> int:
|
||||
"""Get the number of currently active extractions.
|
||||
|
||||
Returns:
|
||||
Number of extractions in progress
|
||||
|
||||
Example:
|
||||
>>> count = service.get_active_extraction_count()
|
||||
>>> print(f"{count} extractions in progress")
|
||||
"""
|
||||
with self._lock:
|
||||
return sum(1 for f in self._active_futures.values() if not f.done())
|
||||
|
||||
def shutdown(self, wait: bool = True):
|
||||
"""Shutdown the metadata service.
|
||||
|
||||
Cancels all pending extractions and shuts down the thread pool.
|
||||
Should be called when the application is closing.
|
||||
|
||||
Args:
|
||||
wait: If True, wait for all threads to complete. If False, cancel immediately.
|
||||
|
||||
Example:
|
||||
# Clean shutdown
|
||||
service.shutdown(wait=True)
|
||||
|
||||
# Force shutdown
|
||||
service.shutdown(wait=False)
|
||||
"""
|
||||
logger.info("Shutting down MetadataService")
|
||||
|
||||
# Cancel all active extractions
|
||||
self.cancel_all_extractions()
|
||||
|
||||
# Shutdown thread pool
|
||||
self.executor.shutdown(wait=wait)
|
||||
|
||||
logger.info("MetadataService shutdown complete")
|
||||
|
||||
def __enter__(self):
|
||||
"""Context manager support."""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager cleanup."""
|
||||
self.shutdown(wait=True)
|
||||
return False
|
||||
@@ -0,0 +1,346 @@
|
||||
"""Rename service for handling file rename operations.
|
||||
|
||||
This service manages the process of renaming files with:
|
||||
- Name validation and sanitization
|
||||
- Proposed name generation
|
||||
- Conflict detection
|
||||
- Atomic rename operations
|
||||
- Error handling and rollback
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable
|
||||
|
||||
from renamer.extractors.extractor import MediaExtractor
|
||||
from renamer.formatters.proposed_name_formatter import ProposedNameFormatter
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RenameService:
|
||||
"""Service for managing file rename operations.
|
||||
|
||||
This service handles:
|
||||
- Proposed name generation from metadata
|
||||
- Name validation and sanitization
|
||||
- File conflict detection
|
||||
- Atomic file rename operations
|
||||
- Rollback on errors
|
||||
|
||||
Example:
|
||||
service = RenameService()
|
||||
|
||||
# Propose a new name
|
||||
new_name = service.propose_name(Path("/media/movie.mkv"))
|
||||
print(f"Proposed: {new_name}")
|
||||
|
||||
# Rename file
|
||||
success, message = service.rename_file(
|
||||
Path("/media/movie.mkv"),
|
||||
new_name
|
||||
)
|
||||
if success:
|
||||
print(f"Renamed successfully")
|
||||
"""
|
||||
|
||||
# Invalid characters for filenames (Windows + Unix)
|
||||
INVALID_CHARS = r'[<>:"|?*\x00-\x1f]'
|
||||
|
||||
# Invalid characters for paths
|
||||
INVALID_PATH_CHARS = r'[<>"|?*\x00-\x1f]'
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the rename service."""
|
||||
logger.debug("RenameService initialized")
|
||||
|
||||
def propose_name(
|
||||
self,
|
||||
file_path: Path,
|
||||
extractor: Optional[MediaExtractor] = None
|
||||
) -> Optional[str]:
|
||||
"""Generate a proposed new filename based on metadata.
|
||||
|
||||
Args:
|
||||
file_path: Current file path
|
||||
extractor: Optional pre-initialized MediaExtractor. If None, creates new one.
|
||||
|
||||
Returns:
|
||||
Proposed filename (without path) or None if generation fails
|
||||
|
||||
Example:
|
||||
>>> service = RenameService()
|
||||
>>> new_name = service.propose_name(Path("/media/movie.2024.mkv"))
|
||||
>>> print(new_name)
|
||||
'Movie Title (2024) [1080p].mkv'
|
||||
"""
|
||||
try:
|
||||
if extractor is None:
|
||||
extractor = MediaExtractor(file_path)
|
||||
|
||||
formatter = ProposedNameFormatter(extractor)
|
||||
# Get the formatted rename line
|
||||
rename_line = formatter.rename_line_formatted(file_path)
|
||||
|
||||
# Extract just the filename from the rename line
|
||||
# Format is typically: "Rename to: [bold]filename[/bold]"
|
||||
if "→" in rename_line:
|
||||
# New format with arrow
|
||||
parts = rename_line.split("→")
|
||||
if len(parts) == 2:
|
||||
# Remove markup tags
|
||||
proposed = self._strip_markup(parts[1].strip())
|
||||
return proposed
|
||||
elif "Rename to:" in rename_line:
|
||||
# Old format
|
||||
parts = rename_line.split("Rename to:")
|
||||
if len(parts) == 2:
|
||||
proposed = self._strip_markup(parts[1].strip())
|
||||
return proposed
|
||||
|
||||
# Fallback: use the whole line after stripping markup
|
||||
return self._strip_markup(rename_line)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to propose name for {file_path}: {e}")
|
||||
return None
|
||||
|
||||
def sanitize_filename(self, filename: str) -> str:
|
||||
"""Sanitize a filename by removing invalid characters.
|
||||
|
||||
Args:
|
||||
filename: The filename to sanitize
|
||||
|
||||
Returns:
|
||||
Sanitized filename safe for all filesystems
|
||||
|
||||
Example:
|
||||
>>> service.sanitize_filename('Movie: Title?')
|
||||
'Movie Title'
|
||||
"""
|
||||
# Remove invalid characters
|
||||
sanitized = re.sub(self.INVALID_CHARS, '', filename)
|
||||
|
||||
# Replace multiple spaces with single space
|
||||
sanitized = re.sub(r'\s+', ' ', sanitized)
|
||||
|
||||
# Strip leading/trailing whitespace and dots
|
||||
sanitized = sanitized.strip('. ')
|
||||
|
||||
return sanitized
|
||||
|
||||
def validate_filename(self, filename: str) -> tuple[bool, Optional[str]]:
|
||||
"""Validate that a filename is safe and legal.
|
||||
|
||||
Args:
|
||||
filename: The filename to validate
|
||||
|
||||
Returns:
|
||||
Tuple of (is_valid, error_message). If valid, error_message is None.
|
||||
|
||||
Example:
|
||||
>>> is_valid, error = service.validate_filename("movie.mkv")
|
||||
>>> if not is_valid:
|
||||
... print(f"Invalid: {error}")
|
||||
"""
|
||||
if not filename:
|
||||
return False, "Filename cannot be empty"
|
||||
|
||||
if len(filename) > 255:
|
||||
return False, "Filename too long (max 255 characters)"
|
||||
|
||||
# Check for invalid characters
|
||||
if re.search(self.INVALID_CHARS, filename):
|
||||
return False, f"Filename contains invalid characters: {filename}"
|
||||
|
||||
# Check for reserved names (Windows)
|
||||
reserved_names = {
|
||||
'CON', 'PRN', 'AUX', 'NUL',
|
||||
'COM1', 'COM2', 'COM3', 'COM4', 'COM5', 'COM6', 'COM7', 'COM8', 'COM9',
|
||||
'LPT1', 'LPT2', 'LPT3', 'LPT4', 'LPT5', 'LPT6', 'LPT7', 'LPT8', 'LPT9',
|
||||
}
|
||||
name_without_ext = Path(filename).stem.upper()
|
||||
if name_without_ext in reserved_names:
|
||||
return False, f"Filename uses reserved name: {name_without_ext}"
|
||||
|
||||
# Check for names ending with dot or space (Windows)
|
||||
if filename.endswith('.') or filename.endswith(' '):
|
||||
return False, "Filename cannot end with dot or space"
|
||||
|
||||
return True, None
|
||||
|
||||
def check_name_conflict(
|
||||
self,
|
||||
source_path: Path,
|
||||
new_filename: str
|
||||
) -> tuple[bool, Optional[str]]:
|
||||
"""Check if a new filename would conflict with existing files.
|
||||
|
||||
Args:
|
||||
source_path: Current file path
|
||||
new_filename: Proposed new filename
|
||||
|
||||
Returns:
|
||||
Tuple of (has_conflict, conflict_message)
|
||||
|
||||
Example:
|
||||
>>> has_conflict, msg = service.check_name_conflict(
|
||||
... Path("/media/old.mkv"),
|
||||
... "new.mkv"
|
||||
... )
|
||||
>>> if has_conflict:
|
||||
... print(msg)
|
||||
"""
|
||||
# Build the new path
|
||||
new_path = source_path.parent / new_filename
|
||||
|
||||
# Check if it's the same file (case-insensitive on some systems)
|
||||
if source_path.resolve() == new_path.resolve():
|
||||
return False, None
|
||||
|
||||
# Check if target already exists
|
||||
if new_path.exists():
|
||||
return True, f"File already exists: {new_filename}"
|
||||
|
||||
return False, None
|
||||
|
||||
def rename_file(
|
||||
self,
|
||||
source_path: Path,
|
||||
new_filename: str,
|
||||
dry_run: bool = False
|
||||
) -> tuple[bool, str]:
|
||||
"""Rename a file to a new filename.
|
||||
|
||||
Args:
|
||||
source_path: Current file path
|
||||
new_filename: New filename (without path)
|
||||
dry_run: If True, validate but don't actually rename
|
||||
|
||||
Returns:
|
||||
Tuple of (success, message). Message contains error or success info.
|
||||
|
||||
Example:
|
||||
>>> success, msg = service.rename_file(
|
||||
... Path("/media/old.mkv"),
|
||||
... "new.mkv"
|
||||
... )
|
||||
>>> print(msg)
|
||||
"""
|
||||
# Validate source file exists
|
||||
if not source_path.exists():
|
||||
error_msg = f"Source file does not exist: {source_path}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
if not source_path.is_file():
|
||||
error_msg = f"Source is not a file: {source_path}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
# Sanitize the new filename
|
||||
sanitized_filename = self.sanitize_filename(new_filename)
|
||||
|
||||
# Validate the new filename
|
||||
is_valid, error = self.validate_filename(sanitized_filename)
|
||||
if not is_valid:
|
||||
logger.error(f"Invalid filename: {error}")
|
||||
return False, error
|
||||
|
||||
# Check for conflicts
|
||||
has_conflict, conflict_msg = self.check_name_conflict(source_path, sanitized_filename)
|
||||
if has_conflict:
|
||||
logger.warning(f"Name conflict: {conflict_msg}")
|
||||
return False, conflict_msg
|
||||
|
||||
# Build the new path
|
||||
new_path = source_path.parent / sanitized_filename
|
||||
|
||||
# Dry run mode - don't actually rename
|
||||
if dry_run:
|
||||
success_msg = f"Would rename: {source_path.name} → {sanitized_filename}"
|
||||
logger.info(success_msg)
|
||||
return True, success_msg
|
||||
|
||||
# Perform the rename
|
||||
try:
|
||||
source_path.rename(new_path)
|
||||
success_msg = f"Renamed: {source_path.name} → {sanitized_filename}"
|
||||
logger.info(success_msg)
|
||||
return True, success_msg
|
||||
|
||||
except PermissionError as e:
|
||||
error_msg = f"Permission denied: {e}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
except OSError as e:
|
||||
error_msg = f"OS error during rename: {e}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Unexpected error during rename: {e}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
def rename_with_callback(
|
||||
self,
|
||||
source_path: Path,
|
||||
new_filename: str,
|
||||
success_callback: Optional[Callable[[Path], None]] = None,
|
||||
error_callback: Optional[Callable[[str], None]] = None,
|
||||
dry_run: bool = False
|
||||
):
|
||||
"""Rename a file with callbacks for success/error.
|
||||
|
||||
Convenience method that performs the rename and calls appropriate callbacks.
|
||||
|
||||
Args:
|
||||
source_path: Current file path
|
||||
new_filename: New filename (without path)
|
||||
success_callback: Called with new_path on success
|
||||
error_callback: Called with error_message on failure
|
||||
dry_run: If True, validate but don't actually rename
|
||||
|
||||
Example:
|
||||
def on_success(new_path):
|
||||
print(f"File renamed to: {new_path}")
|
||||
update_tree_node(new_path)
|
||||
|
||||
def on_error(error):
|
||||
show_error_dialog(error)
|
||||
|
||||
service.rename_with_callback(
|
||||
path, new_name,
|
||||
success_callback=on_success,
|
||||
error_callback=on_error
|
||||
)
|
||||
"""
|
||||
success, message = self.rename_file(source_path, new_filename, dry_run)
|
||||
|
||||
if success:
|
||||
if success_callback:
|
||||
new_path = source_path.parent / self.sanitize_filename(new_filename)
|
||||
success_callback(new_path)
|
||||
else:
|
||||
if error_callback:
|
||||
error_callback(message)
|
||||
|
||||
def _strip_markup(self, text: str) -> str:
|
||||
"""Strip Textual markup tags from text.
|
||||
|
||||
Args:
|
||||
text: Text with markup tags
|
||||
|
||||
Returns:
|
||||
Plain text without markup
|
||||
|
||||
Example:
|
||||
>>> service._strip_markup('[bold]text[/bold]')
|
||||
'text'
|
||||
"""
|
||||
# Remove all markup tags like [bold], [/bold], [green], etc.
|
||||
return re.sub(r'\[/?[^\]]+\]', '', text)
|
||||
Reference in New Issue
Block a user