sbcrumb
e741e4b046
- Add prefixed batch keys (movie:imdbid, tv:imdbid) to prevent IMDb ID collisions - Add path existence validation for Radarr webhooks to reject invalid mappings early - Remove duplicate Radarr webhook handler code - Add debug scripts for troubleshooting webhook and path mapping issues - Create corrected .env template with fixed TV_PATHS and SONARR_ROOT_FOLDERS This fixes the issue where TV path mapping failures caused movie webhooks to process wrong movies due to shared batch queue corruption.
285 lines
12 KiB
Python
285 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
|
|
|
|
from core.logging import _log
|
|
|
|
|
|
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}")
|
|
|
|
# Check if the mapped path actually exists
|
|
from pathlib import Path
|
|
if not Path(container_path).exists():
|
|
_log("WARNING", f"Mapped container path does not exist: {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}") |