282 lines
12 KiB
Python
282 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Path mapping module for NFOGuard
|
|
Handles path translation between container, Radarr/Sonarr, and download client paths
|
|
"""
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
|
|
def _log(level: str, msg: str):
|
|
"""Placeholder logging function - replace with your actual logger"""
|
|
print(f"[{datetime.now().isoformat()}] {level}: {msg}")
|
|
|
|
|
|
class PathMapper:
|
|
"""Handles path mapping between different contexts (container, Radarr/Sonarr, downloads)"""
|
|
|
|
def __init__(self):
|
|
# Load path mappings from environment
|
|
self.container_to_radarr_movie_paths = self._load_movie_path_mappings()
|
|
self.container_to_sonarr_tv_paths = self._load_tv_path_mappings()
|
|
self.download_path_indicators = self._load_download_indicators()
|
|
|
|
# Reverse mappings for efficiency
|
|
self.radarr_to_container_movie_paths = {v: k for k, v in self.container_to_radarr_movie_paths.items()}
|
|
self.sonarr_to_container_tv_paths = {v: k for k, v in self.container_to_sonarr_tv_paths.items()}
|
|
|
|
def _load_movie_path_mappings(self) -> Dict[str, str]:
|
|
"""Load movie path mappings from environment"""
|
|
# Container paths (what NFOGuard sees)
|
|
container_paths_str = os.environ.get("MOVIE_PATHS", "/media/movies,/media/movies6")
|
|
container_paths = [p.strip() for p in container_paths_str.split(",") if p.strip()]
|
|
|
|
# Radarr paths (what Radarr sees)
|
|
radarr_paths_str = os.environ.get("RADARR_ROOT_FOLDERS", "/mnt/unionfs/Media/Movies/movies,/mnt/unionfs/Media/Movies/movies6")
|
|
radarr_paths = [p.strip() for p in radarr_paths_str.split(",") if p.strip()]
|
|
|
|
# Create mappings (container -> radarr)
|
|
mappings = {}
|
|
for i, container_path in enumerate(container_paths):
|
|
if i < len(radarr_paths):
|
|
mappings[container_path] = radarr_paths[i]
|
|
else:
|
|
# Default mapping if not enough Radarr paths specified
|
|
_log("WARNING", f"No Radarr path mapping for container path: {container_path}")
|
|
mappings[container_path] = container_path
|
|
|
|
_log("INFO", f"Movie path mappings: {mappings}")
|
|
return mappings
|
|
|
|
def _load_tv_path_mappings(self) -> Dict[str, str]:
|
|
"""Load TV path mappings from environment"""
|
|
# Container paths
|
|
container_paths_str = os.environ.get("TV_PATHS", "/media/tv,/media/tv6")
|
|
container_paths = [p.strip() for p in container_paths_str.split(",") if p.strip()]
|
|
|
|
# Sonarr paths
|
|
sonarr_paths_str = os.environ.get("SONARR_ROOT_FOLDERS", "/mnt/unionfs/Media/TV/tv,/mnt/unionfs/Media/TV/tv6")
|
|
sonarr_paths = [p.strip() for p in sonarr_paths_str.split(",") if p.strip()]
|
|
|
|
# Create mappings
|
|
mappings = {}
|
|
for i, container_path in enumerate(container_paths):
|
|
if i < len(sonarr_paths):
|
|
mappings[container_path] = sonarr_paths[i]
|
|
else:
|
|
_log("WARNING", f"No Sonarr path mapping for container path: {container_path}")
|
|
mappings[container_path] = container_path
|
|
|
|
_log("INFO", f"TV path mappings: {mappings}")
|
|
return mappings
|
|
|
|
def _load_download_indicators(self) -> List[str]:
|
|
"""Load download path indicators from environment"""
|
|
default_indicators = [
|
|
# nzbget paths
|
|
'/downloads/', '/download/', '/completed/', '/nzb-complete/', '/complete/',
|
|
# Transmission/Deluge
|
|
'/torrents/', '/torrent/', '/seed/', '/seeding/',
|
|
# SABnzbd
|
|
'/sabnzbd/', '/sab/', '/usenet/',
|
|
# qBittorrent
|
|
'/qbittorrent/', '/qbt/',
|
|
# Generic
|
|
'/importing/', '/processing/', '/temp/', '/tmp/', '/staging/',
|
|
# Client names
|
|
'nzbget', 'sabnzbd', 'transmission', 'deluge', 'qbittorrent', 'rtorrent'
|
|
]
|
|
|
|
# Allow custom indicators from environment
|
|
custom_indicators_str = os.environ.get("DOWNLOAD_PATH_INDICATORS", "")
|
|
if custom_indicators_str:
|
|
custom_indicators = [p.strip() for p in custom_indicators_str.split(",") if p.strip()]
|
|
default_indicators.extend(custom_indicators)
|
|
|
|
_log("DEBUG", f"Download path indicators: {default_indicators}")
|
|
return default_indicators
|
|
|
|
def container_path_to_radarr_path(self, container_path: str) -> str:
|
|
"""Convert container path to Radarr path"""
|
|
container_path = str(container_path).rstrip("/")
|
|
|
|
for container_root, radarr_root in self.container_to_radarr_movie_paths.items():
|
|
container_root = container_root.rstrip("/")
|
|
if container_path.startswith(container_root):
|
|
# Replace container root with Radarr root
|
|
relative_path = container_path[len(container_root):].lstrip("/")
|
|
radarr_path = f"{radarr_root.rstrip('/')}/{relative_path}" if relative_path else radarr_root.rstrip("/")
|
|
_log("DEBUG", f"Mapped container path {container_path} -> {radarr_path}")
|
|
return radarr_path
|
|
|
|
_log("WARNING", f"No Radarr mapping found for container path: {container_path}")
|
|
return container_path
|
|
|
|
def radarr_path_to_container_path(self, radarr_path: str) -> str:
|
|
"""Convert Radarr path to container path"""
|
|
radarr_path = str(radarr_path).rstrip("/")
|
|
|
|
for radarr_root, container_root in self.radarr_to_container_movie_paths.items():
|
|
radarr_root = radarr_root.rstrip("/")
|
|
if radarr_path.startswith(radarr_root):
|
|
# Replace Radarr root with container root
|
|
relative_path = radarr_path[len(radarr_root):].lstrip("/")
|
|
container_path = f"{container_root.rstrip('/')}/{relative_path}" if relative_path else container_root.rstrip("/")
|
|
_log("DEBUG", f"Mapped Radarr path {radarr_path} -> {container_path}")
|
|
return container_path
|
|
|
|
_log("WARNING", f"No container mapping found for Radarr path: {radarr_path}")
|
|
return radarr_path
|
|
|
|
def container_path_to_sonarr_path(self, container_path: str) -> str:
|
|
"""Convert container path to Sonarr path"""
|
|
container_path = str(container_path).rstrip("/")
|
|
|
|
for container_root, sonarr_root in self.container_to_sonarr_tv_paths.items():
|
|
container_root = container_root.rstrip("/")
|
|
if container_path.startswith(container_root):
|
|
relative_path = container_path[len(container_root):].lstrip("/")
|
|
sonarr_path = f"{sonarr_root.rstrip('/')}/{relative_path}" if relative_path else sonarr_root.rstrip("/")
|
|
_log("DEBUG", f"Mapped container path {container_path} -> {sonarr_path}")
|
|
return sonarr_path
|
|
|
|
_log("WARNING", f"No Sonarr mapping found for container path: {container_path}")
|
|
return container_path
|
|
|
|
def sonarr_path_to_container_path(self, sonarr_path: str) -> str:
|
|
"""Convert Sonarr path to container path"""
|
|
sonarr_path = str(sonarr_path).rstrip("/")
|
|
|
|
for sonarr_root, container_root in self.sonarr_to_container_tv_paths.items():
|
|
sonarr_root = sonarr_root.rstrip("/")
|
|
if sonarr_path.startswith(sonarr_root):
|
|
relative_path = sonarr_path[len(sonarr_root):].lstrip("/")
|
|
container_path = f"{container_root.rstrip('/')}/{relative_path}" if relative_path else container_root.rstrip("/")
|
|
_log("DEBUG", f"Mapped Sonarr path {sonarr_path} -> {container_path}")
|
|
return container_path
|
|
|
|
_log("WARNING", f"No container mapping found for Sonarr path: {sonarr_path}")
|
|
return sonarr_path
|
|
|
|
def is_download_path(self, path: str) -> bool:
|
|
"""Check if path indicates a download/temporary location"""
|
|
path_lower = str(path).lower()
|
|
return any(indicator in path_lower for indicator in self.download_path_indicators)
|
|
|
|
def analyze_import_source_path(self, source_path: str) -> Tuple[bool, str]:
|
|
"""
|
|
Analyze import source path to determine if it's from downloads.
|
|
|
|
Returns:
|
|
(is_from_downloads, reason)
|
|
"""
|
|
if not source_path:
|
|
return False, "no_source_path"
|
|
|
|
path_lower = str(source_path).lower()
|
|
|
|
# Extract potential IMDb ID from path
|
|
imdb_match = re.search(r'(?:imdb-|tt)(\d{6,8})', path_lower)
|
|
if imdb_match:
|
|
imdb_id = f"tt{imdb_match.group(1)}"
|
|
_log("DEBUG", f"Found IMDb ID in path: {imdb_id}")
|
|
return True, f"imdb_in_path({imdb_id})"
|
|
|
|
# Check for download indicators
|
|
for indicator in self.download_path_indicators:
|
|
if indicator in path_lower:
|
|
return True, f"download_indicator({indicator})"
|
|
|
|
# Check if path is outside media roots (likely a download)
|
|
media_roots = []
|
|
media_roots.extend(self.container_to_radarr_movie_paths.values())
|
|
media_roots.extend(self.container_to_sonarr_tv_paths.values())
|
|
|
|
is_in_media = any(source_path.startswith(root.rstrip('/')) for root in media_roots)
|
|
if not is_in_media:
|
|
return True, "outside_media_roots"
|
|
|
|
return False, "appears_to_be_media_path"
|
|
|
|
def find_container_path_from_webhook(self, webhook_path: str, media_type: str = "movie") -> Optional[Path]:
|
|
"""
|
|
Find container path from webhook path information.
|
|
|
|
Args:
|
|
webhook_path: Path from Radarr/Sonarr webhook
|
|
media_type: "movie" or "tv"
|
|
"""
|
|
if not webhook_path:
|
|
return None
|
|
|
|
if media_type == "movie":
|
|
container_path = self.radarr_path_to_container_path(webhook_path)
|
|
else:
|
|
container_path = self.sonarr_path_to_container_path(webhook_path)
|
|
|
|
path_obj = Path(container_path)
|
|
if path_obj.exists():
|
|
_log("INFO", f"Found container path: {path_obj}")
|
|
return path_obj
|
|
|
|
_log("WARNING", f"Container path does not exist: {path_obj}")
|
|
return None
|
|
|
|
def get_debug_info(self) -> Dict[str, any]:
|
|
"""Get debug information about path mappings"""
|
|
return {
|
|
"movie_mappings": {
|
|
"container_to_radarr": self.container_to_radarr_movie_paths,
|
|
"radarr_to_container": self.radarr_to_container_movie_paths
|
|
},
|
|
"tv_mappings": {
|
|
"container_to_sonarr": self.container_to_sonarr_tv_paths,
|
|
"sonarr_to_container": self.sonarr_to_container_tv_paths
|
|
},
|
|
"download_indicators": self.download_path_indicators
|
|
}
|
|
|
|
|
|
# Global instance
|
|
path_mapper = PathMapper()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test path mapping
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# Set up test environment
|
|
os.environ["MOVIE_PATHS"] = "/media/movies,/media/movies6"
|
|
os.environ["RADARR_ROOT_FOLDERS"] = "/mnt/unionfs/Media/Movies/movies,/mnt/unionfs/Media/Movies/movies6"
|
|
os.environ["TV_PATHS"] = "/media/tv,/media/tv6"
|
|
os.environ["SONARR_ROOT_FOLDERS"] = "/mnt/unionfs/Media/TV/tv,/mnt/unionfs/Media/TV/tv6"
|
|
|
|
mapper = PathMapper()
|
|
|
|
# Test movie path mapping
|
|
container_movie = "/media/movies/Test Movie [imdb-tt1234567]"
|
|
radarr_movie = mapper.container_path_to_radarr_path(container_movie)
|
|
print(f"Container -> Radarr: {container_movie} -> {radarr_movie}")
|
|
|
|
back_to_container = mapper.radarr_path_to_container_path(radarr_movie)
|
|
print(f"Radarr -> Container: {radarr_movie} -> {back_to_container}")
|
|
|
|
# Test download path detection
|
|
test_paths = [
|
|
"/downloads/complete/Test.Movie.2023.1080p.BluRay.x264",
|
|
"/mnt/unionfs/Media/Movies/movies/Test Movie [imdb-tt1234567]",
|
|
"/nzbget/complete/Test.Series.S01E01.1080p.WEB.x264",
|
|
"/torrents/seed/Test.Movie.2023"
|
|
]
|
|
|
|
for path in test_paths:
|
|
is_download, reason = mapper.analyze_import_source_path(path)
|
|
print(f"Path analysis: {path} -> is_download={is_download} ({reason})")
|
|
|
|
# Show debug info
|
|
debug_info = mapper.get_debug_info()
|
|
print(f"\nDebug info: {debug_info}") |