mirror of
https://github.com/shadoll/moma.git
synced 2026-08-28 11:33:25 +00:00
feat: Add OpenScreen for directory input and validation
feat: Introduce poster rendering views with multiple engines feat: Implement ASCII art poster renderer using PIL feat: Create base class for poster renderers feat: Add RichPixels renderer for high-quality terminal image display feat: Implement Viu terminal image viewer renderer feat: Add ProposedFilenameView for generating standardized filenames feat: Create RenameConfirmScreen for renaming files with confirmation feat: Implement SettingsScreen for configuring application settings feat: Add custom PosterWidget for rendering poster images
This commit is contained in:
@@ -0,0 +1,25 @@
|
||||
"""Extractors package - provides metadata extraction from media files.
|
||||
|
||||
This package contains various extractor classes that extract metadata from
|
||||
different sources (filename, MediaInfo, file system, TMDB API, etc.).
|
||||
|
||||
All extractors should implement the DataExtractor protocol defined in base.py.
|
||||
"""
|
||||
|
||||
from .base import DataExtractor
|
||||
from .default_extractor import DefaultExtractor
|
||||
from .filename_extractor import FilenameExtractor
|
||||
from .fileinfo_extractor import FileInfoExtractor
|
||||
from .mediainfo_extractor import MediaInfoExtractor
|
||||
from .metadata_extractor import MetadataExtractor
|
||||
from .tmdb_extractor import TMDBExtractor
|
||||
|
||||
__all__ = [
|
||||
'DataExtractor',
|
||||
'DefaultExtractor',
|
||||
'FilenameExtractor',
|
||||
'FileInfoExtractor',
|
||||
'MediaInfoExtractor',
|
||||
'MetadataExtractor',
|
||||
'TMDBExtractor',
|
||||
]
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Base classes and protocols for extractors.
|
||||
|
||||
This module defines the DataExtractor Protocol that all extractors should implement.
|
||||
The protocol ensures a consistent interface across all extractor types.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Protocol, Optional
|
||||
|
||||
|
||||
class DataExtractor(Protocol):
|
||||
"""Protocol defining the standard interface for all extractors.
|
||||
|
||||
All extractor classes should implement this protocol to ensure consistent
|
||||
behavior across the application. The protocol defines methods for extracting
|
||||
various metadata from media files.
|
||||
|
||||
Attributes:
|
||||
file_path: Path to the file being analyzed
|
||||
|
||||
Example:
|
||||
class MyExtractor:
|
||||
def __init__(self, file_path: Path):
|
||||
self.file_path = file_path
|
||||
|
||||
def extract_title(self) -> Optional[str]:
|
||||
# Implementation here
|
||||
return "Movie Title"
|
||||
"""
|
||||
|
||||
file_path: Path
|
||||
|
||||
def extract_title(self) -> Optional[str]:
|
||||
"""Extract the title of the media file.
|
||||
|
||||
Returns:
|
||||
The extracted title or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_year(self) -> Optional[str]:
|
||||
"""Extract the release year.
|
||||
|
||||
Returns:
|
||||
The year as a string (e.g., "2024") or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_source(self) -> Optional[str]:
|
||||
"""Extract the source/release type (e.g., BluRay, WEB-DL, HDTV).
|
||||
|
||||
Returns:
|
||||
The source type or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_order(self) -> Optional[str]:
|
||||
"""Extract ordering information (e.g., episode number, disc number).
|
||||
|
||||
Returns:
|
||||
The order information or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_resolution(self) -> Optional[str]:
|
||||
"""Extract the video resolution (e.g., 1080p, 2160p, 720p).
|
||||
|
||||
Returns:
|
||||
The resolution or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_hdr(self) -> Optional[str]:
|
||||
"""Extract HDR information (e.g., HDR10, Dolby Vision).
|
||||
|
||||
Returns:
|
||||
The HDR format or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_movie_db(self) -> Optional[str]:
|
||||
"""Extract movie database IDs (e.g., TMDB, IMDB).
|
||||
|
||||
Returns:
|
||||
Database identifiers or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_special_info(self) -> Optional[str]:
|
||||
"""Extract special information (e.g., REPACK, PROPER, Director's Cut).
|
||||
|
||||
Returns:
|
||||
Special release information or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_audio_langs(self) -> Optional[str]:
|
||||
"""Extract audio language codes.
|
||||
|
||||
Returns:
|
||||
Comma-separated language codes or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_meta_type(self) -> Optional[str]:
|
||||
"""Extract metadata type/format information.
|
||||
|
||||
Returns:
|
||||
The metadata type or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_size(self) -> Optional[int]:
|
||||
"""Extract the file size in bytes.
|
||||
|
||||
Returns:
|
||||
File size in bytes or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_modification_time(self) -> Optional[float]:
|
||||
"""Extract the file modification timestamp.
|
||||
|
||||
Returns:
|
||||
Unix timestamp of last modification or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_file_name(self) -> Optional[str]:
|
||||
"""Extract the file name without path.
|
||||
|
||||
Returns:
|
||||
The file name or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_file_path(self) -> Optional[str]:
|
||||
"""Extract the full file path as string.
|
||||
|
||||
Returns:
|
||||
The full file path or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_frame_class(self) -> Optional[str]:
|
||||
"""Extract the frame class/aspect ratio classification.
|
||||
|
||||
Returns:
|
||||
Frame class (e.g., "Widescreen", "Ultra-Widescreen") or None
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_video_tracks(self) -> list[dict]:
|
||||
"""Extract video track information.
|
||||
|
||||
Returns:
|
||||
List of dictionaries containing video track metadata.
|
||||
Returns empty list if no tracks available.
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_audio_tracks(self) -> list[dict]:
|
||||
"""Extract audio track information.
|
||||
|
||||
Returns:
|
||||
List of dictionaries containing audio track metadata.
|
||||
Returns empty list if no tracks available.
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_subtitle_tracks(self) -> list[dict]:
|
||||
"""Extract subtitle track information.
|
||||
|
||||
Returns:
|
||||
List of dictionaries containing subtitle track metadata.
|
||||
Returns empty list if no tracks available.
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_anamorphic(self) -> Optional[str]:
|
||||
"""Extract anamorphic encoding information.
|
||||
|
||||
Returns:
|
||||
Anamorphic status or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_extension(self) -> Optional[str]:
|
||||
"""Extract the file extension.
|
||||
|
||||
Returns:
|
||||
File extension (without dot) or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_tmdb_url(self) -> Optional[str]:
|
||||
"""Extract TMDB URL if available.
|
||||
|
||||
Returns:
|
||||
Full TMDB URL or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_tmdb_id(self) -> Optional[str]:
|
||||
"""Extract TMDB ID if available.
|
||||
|
||||
Returns:
|
||||
TMDB ID as string or None if not available
|
||||
"""
|
||||
...
|
||||
|
||||
def extract_original_title(self) -> Optional[str]:
|
||||
"""Extract the original title (non-localized).
|
||||
|
||||
Returns:
|
||||
The original title or None if not available
|
||||
"""
|
||||
...
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Default extractor providing fallback values.
|
||||
|
||||
This module provides a minimal implementation of the DataExtractor protocol
|
||||
that returns default/empty values for all extraction methods. Used as a
|
||||
fallback when no specific extractor is available.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class DefaultExtractor:
|
||||
"""Extractor that provides default fallback values for all extraction methods.
|
||||
|
||||
This class implements the DataExtractor protocol by returning sensible
|
||||
defaults (None, empty strings, empty lists) for all extraction operations.
|
||||
It's used as a final fallback in the extractor chain when no other
|
||||
extractor can provide data.
|
||||
|
||||
All methods return None or empty values, making it safe to use when
|
||||
no actual data extraction is possible.
|
||||
"""
|
||||
|
||||
def extract_title(self) -> Optional[str]:
|
||||
"""Return default title.
|
||||
|
||||
Returns:
|
||||
Default title string "Unknown Title"
|
||||
"""
|
||||
return "Unknown Title"
|
||||
|
||||
def extract_year(self) -> Optional[str]:
|
||||
"""Return year. Returns None as no year information is available."""
|
||||
return None
|
||||
|
||||
def extract_source(self) -> Optional[str]:
|
||||
"""Return video source. Returns None as no source information is available."""
|
||||
return None
|
||||
|
||||
def extract_order(self) -> Optional[str]:
|
||||
"""Return sequence order. Returns None as no order information is available."""
|
||||
return None
|
||||
|
||||
def extract_resolution(self) -> Optional[str]:
|
||||
"""Return resolution. Returns None as no resolution information is available."""
|
||||
return None
|
||||
|
||||
def extract_hdr(self) -> Optional[str]:
|
||||
"""Return HDR information. Returns None as no HDR information is available."""
|
||||
return None
|
||||
|
||||
def extract_movie_db(self) -> list[str] | None:
|
||||
"""Return movie database ID. Returns None as no database information is available."""
|
||||
return None
|
||||
|
||||
def extract_special_info(self) -> Optional[str]:
|
||||
"""Return special edition info. Returns None as no special info is available."""
|
||||
return None
|
||||
|
||||
def extract_audio_langs(self) -> Optional[str]:
|
||||
"""Return audio languages. Returns None as no language information is available."""
|
||||
return None
|
||||
|
||||
def extract_meta_type(self) -> Optional[str]:
|
||||
"""Return metadata type. Returns None as no type information is available."""
|
||||
return None
|
||||
|
||||
def extract_size(self) -> Optional[int]:
|
||||
"""Return file size. Returns None as no size information is available."""
|
||||
return None
|
||||
|
||||
def extract_modification_time(self) -> Optional[float]:
|
||||
"""Return modification time. Returns None as no timestamp is available."""
|
||||
return None
|
||||
|
||||
def extract_file_name(self) -> Optional[str]:
|
||||
"""Return file name. Returns None as no filename is available."""
|
||||
return None
|
||||
|
||||
def extract_file_path(self) -> Optional[str]:
|
||||
"""Return file path. Returns None as no file path is available."""
|
||||
return None
|
||||
|
||||
def extract_frame_class(self) -> Optional[str]:
|
||||
"""Return frame class. Returns None as no frame class information is available."""
|
||||
return None
|
||||
|
||||
def extract_video_tracks(self) -> list[dict]:
|
||||
"""Return video tracks. Returns empty list as no video tracks are available."""
|
||||
return []
|
||||
|
||||
def extract_audio_tracks(self) -> list[dict]:
|
||||
"""Return audio tracks. Returns empty list as no audio tracks are available."""
|
||||
return []
|
||||
|
||||
def extract_subtitle_tracks(self) -> list[dict]:
|
||||
"""Return subtitle tracks. Returns empty list as no subtitle tracks are available."""
|
||||
return []
|
||||
|
||||
def extract_anamorphic(self) -> Optional[str]:
|
||||
"""Return anamorphic info. Returns None as no anamorphic information is available."""
|
||||
return None
|
||||
|
||||
def extract_extension(self) -> Optional[str]:
|
||||
"""Return file extension. Returns 'ext' as default placeholder."""
|
||||
return "ext"
|
||||
|
||||
def extract_tmdb_url(self) -> Optional[str]:
|
||||
"""Return TMDB URL. Returns None as no TMDB URL is available."""
|
||||
return None
|
||||
|
||||
def extract_tmdb_id(self) -> Optional[str]:
|
||||
"""Return TMDB ID. Returns None as no TMDB ID is available."""
|
||||
return None
|
||||
|
||||
def extract_original_title(self) -> Optional[str]:
|
||||
"""Return original title. Returns None as no original title is available."""
|
||||
return None
|
||||
@@ -0,0 +1,259 @@
|
||||
"""Media metadata extraction coordinator.
|
||||
|
||||
This module provides the MediaExtractor class which coordinates multiple
|
||||
specialized extractors to gather comprehensive metadata about media files.
|
||||
It implements a priority-based extraction system where data is retrieved
|
||||
from the most appropriate source.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from .filename_extractor import FilenameExtractor
|
||||
from .metadata_extractor import MetadataExtractor
|
||||
from .mediainfo_extractor import MediaInfoExtractor
|
||||
from .fileinfo_extractor import FileInfoExtractor
|
||||
from .tmdb_extractor import TMDBExtractor
|
||||
from .default_extractor import DefaultExtractor
|
||||
|
||||
|
||||
class MediaExtractor:
|
||||
"""Coordinator for extracting metadata from media files using multiple specialized extractors.
|
||||
|
||||
This class manages a collection of specialized extractors and provides a unified
|
||||
interface for retrieving metadata. It implements a priority-based system where
|
||||
each type of data is retrieved from the most appropriate source.
|
||||
|
||||
The extraction priority order varies by data type:
|
||||
- Title: TMDB → Metadata → Filename → Default
|
||||
- Year: Filename → Default
|
||||
- Technical info: MediaInfo → Default
|
||||
- File info: FileInfo → Default
|
||||
|
||||
Attributes:
|
||||
file_path: Path to the media file
|
||||
filename_extractor: Extracts metadata from filename patterns
|
||||
metadata_extractor: Extracts embedded metadata tags
|
||||
mediainfo_extractor: Extracts technical media information
|
||||
fileinfo_extractor: Extracts basic file system information
|
||||
tmdb_extractor: Fetches metadata from The Movie Database API
|
||||
default_extractor: Provides fallback default values
|
||||
|
||||
Example:
|
||||
>>> from pathlib import Path
|
||||
>>> extractor = MediaExtractor(Path("Movie (2020) [1080p].mkv"))
|
||||
>>> title = extractor.get("title")
|
||||
>>> year = extractor.get("year")
|
||||
>>> tracks = extractor.get("video_tracks")
|
||||
"""
|
||||
|
||||
def __init__(self, file_path: Path, use_cache: bool = True):
|
||||
self.file_path = file_path
|
||||
|
||||
# Initialize all extractors - they use singleton Cache internally
|
||||
self.filename_extractor = FilenameExtractor(file_path, use_cache)
|
||||
self.metadata_extractor = MetadataExtractor(file_path, use_cache)
|
||||
self.mediainfo_extractor = MediaInfoExtractor(file_path, use_cache)
|
||||
self.fileinfo_extractor = FileInfoExtractor(file_path, use_cache)
|
||||
self.tmdb_extractor = TMDBExtractor(file_path, use_cache)
|
||||
self.default_extractor = DefaultExtractor()
|
||||
|
||||
# Extractor mapping
|
||||
self._extractors = {
|
||||
"Metadata": self.metadata_extractor,
|
||||
"Filename": self.filename_extractor,
|
||||
"MediaInfo": self.mediainfo_extractor,
|
||||
"FileInfo": self.fileinfo_extractor,
|
||||
"TMDB": self.tmdb_extractor,
|
||||
"Default": self.default_extractor,
|
||||
}
|
||||
|
||||
# Define sources and conditions for each data type
|
||||
self._data = {
|
||||
"title": {
|
||||
"sources": [
|
||||
("TMDB", "extract_title"),
|
||||
("Metadata", "extract_title"),
|
||||
("Filename", "extract_title"),
|
||||
("Default", "extract_title"),
|
||||
],
|
||||
},
|
||||
"year": {
|
||||
"sources": [
|
||||
("Filename", "extract_year"),
|
||||
("Default", "extract_year"),
|
||||
],
|
||||
},
|
||||
"source": {
|
||||
"sources": [
|
||||
("Filename", "extract_source"),
|
||||
("Default", "extract_source"),
|
||||
],
|
||||
},
|
||||
"order": {
|
||||
"sources": [
|
||||
("Filename", "extract_order"),
|
||||
("Default", "extract_order"),
|
||||
],
|
||||
},
|
||||
"frame_class": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_frame_class"),
|
||||
("Filename", "extract_frame_class"),
|
||||
("Default", "extract_frame_class"),
|
||||
],
|
||||
},
|
||||
"resolution": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_resolution"),
|
||||
("Default", "extract_resolution"),
|
||||
],
|
||||
},
|
||||
"hdr": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_hdr"),
|
||||
("Filename", "extract_hdr"),
|
||||
("Default", "extract_hdr"),
|
||||
],
|
||||
},
|
||||
"anamorphic": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_anamorphic"),
|
||||
("Default", "extract_anamorphic"),
|
||||
],
|
||||
},
|
||||
"3d_layout": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_3d_layout"),
|
||||
("Default", "extract_3d_layout"),
|
||||
],
|
||||
},
|
||||
"movie_db": {
|
||||
"sources": [
|
||||
("TMDB", "extract_movie_db"),
|
||||
("Filename", "extract_movie_db"),
|
||||
("Default", "extract_movie_db"),
|
||||
],
|
||||
},
|
||||
"special_info": {
|
||||
"sources": [
|
||||
("Filename", "extract_special_info"),
|
||||
("Default", "extract_special_info"),
|
||||
],
|
||||
},
|
||||
"audio_langs": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_audio_langs"),
|
||||
("Filename", "extract_audio_langs"),
|
||||
("Default", "extract_audio_langs"),
|
||||
],
|
||||
},
|
||||
"meta_type": {
|
||||
"sources": [
|
||||
("Metadata", "extract_meta_type"),
|
||||
("Default", "extract_meta_type"),
|
||||
],
|
||||
},
|
||||
"file_size": {
|
||||
"sources": [
|
||||
("FileInfo", "extract_size"),
|
||||
("Default", "extract_size"),
|
||||
],
|
||||
},
|
||||
"modification_time": {
|
||||
"sources": [
|
||||
("FileInfo", "extract_modification_time"),
|
||||
("Default", "extract_modification_time"),
|
||||
],
|
||||
},
|
||||
"file_name": {
|
||||
"sources": [
|
||||
("FileInfo", "extract_file_name"),
|
||||
("Default", "extract_file_name"),
|
||||
],
|
||||
},
|
||||
"file_path": {
|
||||
"sources": [
|
||||
("FileInfo", "extract_file_path"),
|
||||
("Default", "extract_file_path"),
|
||||
],
|
||||
},
|
||||
"extension": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_extension"),
|
||||
("FileInfo", "extract_extension"),
|
||||
("Default", "extract_extension"),
|
||||
],
|
||||
},
|
||||
"video_tracks": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_video_tracks"),
|
||||
("Default", "extract_video_tracks"),
|
||||
],
|
||||
},
|
||||
"audio_tracks": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_audio_tracks"),
|
||||
("Default", "extract_audio_tracks"),
|
||||
],
|
||||
},
|
||||
"subtitle_tracks": {
|
||||
"sources": [
|
||||
("MediaInfo", "extract_subtitle_tracks"),
|
||||
("Default", "extract_subtitle_tracks"),
|
||||
],
|
||||
},
|
||||
"genres": {
|
||||
"sources": [
|
||||
("TMDB", "extract_genres"),
|
||||
("Default", "extract_genres"),
|
||||
],
|
||||
},
|
||||
"production_countries": {
|
||||
"sources": [
|
||||
("TMDB", "extract_production_countries"),
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
def get(self, key: str, source: str | None = None):
|
||||
"""Get metadata value by key, optionally from a specific source.
|
||||
|
||||
Retrieves metadata using a priority-based system. If a source is specified,
|
||||
only that extractor is used. Otherwise, extractors are tried in priority
|
||||
order until a non-None value is found.
|
||||
|
||||
Args:
|
||||
key: The metadata key to retrieve (e.g., "title", "year", "resolution")
|
||||
source: Optional specific extractor to use ("TMDB", "MediaInfo", "Filename", etc.)
|
||||
|
||||
Returns:
|
||||
The extracted metadata value, or None if not found
|
||||
|
||||
Example:
|
||||
>>> extractor = MediaExtractor(Path("movie.mkv"))
|
||||
>>> title = extractor.get("title") # Try all sources in priority order
|
||||
>>> year = extractor.get("year", source="Filename") # Use only filename
|
||||
"""
|
||||
if source:
|
||||
# Specific source requested - find the extractor and call the method directly
|
||||
for extractor_name, extractor in self._extractors.items():
|
||||
if extractor_name.lower() == source.lower():
|
||||
method = f"extract_{key}"
|
||||
if hasattr(extractor, method):
|
||||
val = getattr(extractor, method)()
|
||||
return val if val is not None else None
|
||||
return None
|
||||
|
||||
# Fallback mode - try sources in order
|
||||
if key in self._data:
|
||||
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"]]
|
||||
|
||||
# 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 val is not None:
|
||||
return val
|
||||
return None
|
||||
@@ -0,0 +1,94 @@
|
||||
"""File system information extractor.
|
||||
|
||||
This module provides the FileInfoExtractor class for extracting basic
|
||||
file system metadata such as size, timestamps, paths, and extensions.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
import logging
|
||||
from ..cache import cached_method, Cache
|
||||
from ..logging_config import LoggerConfig # Initialize logging singleton
|
||||
|
||||
|
||||
class FileInfoExtractor:
|
||||
"""Extractor for basic file system information.
|
||||
|
||||
This class extracts file system metadata including size, modification time,
|
||||
file name, path, and extension. All extraction methods are cached for
|
||||
performance.
|
||||
|
||||
Attributes:
|
||||
file_path: Path object pointing to the file
|
||||
_size: Cached file size in bytes
|
||||
_modification_time: Cached modification timestamp
|
||||
_file_name: Cached file name
|
||||
_file_path: Cached full file path as string
|
||||
_cache: Internal cache for method results
|
||||
|
||||
Example:
|
||||
>>> from pathlib import Path
|
||||
>>> extractor = FileInfoExtractor(Path("movie.mkv"))
|
||||
>>> size = extractor.extract_size() # Returns size in bytes
|
||||
>>> name = extractor.extract_file_name() # Returns "movie.mkv"
|
||||
"""
|
||||
|
||||
def __init__(self, file_path: Path, use_cache: bool = True):
|
||||
"""Initialize the FileInfoExtractor.
|
||||
|
||||
Args:
|
||||
file_path: Path object pointing to the file to extract info from
|
||||
use_cache: Whether to use caching (default: True)
|
||||
"""
|
||||
self._file_path = file_path
|
||||
self.file_path = file_path # Expose for cache key generation
|
||||
self.cache = Cache() if use_cache else None # Singleton cache for @cached_method decorator
|
||||
self.settings = None # Will be set by Settings singleton if needed
|
||||
self._stat = file_path.stat()
|
||||
self._cache: dict = {} # Internal cache for method results
|
||||
|
||||
@cached_method()
|
||||
def extract_size(self) -> int:
|
||||
"""Extract file size in bytes.
|
||||
|
||||
Returns:
|
||||
File size in bytes as an integer
|
||||
"""
|
||||
return self._stat.st_size
|
||||
|
||||
@cached_method()
|
||||
def extract_modification_time(self) -> float:
|
||||
"""Extract file modification time.
|
||||
|
||||
Returns:
|
||||
Unix timestamp (seconds since epoch) as a float
|
||||
"""
|
||||
return self._stat.st_mtime
|
||||
|
||||
@cached_method()
|
||||
def extract_file_name(self) -> str:
|
||||
"""Extract file name (basename).
|
||||
|
||||
Returns:
|
||||
File name including extension (e.g., "movie.mkv")
|
||||
"""
|
||||
return self._file_path.name
|
||||
|
||||
@cached_method()
|
||||
def extract_file_path(self) -> str:
|
||||
"""Extract full file path as string.
|
||||
|
||||
Returns:
|
||||
Absolute file path as a string
|
||||
"""
|
||||
return str(self._file_path)
|
||||
|
||||
@cached_method()
|
||||
def extract_extension(self) -> str | None:
|
||||
"""Extract file extension without the dot.
|
||||
|
||||
Returns:
|
||||
File extension in lowercase without leading dot (e.g., "mkv", "mp4"),
|
||||
or None if no extension exists
|
||||
"""
|
||||
ext = self._file_path.suffix.lower().lstrip('.')
|
||||
return ext if ext else None
|
||||
@@ -0,0 +1,488 @@
|
||||
import re
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from collections import Counter
|
||||
from ..constants import (
|
||||
SOURCE_DICT, FRAME_CLASSES, MOVIE_DB_DICT, SPECIAL_EDITIONS, SKIP_WORDS,
|
||||
NON_STANDARD_QUALITY_INDICATORS,
|
||||
is_valid_year,
|
||||
CYRILLIC_TO_ENGLISH
|
||||
)
|
||||
from ..cache import cached_method, Cache
|
||||
from ..utils.pattern_utils import PatternExtractor
|
||||
import langcodes
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FilenameExtractor:
|
||||
"""Class to extract information from filename"""
|
||||
|
||||
def __init__(self, file_path: Path | str, use_cache: bool = True):
|
||||
if isinstance(file_path, str):
|
||||
self.file_path = Path(file_path)
|
||||
self.file_name = file_path
|
||||
else:
|
||||
self.file_path = file_path
|
||||
self.file_name = file_path.name
|
||||
|
||||
self.cache = Cache() if use_cache else None # Singleton cache for @cached_method decorator
|
||||
self.settings = None # Will be set by Settings singleton if needed
|
||||
|
||||
# Initialize utility helper
|
||||
self._pattern_extractor = PatternExtractor()
|
||||
|
||||
def _normalize_cyrillic(self, text: str) -> str:
|
||||
"""Normalize Cyrillic characters to English equivalents for parsing"""
|
||||
for cyr, eng in CYRILLIC_TO_ENGLISH.items():
|
||||
text = text.replace(cyr, eng)
|
||||
return text
|
||||
|
||||
def _get_frame_class_from_height(self, height: int) -> str | None:
|
||||
"""Get frame class from video height using FRAME_CLASSES constant"""
|
||||
for frame_class, info in FRAME_CLASSES.items():
|
||||
if height == info['nominal_height']:
|
||||
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
|
||||
year_pos = -1
|
||||
source_pos = -1
|
||||
quality_pos = -1
|
||||
paren_match = None
|
||||
dot_match = None
|
||||
|
||||
# Find year position (either (YYYY) or .YYYY.)
|
||||
paren_match = re.search(r'\((\d{4})\)', self.file_name)
|
||||
if paren_match:
|
||||
year_pos = paren_match.start()
|
||||
else:
|
||||
dot_match = re.search(r'\.(\d{4})\.', self.file_name)
|
||||
if dot_match:
|
||||
year_pos = dot_match.start()
|
||||
else:
|
||||
# Last resort: any 4-digit number
|
||||
any_match = re.search(r'\b(\d{4})\b', self.file_name)
|
||||
if any_match:
|
||||
year = int(any_match.group(1))
|
||||
# Basic sanity check using constants
|
||||
if is_valid_year(year):
|
||||
year_pos = any_match.start() # Cut before the year for plain years
|
||||
|
||||
# Find source position
|
||||
source = self.extract_source()
|
||||
if source:
|
||||
for alias in SOURCE_DICT[source]:
|
||||
match = re.search(r'\b' + re.escape(alias) + r'\b', self.file_name, re.IGNORECASE)
|
||||
if match:
|
||||
source_pos = match.start()
|
||||
break
|
||||
|
||||
# Find quality bracket position (like [720p,ukr,eng])
|
||||
quality_match = re.search(r'\[[^\]]*(?:720p|1080p|2160p|480p|SD|HD|HDR)[^\]]*\]', self.file_name)
|
||||
if quality_match:
|
||||
quality_pos = quality_match.start()
|
||||
|
||||
# Find the earliest position that's not at the beginning
|
||||
positions = [pos for pos in [year_pos, source_pos, quality_pos] if pos > 0]
|
||||
cut_pos = min(positions) if positions else -1
|
||||
|
||||
# Extract title (everything before the cut position)
|
||||
if cut_pos > 0:
|
||||
title = self.file_name[:cut_pos].strip()
|
||||
else:
|
||||
# No delimiters found after position 0, take everything before the last dot
|
||||
title = self.file_name.rsplit('.', 1)[0].strip()
|
||||
|
||||
# If year is at the beginning, remove it
|
||||
if year_pos == 0:
|
||||
if paren_match and paren_match.start() == 0:
|
||||
title = re.sub(r'^\(\d{4}\)\s*', '', title)
|
||||
elif dot_match and dot_match.start() == 0:
|
||||
title = re.sub(r'^\.\d{4}\.\s*', '', title)
|
||||
|
||||
# Remove common prefixes that are not part of the title
|
||||
# Remove bracketed prefixes like [01.1], [1], etc.
|
||||
title = re.sub(r'^\s*\[[^\]]+\]\s*', '', title)
|
||||
|
||||
# Remove order number prefixes like 01., 1., 1.1 followed by space/underscore
|
||||
# Only remove if the number is multi-digit or has decimal (to avoid removing single digit titles)
|
||||
match = re.match(r'^\s*(\d+(?:\.\d+)?)\.(?=\s|_)', title)
|
||||
if match:
|
||||
order = match.group(1)
|
||||
if len(order) > 1 or '.' in order:
|
||||
title = re.sub(r'^\s*(\d+(?:\.\d+)?)\.(?=\s|_)', '', title)
|
||||
|
||||
# Remove order like 1.9 where 1 is order, 9 is title
|
||||
order = self.extract_order()
|
||||
if order:
|
||||
match = re.match(r'^' + re.escape(order) + r'\.(.+)', title)
|
||||
if match:
|
||||
title = match.group(1)
|
||||
|
||||
# Clean up any remaining leading separators
|
||||
title = title.lstrip('_ \t')
|
||||
|
||||
# Clean up title: remove leading/trailing brackets and dots
|
||||
title = title.strip('[](). ')
|
||||
|
||||
# Replace dots with spaces if they appear to be word separators
|
||||
# Only replace dots that are surrounded by letters/digits (not at edges)
|
||||
title = re.sub(r'(?<=[a-zA-Z0-9À-ÿ])\.(?=[a-zA-Z0-9À-ÿ])', ' ', title)
|
||||
|
||||
# Clean up multiple spaces
|
||||
title = re.sub(r'\s+', ' ', title).strip()
|
||||
|
||||
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)
|
||||
paren_match = re.search(r'\((\d{4})\)', self.file_name)
|
||||
if paren_match:
|
||||
return paren_match.group(1)
|
||||
|
||||
# Fallback: look for year in dots (like .1971.)
|
||||
dot_match = re.search(r'\.(\d{4})\.', self.file_name)
|
||||
if dot_match:
|
||||
return dot_match.group(1)
|
||||
|
||||
# Last resort: any 4-digit number (but this is less reliable)
|
||||
any_match = re.search(r'\b(\d{4})\b', self.file_name)
|
||||
if any_match:
|
||||
year = int(any_match.group(1))
|
||||
# Basic sanity check using constants
|
||||
if is_valid_year(year):
|
||||
year_pos = any_match.start()
|
||||
return str(year)
|
||||
|
||||
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)
|
||||
|
||||
for src, aliases in SOURCE_DICT.items():
|
||||
for alias in aliases:
|
||||
if alias.upper() in temp_name.upper():
|
||||
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
|
||||
# Patterns: [01], [01.1], 01., 1., 1.1 followed by space or underscore
|
||||
|
||||
# Check for bracketed patterns: [01], [01.1], etc.
|
||||
bracket_match = re.match(r'^\[(\d+(?:\.\d+)?)\]', self.file_name)
|
||||
if bracket_match:
|
||||
return bracket_match.group(1)
|
||||
|
||||
# Check for dot patterns: 01., 1., 1.1 followed by title before (
|
||||
dot_match = re.match(r'^(\d+(?:\.\d)*)\.?\s*', self.file_name)
|
||||
if dot_match and '.' in dot_match.group(0):
|
||||
order = dot_match.group(1)
|
||||
if '.' in order:
|
||||
parts = order.split('.')
|
||||
if len(parts) > 1 and parts[-1] != '1':
|
||||
order = parts[0]
|
||||
return order
|
||||
|
||||
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
|
||||
normalized_name = self._normalize_cyrillic(self.file_name)
|
||||
|
||||
# First check for specific numeric resolutions with p/i
|
||||
match = re.search(r'(\d{3,4})([pi])', normalized_name, re.IGNORECASE)
|
||||
if match:
|
||||
height = int(match.group(1))
|
||||
scan_type = match.group(2).lower()
|
||||
frame_class = f"{height}{scan_type}"
|
||||
if frame_class in FRAME_CLASSES:
|
||||
return frame_class
|
||||
# Fallback to height-based if not in constants
|
||||
return self._get_frame_class_from_height(height)
|
||||
|
||||
# If no specific resolution found, check for non-standard quality indicators
|
||||
for indicator in NON_STANDARD_QUALITY_INDICATORS:
|
||||
if re.search(r'\b' + re.escape(indicator) + r'\b', self.file_name, re.IGNORECASE):
|
||||
return None
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_hdr(self) -> str | None:
|
||||
"""Extract HDR information from filename"""
|
||||
# Check for SDR first - indicates no HDR
|
||||
if re.search(r'\bSDR\b', self.file_name, re.IGNORECASE):
|
||||
return None
|
||||
|
||||
# Check for HDR, but not NoHDR
|
||||
if re.search(r'\bHDR\b', self.file_name, re.IGNORECASE) and not re.search(r'\bNoHDR\b', self.file_name, re.IGNORECASE):
|
||||
return 'HDR'
|
||||
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_movie_db(self) -> list[str] | None:
|
||||
"""Extract movie database identifier from filename"""
|
||||
# Use PatternExtractor utility to avoid code duplication
|
||||
db_info = self._pattern_extractor.extract_movie_db_ids(self.file_name)
|
||||
if db_info:
|
||||
return [db_info['type'], db_info['id']]
|
||||
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
|
||||
special_info = []
|
||||
|
||||
for canonical_edition, variants in SPECIAL_EDITIONS.items():
|
||||
for edition in variants:
|
||||
# Check in brackets: [Theatrical Cut], [Director's Cut], etc.
|
||||
bracket_pattern = r'\[([^\]]+)\]'
|
||||
brackets = re.findall(bracket_pattern, self.file_name)
|
||||
for bracket in brackets:
|
||||
# Check if bracket contains comma-separated items
|
||||
items = [item.strip() for item in bracket.split(',')]
|
||||
for item in items:
|
||||
if edition.lower() == item.lower().strip():
|
||||
if canonical_edition not in special_info:
|
||||
special_info.append(canonical_edition)
|
||||
|
||||
# Check as standalone text (case-insensitive)
|
||||
if re.search(r'\b' + re.escape(edition) + r'\b', self.file_name, re.IGNORECASE):
|
||||
if canonical_edition not in special_info:
|
||||
special_info.append(canonical_edition)
|
||||
|
||||
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
|
||||
# Skip subtitle indicators and focus on audio languages
|
||||
|
||||
langs = []
|
||||
|
||||
# First, look for languages inside brackets
|
||||
bracket_pattern = r'\[([^\]]+)\]'
|
||||
brackets = re.findall(bracket_pattern, self.file_name)
|
||||
|
||||
for bracket in brackets:
|
||||
bracket_lower = bracket.lower()
|
||||
|
||||
# Skip brackets that contain movie database patterns
|
||||
if any(db in bracket_lower for db in ['imdb', 'tmdb', 'tvdb']):
|
||||
continue
|
||||
|
||||
# Parse items separated by commas or underscores
|
||||
items = re.split(r'[,_]', bracket)
|
||||
items = [item.strip() for item in items]
|
||||
|
||||
for item in items:
|
||||
# Skip empty items or items that are clearly not languages
|
||||
if not item or len(item) < 2:
|
||||
continue
|
||||
|
||||
item_lower = item.lower()
|
||||
|
||||
# Skip subtitle indicators
|
||||
if item_lower in ['sub', 'subs', 'subtitle']:
|
||||
continue
|
||||
|
||||
# Check if item contains language codes (2-3 letter codes)
|
||||
# Pattern: optional number + optional 'x' + language code
|
||||
# Allow the language code to be at the end of the item
|
||||
lang_match = re.search(r'(?:(\d+)x?)?([a-z]{2,3})$', item_lower)
|
||||
if lang_match:
|
||||
count = int(lang_match.group(1)) if lang_match.group(1) else 1
|
||||
lang_code = lang_match.group(2)
|
||||
|
||||
# Skip if it's a quality/resolution indicator or other skip word
|
||||
if lang_code in SKIP_WORDS:
|
||||
continue
|
||||
|
||||
# Skip if the language code is not at the end or if there are extra letters after
|
||||
# But allow prefixes like numbers and 'x'
|
||||
prefix = item_lower[:-len(lang_code)]
|
||||
if not re.match(r'^(?:\d+x?)?$', prefix):
|
||||
continue
|
||||
|
||||
# Convert to 3-letter ISO code
|
||||
try:
|
||||
lang_obj = langcodes.Language.get(lang_code)
|
||||
iso3_code = lang_obj.to_alpha3()
|
||||
langs.extend([iso3_code] * count)
|
||||
except (LookupError, ValueError, AttributeError) as e:
|
||||
# Skip invalid language codes
|
||||
logger.debug(f"Invalid language code '{lang_code}': {e}")
|
||||
pass
|
||||
|
||||
# Second, look for standalone language codes outside brackets
|
||||
# Remove bracketed content first
|
||||
text_without_brackets = re.sub(r'\[([^\]]+)\]', '', self.file_name)
|
||||
|
||||
# Split on dots, spaces, and underscores
|
||||
parts = re.split(r'[.\s_]+', text_without_brackets)
|
||||
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if not part or len(part) < 2:
|
||||
continue
|
||||
|
||||
part_lower = part.lower()
|
||||
|
||||
# Check if this part is a 2-3 letter code
|
||||
if not re.match(r'^[a-zA-Z]{2,3}$', part):
|
||||
continue
|
||||
|
||||
# Skip title case 2-letter words to avoid false positives like "In" -> "ind"
|
||||
if part.istitle() and len(part) == 2:
|
||||
continue
|
||||
|
||||
# Skip known non-language words
|
||||
if part_lower in SKIP_WORDS:
|
||||
continue
|
||||
|
||||
# Try to validate with langcodes library
|
||||
try:
|
||||
lang_obj = langcodes.Language.get(part_lower)
|
||||
iso3_code = lang_obj.to_alpha3()
|
||||
langs.append(iso3_code)
|
||||
except (LookupError, ValueError, AttributeError) as e:
|
||||
# Not a valid language code, skip
|
||||
logger.debug(f"Invalid language code '{part_lower}': {e}")
|
||||
pass
|
||||
|
||||
if not langs:
|
||||
return ''
|
||||
|
||||
# Count occurrences while preserving order of first appearance
|
||||
lang_counts = {}
|
||||
for lang in langs:
|
||||
if lang not in lang_counts:
|
||||
lang_counts[lang] = 0
|
||||
lang_counts[lang] += 1
|
||||
|
||||
# Format like mediainfo: "2ukr,eng" preserving order
|
||||
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_extension(self) -> str | None:
|
||||
"""Extract file extension from filename"""
|
||||
# Use pathlib to extract extension properly
|
||||
ext = self.file_path.suffix
|
||||
# Remove leading dot and return
|
||||
return ext[1:] if ext else None
|
||||
|
||||
@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
|
||||
|
||||
tracks = []
|
||||
|
||||
# First, look for languages inside brackets
|
||||
bracket_pattern = r'\[([^\]]+)\]'
|
||||
brackets = re.findall(bracket_pattern, self.file_name)
|
||||
|
||||
for bracket in brackets:
|
||||
bracket_lower = bracket.lower()
|
||||
|
||||
# Skip brackets that contain movie database patterns
|
||||
if any(db in bracket_lower for db in ['imdb', 'tmdb', 'tvdb']):
|
||||
continue
|
||||
|
||||
# Parse items separated by commas or underscores
|
||||
items = re.split(r'[,_]', bracket)
|
||||
items = [item.strip() for item in items]
|
||||
|
||||
for item in items:
|
||||
# Skip empty items or items that are clearly not languages
|
||||
if not item or len(item) < 2:
|
||||
continue
|
||||
|
||||
item_lower = item.lower()
|
||||
|
||||
# Skip subtitle indicators
|
||||
if item_lower in ['sub', 'subs', 'subtitle']:
|
||||
continue
|
||||
|
||||
# Check if item contains language codes (2-3 letter codes)
|
||||
# Pattern: optional number + optional 'x' + language code
|
||||
# Allow the language code to be at the end of the item
|
||||
lang_match = re.search(r'(?:(\d+)x?)?([a-z]{2,3})$', item_lower)
|
||||
if lang_match:
|
||||
count = int(lang_match.group(1)) if lang_match.group(1) else 1
|
||||
lang_code = lang_match.group(2)
|
||||
|
||||
# Skip if it's a quality/resolution indicator or other skip word
|
||||
if lang_code in SKIP_WORDS:
|
||||
continue
|
||||
|
||||
# Skip if the language code is not at the end or if there are extra letters after
|
||||
# But allow prefixes like numbers and 'x'
|
||||
prefix = item_lower[:-len(lang_code)]
|
||||
if not re.match(r'^(?:\d+x?)?$', prefix):
|
||||
continue
|
||||
|
||||
# Convert to 3-letter ISO code
|
||||
try:
|
||||
lang_obj = langcodes.Language.get(lang_code)
|
||||
iso3_code = lang_obj.to_alpha3()
|
||||
tracks.append({'language': iso3_code})
|
||||
except (LookupError, ValueError, AttributeError) as e:
|
||||
# Skip invalid language codes
|
||||
logger.debug(f"Invalid language code '{lang_code}': {e}")
|
||||
pass
|
||||
|
||||
# Second, look for standalone language codes outside brackets
|
||||
# Remove bracketed content first
|
||||
text_without_brackets = re.sub(r'\[([^\]]+)\]', '', self.file_name)
|
||||
|
||||
# Split on dots, spaces, and underscores
|
||||
parts = re.split(r'[.\s_]+', text_without_brackets)
|
||||
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if not part or len(part) < 2:
|
||||
continue
|
||||
|
||||
part_lower = part.lower()
|
||||
|
||||
# Check if this part is a 2-3 letter code
|
||||
if not re.match(r'^[a-zA-Z]{2,3}$', part):
|
||||
continue
|
||||
|
||||
# Skip title case 2-letter words to avoid false positives like "In" -> "ind"
|
||||
if part.istitle() and len(part) == 2:
|
||||
continue
|
||||
|
||||
# Skip known non-language words
|
||||
if part_lower in SKIP_WORDS:
|
||||
continue
|
||||
|
||||
# Try to validate with langcodes library
|
||||
try:
|
||||
lang_obj = langcodes.Language.get(part_lower)
|
||||
iso3_code = lang_obj.to_alpha3()
|
||||
tracks.append({'language': iso3_code})
|
||||
except (LookupError, ValueError, AttributeError) as e:
|
||||
# Not a valid language code, skip
|
||||
logger.debug(f"Invalid language code '{part_lower}': {e}")
|
||||
pass
|
||||
|
||||
return tracks
|
||||
@@ -0,0 +1,481 @@
|
||||
from pathlib import Path
|
||||
from pymediainfo import MediaInfo
|
||||
from collections import Counter
|
||||
from ..constants import FRAME_CLASSES, get_extension_from_format
|
||||
from ..cache import cached_method, Cache
|
||||
import langcodes
|
||||
import logging
|
||||
import functools
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def requires_tracks(func):
|
||||
"""Decorator that returns None if media_info has no tracks."""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
media_info = self._get_media_info()
|
||||
if not media_info or not media_info.tracks:
|
||||
return None
|
||||
return func(self, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def requires_tracks_type(track_type: str):
|
||||
"""Decorator that returns None if no tracks of the specified type are available."""
|
||||
|
||||
def decorator(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
tracks = self._get_tracks(track_type=track_type)
|
||||
if tracks is None:
|
||||
return None
|
||||
if type(tracks) is not list:
|
||||
return None
|
||||
if len(tracks) == 0:
|
||||
return None
|
||||
return func(self, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
class MediaInfoExtractor:
|
||||
"""Class to extract information from MediaInfo"""
|
||||
|
||||
def __init__(self, file_path: Path, use_cache: bool = True):
|
||||
self.file_path = file_path
|
||||
self.cache = (
|
||||
Cache() if use_cache else None
|
||||
) # Singleton cache for @cached_method decorator
|
||||
self.settings = None # Will be set by Settings singleton if needed
|
||||
self._cache = {} # Internal cache for method results
|
||||
|
||||
@cached_method()
|
||||
def _get_media_info(self) -> MediaInfo | None:
|
||||
"""Get parsed MediaInfo object, cached. Returns None if no media."""
|
||||
if not self.file_path.exists():
|
||||
return None
|
||||
parsed = MediaInfo.parse(self.file_path)
|
||||
return parsed if parsed else None
|
||||
|
||||
@requires_tracks
|
||||
def _get_tracks(self, track_type="General") -> list | None:
|
||||
"""Return tracks of given type or specific track by ID."""
|
||||
media_info = self._get_media_info()
|
||||
tracks = [t for t in media_info.tracks if t.track_type == track_type]
|
||||
return tracks
|
||||
|
||||
def _get_track(self, track_type="General", track_id: int = 0) -> object | None:
|
||||
"""Return tracks of given type or specific track by ID."""
|
||||
tracks = self._get_tracks(track_type=track_type)
|
||||
if tracks is None:
|
||||
return None
|
||||
return tracks[track_id] if track_id < len(tracks) else None
|
||||
|
||||
@requires_tracks_type("General")
|
||||
def extract_general_track(self) -> dict | None:
|
||||
"""Extract general track data"""
|
||||
general = self._get_track(track_type="General", track_id=0)
|
||||
result = {
|
||||
"format": getattr(general, "format", None) or "unknown",
|
||||
"file_size": getattr(general, "file_size", None),
|
||||
"duration": getattr(general, "duration", 0) / 1000
|
||||
if getattr(general, "duration", None)
|
||||
else None,
|
||||
"overall_bit_rate": getattr(general, "overall_bit_rate", None),
|
||||
"movie_name": getattr(general, "movie_name", None),
|
||||
"encoded_date": getattr(general, "encoded_date", None),
|
||||
}
|
||||
return result
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_video_tracks(self) -> list[dict] | None:
|
||||
"""Extract video track data"""
|
||||
tracks = self._get_tracks(track_type="Video")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
result = []
|
||||
for v in tracks[:2]: # Up to 2 videos
|
||||
track_data = {
|
||||
"codec": getattr(v, "format", None)
|
||||
or getattr(v, "codec", None)
|
||||
or "unknown",
|
||||
"width": getattr(v, "width", None),
|
||||
"height": getattr(v, "height", None),
|
||||
"bitrate": getattr(v, "bit_rate", None),
|
||||
"fps": getattr(v, "frame_rate", None),
|
||||
"profile": getattr(v, "format_profile", None) or "",
|
||||
"interlaced": getattr(v, "interlaced", None) == "Yes",
|
||||
"anamorphic": getattr(v, "anamorphic", None) == "Yes",
|
||||
}
|
||||
result.append(track_data)
|
||||
return result if result else None
|
||||
|
||||
@requires_tracks_type("Audio")
|
||||
def extract_audio_tracks(self) -> list[dict] | None:
|
||||
"""Extract audio track data"""
|
||||
tracks = self._get_tracks(track_type="Audio")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
result = []
|
||||
for a in tracks[:10]: # Up to 10 audios
|
||||
track_data = {
|
||||
"codec": getattr(a, "format", None)
|
||||
or getattr(a, "codec", None)
|
||||
or "unknown",
|
||||
"channels": getattr(a, "channel_s", None),
|
||||
"language": getattr(a, "language", "und"),
|
||||
"bitrate": getattr(a, "bit_rate", None),
|
||||
}
|
||||
result.append(track_data)
|
||||
return result if result else None
|
||||
|
||||
@requires_tracks_type("Text")
|
||||
def extract_subtitle_tracks(self) -> list[dict] | None:
|
||||
"""Extract subtitle track data"""
|
||||
tracks = self._get_tracks(track_type="Text")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
result = []
|
||||
for s in tracks[:10]: # Up to 10 subs
|
||||
track_data = {
|
||||
"language": getattr(s, "language", "und"),
|
||||
"format": getattr(s, "format", None)
|
||||
or getattr(s, "codec", None)
|
||||
or "unknown",
|
||||
"forced": getattr(s, "forced", None) == "Yes",
|
||||
"default": getattr(s, "default", None) == "Yes",
|
||||
}
|
||||
result.append(track_data)
|
||||
return result if result else None
|
||||
|
||||
@requires_tracks
|
||||
@requires_tracks_type("General")
|
||||
def extract_duration(self) -> float | None:
|
||||
"""Extract duration from media info in seconds"""
|
||||
tracks = self._get_tracks(track_type="General")
|
||||
# Type assertion: decorators guarantee tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
for track in tracks:
|
||||
return (
|
||||
getattr(track, "duration", 0) / 1000
|
||||
if getattr(track, "duration", None)
|
||||
else None
|
||||
)
|
||||
return None
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_resolution(self) -> tuple[int, int] | None:
|
||||
"""Extract actual video resolution as (width, height) tuple from media info"""
|
||||
track = self._get_track(track_type="Video", track_id=0)
|
||||
width = getattr(track, "width", None)
|
||||
height = getattr(track, "height", None)
|
||||
if width is not None and height is not None:
|
||||
try:
|
||||
return int(width), int(height)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
return None
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_frame_class(self) -> str | None:
|
||||
"""Extract frame class from media info (480p, 720p, 1080p, etc.)"""
|
||||
track = self._get_track(track_type="Video", track_id=0)
|
||||
|
||||
scan_type_attr = getattr(track, "scan_type", None)
|
||||
|
||||
interlaced = self.extract_interlaced()
|
||||
scan_order = getattr(track, "scan_order", None)
|
||||
|
||||
resolution = self.extract_resolution()
|
||||
if not resolution:
|
||||
return None
|
||||
height, width = resolution
|
||||
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Frame class detection - Resolution: {width}x{height}"
|
||||
)
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_type attribute: {scan_type_attr!r} (type: {type(scan_type_attr).__name__})"
|
||||
)
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] interlaced attribute: {interlaced!r} (type: {type(interlaced).__name__})"
|
||||
)
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_order attribute: {scan_order!r} (type: {type(scan_order).__name__})"
|
||||
)
|
||||
|
||||
# Determine scan type from available attributes
|
||||
# Check scan_type first (e.g., "Interlaced", "Progressive", "MBAFF")
|
||||
if scan_type_attr and isinstance(scan_type_attr, str):
|
||||
scan_type = "i" if "interlaced" in scan_type_attr.lower() else "p"
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Using scan_type: {scan_type_attr!r} -> scan_type={scan_type!r}"
|
||||
)
|
||||
# Check scan_order (e.g., "TFF", "BFF" for interlaced, "Progressive" for progressive)
|
||||
elif scan_order and isinstance(scan_order, str):
|
||||
scan_type = "i" if scan_order.upper() in ["TFF", "BFF"] else "p"
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Using scan_order: {scan_order!r} -> scan_type={scan_type!r}"
|
||||
)
|
||||
# Then check interlaced flag from extract_interlaced() method
|
||||
elif interlaced is True:
|
||||
scan_type = "i"
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Using interlaced: True -> scan_type=i"
|
||||
)
|
||||
elif interlaced is False:
|
||||
scan_type = "p"
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Using interlaced: False -> scan_type=p"
|
||||
)
|
||||
else:
|
||||
# Default to progressive if no information available
|
||||
scan_type = "p"
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] No scan type info, defaulting to progressive"
|
||||
)
|
||||
|
||||
# Calculate effective height for frame class determination
|
||||
aspect_ratio = 16 / 9
|
||||
if height > width:
|
||||
effective_height = height / aspect_ratio
|
||||
else:
|
||||
effective_height = height
|
||||
|
||||
# First, try to match width to typical widths
|
||||
# Use a larger tolerance (10 pixels) to handle cinema/ultrawide aspect ratios
|
||||
width_matches = []
|
||||
for frame_class, info in FRAME_CLASSES.items():
|
||||
for tw in info["typical_widths"]:
|
||||
if abs(width - tw) <= 10 and frame_class.endswith(scan_type):
|
||||
diff = abs(height - info["nominal_height"])
|
||||
width_matches.append((frame_class, diff))
|
||||
|
||||
if width_matches:
|
||||
# Choose the frame class with the smallest height difference
|
||||
width_matches.sort(key=lambda x: x[1])
|
||||
result = width_matches[0][0]
|
||||
logger.debug(f"[{self.file_path.name}] Result (width match): {result!r}")
|
||||
return result
|
||||
|
||||
# If no width match, fall back to height-based matching
|
||||
# First try exact match with standard frame classes
|
||||
frame_class = f"{int(round(effective_height))}{scan_type}"
|
||||
if frame_class in FRAME_CLASSES:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result (exact height match): {frame_class!r}"
|
||||
)
|
||||
return frame_class
|
||||
|
||||
# Find closest standard height match
|
||||
closest_class = None
|
||||
min_diff = float("inf")
|
||||
for fc, info in FRAME_CLASSES.items():
|
||||
if fc.endswith(scan_type):
|
||||
diff = abs(effective_height - info["nominal_height"])
|
||||
if diff < min_diff:
|
||||
min_diff = diff
|
||||
closest_class = fc
|
||||
|
||||
# Return closest standard match if within reasonable distance (20 pixels)
|
||||
if closest_class and min_diff <= 20:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result (closest match, diff={min_diff}): {closest_class!r}"
|
||||
)
|
||||
return closest_class
|
||||
|
||||
# For non-standard resolutions, create a custom frame class
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result (custom/non-standard): {frame_class!r}"
|
||||
)
|
||||
return frame_class
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_aspect_ratio(self) -> str | None:
|
||||
"""Extract video aspect ratio from media info"""
|
||||
tracks = self._get_tracks(track_type="Video")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
track = tracks[0]
|
||||
aspect_ratio = getattr(track, "display_aspect_ratio", None)
|
||||
if aspect_ratio:
|
||||
return str(aspect_ratio)
|
||||
return None
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_hdr(self) -> str | None:
|
||||
"""Extract HDR info from media info"""
|
||||
tracks = self._get_tracks(track_type="Video")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
track = tracks[0]
|
||||
profile = getattr(track, "format_profile", "") or ""
|
||||
if "HDR" in profile.upper():
|
||||
return "HDR"
|
||||
return None
|
||||
|
||||
@requires_tracks
|
||||
@requires_tracks_type("Audio")
|
||||
def extract_audio_langs(self) -> str | None:
|
||||
"""Extract audio languages from media info"""
|
||||
tracks = self._get_tracks(track_type="Audio")
|
||||
if not isinstance(tracks, list):
|
||||
return None
|
||||
langs = []
|
||||
for a in tracks:
|
||||
lang_code = getattr(a, "language", "und") or "und"
|
||||
try:
|
||||
# Try to get the 3-letter code
|
||||
lang_obj = langcodes.Language.get(lang_code.lower())
|
||||
alpha3 = lang_obj.to_alpha3()
|
||||
langs.append(alpha3)
|
||||
except (LookupError, ValueError, AttributeError) as e:
|
||||
# If conversion fails, use the original code
|
||||
logger.debug(f"Invalid language code '{lang_code}': {e}")
|
||||
langs.append(lang_code.lower()[:3])
|
||||
|
||||
lang_counts = Counter(langs)
|
||||
audio_langs = [
|
||||
f"{count}{lang}" if count > 1 else lang
|
||||
for lang, count in lang_counts.items()
|
||||
]
|
||||
return ",".join(audio_langs)
|
||||
|
||||
def is_3d(self) -> bool:
|
||||
"""Check if the video is 3D"""
|
||||
track = self._get_track("Video", 0)
|
||||
if not track:
|
||||
return False
|
||||
multi_view = getattr(track, "multi_view_count", None)
|
||||
if multi_view and int(multi_view) > 1:
|
||||
return True
|
||||
stereoscopic = getattr(track, "stereoscopic", None)
|
||||
if stereoscopic == "Yes":
|
||||
return True
|
||||
return False
|
||||
|
||||
@requires_tracks_type("General")
|
||||
def extract_extension(self) -> str | None:
|
||||
"""Extract file extension based on container format.
|
||||
|
||||
Uses MediaInfo's format field to determine the appropriate file extension.
|
||||
Handles special cases like Matroska 3D (mk3d vs mkv).
|
||||
|
||||
Returns:
|
||||
File extension (e.g., "mp4", "mkv") or None if format is unknown
|
||||
"""
|
||||
|
||||
general_track = self._get_track(track_type="General", track_id=0)
|
||||
format_ = getattr(general_track, "format", None)
|
||||
if not format_:
|
||||
return None
|
||||
|
||||
# Use the constants function to get extension from format
|
||||
ext = get_extension_from_format(format_)
|
||||
|
||||
# Special case: Matroska 3D uses mk3d extension
|
||||
if ext == "mkv" and self.is_3d():
|
||||
return "mk3d"
|
||||
|
||||
return ext
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_3d_layout(self) -> str | None:
|
||||
"""Extract 3D stereoscopic layout from MediaInfo"""
|
||||
if not self.is_3d():
|
||||
return None
|
||||
tracks = self._get_tracks(track_type="Video")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
track = tracks[0]
|
||||
stereoscopic = getattr(track, "stereoscopic", None)
|
||||
return stereoscopic if stereoscopic else None
|
||||
|
||||
@requires_tracks_type("Video")
|
||||
def extract_interlaced(self) -> bool | None:
|
||||
"""Determine if the video is interlaced.
|
||||
|
||||
Returns:
|
||||
True: Video is interlaced
|
||||
False: Video is progressive (explicitly set)
|
||||
None: Information not available in MediaInfo
|
||||
"""
|
||||
tracks = self._get_tracks(track_type="Video")
|
||||
# Type assertion: decorator guarantees tracks is a list
|
||||
assert isinstance(tracks, list)
|
||||
track = tracks[0]
|
||||
scan_type_attr = getattr(track, "scan_type", None)
|
||||
interlaced = getattr(track, "interlaced", None)
|
||||
scan_order = getattr(track, "scan_order", None)
|
||||
|
||||
logger.debug(f"[{self.file_path.name}] Interlaced detection:")
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_type: {scan_type_attr!r} (type: {type(scan_type_attr).__name__})"
|
||||
)
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] interlaced: {interlaced!r} (type: {type(interlaced).__name__})"
|
||||
)
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_order: {scan_order!r} (type: {type(scan_order).__name__})"
|
||||
)
|
||||
|
||||
# Check scan_type attribute first (e.g., "Interlaced", "Progressive", "MBAFF")
|
||||
if scan_type_attr and isinstance(scan_type_attr, str):
|
||||
scan_lower = scan_type_attr.lower()
|
||||
if "interlaced" in scan_lower or "mbaff" in scan_lower:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: True (from scan_type={scan_type_attr!r})"
|
||||
)
|
||||
return True
|
||||
elif "progressive" in scan_lower:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: False (from scan_type={scan_type_attr!r})"
|
||||
)
|
||||
return False
|
||||
# If scan_type has some other value, fall through to check other attributes
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_type unrecognized, checking other attributes"
|
||||
)
|
||||
|
||||
# Check scan_order attribute (e.g., "TFF", "BFF" for interlaced, "Progressive" for progressive)
|
||||
if scan_order and isinstance(scan_order, str):
|
||||
scan_order_upper = scan_order.upper()
|
||||
if scan_order_upper in ["TFF", "BFF"]:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: True (from scan_order={scan_order!r})"
|
||||
)
|
||||
return True
|
||||
elif scan_order_upper == "PROGRESSIVE":
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: False (from scan_order={scan_order!r})"
|
||||
)
|
||||
return False
|
||||
# If scan_order has some other value, fall through
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] scan_order unrecognized, checking interlaced attribute"
|
||||
)
|
||||
|
||||
# Check interlaced attribute (e.g., "Yes", "No")
|
||||
if interlaced and isinstance(interlaced, str):
|
||||
interlaced_lower = interlaced.lower()
|
||||
if interlaced_lower in ["yes", "true", "1"]:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: True (from interlaced={interlaced!r})"
|
||||
)
|
||||
return True
|
||||
elif interlaced_lower in ["no", "false", "0"]:
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: False (from interlaced={interlaced!r})"
|
||||
)
|
||||
return False
|
||||
|
||||
# No information available
|
||||
logger.debug(
|
||||
f"[{self.file_path.name}] Result: None (no information available)"
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Embedded metadata extractor using Mutagen.
|
||||
|
||||
This module provides the MetadataExtractor class for reading embedded
|
||||
metadata tags from media files using the Mutagen library.
|
||||
"""
|
||||
|
||||
import mutagen
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from ..constants import MEDIA_TYPES
|
||||
from ..cache import cached_method, Cache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MetadataExtractor:
|
||||
"""Extractor for embedded metadata tags from media files.
|
||||
|
||||
This class uses the Mutagen library to read embedded metadata tags
|
||||
such as title, artist, and duration. Falls back to MIME type detection
|
||||
when Mutagen cannot read the file.
|
||||
|
||||
Attributes:
|
||||
file_path: Path object pointing to the file
|
||||
info: Mutagen file info object, or None if file cannot be read
|
||||
_cache: Internal cache for method results
|
||||
|
||||
Example:
|
||||
>>> from pathlib import Path
|
||||
>>> extractor = MetadataExtractor(Path("movie.mkv"))
|
||||
>>> title = extractor.extract_title()
|
||||
>>> duration = extractor.extract_duration()
|
||||
"""
|
||||
|
||||
def __init__(self, file_path: Path, use_cache: bool = True):
|
||||
"""Initialize the MetadataExtractor.
|
||||
|
||||
Args:
|
||||
file_path: Path object pointing to the media file
|
||||
use_cache: Whether to use caching (default: True)
|
||||
"""
|
||||
self.file_path = file_path
|
||||
self.cache = Cache() if use_cache else None # Singleton cache for @cached_method decorator
|
||||
self.settings = None # Will be set by Settings singleton if needed
|
||||
self._cache: dict = {} # Internal cache for method results
|
||||
try:
|
||||
self.info = mutagen.File(file_path) # type: ignore
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to read metadata from {file_path}: {e}")
|
||||
self.info = None
|
||||
|
||||
@cached_method()
|
||||
def extract_title(self) -> str | None:
|
||||
"""Extract title from embedded metadata tags.
|
||||
|
||||
Returns:
|
||||
Title string if found in metadata, None otherwise
|
||||
"""
|
||||
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.
|
||||
|
||||
Returns:
|
||||
Duration in seconds as a float, or None if not available
|
||||
"""
|
||||
if self.info:
|
||||
return getattr(self.info, 'length', None)
|
||||
return None
|
||||
|
||||
@cached_method()
|
||||
def extract_artist(self) -> str | None:
|
||||
"""Extract artist from embedded metadata tags.
|
||||
|
||||
Returns:
|
||||
Artist string if found in metadata, None otherwise
|
||||
"""
|
||||
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 metadata container type.
|
||||
|
||||
Returns the Mutagen class name (e.g., "FLAC", "MP4") if available,
|
||||
otherwise falls back to MIME type detection.
|
||||
|
||||
Returns:
|
||||
Container type name, or "Unknown" if cannot be determined
|
||||
"""
|
||||
if self.info:
|
||||
return type(self.info).__name__
|
||||
return self._detect_by_mime()
|
||||
|
||||
def _detect_by_mime(self) -> str:
|
||||
"""Detect metadata type by MIME type.
|
||||
|
||||
Uses python-magic library to detect file MIME type and maps it
|
||||
to a metadata container type.
|
||||
|
||||
Returns:
|
||||
Container type name based on MIME type, or "Unknown" if detection fails
|
||||
"""
|
||||
try:
|
||||
import magic
|
||||
mime = magic.from_file(str(self.file_path), mime=True)
|
||||
for ext, info in MEDIA_TYPES.items():
|
||||
if info['mime'] == mime:
|
||||
return info['meta_type']
|
||||
return 'Unknown'
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to detect MIME type for {self.file_path}: {e}")
|
||||
return 'Unknown'
|
||||
@@ -0,0 +1,297 @@
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import hashlib
|
||||
import requests
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Tuple, Any
|
||||
from ..secrets import TMDB_API_KEY, TMDB_ACCESS_TOKEN
|
||||
from ..cache import Cache
|
||||
from ..settings import Settings
|
||||
|
||||
class TMDBExtractor:
|
||||
"""Class to extract TMDB movie information"""
|
||||
|
||||
def __init__(self, file_path: Path, use_cache: bool = True):
|
||||
self.file_path = file_path
|
||||
self.cache = Cache() if use_cache else None # Singleton cache
|
||||
self.settings = Settings() # Singleton settings
|
||||
self.ttl_seconds = self.settings.get("cache_ttl_extractors", 21600)
|
||||
self._movie_db_info = None
|
||||
|
||||
def _get_cached_data(self, cache_key: str) -> Optional[Dict[str, Any]]:
|
||||
"""Get data from cache if valid"""
|
||||
if self.cache:
|
||||
return self.cache.get_object(f"tmdb_{cache_key}")
|
||||
return None
|
||||
|
||||
def _set_cached_data(self, cache_key: str, data: Dict[str, Any]):
|
||||
"""Store data in cache"""
|
||||
if self.cache:
|
||||
self.cache.set_object(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"""
|
||||
base_url = "https://api.themoviedb.org/3"
|
||||
url = f"{base_url}{endpoint}"
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {TMDB_ACCESS_TOKEN}",
|
||||
"accept": "application/json"
|
||||
}
|
||||
|
||||
if params is None:
|
||||
params = {}
|
||||
params['api_key'] = TMDB_API_KEY
|
||||
|
||||
try:
|
||||
response = requests.get(url, headers=headers, params=params, timeout=10)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except (requests.RequestException, ValueError) as e:
|
||||
logging.warning(f"TMDB API request failed for {url}: {e}")
|
||||
return None
|
||||
|
||||
def _search_movie_by_title_year(self, title: str, year: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||||
"""Search for movie by title and optionally year"""
|
||||
cache_key = f"search_{title}_{year or 'no_year'}"
|
||||
|
||||
# Check cache first
|
||||
cached = self._get_cached_data(cache_key)
|
||||
if cached is not None:
|
||||
logging.info(f"TMDB cache hit for search: {title} ({year})")
|
||||
return cached
|
||||
|
||||
logging.info(f"TMDB cache miss for search: {title} ({year}), making request")
|
||||
params = {'query': title}
|
||||
if year:
|
||||
params['year'] = year
|
||||
|
||||
result = self._make_tmdb_request('/search/movie', params)
|
||||
if result and result.get('results'):
|
||||
movies = result['results']
|
||||
|
||||
# If year provided, try exact match first
|
||||
if year:
|
||||
exact_matches = [m for m in movies if str(m.get('release_date', ''))[:4] == year]
|
||||
if exact_matches:
|
||||
movie = exact_matches[0]
|
||||
else:
|
||||
# Try ±1 year
|
||||
year_int = int(year)
|
||||
close_matches = [m for m in movies if abs(int(str(m.get('release_date', ''))[:4]) - year_int) <= 1]
|
||||
if close_matches:
|
||||
movie = close_matches[0]
|
||||
else:
|
||||
movie = movies[0] # Fallback to first result
|
||||
else:
|
||||
movie = movies[0] # No year filter, take first result
|
||||
|
||||
# Cache the result
|
||||
self._set_cached_data(cache_key, movie)
|
||||
return movie
|
||||
|
||||
return None
|
||||
|
||||
def _get_movie_details(self, movie_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""Get detailed movie information by ID"""
|
||||
cache_key = f"movie_{movie_id}"
|
||||
|
||||
# Check cache first
|
||||
cached = self._get_cached_data(cache_key)
|
||||
if cached is not None:
|
||||
logging.info(f"TMDB cache hit for movie details: {movie_id}")
|
||||
return cached
|
||||
|
||||
logging.info(f"TMDB cache miss for movie details: {movie_id}, making request")
|
||||
result = self._make_tmdb_request(f'/movie/{movie_id}')
|
||||
if result:
|
||||
# Cache the result
|
||||
self._set_cached_data(cache_key, result)
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
def _extract_movie_db_from_filename(self) -> Optional[Tuple[str, str]]:
|
||||
"""Extract movie database ID from filename (similar to FilenameExtractor.extract_movie_db)"""
|
||||
import re
|
||||
from ..constants import MOVIE_DB_DICT
|
||||
|
||||
file_name = self.file_path.name
|
||||
|
||||
# Look for patterns at the end of filename in brackets or braces
|
||||
# Patterns: [tmdbid-123] {imdb-tt123} [imdbid-tt123] etc.
|
||||
|
||||
# Match patterns like [tmdbid-123456] or {imdb-tt1234567}
|
||||
pattern = r'[\[\{]([a-zA-Z]+(?:id)?)[-\s]*([a-zA-Z0-9]+)[\]\}]'
|
||||
matches = re.findall(pattern, file_name)
|
||||
|
||||
if matches:
|
||||
# Take the last match (closest to end of filename)
|
||||
db_type, db_id = matches[-1]
|
||||
|
||||
# Normalize database type
|
||||
db_type_lower = db_type.lower()
|
||||
for db_key, db_info in MOVIE_DB_DICT.items():
|
||||
if any(db_type_lower.startswith(pattern.rstrip('-')) for pattern in db_info['patterns']):
|
||||
return (db_key, db_id)
|
||||
|
||||
return None
|
||||
|
||||
def _get_movie_info(self) -> Optional[Dict[str, Any]]:
|
||||
"""Get movie information from TMDB"""
|
||||
if self._movie_db_info is not None:
|
||||
return self._movie_db_info
|
||||
|
||||
# First, check if we have a TMDB ID in the filename
|
||||
movie_db = self._extract_movie_db_from_filename()
|
||||
if movie_db and movie_db[0] == 'tmdb':
|
||||
try:
|
||||
movie_id = int(movie_db[1])
|
||||
movie_data = self._get_movie_details(movie_id)
|
||||
if movie_data:
|
||||
self._movie_db_info = movie_data
|
||||
return movie_data
|
||||
except ValueError:
|
||||
pass # Invalid ID format
|
||||
|
||||
# If no TMDB ID or failed to get details, try searching by title/year
|
||||
# We need title and year from filename extraction
|
||||
from .filename_extractor import FilenameExtractor
|
||||
filename_extractor = FilenameExtractor(self.file_path)
|
||||
title = filename_extractor.extract_title()
|
||||
year = filename_extractor.extract_year()
|
||||
|
||||
if title:
|
||||
search_result = self._search_movie_by_title_year(title, year)
|
||||
if search_result and search_result.get('id'):
|
||||
# Fetch full movie details using the ID from search results
|
||||
movie_id = search_result['id']
|
||||
movie_data = self._get_movie_details(movie_id)
|
||||
if movie_data:
|
||||
self._movie_db_info = movie_data
|
||||
return movie_data
|
||||
|
||||
self._movie_db_info = None
|
||||
return None
|
||||
|
||||
def extract_tmdb_id(self) -> Optional[str]:
|
||||
"""Extract TMDB ID"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return str(movie_info.get('id'))
|
||||
return None
|
||||
|
||||
def extract_title(self) -> Optional[str]:
|
||||
"""Extract TMDB title"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return movie_info.get('title')
|
||||
return None
|
||||
|
||||
def extract_original_title(self) -> Optional[str]:
|
||||
"""Extract TMDB original title"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info:
|
||||
return f"({movie_info.get('original_language')}) {movie_info.get('original_title')}"
|
||||
return None
|
||||
|
||||
def extract_year(self) -> Optional[str]:
|
||||
"""Extract TMDB release year"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info and movie_info.get('release_date'):
|
||||
return movie_info['release_date'][:4]
|
||||
return None
|
||||
|
||||
def extract_tmdb_url(self) -> Optional[str]:
|
||||
"""Extract TMDB movie URL"""
|
||||
movie_id = self.extract_tmdb_id()
|
||||
if movie_id:
|
||||
return f"https://www.themoviedb.org/movie/{movie_id}"
|
||||
return None
|
||||
|
||||
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_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)
|
||||
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_production_countries(self) -> Optional[str]:
|
||||
"""Extract TMDB production countries"""
|
||||
movie_info = self._get_movie_info()
|
||||
if movie_info and movie_info.get('production_countries'):
|
||||
return ', '.join(country['name'] for country in movie_info['production_countries'])
|
||||
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 as e:
|
||||
logging.warning(f"Failed to download poster from {poster_url}: {e}")
|
||||
return None
|
||||
Reference in New Issue
Block a user