refactor
This commit is contained in:
@@ -0,0 +1,403 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Database operations module for NFOGuard
|
||||
"""
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
from typing import Dict, List, Optional, Tuple, Any
|
||||
|
||||
|
||||
def _log(level: str, msg: str):
|
||||
"""Placeholder logging function - replace with your actual logger"""
|
||||
print(f"[{datetime.now().isoformat()}] {level}: {msg}")
|
||||
|
||||
|
||||
class NFOGuardDatabase:
|
||||
"""Database operations for TV series, episodes, movies, and their metadata"""
|
||||
|
||||
# Database schemas
|
||||
TV_SCHEMA = """
|
||||
PRAGMA journal_mode=WAL;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS series (
|
||||
imdb_id TEXT PRIMARY KEY,
|
||||
series_path TEXT NOT NULL,
|
||||
last_applied TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS episode_dates (
|
||||
imdb_id TEXT NOT NULL,
|
||||
season INTEGER NOT NULL,
|
||||
episode INTEGER NOT NULL,
|
||||
aired TEXT,
|
||||
dateadded TEXT,
|
||||
source TEXT,
|
||||
has_video_file INTEGER DEFAULT 0,
|
||||
PRIMARY KEY (imdb_id, season, episode)
|
||||
);
|
||||
"""
|
||||
|
||||
MOVIE_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS movies (
|
||||
imdb_id TEXT PRIMARY KEY,
|
||||
movie_path TEXT NOT NULL,
|
||||
last_applied TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS movie_dates (
|
||||
imdb_id TEXT PRIMARY KEY,
|
||||
released TEXT,
|
||||
dateadded TEXT,
|
||||
source TEXT,
|
||||
has_video_file INTEGER DEFAULT 0
|
||||
);
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: Path):
|
||||
self.db_path = Path(db_path)
|
||||
self._ensure_database()
|
||||
|
||||
def _ensure_database(self):
|
||||
"""Create database and tables if they don't exist"""
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with self.get_connection() as conn:
|
||||
# Check and create TV schema
|
||||
try:
|
||||
result = conn.execute("PRAGMA table_info(episode_dates);").fetchall()
|
||||
columns = [row[1] for row in result]
|
||||
required_columns = ['imdb_id', 'season', 'episode', 'aired', 'dateadded', 'source']
|
||||
|
||||
if not all(col in columns for col in required_columns):
|
||||
_log("WARNING", "TV database schema outdated, recreating tables...")
|
||||
conn.execute("DROP TABLE IF EXISTS episode_dates;")
|
||||
conn.execute("DROP TABLE IF EXISTS series;")
|
||||
conn.execute("DROP TABLE IF EXISTS meta;")
|
||||
elif 'has_video_file' not in columns:
|
||||
_log("INFO", "Adding has_video_file column to episode_dates table")
|
||||
conn.execute("ALTER TABLE episode_dates ADD COLUMN has_video_file INTEGER DEFAULT 0;")
|
||||
|
||||
except sqlite3.OperationalError:
|
||||
# Tables don't exist, which is fine
|
||||
pass
|
||||
|
||||
# Create TV schema
|
||||
conn.executescript(self.TV_SCHEMA)
|
||||
|
||||
# Check and create movie schema
|
||||
try:
|
||||
result = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='movies'").fetchone()
|
||||
if not result:
|
||||
_log("INFO", "Creating movie tables in database")
|
||||
conn.executescript(self.MOVIE_SCHEMA)
|
||||
except sqlite3.Error as e:
|
||||
_log("ERROR", f"Failed to create movie tables: {e}")
|
||||
|
||||
conn.commit()
|
||||
_log("DEBUG", "Database schema verified and up to date")
|
||||
|
||||
def get_connection(self) -> sqlite3.Connection:
|
||||
"""Get database connection with proper settings"""
|
||||
conn = sqlite3.Connection(str(self.db_path))
|
||||
conn.execute("PRAGMA foreign_keys=ON;")
|
||||
conn.execute("PRAGMA busy_timeout = 30000;") # 30 second timeout
|
||||
return conn
|
||||
|
||||
# TV Series Operations
|
||||
def upsert_series(self, imdb_id: str, series_path: str) -> None:
|
||||
"""Insert or update series record"""
|
||||
with self.get_connection() as conn:
|
||||
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
conn.execute(
|
||||
"INSERT INTO series(imdb_id, series_path, last_applied) VALUES(?,?,?) "
|
||||
"ON CONFLICT(imdb_id) DO UPDATE SET series_path=excluded.series_path, last_applied=excluded.last_applied",
|
||||
(imdb_id, series_path, now)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def get_series_episodes(self, imdb_id: str, has_video_file_only: bool = False) -> List[Dict[str, Any]]:
|
||||
"""Get all episodes for a series"""
|
||||
with self.get_connection() as conn:
|
||||
if has_video_file_only:
|
||||
query = "SELECT season, episode, aired, dateadded, source FROM episode_dates WHERE imdb_id = ? AND has_video_file = 1"
|
||||
else:
|
||||
query = "SELECT season, episode, aired, dateadded, source FROM episode_dates WHERE imdb_id = ?"
|
||||
|
||||
rows = conn.execute(query, (imdb_id,)).fetchall()
|
||||
return [
|
||||
{
|
||||
"season": row[0],
|
||||
"episode": row[1],
|
||||
"aired": row[2],
|
||||
"dateadded": row[3],
|
||||
"source": row[4]
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
def upsert_episode_date(
|
||||
self,
|
||||
imdb_id: str,
|
||||
season: int,
|
||||
episode: int,
|
||||
aired: Optional[str] = None,
|
||||
dateadded: Optional[str] = None,
|
||||
source: Optional[str] = None,
|
||||
has_video_file: bool = False
|
||||
) -> None:
|
||||
"""Insert or update episode date record"""
|
||||
with self.get_connection() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO episode_dates(imdb_id, season, episode, aired, dateadded, source, has_video_file) "
|
||||
"VALUES(?,?,?,?,?,?,?) "
|
||||
"ON CONFLICT(imdb_id, season, episode) DO UPDATE SET "
|
||||
"aired=excluded.aired, dateadded=excluded.dateadded, source=excluded.source, has_video_file=excluded.has_video_file",
|
||||
(imdb_id, season, episode, aired, dateadded, source, 1 if has_video_file else 0)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def cleanup_orphaned_episodes(self, existing_episodes: Dict[str, List[Tuple[int, int]]]) -> int:
|
||||
"""
|
||||
Remove episode records that no longer have video files on disk.
|
||||
|
||||
Args:
|
||||
existing_episodes: Dict mapping imdb_id -> [(season, episode), ...]
|
||||
"""
|
||||
removed_count = 0
|
||||
|
||||
with self.get_connection() as conn:
|
||||
# Get all series from database
|
||||
series_rows = conn.execute("SELECT imdb_id FROM series").fetchall()
|
||||
|
||||
for (imdb_id,) in series_rows:
|
||||
if imdb_id not in existing_episodes:
|
||||
# Series no longer exists on disk
|
||||
removed = conn.execute("DELETE FROM episode_dates WHERE imdb_id = ?", (imdb_id,)).rowcount
|
||||
removed_count += removed
|
||||
_log("INFO", f"Removed {removed} episodes for missing series {imdb_id}")
|
||||
continue
|
||||
|
||||
# Get episodes in database for this series
|
||||
db_episodes = conn.execute(
|
||||
"SELECT season, episode FROM episode_dates WHERE imdb_id = ?",
|
||||
(imdb_id,)
|
||||
).fetchall()
|
||||
|
||||
disk_episodes = set(existing_episodes[imdb_id])
|
||||
|
||||
for season, episode in db_episodes:
|
||||
if (season, episode) not in disk_episodes:
|
||||
conn.execute(
|
||||
"DELETE FROM episode_dates WHERE imdb_id = ? AND season = ? AND episode = ?",
|
||||
(imdb_id, season, episode)
|
||||
)
|
||||
removed_count += 1
|
||||
_log("DEBUG", f"Removed orphaned episode {imdb_id} S{season:02d}E{episode:02d}")
|
||||
|
||||
conn.commit()
|
||||
|
||||
return removed_count
|
||||
|
||||
# Movie Operations
|
||||
def upsert_movie(self, imdb_id: str, movie_path: str) -> None:
|
||||
"""Insert or update movie record"""
|
||||
with self.get_connection() as conn:
|
||||
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
conn.execute(
|
||||
"INSERT INTO movies(imdb_id, movie_path, last_applied) VALUES(?,?,?) "
|
||||
"ON CONFLICT(imdb_id) DO UPDATE SET movie_path=excluded.movie_path, last_applied=excluded.last_applied",
|
||||
(imdb_id, movie_path, now)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def get_movie_dates(self, imdb_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""Get stored movie dates"""
|
||||
with self.get_connection() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT released, dateadded, source, has_video_file FROM movie_dates WHERE imdb_id = ?",
|
||||
(imdb_id,)
|
||||
).fetchone()
|
||||
|
||||
if row:
|
||||
return {
|
||||
"released": row[0],
|
||||
"dateadded": row[1],
|
||||
"source": row[2],
|
||||
"has_video_file": bool(row[3])
|
||||
}
|
||||
return None
|
||||
|
||||
def upsert_movie_dates(
|
||||
self,
|
||||
imdb_id: str,
|
||||
released: Optional[str] = None,
|
||||
dateadded: Optional[str] = None,
|
||||
source: Optional[str] = None,
|
||||
has_video_file: bool = False
|
||||
) -> None:
|
||||
"""Insert or update movie dates"""
|
||||
with self.get_connection() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO movie_dates(imdb_id, released, dateadded, source, has_video_file) VALUES(?,?,?,?,?) "
|
||||
"ON CONFLICT(imdb_id) DO UPDATE SET "
|
||||
"released=excluded.released, dateadded=excluded.dateadded, source=excluded.source, has_video_file=excluded.has_video_file",
|
||||
(imdb_id, released, dateadded, source, 1 if has_video_file else 0)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def cleanup_orphaned_movies(self, existing_movies: List[str]) -> int:
|
||||
"""Remove movie records that no longer exist on disk"""
|
||||
if not existing_movies:
|
||||
return 0
|
||||
|
||||
removed_count = 0
|
||||
existing_set = set(existing_movies)
|
||||
|
||||
with self.get_connection() as conn:
|
||||
# Get all movies from database
|
||||
db_movies = conn.execute("SELECT imdb_id FROM movies").fetchall()
|
||||
|
||||
for (imdb_id,) in db_movies:
|
||||
if imdb_id not in existing_set:
|
||||
# Movie no longer exists on disk
|
||||
conn.execute("DELETE FROM movies WHERE imdb_id = ?", (imdb_id,))
|
||||
conn.execute("DELETE FROM movie_dates WHERE imdb_id = ?", (imdb_id,))
|
||||
removed_count += 1
|
||||
_log("DEBUG", f"Removed orphaned movie {imdb_id}")
|
||||
|
||||
conn.commit()
|
||||
|
||||
return removed_count
|
||||
|
||||
# Statistics and Reporting
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
"""Get database statistics"""
|
||||
with self.get_connection() as conn:
|
||||
stats = {}
|
||||
|
||||
# TV stats
|
||||
stats["tv_series_count"] = conn.execute("SELECT COUNT(*) FROM series").fetchone()[0]
|
||||
stats["tv_episode_count"] = conn.execute("SELECT COUNT(*) FROM episode_dates").fetchone()[0]
|
||||
stats["tv_episodes_with_files"] = conn.execute("SELECT COUNT(*) FROM episode_dates WHERE has_video_file = 1").fetchone()[0]
|
||||
|
||||
# Movie stats
|
||||
stats["movie_count"] = conn.execute("SELECT COUNT(*) FROM movies").fetchone()[0]
|
||||
stats["movies_with_files"] = conn.execute("SELECT COUNT(*) FROM movie_dates WHERE has_video_file = 1").fetchone()[0]
|
||||
|
||||
# Source breakdowns
|
||||
tv_sources = conn.execute("""
|
||||
SELECT source, COUNT(*) as count
|
||||
FROM episode_dates
|
||||
WHERE has_video_file = 1
|
||||
GROUP BY source
|
||||
ORDER BY count DESC
|
||||
""").fetchall()
|
||||
stats["tv_sources"] = dict(tv_sources)
|
||||
|
||||
movie_sources = conn.execute("""
|
||||
SELECT source, COUNT(*) as count
|
||||
FROM movie_dates
|
||||
WHERE has_video_file = 1
|
||||
GROUP BY source
|
||||
ORDER BY count DESC
|
||||
""").fetchall()
|
||||
stats["movie_sources"] = dict(movie_sources)
|
||||
|
||||
return stats
|
||||
|
||||
def vacuum_database(self) -> None:
|
||||
"""Vacuum database to reclaim space"""
|
||||
with self.get_connection() as conn:
|
||||
conn.execute("VACUUM;")
|
||||
_log("INFO", "Database vacuumed")
|
||||
|
||||
# Utility Methods
|
||||
def clear_all_data(self) -> None:
|
||||
"""Clear all data (keep schema) - useful for testing"""
|
||||
with self.get_connection() as conn:
|
||||
conn.execute("DELETE FROM episode_dates;")
|
||||
conn.execute("DELETE FROM series;")
|
||||
conn.execute("DELETE FROM movie_dates;")
|
||||
conn.execute("DELETE FROM movies;")
|
||||
conn.execute("DELETE FROM meta;")
|
||||
conn.commit()
|
||||
_log("WARNING", "All database data cleared")
|
||||
|
||||
def export_series_episodes(self, imdb_id: str) -> List[Dict[str, Any]]:
|
||||
"""Export all episode data for a series (for debugging)"""
|
||||
episodes = self.get_series_episodes(imdb_id)
|
||||
_log("INFO", f"Series {imdb_id} has {len(episodes)} episodes in database")
|
||||
return episodes
|
||||
|
||||
def check_database_health(self) -> Dict[str, Any]:
|
||||
"""Check database health and integrity"""
|
||||
health = {"status": "healthy", "issues": []}
|
||||
|
||||
try:
|
||||
with self.get_connection() as conn:
|
||||
# Check integrity
|
||||
integrity = conn.execute("PRAGMA integrity_check;").fetchone()[0]
|
||||
if integrity != "ok":
|
||||
health["status"] = "corrupted"
|
||||
health["issues"].append(f"Integrity check failed: {integrity}")
|
||||
|
||||
# Check for orphaned records
|
||||
orphaned_episodes = conn.execute("""
|
||||
SELECT COUNT(*) FROM episode_dates
|
||||
WHERE imdb_id NOT IN (SELECT imdb_id FROM series)
|
||||
""").fetchone()[0]
|
||||
|
||||
if orphaned_episodes > 0:
|
||||
health["issues"].append(f"{orphaned_episodes} orphaned episodes")
|
||||
|
||||
orphaned_movie_dates = conn.execute("""
|
||||
SELECT COUNT(*) FROM movie_dates
|
||||
WHERE imdb_id NOT IN (SELECT imdb_id FROM movies)
|
||||
""").fetchone()[0]
|
||||
|
||||
if orphaned_movie_dates > 0:
|
||||
health["issues"].append(f"{orphaned_movie_dates} orphaned movie dates")
|
||||
|
||||
health["orphaned_episodes"] = orphaned_episodes
|
||||
health["orphaned_movie_dates"] = orphaned_movie_dates
|
||||
|
||||
except Exception as e:
|
||||
health["status"] = "error"
|
||||
health["error"] = str(e)
|
||||
|
||||
return health
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Test the database
|
||||
test_db = NFOGuardDatabase(Path("/tmp/test_nfoguard.db"))
|
||||
|
||||
# Test TV operations
|
||||
test_db.upsert_series("tt1234567", "/media/tv/Test Series [imdb-tt1234567]")
|
||||
test_db.upsert_episode_date("tt1234567", 1, 1, "2020-01-01", "2025-07-07T12:00:00+00:00", "test:import", True)
|
||||
|
||||
episodes = test_db.get_series_episodes("tt1234567")
|
||||
print(f"Test series episodes: {episodes}")
|
||||
|
||||
# Test movie operations
|
||||
test_db.upsert_movie("tt7654321", "/media/movies/Test Movie [imdb-tt7654321]")
|
||||
test_db.upsert_movie_dates("tt7654321", "2020-01-01", "2025-07-07T12:00:00+00:00", "test:import", True)
|
||||
|
||||
movie_dates = test_db.get_movie_dates("tt7654321")
|
||||
print(f"Test movie dates: {movie_dates}")
|
||||
|
||||
# Get stats
|
||||
stats = test_db.get_stats()
|
||||
print(f"Database stats: {stats}")
|
||||
|
||||
# Check health
|
||||
health = test_db.check_database_health()
|
||||
print(f"Database health: {health}")
|
||||
|
||||
print("Database test completed successfully!")
|
||||
@@ -0,0 +1,387 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
NFO file management module for movies and TV episodes
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
|
||||
def _log(level: str, msg: str):
|
||||
"""Placeholder logging function - replace with your actual logger"""
|
||||
print(f"[{datetime.now().isoformat()}] {level}: {msg}")
|
||||
|
||||
|
||||
class NFOManager:
|
||||
"""Handles creation and updating of NFO files for movies and TV episodes"""
|
||||
|
||||
# Regex patterns
|
||||
IMDB_TAG_PATTERN = re.compile(r"\[imdb-(tt\d+)\]", re.IGNORECASE)
|
||||
LONG_NFO_PATTERN = re.compile(r".*-S\d{2}E\d{2}-.*\.nfo$", re.IGNORECASE)
|
||||
SHORT_NFO_PATTERN = re.compile(r"^S(\d{2})E(\d{2})\.nfo$", re.IGNORECASE)
|
||||
|
||||
def __init__(self, manager_brand: str = "NFOGuard"):
|
||||
self.manager_brand = manager_brand
|
||||
|
||||
@staticmethod
|
||||
def parse_imdb_from_path(path: Path) -> Optional[str]:
|
||||
"""Extract IMDb ID from directory or filename"""
|
||||
match = NFOManager.IMDB_TAG_PATTERN.search(path.name)
|
||||
return match.group(1).lower() if match else None
|
||||
|
||||
@staticmethod
|
||||
def parse_imdb_from_nfo(nfo_path: Path) -> Optional[str]:
|
||||
"""Extract IMDb ID from existing NFO file"""
|
||||
if not nfo_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
tree = ET.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Check uniqueid elements
|
||||
for uniqueid in root.findall("uniqueid"):
|
||||
if (uniqueid.attrib.get("type") or "").lower() == "imdb":
|
||||
imdb_id = (uniqueid.text or "").strip()
|
||||
if imdb_id.startswith("tt"):
|
||||
return imdb_id.lower()
|
||||
|
||||
# Check direct imdbid elements (older format)
|
||||
imdbid_elem = root.find("imdbid")
|
||||
if imdbid_elem is not None and imdbid_elem.text:
|
||||
imdb_id = imdbid_elem.text.strip()
|
||||
if imdb_id.startswith("tt"):
|
||||
return imdb_id.lower()
|
||||
|
||||
except Exception as e:
|
||||
_log("DEBUG", f"Could not parse IMDb from NFO {nfo_path}: {e}")
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _ensure_child_text(parent: ET.Element, tag: str, text: str, overwrite: bool = True) -> ET.Element:
|
||||
"""Ensure child element exists with specified text"""
|
||||
child = parent.find(tag)
|
||||
if child is None:
|
||||
child = ET.SubElement(parent, tag)
|
||||
child.text = text
|
||||
else:
|
||||
if overwrite or (child.text is None or str(child.text).strip() == ""):
|
||||
child.text = text
|
||||
return child
|
||||
|
||||
@staticmethod
|
||||
def _ensure_uniqueid_imdb(root: ET.Element, imdb_id: str):
|
||||
"""Ensure IMDb uniqueid element exists"""
|
||||
imdb_elem = None
|
||||
for uid in root.findall("uniqueid"):
|
||||
if (uid.attrib.get("type") or "").lower() == "imdb":
|
||||
imdb_elem = uid
|
||||
break
|
||||
|
||||
if imdb_elem is None:
|
||||
imdb_elem = ET.SubElement(root, "uniqueid", {"type": "imdb", "default": "true"})
|
||||
else:
|
||||
if "default" not in imdb_elem.attrib:
|
||||
imdb_elem.set("default", "true")
|
||||
|
||||
imdb_elem.text = imdb_id
|
||||
|
||||
@staticmethod
|
||||
def _indent_xml(elem: ET.Element, level: int = 0, space: str = " "):
|
||||
"""Pretty-print XML with proper indentation"""
|
||||
i = "\n" + level * space
|
||||
if len(elem):
|
||||
if not elem.text or not elem.text.strip():
|
||||
elem.text = i + space
|
||||
for idx, e in enumerate(list(elem)):
|
||||
NFOManager._indent_xml(e, level + 1, space)
|
||||
if not e.tail or not e.tail.strip():
|
||||
e.tail = i + (space if idx < len(elem) - 1 else "")
|
||||
if not elem.tail or not elem.tail.strip():
|
||||
elem.tail = "\n" + (level - 1) * space if level else "\n"
|
||||
else:
|
||||
if not elem.text or not elem.text.strip():
|
||||
elem.text = ""
|
||||
if not elem.tail or not elem.tail.strip():
|
||||
elem.tail = "\n" + (level - 1) * space if level else "\n"
|
||||
|
||||
@staticmethod
|
||||
def _strip_existing_footer_comments(root: ET.Element):
|
||||
"""Remove existing manager comments from XML"""
|
||||
for node in list(root):
|
||||
if hasattr(node, 'text') and isinstance(node.text, str):
|
||||
text_lower = node.text.lower()
|
||||
if "source:" in text_lower or "managed by" in text_lower:
|
||||
root.remove(node)
|
||||
|
||||
def create_movie_nfo(
|
||||
self,
|
||||
movie_dir: Path,
|
||||
imdb_id: str,
|
||||
dateadded: Optional[str] = None,
|
||||
released: Optional[str] = None,
|
||||
source_detail: str = "unknown",
|
||||
lock_metadata: bool = True
|
||||
):
|
||||
"""Create or update movie.nfo file"""
|
||||
canonical_path = movie_dir / "movie.nfo"
|
||||
alt_path = movie_dir / f"{movie_dir.name}.nfo"
|
||||
|
||||
# Try to load existing NFO
|
||||
root = None
|
||||
for nfo_path in (canonical_path, alt_path):
|
||||
if nfo_path.exists():
|
||||
try:
|
||||
tree = ET.parse(str(nfo_path))
|
||||
existing_root = tree.getroot()
|
||||
if existing_root is not None and existing_root.tag.lower() == "movie":
|
||||
root = existing_root
|
||||
break
|
||||
except Exception as e:
|
||||
_log("DEBUG", f"Could not parse existing NFO {nfo_path}: {e}")
|
||||
|
||||
if root is None:
|
||||
root = ET.Element("movie")
|
||||
|
||||
# Ensure IMDb ID format
|
||||
if imdb_id and not imdb_id.lower().startswith("tt"):
|
||||
imdb_id = f"tt{imdb_id}"
|
||||
|
||||
# Add/update elements
|
||||
if imdb_id:
|
||||
self._ensure_uniqueid_imdb(root, imdb_id)
|
||||
|
||||
if released:
|
||||
# Extract date part from ISO string
|
||||
release_date = released.split("T")[0]
|
||||
self._ensure_child_text(root, "premiered", release_date, overwrite=False)
|
||||
self._ensure_child_text(root, "year", release_date.split("-")[0], overwrite=False)
|
||||
|
||||
if dateadded:
|
||||
if dateadded == "MANUAL_REVIEW_NEEDED":
|
||||
self._ensure_child_text(root, "dateadded", "MANUAL_REVIEW_NEEDED", overwrite=True)
|
||||
_log("WARNING", f"Set MANUAL_REVIEW_NEEDED marker in NFO: {movie_dir}/movie.nfo")
|
||||
else:
|
||||
self._ensure_child_text(root, "dateadded", dateadded, overwrite=True)
|
||||
|
||||
if lock_metadata:
|
||||
self._ensure_child_text(root, "lockdata", "true", overwrite=True)
|
||||
|
||||
# Add footer comments
|
||||
self._strip_existing_footer_comments(root)
|
||||
root.append(ET.Comment(f" source: Movie; {source_detail} "))
|
||||
root.append(ET.Comment(f" managed by {self.manager_brand} "))
|
||||
|
||||
# Pretty-print and save
|
||||
self._indent_xml(root)
|
||||
|
||||
# Write to both canonical path and alt path if it exists
|
||||
for target in [canonical_path] + ([alt_path] if alt_path.exists() else []):
|
||||
try:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
ET.ElementTree(root).write(str(target), encoding="utf-8", xml_declaration=True)
|
||||
_log("DEBUG", f"Updated NFO: {target}")
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed writing {target}: {e}")
|
||||
|
||||
def create_episode_nfo(
|
||||
self,
|
||||
season_dir: Path,
|
||||
season_num: int,
|
||||
episode_num: int,
|
||||
aired: Optional[str] = None,
|
||||
dateadded: Optional[str] = None,
|
||||
source_detail: str = "unknown",
|
||||
lock_metadata: bool = True
|
||||
):
|
||||
"""Create episode NFO file (S01E01.nfo format)"""
|
||||
try:
|
||||
season_dir.mkdir(parents=True, exist_ok=True)
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed creating season directory {season_dir}: {e}")
|
||||
return
|
||||
|
||||
nfo_path = season_dir / f"S{season_num:02d}E{episode_num:02d}.nfo"
|
||||
|
||||
# Build XML content
|
||||
xml_lines = [
|
||||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
|
||||
"<episodedetails>",
|
||||
f" <season>{season_num}</season>",
|
||||
f" <episode>{episode_num}</episode>",
|
||||
]
|
||||
|
||||
if aired:
|
||||
# Extract date part from ISO string
|
||||
air_date = aired.split('T')[0]
|
||||
xml_lines.append(f" <aired>{air_date}</aired>")
|
||||
|
||||
if dateadded:
|
||||
xml_lines.append(f" <dateadded>{dateadded}</dateadded>")
|
||||
# Add premiered tag for broader compatibility
|
||||
premiered_date = dateadded.split('T')[0]
|
||||
xml_lines.append(f" <premiered>{premiered_date}</premiered>")
|
||||
|
||||
if lock_metadata:
|
||||
xml_lines.append(" <lockdata>true</lockdata>")
|
||||
|
||||
# Add source comments
|
||||
if source_detail:
|
||||
xml_lines.append(f" <!-- source: TV Episode; {source_detail} -->")
|
||||
xml_lines.append(f" <!-- managed by {self.manager_brand} -->")
|
||||
|
||||
xml_lines.append("</episodedetails>")
|
||||
xml_lines.append("") # Final newline
|
||||
|
||||
content = "\n".join(xml_lines)
|
||||
|
||||
try:
|
||||
nfo_path.write_text(content, encoding="utf-8")
|
||||
_log("DEBUG", f"Created episode NFO: {nfo_path}")
|
||||
except Exception as e:
|
||||
_log("ERROR", f"Failed writing {nfo_path}: {e}")
|
||||
|
||||
def create_season_nfo(self, season_dir: Path, season_num: int):
|
||||
"""Create season.nfo if it doesn't exist"""
|
||||
season_nfo = season_dir / "season.nfo"
|
||||
if season_nfo.exists():
|
||||
return
|
||||
|
||||
xml_content = f"""<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
|
||||
<season>
|
||||
<seasonnumber>{season_num}</seasonnumber>
|
||||
</season>
|
||||
"""
|
||||
try:
|
||||
season_nfo.write_text(xml_content, encoding="utf-8")
|
||||
_log("DEBUG", f"Created season NFO: {season_nfo}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Could not write season.nfo in {season_dir}: {e}")
|
||||
|
||||
def create_tvshow_nfo(self, series_dir: Path, imdb_id: str):
|
||||
"""Create tvshow.nfo if it doesn't exist"""
|
||||
tvshow_nfo = series_dir / "tvshow.nfo"
|
||||
if tvshow_nfo.exists():
|
||||
return
|
||||
|
||||
xml_content = f"""<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
|
||||
<tvshow>
|
||||
<id>{imdb_id}</id>
|
||||
<uniqueid type="imdb" default="true">{imdb_id}</uniqueid>
|
||||
</tvshow>
|
||||
"""
|
||||
try:
|
||||
tvshow_nfo.write_text(xml_content, encoding="utf-8")
|
||||
_log("DEBUG", f"Created tvshow NFO: {tvshow_nfo}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Could not create tvshow.nfo in {series_dir}: {e}")
|
||||
|
||||
def remove_long_episode_nfos(self, season_dir: Path) -> int:
|
||||
"""Remove long-format episode NFO files (e.g., Show.Name.S01E01.nfo)"""
|
||||
count = 0
|
||||
for nfo_file in season_dir.glob("*.nfo"):
|
||||
if self.LONG_NFO_PATTERN.match(nfo_file.name):
|
||||
try:
|
||||
nfo_file.unlink()
|
||||
count += 1
|
||||
_log("DEBUG", f"Removed long NFO: {nfo_file}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to remove long NFO {nfo_file}: {e}")
|
||||
return count
|
||||
|
||||
def remove_orphaned_nfos(self, season_dir: Path, existing_episodes: set) -> int:
|
||||
"""Remove NFO files that don't have corresponding video files"""
|
||||
count = 0
|
||||
for nfo_file in season_dir.glob("S??E??.nfo"):
|
||||
match = self.SHORT_NFO_PATTERN.match(nfo_file.name)
|
||||
if match:
|
||||
nfo_season = int(match.group(1))
|
||||
nfo_episode = int(match.group(2))
|
||||
|
||||
# Check if we have a video file for this episode
|
||||
if (nfo_season, nfo_episode) not in existing_episodes:
|
||||
try:
|
||||
nfo_file.unlink()
|
||||
count += 1
|
||||
_log("DEBUG", f"Removed orphaned NFO: {nfo_file}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to remove orphaned NFO {nfo_file}: {e}")
|
||||
return count
|
||||
|
||||
@staticmethod
|
||||
def set_file_mtime(file_path: Path, iso_timestamp: str):
|
||||
"""Set file modification time to match the timestamp"""
|
||||
try:
|
||||
timestamp = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")).timestamp()
|
||||
os.utime(file_path, (timestamp, timestamp), follow_symlinks=False)
|
||||
_log("DEBUG", f"Updated mtime: {file_path} -> {iso_timestamp}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to set mtime on {file_path}: {e}")
|
||||
|
||||
def set_directory_mtime(self, dir_path: Path, iso_timestamp: str):
|
||||
"""Set directory modification time"""
|
||||
try:
|
||||
timestamp = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")).timestamp()
|
||||
os.utime(dir_path, (timestamp, timestamp), follow_symlinks=False)
|
||||
_log("DEBUG", f"Updated directory mtime: {dir_path} -> {iso_timestamp}")
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to set mtime on directory {dir_path}: {e}")
|
||||
|
||||
def update_movie_files_mtime(self, movie_dir: Path, iso_timestamp: str):
|
||||
"""Update modification time for all files in movie directory"""
|
||||
if not iso_timestamp or iso_timestamp == "MANUAL_REVIEW_NEEDED":
|
||||
return
|
||||
|
||||
try:
|
||||
timestamp = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")).timestamp()
|
||||
|
||||
files_updated = 0
|
||||
for file_path in movie_dir.iterdir():
|
||||
if file_path.is_file():
|
||||
try:
|
||||
os.utime(file_path, (timestamp, timestamp), follow_symlinks=False)
|
||||
files_updated += 1
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to update mtime on {file_path.name}: {e}")
|
||||
|
||||
# Update directory timestamp
|
||||
os.utime(movie_dir, (timestamp, timestamp), follow_symlinks=False)
|
||||
_log("INFO", f"Updated {files_updated} files and directory mtime: {movie_dir.name} -> {iso_timestamp}")
|
||||
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Failed to update movie directory mtimes: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Test the NFO manager
|
||||
nfo_manager = NFOManager("NFOGuard-Test")
|
||||
|
||||
# Test movie NFO creation
|
||||
test_movie_dir = Path("/tmp/test_movie")
|
||||
test_movie_dir.mkdir(exist_ok=True)
|
||||
|
||||
nfo_manager.create_movie_nfo(
|
||||
test_movie_dir,
|
||||
"tt1234567",
|
||||
"2025-07-07T12:00:00+00:00",
|
||||
"2020-01-01T00:00:00+00:00",
|
||||
"test:movie_creation"
|
||||
)
|
||||
|
||||
# Test episode NFO creation
|
||||
test_season_dir = Path("/tmp/test_series/Season 01")
|
||||
test_season_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
nfo_manager.create_episode_nfo(
|
||||
test_season_dir,
|
||||
1, 1,
|
||||
"2020-01-01T00:00:00+00:00",
|
||||
"2025-07-07T12:00:00+00:00",
|
||||
"test:episode_creation"
|
||||
)
|
||||
|
||||
print("NFO test files created in /tmp/")
|
||||
@@ -0,0 +1,273 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Path mapping module for NFOGuard
|
||||
Handles path translation between container, Radarr/Sonarr, and download client paths
|
||||
"""
|
||||
import os
|
||||
from pathlib import Path
|
||||
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 = source_path.lower()
|
||||
|
||||
# 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}")
|
||||
Reference in New Issue
Block a user