nfoguard/utils/file_utils.py

"""
File utility functions for NFOGuard
Common file operations to eliminate code duplication
"""
import glob
import re
from pathlib import Path
from typing import Optional, List, Dict, Tuple, Union

from utils.logging import _log


# Video file extensions used throughout the application
VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.ts'}

# Episode pattern for TV series files
EPISODE_PATTERN = re.compile(
    r'.*[sS](\d{1,2})[eE](\d{1,3}).*|.*(\d{1,2})x(\d{1,3}).*'
)


def find_media_path_by_imdb_and_title(
    title: str, 
    imdb_id: str, 
    search_paths: List[Path], 
    webhook_path: Optional[str] = None,
    path_mapper = None
) -> Optional[Path]:
    """
    Unified media path finder for both TV series and movies
    
    Args:
        title: Media title to search for
        imdb_id: IMDb ID to search for
        search_paths: List of paths to search in (tv_paths or movie_paths)
        webhook_path: Optional webhook path to try first
        path_mapper: Optional path mapper for webhook path conversion
    
    Returns:
        Path to media directory if found, None otherwise
    """
    # Try webhook path first if provided
    print(f"🔍 Webhook path provided: {webhook_path}")
    print(f"🔍 Path mapper available: {path_mapper is not None}")
    if webhook_path and path_mapper:
        try:
            if hasattr(path_mapper, 'sonarr_path_to_container_path'):
                container_path = path_mapper.sonarr_path_to_container_path(webhook_path)
                print(f"🔍 Sonarr path mapped: {webhook_path} → {container_path}")
            elif hasattr(path_mapper, 'radarr_path_to_container_path'):
                container_path = path_mapper.radarr_path_to_container_path(webhook_path)
                print(f"🔍 Radarr path mapped: {webhook_path} → {container_path}")
            else:
                container_path = webhook_path
                print(f"🔍 No mapper, using direct path: {container_path}")
                
            path_obj = Path(container_path)
            print(f"🔍 Mapped path exists: {path_obj.exists()} - {path_obj}")
            if path_obj.exists():
                print(f"✅ Using webhook path: {path_obj}")
                return path_obj
            else:
                print(f"❌ Webhook path doesn't exist, falling back to search")
        except Exception as e:
            print(f"❌ Failed to process webhook path {webhook_path}: {e}")
            _log("WARNING", f"Failed to process webhook path {webhook_path}: {e}")
    else:
        if not webhook_path:
            print(f"❌ No webhook path provided")
        if not path_mapper:
            print(f"❌ No path mapper available")
    
    # Search by IMDb ID or title in configured paths
    print(f"🔍 Total search paths to check: {len(search_paths)} - {[str(p) for p in search_paths]}")
    for media_path in search_paths:
        print(f"🔍 Checking path: {media_path} (exists: {media_path.exists()})")
        if not media_path.exists():
            print(f"❌ Skipping non-existent path: {media_path}")
            continue
        
        # Search by IMDb ID first (more reliable)
        if imdb_id:
            # Use proper glob pattern - escape brackets to match literal [imdb-ID]
            pattern = str(media_path / f"*\\[imdb-{imdb_id}\\]*")
            print(f"🔍 Searching with pattern: {pattern}")
            matches = glob.glob(pattern)
            print(f"🔍 Pattern matches found: {len(matches)} - {matches[:3] if matches else 'None'}")
            if matches:
                print(f"✅ Returning first match: {matches[0]}")
                return Path(matches[0])
        
        # Search by title as fallback
        if title:
            title_clean = clean_title_for_search(title)
            print(f"🔍 Title fallback: searching for '{title}' → cleaned: '{title_clean}'")
            for item in media_path.iterdir():
                if item.is_dir() and "[imdb-" in item.name.lower():
                    item_clean = clean_title_for_search(item.name)
                    print(f"🔍 Checking directory: '{item.name}' → cleaned: '{item_clean}'")
                    if title_clean in item_clean:
                        print(f"✅ Title match found: '{title_clean}' in '{item_clean}' → returning {item}")
                        return item
    
    return None


def clean_title_for_search(title: str) -> str:
    """
    Clean title for fuzzy matching
    
    Args:
        title: Raw title string
        
    Returns:
        Cleaned title for comparison
    """
    return title.lower().replace(" ", "").replace("-", "").replace(".", "")


def find_video_files(directory: Path, recursive: bool = True) -> List[Path]:
    """
    Find all video files in a directory
    
    Args:
        directory: Directory to search
        recursive: Whether to search recursively
        
    Returns:
        List of video file paths
    """
    if not directory.exists():
        return []
    
    video_files = []
    
    if recursive:
        for item in directory.rglob('*'):
            if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
                video_files.append(item)
    else:
        for item in directory.iterdir():
            if item.is_file() and item.suffix.lower() in VIDEO_EXTENSIONS:
                video_files.append(item)
    
    return video_files


def extract_episode_info(filename: str) -> Optional[Tuple[int, int]]:
    """
    Extract season and episode numbers from filename
    
    Args:
        filename: Video filename to parse
        
    Returns:
        Tuple of (season, episode) if found, None otherwise
    """
    match = EPISODE_PATTERN.match(filename)
    if not match:
        return None
    
    if match.group(1) and match.group(2):  # SxxExx format
        season = int(match.group(1))
        episode = int(match.group(2))
    elif match.group(3) and match.group(4):  # NxNN format
        season = int(match.group(3))
        episode = int(match.group(4))
    else:
        return None
    
    return (season, episode)


def find_episodes_on_disk(series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
    """
    Find all episodes on disk and return mapping of (season, episode) -> [video_files]
    
    Args:
        series_path: Path to series directory
        
    Returns:
        Dictionary mapping (season, episode) tuples to lists of video files
    """
    episodes = {}
    
    if not series_path.exists():
        return episodes
    
    for video_file in find_video_files(series_path, recursive=True):
        episode_info = extract_episode_info(video_file.name)
        if episode_info:
            season, episode = episode_info
            key = (season, episode)
            if key not in episodes:
                episodes[key] = []
            episodes[key].append(video_file)
    
    return episodes


def extract_title_from_directory_name(directory_name: str) -> Optional[str]:
    """
    Extract clean title from directory name, removing year and IMDb ID
    
    Args:
        directory_name: Directory name to parse
        
    Returns:
        Cleaned title or None if no title found
    """
    name = directory_name
    
    # Remove IMDb ID part: [imdb-ttXXXXXX] or [ttXXXXXX]
    name = re.sub(r'\s*\[imdb-?tt\d+\]', '', name, flags=re.IGNORECASE)
    name = re.sub(r'\s*\[tt\d+\]', '', name, flags=re.IGNORECASE)
    
    # Remove year in parentheses: (YYYY)
    name = re.sub(r'\s*\(\d{4}\)', '', name)
    
    # Clean up extra spaces
    name = ' '.join(name.split())
    
    return name.strip() if name.strip() else None


def extract_imdb_id_from_path(path: Union[str, Path]) -> Optional[str]:
    """
    Extract IMDb ID from directory or file path
    
    Args:
        path: Path to examine (string or Path object)
        
    Returns:
        IMDb ID if found, None otherwise
    """
    path_str = str(path)
    
    # Look for [imdb-ttXXXXXX] or [ttXXXXXX] patterns
    patterns = [
        r'\[imdb-(tt\d+)\]',  # [imdb-tt1234567]
        r'\[(tt\d+)\]',       # [tt1234567]
        r'imdb[_-]?(tt\d+)',  # imdb_tt1234567 or imdb-tt1234567
        r'(tt\d{7,})',        # standalone tt1234567 (7+ digits)
    ]
    
    for pattern in patterns:
        match = re.search(pattern, path_str, re.IGNORECASE)
        if match:
            imdb_id = match.group(1)
            # Ensure it starts with 'tt'
            if not imdb_id.startswith('tt'):
                imdb_id = f'tt{imdb_id}'
            return imdb_id
    
    return None


def is_video_file(file_path: Path) -> bool:
    """
    Check if a file is a video file based on extension
    
    Args:
        file_path: Path to check
        
    Returns:
        True if it's a video file, False otherwise
    """
    return file_path.suffix.lower() in VIDEO_EXTENSIONS


def safe_directory_scan(directory: Path, pattern: str = "*") -> List[Path]:
    """
    Safely scan directory with error handling
    
    Args:
        directory: Directory to scan
        pattern: Glob pattern to match
        
    Returns:
        List of matching paths, empty list if scan fails
    """
    try:
        if not directory.exists():
            return []
        return list(directory.glob(pattern))
    except (PermissionError, OSError) as e:
        _log("WARNING", f"Failed to scan directory {directory}: {e}")
        return []


def normalize_path_separators(path: str) -> str:
    """
    Normalize path separators for cross-platform compatibility
    
    Args:
        path: Path string to normalize
        
    Returns:
        Normalized path string
    """
    return str(Path(path).as_posix())