removal of hard codeing
This commit is contained in:
+178
-475
@@ -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 = [
|
||||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
|
||||
"<episodedetails>",
|
||||
f" <season>{season_num}</season>",
|
||||
f" <episode>{episode_num}</episode>",
|
||||
]
|
||||
|
||||
# Add enhanced metadata if available
|
||||
if enhanced_metadata:
|
||||
if enhanced_metadata.get("title"):
|
||||
xml_lines.append(f" <title>{self._escape_xml(enhanced_metadata['title'])}</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" <plot>{self._escape_xml(enhanced_metadata['overview'])}</plot>")
|
||||
# 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" <runtime>{enhanced_metadata['runtime']}</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" <rating>{imdb_rating}</rating>")
|
||||
xml_lines.append(f" <votes>{imdb_votes}</votes>")
|
||||
|
||||
# Add historical air date and premiere date (should be the same)
|
||||
if aired:
|
||||
air_date = aired.split('T')[0]
|
||||
xml_lines.append(f" <aired>{air_date}</aired>")
|
||||
xml_lines.append(f" <premiered>{air_date}</premiered>")
|
||||
|
||||
# Add import/download date (when user actually got the episode)
|
||||
if dateadded:
|
||||
xml_lines.append(f" <dateadded>{dateadded}</dateadded>")
|
||||
|
||||
if lock_metadata:
|
||||
xml_lines.append(" <lockdata>true</lockdata>")
|
||||
|
||||
# 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" <!-- source: TV Episode; {source_detail} -->")
|
||||
xml_lines.append(f" <!-- managed by {self.manager_brand} at {current_time} -->")
|
||||
|
||||
xml_lines.append("</episodedetails>")
|
||||
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"""<?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, 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' <uniqueid type="tvdb" default="true">{tvdb_id}</uniqueid>')
|
||||
if imdb_id:
|
||||
default_attr = "" if tvdb_id else ' default="true"'
|
||||
uniqueid_elements.append(f' <uniqueid type="imdb"{default_attr}>{imdb_id}</uniqueid>')
|
||||
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"""<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
|
||||
<tvshow>
|
||||
<id>{main_id}</id>
|
||||
{chr(10).join(uniqueid_elements)}
|
||||
<!-- NFOGuard: TVDB ID required for proper Emby metadata loading -->
|
||||
<!-- If missing TVDB ID, use: Dashboard > Series > Edit Metadata > External IDs -->
|
||||
</tvshow>
|
||||
"""
|
||||
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/")
|
||||
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)
|
||||
Reference in New Issue
Block a user