Add comprehensive TV show enhancements and URL-safe endpoints

✨ Features Added: • Single season/episode processing capability via new endpoints • Enhanced NFO generation with full Sonarr API metadata (titles, plots, ratings) • NFOGuard timestamp tracking in all NFO source comments • URL-safe /tv/scan-season and /tv/scan-episode endpoints 🔧 Technical Improvements: • Fixed URL encoding issues for paths with spaces and special characters • Enhanced TVProcessor with process_season() and process_episode_file() methods • Rich metadata extraction from Sonarr API (episode titles, plots, runtime, ratings) • XML escaping for special characters in metadata • Comprehensive episode parsing (SxxExx and numeric formats) 📚 Documentation: • Updated README with new TV processing endpoints • Added enhanced NFO generation examples showing before/after • Documented URL-safe alternatives to manual scan paths 🎯 User Benefits: • Can now process single seasons: Chicago Fire Season 13 • Can now process single episodes: S13E21 files • Enhanced NFO files with full episode metadata for better Emby/Plex experience • Timestamps show when NFOGuard processed each file
2025-09-11 09:39:42 -04:00
parent c7c783570c
commit c705fb4d6d
3 changed files with 426 additions and 12 deletions
@@ -416,6 +416,230 @@ class TVProcessor:
        except Exception:
            return None

+    def process_season(self, series_path: Path, season_path: Path) -> None:
+        """Process a specific TV season directory"""
+        imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
+        if not imdb_id:
+            _log("ERROR", f"No IMDb ID found in series path: {series_path}")
+            return
+        
+        # Extract season number from path
+        season_match = re.search(r'season\s*(\d+)', season_path.name, re.IGNORECASE)
+        if not season_match:
+            _log("ERROR", f"Could not extract season number from: {season_path.name}")
+            return
+        
+        season_num = int(season_match.group(1))
+        _log("INFO", f"Processing TV season {season_num}: {season_path}")
+        
+        # Update database
+        self.db.upsert_series(imdb_id, str(series_path))
+        
+        # Find video files in this season only
+        disk_episodes = self._find_season_episodes(season_path, season_num)
+        _log("INFO", f"Found {len(disk_episodes)} episodes in season {season_num}")
+        
+        # Get enhanced metadata from Sonarr
+        series_metadata = self._get_sonarr_series_metadata(imdb_id)
+        
+        # Get episode dates for this season
+        episode_dates = self._gather_season_episode_dates(series_path, imdb_id, season_num, disk_episodes, series_metadata)
+        
+        # Process episodes
+        for (season, episode), (aired, dateadded, source) in episode_dates.items():
+            if (season, episode) in disk_episodes:
+                # Get enhanced episode metadata
+                enhanced_metadata = self._get_episode_metadata(series_metadata, season, episode) if series_metadata else None
+                
+                # Create NFO
+                if config.manage_nfo:
+                    self.nfo_manager.create_episode_nfo(
+                        season_path,
+                        season, episode, aired, dateadded, source, config.lock_metadata,
+                        enhanced_metadata
+                    )
+                
+                # Update file mtimes
+                if config.fix_dir_mtimes and dateadded:
+                    video_files = disk_episodes[(season, episode)]
+                    for video_file in video_files:
+                        self.nfo_manager.set_file_mtime(video_file, dateadded)
+                
+                # Save to database
+                self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True)
+        
+        # Create season NFO
+        if config.manage_nfo:
+            self.nfo_manager.create_season_nfo(season_path, season_num)
+            self.nfo_manager.create_tvshow_nfo(series_path, imdb_id)
+        
+        _log("INFO", f"Completed processing season {season_num}")
+
+    def process_episode_file(self, series_path: Path, season_path: Path, episode_file: Path) -> None:
+        """Process a single TV episode file"""
+        imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
+        if not imdb_id:
+            _log("ERROR", f"No IMDb ID found in series path: {series_path}")
+            return
+        
+        # Parse episode info from filename
+        episode_info = self._parse_episode_from_filename(episode_file.name)
+        if not episode_info:
+            _log("ERROR", f"Could not parse episode info from: {episode_file.name}")
+            return
+        
+        season_num, episode_num = episode_info
+        _log("INFO", f"Processing single episode S{season_num:02d}E{episode_num:02d}: {episode_file.name}")
+        
+        # Update database
+        self.db.upsert_series(imdb_id, str(series_path))
+        
+        # Get enhanced metadata from Sonarr
+        series_metadata = self._get_sonarr_series_metadata(imdb_id)
+        enhanced_metadata = self._get_episode_metadata(series_metadata, season_num, episode_num) if series_metadata else None
+        
+        # Get episode date
+        aired, dateadded, source = self._get_single_episode_date(imdb_id, season_num, episode_num, series_metadata)
+        
+        # Create NFO
+        if config.manage_nfo and dateadded:
+            self.nfo_manager.create_episode_nfo(
+                season_path,
+                season_num, episode_num, aired, dateadded, source, config.lock_metadata,
+                enhanced_metadata
+            )
+        
+        # Update file mtime
+        if config.fix_dir_mtimes and dateadded:
+            self.nfo_manager.set_file_mtime(episode_file, dateadded)
+        
+        # Save to database
+        self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True)
+        
+        _log("INFO", f"Completed processing episode S{season_num:02d}E{episode_num:02d}")
+
+    def _find_season_episodes(self, season_path: Path, season_num: int) -> Dict[Tuple[int, int], List[Path]]:
+        """Find all episode video files in a single season"""
+        disk_episodes = {}
+        video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v", ".ts", ".m2ts")
+        
+        for file_path in season_path.iterdir():
+            if not file_path.is_file() or file_path.suffix.lower() not in video_exts:
+                continue
+            
+            episode_info = self._parse_episode_from_filename(file_path.name)
+            if episode_info and episode_info[0] == season_num:
+                season, episode = episode_info
+                key = (season, episode)
+                if key not in disk_episodes:
+                    disk_episodes[key] = []
+                disk_episodes[key].append(file_path)
+        
+        return disk_episodes
+
+    def _parse_episode_from_filename(self, filename: str) -> Optional[Tuple[int, int]]:
+        """Parse season and episode numbers from filename"""
+        # Try SxxExx format
+        match = re.search(r'S(\d{1,2})E(\d{1,2})', filename, re.IGNORECASE)
+        if match:
+            return int(match.group(1)), int(match.group(2))
+        
+        # Try season.episode format
+        match = re.search(r'(\d{1,2})\.(\d{1,2})', filename)
+        if match:
+            return int(match.group(1)), int(match.group(2))
+        
+        return None
+
+    def _get_sonarr_series_metadata(self, imdb_id: str) -> Optional[Dict[str, Any]]:
+        """Get enhanced series metadata from Sonarr API"""
+        try:
+            if not self.sonarr.enabled:
+                return None
+            
+            series = self.sonarr.series_by_imdb(imdb_id)
+            if not series:
+                _log("DEBUG", f"No Sonarr series found for IMDb {imdb_id}")
+                return None
+            
+            # Get episodes for the series
+            series_id = series.get("id")
+            if series_id:
+                episodes = self.sonarr.episodes_for_series(series_id)
+                series["episodes"] = episodes
+                _log("DEBUG", f"Got {len(episodes)} episodes from Sonarr for series {series.get('title')}")
+            
+            return series
+            
+        except Exception as e:
+            _log("ERROR", f"Error getting Sonarr metadata for {imdb_id}: {e}")
+            return None
+
+    def _get_episode_metadata(self, series_metadata: Dict[str, Any], season_num: int, episode_num: int) -> Optional[Dict[str, Any]]:
+        """Extract specific episode metadata from series data"""
+        if not series_metadata or "episodes" not in series_metadata:
+            return None
+        
+        for episode in series_metadata["episodes"]:
+            if episode.get("seasonNumber") == season_num and episode.get("episodeNumber") == episode_num:
+                return {
+                    "title": episode.get("title"),
+                    "overview": episode.get("overview"),
+                    "runtime": episode.get("runtime"),
+                    "ratings": episode.get("ratings", {})
+                }
+        
+        return None
+
+    def _gather_season_episode_dates(self, series_path: Path, imdb_id: str, season_num: int, disk_episodes: Dict[Tuple[int, int], List[Path]], series_metadata: Optional[Dict[str, Any]] = None) -> Dict[Tuple[int, int], Tuple[Optional[str], Optional[str], str]]:
+        """Get episode dates for a specific season"""
+        episode_dates = {}
+        
+        for (season, episode) in disk_episodes.keys():
+            if season != season_num:
+                continue
+            
+            aired, dateadded, source = self._get_single_episode_date(imdb_id, season, episode, series_metadata)
+            episode_dates[(season, episode)] = (aired, dateadded, source)
+        
+        return episode_dates
+
+    def _get_single_episode_date(self, imdb_id: str, season_num: int, episode_num: int, series_metadata: Optional[Dict[str, Any]] = None) -> Tuple[Optional[str], Optional[str], str]:
+        """Get date info for a single episode"""
+        # Try database first
+        existing = self.db.get_episode_date(imdb_id, season_num, episode_num)
+        if existing and existing.get("dateadded"):
+            return existing.get("aired"), existing["dateadded"], existing.get("source", "database")
+        
+        # Try Sonarr API if metadata available
+        if series_metadata and "episodes" in series_metadata:
+            for episode in series_metadata["episodes"]:
+                if episode.get("seasonNumber") == season_num and episode.get("episodeNumber") == episode_num:
+                    aired = episode.get("airDateUtc")
+                    if aired:
+                        return aired, aired, "sonarr:episode.airDateUtc"
+        
+        # Try Sonarr API history
+        if self.sonarr.enabled:
+            try:
+                # Get episode ID from Sonarr
+                series = self.sonarr.series_by_imdb(imdb_id)
+                if series:
+                    episodes = self.sonarr.episodes_for_series(series["id"])
+                    for ep in episodes:
+                        if ep.get("seasonNumber") == season_num and ep.get("episodeNumber") == episode_num:
+                            import_date = self.sonarr.get_episode_import_history(ep["id"])
+                            if import_date:
+                                return ep.get("airDateUtc"), import_date, "sonarr:history.import"
+                            elif ep.get("airDateUtc"):
+                                return ep.get("airDateUtc"), ep.get("airDateUtc"), "sonarr:episode.airDateUtc"
+            except Exception as e:
+                _log("DEBUG", f"Sonarr API error for episode S{season_num:02d}E{episode_num:02d}: {e}")
+        
+        # Fallback to current time
+        current_time = datetime.now(timezone.utc).isoformat(timespec="seconds")
+        return None, current_time, "fallback:current_time"
+

 class MovieProcessor:
    """Handles movie processing"""
@@ -1286,13 +1510,35 @@ async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = N
            if not scan_path.exists():
                continue
                
-            if scan_type in ["both", "tv"] and scan_path in config.tv_paths:
-                for item in scan_path.iterdir():
-                    if item.is_dir() and nfo_manager.parse_imdb_from_path(item):
+            if scan_type in ["both", "tv"] and (scan_path in config.tv_paths or path):
+                # Handle specific season/episode path
+                if path and scan_path.name.lower().startswith('season'):
+                    # Single season processing
+                    series_path = scan_path.parent
+                    if nfo_manager.parse_imdb_from_path(series_path):
+                        _log("INFO", f"Processing single season: {scan_path}")
                        try:
-                            tv_processor.process_series(item)
+                            tv_processor.process_season(series_path, scan_path)
                        except Exception as e:
-                            _log("ERROR", f"Failed processing TV series {item}: {e}")
+                            _log("ERROR", f"Failed processing season {scan_path}: {e}")
+                elif path and scan_path.is_file() and scan_path.suffix.lower() in ('.mkv', '.mp4', '.avi'):
+                    # Single episode processing
+                    season_path = scan_path.parent
+                    series_path = season_path.parent
+                    if nfo_manager.parse_imdb_from_path(series_path):
+                        _log("INFO", f"Processing single episode: {scan_path}")
+                        try:
+                            tv_processor.process_episode_file(series_path, season_path, scan_path)
+                        except Exception as e:
+                            _log("ERROR", f"Failed processing episode {scan_path}: {e}")
+                else:
+                    # Full series processing
+                    for item in scan_path.iterdir():
+                        if item.is_dir() and nfo_manager.parse_imdb_from_path(item):
+                            try:
+                                tv_processor.process_series(item)
+                            except Exception as e:
+                                _log("ERROR", f"Failed processing TV series {item}: {e}")
            
            if scan_type in ["both", "movies"] and scan_path in config.movie_paths:
                _log("INFO", f"Scanning movies in: {scan_path}")
@@ -1310,6 +1556,65 @@ async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = N
    background_tasks.add_task(run_scan)
    return {"status": "started", "message": f"Manual {scan_type} scan started"}

+@app.post("/tv/scan-season")
+async def scan_tv_season(background_tasks: BackgroundTasks, series_path: str, season_name: str):
+    """Scan a specific TV season - URL-safe endpoint"""
+    try:
+        series_dir = Path(series_path)
+        season_dir = series_dir / season_name
+        
+        if not series_dir.exists():
+            raise HTTPException(status_code=404, detail=f"Series path not found: {series_path}")
+        if not season_dir.exists():
+            raise HTTPException(status_code=404, detail=f"Season path not found: {season_dir}")
+        
+        imdb_id = nfo_manager.parse_imdb_from_path(series_dir)
+        if not imdb_id:
+            raise HTTPException(status_code=400, detail="No IMDb ID found in series path")
+        
+        async def process_season():
+            _log("INFO", f"Processing TV season: {season_dir}")
+            try:
+                tv_processor.process_season(series_dir, season_dir)
+            except Exception as e:
+                _log("ERROR", f"Failed processing season {season_dir}: {e}")
+        
+        background_tasks.add_task(process_season)
+        return {"status": "started", "message": f"Season scan started for {season_name}"}
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/tv/scan-episode")
+async def scan_tv_episode(background_tasks: BackgroundTasks, series_path: str, season_name: str, episode_name: str):
+    """Scan a specific TV episode - URL-safe endpoint"""
+    try:
+        series_dir = Path(series_path)
+        season_dir = series_dir / season_name
+        episode_file = season_dir / episode_name
+        
+        if not series_dir.exists():
+            raise HTTPException(status_code=404, detail=f"Series path not found: {series_path}")
+        if not episode_file.exists():
+            raise HTTPException(status_code=404, detail=f"Episode file not found: {episode_file}")
+        
+        imdb_id = nfo_manager.parse_imdb_from_path(series_dir)
+        if not imdb_id:
+            raise HTTPException(status_code=400, detail="No IMDb ID found in series path")
+        
+        async def process_episode():
+            _log("INFO", f"Processing TV episode: {episode_file}")
+            try:
+                tv_processor.process_episode_file(series_dir, season_dir, episode_file)
+            except Exception as e:
+                _log("ERROR", f"Failed processing episode {episode_file}: {e}")
+        
+        background_tasks.add_task(process_episode)
+        return {"status": "started", "message": f"Episode scan started for {episode_name}"}
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
@app.post("/test/bulk-update")
 async def test_bulk_update():
    """Test bulk update functionality without modifying data"""