push updates to paths again
This commit is contained in:
+20
-270
@@ -1,285 +1,35 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Path mapping module for NFOGuard
|
||||
Handles path translation between container, Radarr/Sonarr, and download client paths
|
||||
Path mapping utilities for NFOGuard
|
||||
Handles conversion between external service paths and container 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)"""
|
||||
"""Handles path mapping between different environments"""
|
||||
|
||||
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
|
||||
|
||||
pass
|
||||
|
||||
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}")
|
||||
# Replace /mnt/unionfs/Media with /media
|
||||
if sonarr_path.startswith("/mnt/unionfs/Media"):
|
||||
return sonarr_path.replace("/mnt/unionfs/Media", "/media")
|
||||
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"
|
||||
def radarr_path_to_container_path(self, radarr_path: str) -> str:
|
||||
"""Convert Radarr path to container path"""
|
||||
# Replace /mnt/unionfs/Media with /media
|
||||
if radarr_path.startswith("/mnt/unionfs/Media"):
|
||||
container_path = radarr_path.replace("/mnt/unionfs/Media", "/media")
|
||||
# Keep movies6 as movies6, don't convert to movies/6
|
||||
return container_path
|
||||
return radarr_path
|
||||
|
||||
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}")
|
||||
def container_path_to_host_path(self, container_path: str) -> str:
|
||||
"""Convert container path back to host path if needed"""
|
||||
# This might be needed for file operations
|
||||
return container_path
|
||||
Reference in New Issue
Block a user