diff --git a/api/routes.py b/api/routes.py index 4d40f85..22c1495 100644 --- a/api/routes.py +++ b/api/routes.py @@ -355,4 +355,485 @@ def register_routes( } except Exception as e: _log("ERROR", f"Lockdata stripping failed: {e}") - raise HTTPException(status_code=500, detail=f"Lockdata stripping failed: {str(e)}") \ No newline at end of file + raise HTTPException(status_code=500, detail=f"Lockdata stripping failed: {str(e)}") + + @app.get("/debug/nfo/{path:path}") + async def debug_nfo_content(path: str): + """Debug NFO file content to troubleshoot Emby title issues""" + try: + from pathlib import Path + import xml.etree.ElementTree as ET + + # Construct full path + nfo_path = Path("/" + path.lstrip("/")) + if not nfo_path.exists(): + raise HTTPException(status_code=404, detail=f"NFO file not found: {nfo_path}") + + if not str(nfo_path).endswith('.nfo'): + raise HTTPException(status_code=400, detail="Path must point to a .nfo file") + + # Parse NFO content + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + # Extract key elements that Emby uses + result = { + "file_path": str(nfo_path), + "root_element": root.tag, + "elements_found": {}, + "nfoguard_data": {}, + "raw_xml": None + } + + # Check for key elements + key_elements = ["title", "plot", "premiered", "dateadded", "lockdata", "uniqueid", "year", "runtime"] + for element_name in key_elements: + elem = root.find(f".//{element_name}") + if elem is not None: + result["elements_found"][element_name] = { + "text": elem.text, + "attributes": elem.attrib if elem.attrib else None + } + + # Check NFOGuard specific data + nfoguard_source = root.find(".//nfoguard_source") + if nfoguard_source is not None: + result["nfoguard_data"]["source"] = nfoguard_source.text + + # Get first few lines of raw XML for inspection + with open(nfo_path, 'r', encoding='utf-8') as f: + raw_content = f.read() + lines = raw_content.split('\n') + result["raw_xml"] = lines[:20] # First 20 lines + + return result + + except ET.ParseError as e: + # If XML parsing fails, return raw content + with open(nfo_path, 'r', encoding='utf-8') as f: + raw_content = f.read() + + return { + "file_path": str(nfo_path), + "parse_error": str(e), + "raw_content": raw_content[:1000] # First 1000 chars + } + + except Exception as e: + _log("ERROR", f"NFO debug failed: {e}") + raise HTTPException(status_code=500, detail=f"NFO debug failed: {str(e)}") + + @app.get("/debug/emby-title-test/{imdb_id}") + async def test_emby_title_format(imdb_id: str): + """Test different NFO title formats for Emby compatibility""" + try: + # Create a test NFO with different title positioning to see what Emby prefers + test_formats = { + "title_first": """ + + Test Movie Title + Test plot description + {imdb_id} + 2023-01-01T12:00:00 +""", + "title_after_plot": """ + + Test plot description + Test Movie Title + {imdb_id} + 2023-01-01T12:00:00 +""", + "title_before_uniqueid": """ + + Test plot description + {imdb_id} + Test Movie Title + 2023-01-01T12:00:00 +""" + } + + for format_name, xml_content in test_formats.items(): + test_formats[format_name] = xml_content.format(imdb_id=imdb_id) + + return { + "imdb_id": imdb_id, + "test_formats": test_formats, + "recommendation": "Try creating a test NFO file with title as the first element after " + } + + except Exception as e: + _log("ERROR", f"Emby title test failed: {e}") + raise HTTPException(status_code=500, detail=f"Emby title test failed: {str(e)}") + + @app.get("/debug/tv-episode/{path:path}") + async def debug_tv_episode_nfo(path: str): + """Debug TV episode NFO specifically for Emby title issues""" + try: + from pathlib import Path + import xml.etree.ElementTree as ET + + # Handle paths like "media/TV/ShowName/Season 01/S01E01.nfo" + nfo_path = Path("/" + path.lstrip("/")) + if not nfo_path.exists(): + raise HTTPException(status_code=404, detail=f"Episode NFO not found: {nfo_path}") + + if not str(nfo_path).endswith('.nfo'): + raise HTTPException(status_code=400, detail="Path must point to a .nfo file") + + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + # Check if it's an episode NFO + if root.tag != "episodedetails": + return {"error": f"Not a TV episode NFO (root is {root.tag}, expected episodedetails)"} + + # Extract TV-specific elements + result = { + "file_path": str(nfo_path), + "nfo_type": "TV Episode", + "emby_compatibility_check": {}, + "elements": {}, + "video_file_info": None, + "recommendations": [] + } + + # Check critical elements for TV episodes + tv_elements = ["title", "season", "episode", "aired", "plot", "uniqueid"] + for elem_name in tv_elements: + elem = root.find(elem_name) + if elem is not None: + result["elements"][elem_name] = { + "text": elem.text, + "attributes": elem.attrib if elem.attrib else {} + } + + # Emby compatibility checks + title_elem = root.find("title") + if title_elem is not None and title_elem.text: + result["emby_compatibility_check"]["has_title"] = True + result["emby_compatibility_check"]["title_value"] = title_elem.text.strip() + result["emby_compatibility_check"]["title_position"] = list(root).index(title_elem) + + if list(root).index(title_elem) == 0: + result["emby_compatibility_check"]["title_is_first"] = True + else: + result["emby_compatibility_check"]["title_is_first"] = False + result["recommendations"].append("Consider moving to be the first element after <episodedetails>") + else: + result["emby_compatibility_check"]["has_title"] = False + result["recommendations"].append("CRITICAL: No <title> element found!") + + # Check for video file in same directory + season_dir = nfo_path.parent + episode_pattern = nfo_path.stem # S01E01 from S01E01.nfo + video_exts = [".mkv", ".mp4", ".avi", ".mov", ".m4v"] + + for video_ext in video_exts: + potential_video = season_dir / f"{episode_pattern}{video_ext}" + if potential_video.exists(): + result["video_file_info"] = { + "filename": potential_video.name, + "matches_nfo": True + } + break + + # Look for any video file with the pattern + if not result["video_file_info"]: + for video_file in season_dir.glob(f"*{episode_pattern}*"): + if video_file.suffix.lower() in video_exts: + result["video_file_info"] = { + "filename": video_file.name, + "matches_nfo": episode_pattern in video_file.stem + } + if not (episode_pattern in video_file.stem): + result["recommendations"].append(f"Video filename '{video_file.name}' doesn't cleanly match NFO pattern") + break + + # Element order analysis + element_order = [child.tag for child in root] + result["element_order"] = element_order + + # Emby-specific title positioning checks + title_position = element_order.index('title') + 1 if 'title' in element_order else -1 + result["title_position"] = title_position + + if title_position > 1: + result["recommendations"].append(f"Title element is at position {title_position}. Emby prefers title as first element.") + + # Check for common Emby compatibility issues + emby_checks = { + "title_first": element_order[0] == "title" if element_order else False, + "has_plot": "plot" in element_order, + "has_aired": "aired" in element_order, + "has_season": "season" in element_order, + "has_episode": "episode" in element_order, + "no_lockdata": "lockdata" not in element_order + } + result["emby_compatibility"] = emby_checks + + # Additional Emby recommendations + if not emby_checks["title_first"]: + result["recommendations"].append("Move <title> to be the first element for best Emby compatibility") + + if not emby_checks["no_lockdata"]: + result["recommendations"].append("Remove <lockdata> elements - they can cause Emby metadata issues") + + # Check title content quality + if title_elem is not None and title_elem.text: + title_text = title_elem.text.strip() + # Check if title looks like filename + video_filename_base = None + for video_file in video_files: + if episode_pattern in video_file.stem: + video_filename_base = video_file.stem + break + + if video_filename_base and title_text.lower() in video_filename_base.lower(): + result["recommendations"].append(f"Title '{title_text}' appears to be derived from filename - ensure it's a proper episode title") + + return result + + except ET.ParseError as e: + with open(nfo_path, 'r', encoding='utf-8') as f: + raw_content = f.read() + return { + "file_path": str(nfo_path), + "error": "XML Parse Error", + "parse_error": str(e), + "raw_content_sample": raw_content[:500] + } + + except Exception as e: + _log("ERROR", f"TV episode debug failed: {e}") + raise HTTPException(status_code=500, detail=f"TV episode debug failed: {str(e)}") + + @app.post("/debug/validate-series-nfos") + async def validate_series_nfos(request: Dict[str, Any]): + """Validate NFO files for an entire TV series for Emby compatibility""" + try: + series_path_str = request.get("series_path") + if not series_path_str: + raise HTTPException(status_code=400, detail="series_path required") + + series_path = Path(series_path_str) + if not series_path.exists(): + raise HTTPException(status_code=404, detail="Series path not found") + + _log("INFO", f"Validating series NFOs: {series_path}") + + validation_results = { + "series_path": str(series_path), + "series_nfo": None, + "seasons": {}, + "episodes": [], + "summary": { + "total_nfos": 0, + "emby_compatible": 0, + "title_position_issues": 0, + "lockdata_issues": 0, + "common_issues": [] + } + } + + # Check series NFO + tvshow_nfo = series_path / "tvshow.nfo" + if tvshow_nfo.exists(): + validation_results["series_nfo"] = await debug_nfo_structure(str(tvshow_nfo), "series") + validation_results["summary"]["total_nfos"] += 1 + + # Check season NFOs and episodes + from core.fs_cache import fs_cache + season_dirs = [d for d in fs_cache.get_directory_contents(series_path) + if d.is_dir() and d.name.lower().startswith("season")] + + for season_dir in season_dirs: + season_name = season_dir.name + season_data = {"path": str(season_dir), "nfo": None, "episodes": []} + + # Check season NFO + season_nfo = season_dir / "season.nfo" + if season_nfo.exists(): + season_data["nfo"] = await debug_nfo_structure(str(season_nfo), "season") + validation_results["summary"]["total_nfos"] += 1 + + # Check episode NFOs + video_files = fs_cache.find_video_files(season_dir) + for video_file in video_files: + nfo_file = video_file.with_suffix('.nfo') + if nfo_file.exists(): + episode_debug = await debug_tv_episode_nfo(str(nfo_file).replace(str(series_path) + "/", "")) + season_data["episodes"].append(episode_debug) + validation_results["episodes"].append(episode_debug) + validation_results["summary"]["total_nfos"] += 1 + + # Analyze for summary + if "emby_compatibility" in episode_debug: + emby_compat = episode_debug["emby_compatibility"] + if all(emby_compat.values()): + validation_results["summary"]["emby_compatible"] += 1 + if not emby_compat.get("title_first", False): + validation_results["summary"]["title_position_issues"] += 1 + if not emby_compat.get("no_lockdata", True): + validation_results["summary"]["lockdata_issues"] += 1 + + validation_results["seasons"][season_name] = season_data + + # Generate common issues summary + if validation_results["summary"]["title_position_issues"] > 0: + validation_results["summary"]["common_issues"].append( + f"{validation_results['summary']['title_position_issues']} NFOs have title positioning issues" + ) + + if validation_results["summary"]["lockdata_issues"] > 0: + validation_results["summary"]["common_issues"].append( + f"{validation_results['summary']['lockdata_issues']} NFOs contain lockdata elements" + ) + + compat_rate = (validation_results["summary"]["emby_compatible"] / + validation_results["summary"]["total_nfos"] * 100) if validation_results["summary"]["total_nfos"] > 0 else 0 + + validation_results["summary"]["compatibility_rate"] = round(compat_rate, 1) + + return validation_results + + except Exception as e: + _log("ERROR", f"Series NFO validation failed: {e}") + raise HTTPException(status_code=500, detail=f"Series NFO validation failed: {str(e)}") + + async def debug_nfo_structure(nfo_path_str: str, nfo_type: str) -> Dict[str, Any]: + """Helper function to debug NFO structure""" + try: + nfo_path = Path(nfo_path_str) + if not nfo_path.exists(): + return {"error": "NFO file not found", "path": nfo_path_str} + + import xml.etree.ElementTree as ET + + tree = ET.parse(nfo_path) + root = tree.getroot() + + # Basic structure analysis + result = { + "file_path": str(nfo_path), + "root_element": root.tag, + "element_count": len(root), + "elements": {}, + "recommendations": [] + } + + # Element analysis + for child in root: + if child.text: + result["elements"][child.tag] = child.text.strip()[:100] # Truncate long values + else: + result["elements"][child.tag] = None + + # Element order analysis + element_order = [child.tag for child in root] + result["element_order"] = element_order + + # Common checks + if "title" in element_order: + title_position = element_order.index('title') + 1 + result["title_position"] = title_position + if title_position > 1: + result["recommendations"].append(f"Title element is at position {title_position}. Consider moving to position 1.") + + if "lockdata" in element_order: + result["recommendations"].append("Contains lockdata element - consider removing for Emby compatibility") + + return result + + except Exception as e: + return {"error": str(e), "path": nfo_path_str} + + @app.post("/debug/fix-emby-compatibility") + async def fix_emby_compatibility(request: Dict[str, Any]): + """Apply Emby compatibility fixes to NFO files""" + try: + file_path_str = request.get("file_path") + series_path_str = request.get("series_path") + + if not file_path_str and not series_path_str: + raise HTTPException(status_code=400, detail="Either file_path or series_path required") + + results = { + "fixed_files": [], + "total_processed": 0, + "total_updated": 0, + "errors": [] + } + + nfo_files = [] + + # Single file fix + if file_path_str: + file_path = Path(file_path_str) + if not file_path.exists(): + raise HTTPException(status_code=404, detail="File not found") + if file_path.suffix.lower() == '.nfo': + nfo_files.append(file_path) + + # Series directory fix + if series_path_str: + series_path = Path(series_path_str) + if not series_path.exists(): + raise HTTPException(status_code=404, detail="Series path not found") + + # Find all NFO files in series directory + from core.fs_cache import fs_cache + + # Series NFO + tvshow_nfo = series_path / "tvshow.nfo" + if tvshow_nfo.exists(): + nfo_files.append(tvshow_nfo) + + # Season and episode NFOs + season_dirs = [d for d in fs_cache.get_directory_contents(series_path) + if d.is_dir() and d.name.lower().startswith("season")] + + for season_dir in season_dirs: + # Season NFO + season_nfo = season_dir / "season.nfo" + if season_nfo.exists(): + nfo_files.append(season_nfo) + + # Episode NFOs + video_files = fs_cache.find_video_files(season_dir) + for video_file in video_files: + nfo_file = video_file.with_suffix('.nfo') + if nfo_file.exists(): + nfo_files.append(nfo_file) + + # Apply fixes to all found NFO files + for nfo_file in nfo_files: + try: + fix_results = nfo_manager.fix_nfo_for_emby_compatibility(nfo_file) + + file_result = { + "file_path": str(nfo_file), + "lockdata_removed": fix_results["lockdata_removed"], + "title_reordered": fix_results["title_reordered"], + "updated": fix_results["file_updated"] + } + + results["fixed_files"].append(file_result) + results["total_processed"] += 1 + + if fix_results["file_updated"]: + results["total_updated"] += 1 + + except Exception as e: + error_msg = f"Error fixing {nfo_file}: {str(e)}" + results["errors"].append(error_msg) + _log("ERROR", error_msg) + + _log("INFO", f"Emby compatibility fixes complete: {results['total_updated']} of {results['total_processed']} files updated") + + return results + + except Exception as e: + _log("ERROR", f"Emby compatibility fix failed: {e}") + raise HTTPException(status_code=500, detail=f"Emby compatibility fix failed: {str(e)}") \ No newline at end of file diff --git a/core/nfo_manager.py b/core/nfo_manager.py index 9a187ba..0c650fe 100644 --- a/core/nfo_manager.py +++ b/core/nfo_manager.py @@ -9,6 +9,7 @@ from pathlib import Path from datetime import datetime from typing import Optional, Dict, Any import re +from .logging import _log class NFOManager: @@ -858,6 +859,71 @@ class NFOManager: return False + def reorder_nfo_for_emby(self, nfo_path: Path) -> bool: + """Reorder NFO elements to put title first for better Emby compatibility""" + if not nfo_path.exists(): + return False + + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + # Find title element + title_elem = root.find('title') + if title_elem is None: + return False # No title element to reorder + + # Check if title is already first + if len(root) > 0 and root[0].tag == 'title': + return False # Title is already first + + # Remove title from current position + root.remove(title_elem) + + # Insert title as first element + root.insert(0, title_elem) + + # Write back the modified NFO + ET.indent(tree, space=" ", level=0) + tree.write(nfo_path, encoding="utf-8", xml_declaration=True) + _log("INFO", f"Reordered NFO elements - moved title to first position: {nfo_path.name}") + return True + + except Exception as e: + _log("ERROR", f"Error reordering NFO {nfo_path}: {e}") + + return False + + def fix_nfo_for_emby_compatibility(self, nfo_path: Path) -> Dict[str, bool]: + """Apply all Emby compatibility fixes to an NFO file""" + results = { + "lockdata_removed": False, + "title_reordered": False, + "file_updated": False + } + + if not nfo_path.exists(): + return results + + try: + # Strip lockdata elements + results["lockdata_removed"] = self.strip_lockdata_from_nfo(nfo_path) + + # Reorder title to first position + results["title_reordered"] = self.reorder_nfo_for_emby(nfo_path) + + results["file_updated"] = results["lockdata_removed"] or results["title_reordered"] + + if results["file_updated"]: + _log("INFO", f"Applied Emby compatibility fixes to {nfo_path.name}: " + f"lockdata_removed={results['lockdata_removed']}, " + f"title_reordered={results['title_reordered']}") + + except Exception as e: + _log("ERROR", f"Error applying Emby fixes to {nfo_path}: {e}") + + return results + def strip_lockdata_from_directory(self, directory_path: Path) -> int: """Strip lockdata from all NFO files in a directory""" if not directory_path.exists(): @@ -874,4 +940,111 @@ class NFOManager: _log("INFO", f"Stripped lockdata from {processed_count} NFO files in {directory_path}") return processed_count + + def update_episode_nfo_preserving_name(self, nfo_path: Path, season_num: int, episode_num: int, + aired: Optional[str], dateadded: Optional[str], source: str) -> bool: + """Update existing TV episode NFO file while preserving its original filename""" + if not nfo_path.exists(): + return False + + try: + # Parse the existing NFO file + tree = ET.parse(nfo_path) + episode = tree.getroot() + + # Ensure root element is <episodedetails> + if episode.tag != "episodedetails": + print(f"⚠️ NFO file {nfo_path.name} does not have episodedetails root element") + return False + + # Remove any existing NFOGuard fields to re-add them at the bottom + nfoguard_fields = ["aired", "dateadded", "season", "episode"] + for field_name in nfoguard_fields: + existing_field = episode.find(field_name) + if existing_field is not None: + episode.remove(existing_field) + + # Remove any existing NFOGuard comments + for child in episode: + if isinstance(child, ET.Comment): + if self.manager_brand in str(child): + episode.remove(child) + + # Add NFOGuard fields at the bottom + + # Basic episode info at the end + 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 at the end + 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 + + # Add NFOGuard comment at the bottom with other NFOGuard fields + nfoguard_comment = ET.Comment(f" Updated by {self.manager_brand} - Source: {source} ") + episode.append(nfoguard_comment) + + # Write file with proper formatting + tree = ET.ElementTree(episode) + ET.indent(tree, space=" ", level=0) + tree.write(nfo_path, encoding='utf-8', xml_declaration=True) + + print(f"✅ Successfully updated episode NFO (preserving name): {nfo_path.name}") + print(f" S{season_num:02d}E{episode_num:02d}, Aired: {aired}, Date Added: {dateadded}") + + return True + + except Exception as e: + print(f"❌ Error updating episode NFO {nfo_path.name}: {e}") + return False + + def find_episode_nfo_matching_video(self, season_dir: Path, season_num: int, episode_num: int) -> Optional[Path]: + """Find episode NFO that matches the naming pattern of corresponding video files""" + if not season_dir.exists(): + return None + + # Find video files for this episode (check common patterns) + season_episode_pattern = f"S{season_num:02d}E{episode_num:02d}" + + # Look for video files with season/episode pattern + video_extensions = ['.mkv', '.mp4', '.avi', '.m4v'] + for ext in video_extensions: + video_files = season_dir.glob(f"*{season_episode_pattern}*{ext}") + for video_file in video_files: + # For each video file, look for an NFO with the same base name + nfo_path = video_file.with_suffix('.nfo') + if nfo_path.exists(): + # Verify this NFO contains the right season/episode + try: + tree = ET.parse(nfo_path) + root = tree.getroot() + + if root.tag == "episodedetails": + season_elem = root.find("season") + episode_elem = root.find("episode") + + if (season_elem is not None and episode_elem is not None and + season_elem.text and episode_elem.text): + try: + file_season = int(season_elem.text) + file_episode = int(episode_elem.text) + + if file_season == season_num and file_episode == episode_num: + print(f"🎯 Found episode NFO matching video file: {nfo_path.name}") + return nfo_path + except ValueError: + continue + + except (ET.ParseError, Exception): + continue + + return None diff --git a/processors/tv_processor.py b/processors/tv_processor.py index 6177fcb..5789af8 100644 --- a/processors/tv_processor.py +++ b/processors/tv_processor.py @@ -87,13 +87,31 @@ class TVProcessor: # Process episodes for (season, episode), (aired, dateadded, source) in episode_dates.items(): if (season, episode) in disk_episodes: - # Create NFO + # Create or update NFO if config.manage_nfo: season_dir = series_path / config.tv_season_dir_format.format(season=season) - self.nfo_manager.create_episode_nfo( - season_dir, - season, episode, aired, dateadded, source, config.lock_metadata - ) + + # Check if this is an existing NFO that needs updating while preserving name + if source == "nfo_update_required": + # Find the existing NFO file and update it in place + existing_nfo = self.nfo_manager.find_episode_nfo_matching_video(season_dir, season, episode) + if existing_nfo: + # Update existing NFO while preserving its filename + self.nfo_manager.update_episode_nfo_preserving_name( + existing_nfo, season, episode, aired, dateadded, "file_update" + ) + else: + # Fallback to standard creation if NFO not found + self.nfo_manager.create_episode_nfo( + season_dir, + season, episode, aired, dateadded, source, config.lock_metadata + ) + else: + # Standard NFO creation for new episodes + self.nfo_manager.create_episode_nfo( + season_dir, + season, episode, aired, dateadded, source, config.lock_metadata + ) # Update file mtimes if config.fix_dir_mtimes and dateadded: @@ -170,21 +188,21 @@ class TVProcessor: if key in disk_episodes: # Only use cached data for episodes we have episode_dates[key] = (ep["aired"], ep["dateadded"], ep["source"]) - # Check for existing NFO files (including long-named ones) for migration + # Check for existing NFO files that match video file names (preserve long names) nfo_episodes_found = 0 for (season_num, episode_num) in disk_episodes.keys(): if (season_num, episode_num) not in episode_dates: - # Check if this episode has an existing NFO file that needs migration + # Check if this episode has an NFO file matching its video file name season_dir = series_path / config.tv_season_dir_format.format(season=season_num) - existing_nfo = self.nfo_manager.find_existing_episode_nfo(season_dir, season_num, episode_num) + existing_nfo = self.nfo_manager.find_episode_nfo_matching_video(season_dir, season_num, episode_num) if existing_nfo: - # Force processing of this episode for NFO migration - episode_dates[(season_num, episode_num)] = (None, None, "nfo_migration_required") + # Force processing of this episode to update NFO with NFOGuard data + episode_dates[(season_num, episode_num)] = (None, None, "nfo_update_required") nfo_episodes_found += 1 if nfo_episodes_found > 0: - _log("INFO", f"Found {nfo_episodes_found} episodes with existing NFO files requiring migration") + _log("INFO", f"Found {nfo_episodes_found} episodes with existing NFO files to update (preserving names)") # Find missing episodes (not in cache and no existing NFO) cached_keys = set(episode_dates.keys())