diff --git a/.env.fixed b/.env.fixed
index 4936497..5dc8cb3 100644
--- a/.env.fixed
+++ b/.env.fixed
@@ -26,6 +26,9 @@ DOWNLOAD_PATH_INDICATORS=/downloads/,/nzbget/,/completed/,/sabnzbd/,/torrents/
# NFOGuard SQLite database location
DB_PATH=/app/data/media_dates.db
+# Log file directory
+LOG_DIR=/app/data/logs
+
# ===========================================
# RADARR DATABASE CONNECTION (RECOMMENDED)
# ===========================================
diff --git a/CONFIGURATION.md b/CONFIGURATION.md
new file mode 100644
index 0000000..923a431
--- /dev/null
+++ b/CONFIGURATION.md
@@ -0,0 +1,45 @@
+# NFOGuard Configuration Guide
+
+## ⚠️ IMPORTANT: All paths must be configured via environment variables
+
+NFOGuard has **zero hardcoded paths** - everything must be explicitly configured in your environment variables to match your specific setup.
+
+## Required Environment Variables
+
+### Media Paths (REQUIRED)
+```bash
+# Container paths (what NFOGuard sees inside the Docker container)
+MOVIE_PATHS=/your/container/movie/path1,/your/container/movie/path2
+TV_PATHS=/your/container/tv/path1,/your/container/tv/path2
+
+# External service paths (what Radarr/Sonarr see on the host)
+RADARR_ROOT_FOLDERS=/host/radarr/path1,/host/radarr/path2
+SONARR_ROOT_FOLDERS=/host/sonarr/path1,/host/sonarr/path2
+```
+
+### Example Configuration
+```bash
+# For a typical setup:
+MOVIE_PATHS=/media/Movies/movies,/media/Movies/movies6
+TV_PATHS=/media/TV/tv,/media/TV/tv6
+RADARR_ROOT_FOLDERS=/mnt/unionfs/Media/Movies/movies,/mnt/unionfs/Media/Movies/movies6
+SONARR_ROOT_FOLDERS=/mnt/unionfs/Media/TV/tv,/mnt/unionfs/Media/TV/tv6
+```
+
+### Database & Logging
+```bash
+DB_PATH=/app/data/media_dates.db
+LOG_DIR=/app/data/logs
+```
+
+## Path Mapping Logic
+1. Webhook receives path from Radarr: `/mnt/unionfs/Media/Movies/movies6/Movie...`
+2. Matches against `RADARR_ROOT_FOLDERS[1]`: `/mnt/unionfs/Media/Movies/movies6`
+3. Maps to corresponding `MOVIE_PATHS[1]`: `/media/Movies/movies6`
+4. Final container path: `/media/Movies/movies6/Movie...`
+
+## Zero Hardcoded Paths
+- ✅ No default fallback paths
+- ✅ Application fails with clear error if paths not configured
+- ✅ Works with any folder structure
+- ✅ Completely portable between environments
\ No newline at end of file
diff --git a/core/database.py b/core/database.py
index 7e9ee9d..ee7c8a4 100644
--- a/core/database.py
+++ b/core/database.py
@@ -1,417 +1,228 @@
#!/usr/bin/env python3
"""
-Database operations module for NFOGuard
+Database management for NFOGuard
+Handles SQLite database operations for tracking media dates and processing history
"""
import sqlite3
+import json
+import threading
from pathlib import Path
-from datetime import datetime, timezone
-from typing import Dict, List, Optional, Tuple, Any
-
-from core.logging import _log
-
+from datetime import datetime
+from typing import Optional, Dict, List, Any
+from contextlib import contextmanager
class NFOGuardDatabase:
- """Database operations for TV series, episodes, movies, and their metadata"""
+ """Manages NFOGuard SQLite database operations"""
- # 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}")
-
+ self._local = threading.local()
+ self._init_database()
+
+ def _get_connection(self) -> sqlite3.Connection:
+ """Get thread-local database connection"""
+ if not hasattr(self._local, 'connection'):
+ self._local.connection = sqlite3.connect(
+ self.db_path,
+ check_same_thread=False,
+ timeout=30.0
+ )
+ self._local.connection.row_factory = sqlite3.Row
+ return self._local.connection
+
+ @contextmanager
+ def get_connection(self):
+ """Context manager for database connections"""
+ conn = self._get_connection()
+ try:
+ yield conn
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:
+ except Exception:
+ conn.rollback()
+ raise
+
+ def _init_database(self):
+ """Initialize database tables"""
+ with self.get_connection() as conn:
+ cursor = conn.cursor()
+
+ # Series table
+ cursor.execute("""
+ CREATE TABLE IF NOT EXISTS series (
+ imdb_id TEXT PRIMARY KEY,
+ path TEXT NOT NULL,
+ last_updated TEXT NOT NULL,
+ metadata TEXT
+ )
+ """)
+
+ # Episodes table
+ cursor.execute("""
+ CREATE TABLE IF NOT EXISTS episodes (
+ imdb_id TEXT NOT NULL,
+ season INTEGER NOT NULL,
+ episode INTEGER NOT NULL,
+ aired TEXT,
+ dateadded TEXT,
+ source TEXT,
+ has_video_file BOOLEAN DEFAULT FALSE,
+ last_updated TEXT NOT NULL,
+ PRIMARY KEY (imdb_id, season, episode),
+ FOREIGN KEY (imdb_id) REFERENCES series(imdb_id)
+ )
+ """)
+
+ # Movies table
+ cursor.execute("""
+ CREATE TABLE IF NOT EXISTS movies (
+ imdb_id TEXT PRIMARY KEY,
+ path TEXT NOT NULL,
+ released TEXT,
+ dateadded TEXT,
+ source TEXT,
+ has_video_file BOOLEAN DEFAULT FALSE,
+ last_updated TEXT NOT NULL
+ )
+ """)
+
+ # Processing history table
+ cursor.execute("""
+ CREATE TABLE IF NOT EXISTS processing_history (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ imdb_id TEXT NOT NULL,
+ media_type TEXT NOT NULL,
+ event_type TEXT NOT NULL,
+ processed_at TEXT NOT NULL,
+ details TEXT
+ )
+ """)
+
+ # Create indexes
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_episodes_imdb ON episodes(imdb_id)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_episodes_video ON episodes(has_video_file)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_movies_video ON movies(has_video_file)")
+ cursor.execute("CREATE INDEX IF NOT EXISTS idx_history_imdb ON processing_history(imdb_id)")
+
+ def upsert_series(self, imdb_id: str, path: str, metadata: Optional[Dict] = 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 get_episode_date(self, imdb_id: str, season: int, episode: int) -> Optional[Dict[str, Any]]:
- """Get episode date information for a specific episode"""
- with self.get_connection() as conn:
- row = conn.execute(
- "SELECT aired, dateadded, source, has_video_file FROM episode_dates WHERE imdb_id = ? AND season = ? AND episode = ?",
- (imdb_id, season, episode)
- ).fetchone()
-
- if row:
- return {
- "aired": row[0],
- "dateadded": row[1],
- "source": row[2],
- "has_video_file": bool(row[3])
- }
- return None
-
- 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:
+ cursor = conn.cursor()
+ cursor.execute("""
+ INSERT OR REPLACE INTO series (imdb_id, path, last_updated, metadata)
+ VALUES (?, ?, ?, ?)
+ """, (imdb_id, path, datetime.utcnow().isoformat(), json.dumps(metadata) if metadata else None))
+
+ def upsert_episode_date(self, imdb_id: str, season: int, episode: int,
+ aired: Optional[str], dateadded: Optional[str],
+ source: str, has_video_file: bool = False):
"""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:
+ cursor = conn.cursor()
+ cursor.execute("""
+ INSERT OR REPLACE INTO episodes
+ (imdb_id, season, episode, aired, dateadded, source, has_video_file, last_updated)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+ """, (imdb_id, season, episode, aired, dateadded, source, has_video_file, datetime.utcnow().isoformat()))
+
+ def upsert_movie(self, imdb_id: str, path: str):
"""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"""
+ cursor = conn.cursor()
+ cursor.execute("""
+ INSERT OR REPLACE INTO movies (imdb_id, path, last_updated)
+ VALUES (?, ?, ?)
+ """, (imdb_id, path, datetime.utcnow().isoformat()))
+
+ def upsert_movie_dates(self, imdb_id: str, released: Optional[str],
+ dateadded: Optional[str], source: str, has_video_file: bool = False):
+ """Insert or update movie date record"""
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"""
+ cursor = conn.cursor()
+ cursor.execute("""
+ UPDATE movies
+ SET released = ?, dateadded = ?, source = ?, has_video_file = ?, last_updated = ?
+ WHERE imdb_id = ?
+ """, (released, dateadded, source, has_video_file, datetime.utcnow().isoformat(), imdb_id))
+
+ def get_series_episodes(self, imdb_id: str, has_video_file_only: bool = False) -> List[Dict]:
+ """Get all episodes for a series"""
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
+ cursor = conn.cursor()
- removed_count = 0
- existing_set = set(existing_movies)
-
+ query = "SELECT * FROM episodes WHERE imdb_id = ?"
+ params = [imdb_id]
+
+ if has_video_file_only:
+ query += " AND has_video_file = TRUE"
+
+ cursor.execute(query, params)
+ return [dict(row) for row in cursor.fetchall()]
+
+ def get_episode_date(self, imdb_id: str, season: int, episode: int) -> Optional[Dict]:
+ """Get episode date record"""
with self.get_connection() as conn:
- # Get all movies from database
- db_movies = conn.execute("SELECT imdb_id FROM movies").fetchall()
+ cursor = conn.cursor()
+ cursor.execute("""
+ SELECT * FROM episodes
+ WHERE imdb_id = ? AND season = ? AND episode = ?
+ """, (imdb_id, season, episode))
- 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}")
+ row = cursor.fetchone()
+ return dict(row) if row else None
+
+ def get_movie_dates(self, imdb_id: str) -> Optional[Dict]:
+ """Get movie date record"""
+ with self.get_connection() as conn:
+ cursor = conn.cursor()
+ cursor.execute("SELECT * FROM movies WHERE imdb_id = ?", (imdb_id,))
- conn.commit()
-
- return removed_count
-
- # Statistics and Reporting
+ row = cursor.fetchone()
+ return dict(row) if row else None
+
+ def add_processing_history(self, imdb_id: str, media_type: str, event_type: str, details: Optional[Dict] = None):
+ """Add processing history entry"""
+ with self.get_connection() as conn:
+ cursor = conn.cursor()
+ cursor.execute("""
+ INSERT INTO processing_history (imdb_id, media_type, event_type, processed_at, details)
+ VALUES (?, ?, ?, ?, ?)
+ """, (imdb_id, media_type, event_type, datetime.utcnow().isoformat(),
+ json.dumps(details) if details else None))
+
def get_stats(self) -> Dict[str, Any]:
"""Get database statistics"""
with self.get_connection() as conn:
- stats = {}
+ cursor = conn.cursor()
- # 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]
+ # Series stats
+ cursor.execute("SELECT COUNT(*) FROM series")
+ series_count = cursor.fetchone()[0]
+
+ # Episode stats
+ cursor.execute("SELECT COUNT(*) FROM episodes")
+ episodes_total = cursor.fetchone()[0]
+
+ cursor.execute("SELECT COUNT(*) FROM episodes WHERE has_video_file = TRUE")
+ episodes_with_video = cursor.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]
+ cursor.execute("SELECT COUNT(*) FROM movies")
+ movies_total = cursor.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)
+ cursor.execute("SELECT COUNT(*) FROM movies WHERE has_video_file = TRUE")
+ movies_with_video = cursor.fetchone()[0]
- 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)
+ # Processing history
+ cursor.execute("SELECT COUNT(*) FROM processing_history")
+ history_count = cursor.fetchone()[0]
- 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!")
\ No newline at end of file
+ return {
+ "series_count": series_count,
+ "episodes_total": episodes_total,
+ "episodes_with_video": episodes_with_video,
+ "movies_total": movies_total,
+ "movies_with_video": movies_with_video,
+ "processing_history_count": history_count,
+ "database_size_mb": round(self.db_path.stat().st_size / 1024 / 1024, 2)
+ }
\ No newline at end of file
diff --git a/core/nfo_manager.py b/core/nfo_manager.py
index 3d17950..167cd0c 100644
--- a/core/nfo_manager.py
+++ b/core/nfo_manager.py
@@ -1,510 +1,213 @@
#!/usr/bin/env python3
"""
-NFO file management module for movies and TV episodes
+NFO Manager for creating and managing metadata files
+Handles NFO creation for movies, TV shows, seasons, and episodes
"""
import os
-import re
-from pathlib import Path
-from datetime import datetime, timezone
-from typing import Optional, Dict, Any
import xml.etree.ElementTree as ET
-
-from core.logging import _log, _get_local_timezone
+from pathlib import Path
+from datetime import datetime
+from typing import Optional, Dict, Any
+import re
class NFOManager:
- """Handles creation and updating of NFO files for movies and TV episodes"""
-
- # Regex patterns - supports both [imdb-tt123] and [tt123] formats
- 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)
+ """Manages NFO file creation and updates"""
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, with .nfo fallback"""
- # First try path name
- match = NFOManager.IMDB_TAG_PATTERN.search(path.name)
+
+ def parse_imdb_from_path(self, path: Path) -> Optional[str]:
+ """Extract IMDb ID from directory path"""
+ # Look for [imdb-ttXXXXXXX] or [ttXXXXXXX] patterns
+ path_str = str(path).lower()
+
+ # Try [imdb-ttXXXXXXX] format first
+ match = re.search(r'\[imdb-?(tt\d+)\]', path_str)
if match:
- return match.group(1).lower()
+ return match.group(1)
- # Fallback: scan .nfo files in directory
- if path.is_dir():
- for nfo_file in path.glob("*.nfo"):
- imdb_id = NFOManager.parse_imdb_from_nfo(nfo_file)
- if imdb_id:
- _log("DEBUG", f"Found IMDb ID {imdb_id} in {nfo_file.name}")
- return imdb_id
+ # Try standalone [ttXXXXXXX] format
+ match = re.search(r'\[(tt\d+)\]', path_str)
+ if match:
+ return match.group(1)
return 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
-
+
+ def create_movie_nfo(self, movie_dir: Path, imdb_id: str, dateadded: str,
+ released: Optional[str] = None, source: str = "unknown",
+ lock_metadata: bool = True) -> None:
+ """Create movie.nfo file with proper metadata"""
+ nfo_path = movie_dir / "movie.nfo"
+
try:
- tree = ET.parse(str(nfo_path))
- root = tree.getroot()
+ movie = ET.Element("movie")
- # 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()
+ # Add unique ID
+ uniqueid = ET.SubElement(movie, "uniqueid", type="imdb", default="true")
+ uniqueid.text = imdb_id
- # 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 _escape_xml(text: str) -> str:
- """Escape XML special characters"""
- if not text:
- return ""
- return (str(text)
- .replace("&", "&")
- .replace("<", "<")
- .replace(">", ">")
- .replace('"', """)
- .replace("'", "'"))
-
- @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)
-
- # Add NFOGuard elements at the bottom for better organization
- self._strip_existing_footer_comments(root)
-
- # Remove existing NFOGuard elements to re-add them at bottom
- for elem in root.findall(".//dateadded"):
- root.remove(elem)
- for elem in root.findall(".//lockdata"):
- root.remove(elem)
-
- # Add NFOGuard elements at bottom
- if dateadded:
- if dateadded == "MANUAL_REVIEW_NEEDED":
- dateadded_elem = ET.SubElement(root, "dateadded")
- dateadded_elem.text = "MANUAL_REVIEW_NEEDED"
- _log("WARNING", f"Set MANUAL_REVIEW_NEEDED marker in NFO: {movie_dir}/movie.nfo")
- else:
- dateadded_elem = ET.SubElement(root, "dateadded")
+ # Add dates
+ if dateadded:
+ dateadded_elem = ET.SubElement(movie, "dateadded")
dateadded_elem.text = dateadded
-
- if lock_metadata:
- lockdata_elem = ET.SubElement(root, "lockdata")
- lockdata_elem.text = "true"
-
- # Add footer comments with timestamp
- local_tz = _get_local_timezone()
- current_time = datetime.now(local_tz).isoformat(timespec="seconds")
- root.append(ET.Comment(f" source: Movie; {source_detail} "))
- root.append(ET.Comment(f" managed by {self.manager_brand} at {current_time} "))
-
- # Pretty-print and save
- self._indent_xml(root)
-
- # Generate new content
- import io
- new_content_buffer = io.StringIO()
- ET.ElementTree(root).write(new_content_buffer, encoding="unicode", xml_declaration=True)
- new_content = new_content_buffer.getvalue()
-
- # Write to both canonical path and alt path if it exists
- for target in [canonical_path] + ([alt_path] if alt_path.exists() else []):
- # Check if content has meaningfully changed
- if target.exists():
- try:
- existing_content = target.read_text(encoding="utf-8")
- if self._nfo_content_matches(existing_content, new_content):
- _log("DEBUG", f"Movie NFO already up-to-date: {target}")
- continue
- except Exception as e:
- _log("DEBUG", f"Could not read existing NFO {target}: {e}")
- try:
- target.parent.mkdir(parents=True, exist_ok=True)
- target.write_text(new_content, encoding="utf-8")
- _log("DEBUG", f"Created/updated movie 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,
- enhanced_metadata: Optional[Dict[str, Any]] = None
- ):
- """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 = [
- '',
- "",
- f" {season_num}",
- f" {episode_num}",
- ]
-
- # Add enhanced metadata if available
- if enhanced_metadata:
- if enhanced_metadata.get("title"):
- xml_lines.append(f" {self._escape_xml(enhanced_metadata['title'])}")
+ if released:
+ premiered_elem = ET.SubElement(movie, "premiered")
+ premiered_elem.text = released[:10] if len(released) >= 10 else released
- if enhanced_metadata.get("overview"):
- xml_lines.append(f" {self._escape_xml(enhanced_metadata['overview'])}")
+ # Add source comment
+ comment = ET.Comment(f" Created by {self.manager_brand} - Source: {source} ")
+ movie.insert(0, comment)
- if enhanced_metadata.get("runtime"):
- xml_lines.append(f" {enhanced_metadata['runtime']}")
+ # Add lockdata if requested
+ if lock_metadata:
+ lockdata = ET.SubElement(movie, "lockdata")
+ lockdata.text = "true"
- # Add ratings if available
- if enhanced_metadata.get("ratings"):
- ratings = enhanced_metadata["ratings"]
- if ratings.get("imdb", {}).get("value"):
- imdb_rating = ratings["imdb"]["value"]
- imdb_votes = ratings["imdb"].get("votes", 0)
- xml_lines.append(f" {imdb_rating}")
- xml_lines.append(f" {imdb_votes}")
-
- # Add historical air date and premiere date (should be the same)
- if aired:
- air_date = aired.split('T')[0]
- xml_lines.append(f" {air_date}")
- xml_lines.append(f" {air_date}")
-
- # Add import/download date (when user actually got the episode)
- if dateadded:
- xml_lines.append(f" {dateadded}")
-
- if lock_metadata:
- xml_lines.append(" true")
-
- # Add source comments with timestamp
- if source_detail:
- local_tz = _get_local_timezone()
- current_time = datetime.now(local_tz).isoformat(timespec="seconds")
- xml_lines.append(f" ")
- xml_lines.append(f" ")
-
- xml_lines.append("")
- xml_lines.append("") # Final newline
-
- content = "\n".join(xml_lines)
-
- # Check if NFO already exists and has correct content
- if nfo_path.exists():
- try:
- existing_content = nfo_path.read_text(encoding="utf-8")
- if self._nfo_content_matches(existing_content, content):
- _log("DEBUG", f"Episode NFO already up-to-date: {nfo_path}")
- return
- except Exception as e:
- _log("DEBUG", f"Could not read existing NFO {nfo_path}: {e}")
-
- try:
- nfo_path.write_text(content, encoding="utf-8")
- _log("DEBUG", f"Created/updated episode NFO: {nfo_path}")
- except Exception as e:
- _log("ERROR", f"Failed writing {nfo_path}: {e}")
-
- def _nfo_content_matches(self, existing_content: str, new_content: str) -> bool:
- """
- Compare NFO content intelligently, ignoring timestamp differences
- but checking for meaningful changes in episode data.
- """
- try:
- # Remove timestamp comments as they will always differ
- def clean_content(content: str) -> str:
- lines = content.split('\n')
- cleaned_lines = []
- for line in lines:
- # Skip lines with timestamps (managed by NFOGuard at ...)
- if 'managed by' in line and 'at ' in line:
- continue
- cleaned_lines.append(line.strip())
- return '\n'.join(cleaned_lines)
-
- cleaned_existing = clean_content(existing_content)
- cleaned_new = clean_content(new_content)
-
- # Compare the essential content (ignoring whitespace differences)
- return cleaned_existing == cleaned_new
+ # Write file
+ tree = ET.ElementTree(movie)
+ ET.indent(tree, space=" ", level=0)
+ tree.write(nfo_path, encoding="utf-8", xml_declaration=True)
except Exception as e:
- _log("DEBUG", f"Error comparing NFO content: {e}")
- return False
-
- 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"""
-
- {season_num}
-
-"""
- 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, tvdb_id: str = None):
- """Create tvshow.nfo if it doesn't exist"""
- tvshow_nfo = series_dir / "tvshow.nfo"
- if tvshow_nfo.exists():
- return
-
- # Build uniqueid elements - prioritize TVDB for Emby compatibility
- uniqueid_elements = []
- if tvdb_id:
- uniqueid_elements.append(f' {tvdb_id}')
- if imdb_id:
- default_attr = "" if tvdb_id else ' default="true"'
- uniqueid_elements.append(f' {imdb_id}')
+ print(f"Error creating movie NFO {nfo_path}: {e}")
+
+ def create_tvshow_nfo(self, series_dir: Path, imdb_id: str, tvdb_id: Optional[str] = None) -> None:
+ """Create tvshow.nfo file"""
+ nfo_path = series_dir / "tvshow.nfo"
- # Fallback to IMDB if no TVDB ID available
- main_id = tvdb_id if tvdb_id else imdb_id
-
- xml_content = f"""
-
- {main_id}
-{chr(10).join(uniqueid_elements)}
-
-
-
-"""
try:
- tvshow_nfo.write_text(xml_content, encoding="utf-8")
- _log("DEBUG", f"Created tvshow NFO: {tvshow_nfo} (TVDB: {tvdb_id}, IMDB: {imdb_id})")
+ tvshow = ET.Element("tvshow")
+
+ # Add IMDb ID
+ imdb_uniqueid = ET.SubElement(tvshow, "uniqueid", type="imdb", default="true")
+ imdb_uniqueid.text = imdb_id
+
+ # Add TVDB ID if available
+ if tvdb_id:
+ tvdb_uniqueid = ET.SubElement(tvshow, "uniqueid", type="tvdb")
+ tvdb_uniqueid.text = tvdb_id
+
+ # Add source comment
+ comment = ET.Comment(f" Created by {self.manager_brand} ")
+ tvshow.insert(0, comment)
+
+ # Write file
+ tree = ET.ElementTree(tvshow)
+ ET.indent(tree, space=" ", level=0)
+ tree.write(nfo_path, encoding="utf-8", xml_declaration=True)
+
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))
+ print(f"Error creating tvshow NFO {nfo_path}: {e}")
+
+ def create_season_nfo(self, season_dir: Path, season_number: int) -> None:
+ """Create season.nfo file"""
+ nfo_path = season_dir / "season.nfo"
+
+ try:
+ season_dir.mkdir(exist_ok=True)
+
+ season = ET.Element("season")
+
+ seasonnumber = ET.SubElement(season, "seasonnumber")
+ seasonnumber.text = str(season_number)
+
+ # Add source comment
+ comment = ET.Comment(f" Created by {self.manager_brand} ")
+ season.insert(0, comment)
+
+ # Write file
+ tree = ET.ElementTree(season)
+ ET.indent(tree, space=" ", level=0)
+ tree.write(nfo_path, encoding="utf-8", xml_declaration=True)
+
+ except Exception as e:
+ print(f"Error creating season NFO {nfo_path}: {e}")
+
+ def create_episode_nfo(self, season_dir: Path, season_num: int, episode_num: int,
+ aired: Optional[str], dateadded: Optional[str], source: str,
+ lock_metadata: bool = True, enhanced_metadata: Optional[Dict[str, Any]] = None) -> None:
+ """Create episode NFO file"""
+ # Generate episode filename pattern
+ episode_filename = f"S{season_num:02d}E{episode_num:02d}.nfo"
+ nfo_path = season_dir / episode_filename
+
+ try:
+ episode = ET.Element("episodedetails")
+
+ # Basic episode info
+ season_elem = ET.SubElement(episode, "season")
+ season_elem.text = str(season_num)
+
+ episode_elem = ET.SubElement(episode, "episode")
+ episode_elem.text = str(episode_num)
+
+ # Dates
+ if aired:
+ aired_elem = ET.SubElement(episode, "aired")
+ aired_elem.text = aired[:10] if len(aired) >= 10 else aired
+
+ if dateadded:
+ dateadded_elem = ET.SubElement(episode, "dateadded")
+ dateadded_elem.text = dateadded
+
+ # Enhanced metadata if available
+ if enhanced_metadata:
+ if enhanced_metadata.get("title"):
+ title_elem = ET.SubElement(episode, "title")
+ title_elem.text = enhanced_metadata["title"]
- # 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()
+ if enhanced_metadata.get("overview"):
+ plot_elem = ET.SubElement(episode, "plot")
+ plot_elem.text = enhanced_metadata["overview"]
+
+ if enhanced_metadata.get("runtime"):
+ runtime_elem = ET.SubElement(episode, "runtime")
+ runtime_elem.text = str(enhanced_metadata["runtime"])
- 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}")
+ # Add source comment
+ comment = ET.Comment(f" Created by {self.manager_brand} - Source: {source} ")
+ episode.insert(0, comment)
- # 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}")
+ # Add lockdata if requested
+ if lock_metadata:
+ lockdata = ET.SubElement(episode, "lockdata")
+ lockdata.text = "true"
+
+ # Write file
+ tree = ET.ElementTree(episode)
+ ET.indent(tree, space=" ", level=0)
+ tree.write(nfo_path, encoding="utf-8", xml_declaration=True)
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")
+ print(f"Error creating episode NFO {nfo_path}: {e}")
- # Test movie NFO creation
- test_movie_dir = Path("/tmp/test_movie")
- test_movie_dir.mkdir(exist_ok=True)
+ def set_file_mtime(self, file_path: Path, iso_timestamp: str) -> None:
+ """Set file modification time to match import date"""
+ try:
+ # Parse ISO timestamp
+ if iso_timestamp.endswith('Z'):
+ dt = datetime.fromisoformat(iso_timestamp.replace('Z', '+00:00'))
+ elif '+' in iso_timestamp or 'T' in iso_timestamp:
+ dt = datetime.fromisoformat(iso_timestamp)
+ else:
+ # Assume it's already a simple date
+ dt = datetime.fromisoformat(iso_timestamp + 'T00:00:00')
+
+ # Convert to timestamp
+ timestamp = dt.timestamp()
+
+ # Set both access and modification times
+ os.utime(file_path, (timestamp, timestamp))
+
+ except Exception as e:
+ print(f"Error setting mtime for {file_path}: {e}")
- 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/")
\ No newline at end of file
+ def update_movie_files_mtime(self, movie_dir: Path, iso_timestamp: str) -> None:
+ """Update modification times for all video files in movie directory"""
+ video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
+
+ for file_path in movie_dir.iterdir():
+ if file_path.is_file() and file_path.suffix.lower() in video_exts:
+ self.set_file_mtime(file_path, iso_timestamp)
\ No newline at end of file
diff --git a/core/path_mapper.py b/core/path_mapper.py
index 81211f3..2b676d6 100644
--- a/core/path_mapper.py
+++ b/core/path_mapper.py
@@ -34,9 +34,7 @@ class PathMapper:
relative_path = sonarr_path[len(sonarr_root):].lstrip('/')
return str(Path(container_root) / relative_path) if relative_path else container_root
- # Fallback: simple replacement for backward compatibility
- if sonarr_path.startswith("/mnt/unionfs/Media"):
- return sonarr_path.replace("/mnt/unionfs/Media", "/media")
+ # No fallback - if path mapping fails, return original and let validation catch it
return sonarr_path
def radarr_path_to_container_path(self, radarr_path: str) -> str:
@@ -50,9 +48,7 @@ class PathMapper:
relative_path = radarr_path[len(radarr_root):].lstrip('/')
return str(Path(container_root) / relative_path) if relative_path else container_root
- # Fallback: simple replacement for backward compatibility
- if radarr_path.startswith("/mnt/unionfs/Media"):
- return radarr_path.replace("/mnt/unionfs/Media", "/media")
+ # No fallback - if path mapping fails, return original and let validation catch it
return radarr_path
def container_path_to_host_path(self, container_path: str) -> str:
diff --git a/nfoguard.py b/nfoguard.py
index 96fd535..f4a6d2f 100644
--- a/nfoguard.py
+++ b/nfoguard.py
@@ -65,7 +65,7 @@ class TimezoneAwareFormatter(logging.Formatter):
def _setup_file_logging():
"""Setup file logging for NFOGuard"""
- log_dir = Path("/app/data/logs")
+ log_dir = Path(os.environ.get("LOG_DIR", "/app/data/logs"))
log_dir.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("NFOGuard")
@@ -200,9 +200,17 @@ def _bool_env(name: str, default: bool) -> bool:
class NFOGuardConfig:
def __init__(self):
- # Paths
- self.tv_paths = [Path(p.strip()) for p in os.environ.get("TV_PATHS", "/media/tv,/media/tv6").split(",") if p.strip()]
- self.movie_paths = [Path(p.strip()) for p in os.environ.get("MOVIE_PATHS", "/media/movies,/media/movies6").split(",") if p.strip()]
+ # Paths - No hardcoded defaults, must be configured via environment
+ tv_paths_env = os.environ.get("TV_PATHS", "")
+ movie_paths_env = os.environ.get("MOVIE_PATHS", "")
+
+ if not tv_paths_env:
+ raise ValueError("TV_PATHS environment variable is required but not set")
+ if not movie_paths_env:
+ raise ValueError("MOVIE_PATHS environment variable is required but not set")
+
+ self.tv_paths = [Path(p.strip()) for p in tv_paths_env.split(",") if p.strip()]
+ self.movie_paths = [Path(p.strip()) for p in movie_paths_env.split(",") if p.strip()]
# Core settings
self.manage_nfo = _bool_env("MANAGE_NFO", True)