feat: Implement NFOService.update_tvshow_nfo()
- Parse existing NFO to extract TMDB ID from uniqueid or tmdbid element - Fetch fresh metadata from TMDB API - Regenerate NFO with updated data - Optionally re-download media files - Add comprehensive error handling (missing NFO, no TMDB ID, invalid XML) - Add unit tests for XML parsing logic (4 tests, all passing) - Add integration test script (requires TMDB API key)
This commit is contained in:
@@ -258,7 +258,7 @@ Task 3 is **95% Complete** and **Production Ready**.
|
||||
**⚠️ Documentation Remaining (5%):**
|
||||
|
||||
- 📝 TMDB API setup guide (10 min)
|
||||
- 📝 Configuration examples for README (10 min)
|
||||
- 📝 Configuration examples for README (10 min)
|
||||
- 📝 ARCHITECTURE.md component diagram (10 min)
|
||||
|
||||
**Optional Future Work (Not blocking):**
|
||||
|
||||
183
scripts/test_nfo_update.py
Normal file
183
scripts/test_nfo_update.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""Test script for NFOService.update_tvshow_nfo() functionality.
|
||||
|
||||
This script tests the update functionality by:
|
||||
1. Creating a test NFO file with TMDB ID
|
||||
2. Updating it with fresh data from TMDB
|
||||
3. Verifying the update worked correctly
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from src.config.settings import settings
|
||||
from src.core.services.nfo_service import NFOService
|
||||
from src.core.services.tmdb_client import TMDBAPIError
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def test_update_nfo():
|
||||
"""Test NFO update functionality."""
|
||||
|
||||
# Check if TMDB API key is configured
|
||||
if not settings.tmdb_api_key:
|
||||
logger.error("TMDB_API_KEY not configured in environment")
|
||||
logger.error("Set TMDB_API_KEY in .env file or environment variables")
|
||||
return False
|
||||
|
||||
# Test series: Attack on Titan (TMDB ID: 1429)
|
||||
test_serie_name = "Attack on Titan"
|
||||
test_serie_folder = "test_update_nfo"
|
||||
test_tmdb_id = 1429
|
||||
|
||||
# Create test folder
|
||||
test_folder = Path(settings.anime_directory) / test_serie_folder
|
||||
test_folder.mkdir(parents=True, exist_ok=True)
|
||||
logger.info(f"Created test folder: {test_folder}")
|
||||
|
||||
# Initialize NFO service
|
||||
nfo_service = NFOService(
|
||||
tmdb_api_key=settings.tmdb_api_key,
|
||||
anime_directory=settings.anime_directory,
|
||||
image_size=settings.nfo_image_size
|
||||
)
|
||||
|
||||
try:
|
||||
# Step 1: Create initial NFO
|
||||
logger.info("=" * 60)
|
||||
logger.info("STEP 1: Creating initial NFO")
|
||||
logger.info("=" * 60)
|
||||
|
||||
nfo_path = await nfo_service.create_tvshow_nfo(
|
||||
serie_name=test_serie_name,
|
||||
serie_folder=test_serie_folder,
|
||||
year=2013,
|
||||
download_poster=False, # Skip downloads for faster testing
|
||||
download_logo=False,
|
||||
download_fanart=False
|
||||
)
|
||||
|
||||
logger.info(f"✓ Initial NFO created: {nfo_path}")
|
||||
|
||||
# Read initial NFO content
|
||||
initial_content = nfo_path.read_text(encoding="utf-8")
|
||||
logger.info(f"Initial NFO size: {len(initial_content)} bytes")
|
||||
|
||||
# Verify TMDB ID is in the file
|
||||
if str(test_tmdb_id) not in initial_content:
|
||||
logger.error(f"TMDB ID {test_tmdb_id} not found in NFO!")
|
||||
return False
|
||||
logger.info(f"✓ TMDB ID {test_tmdb_id} found in NFO")
|
||||
|
||||
# Step 2: Update the NFO
|
||||
logger.info("")
|
||||
logger.info("=" * 60)
|
||||
logger.info("STEP 2: Updating NFO with fresh data")
|
||||
logger.info("=" * 60)
|
||||
|
||||
updated_path = await nfo_service.update_tvshow_nfo(
|
||||
serie_folder=test_serie_folder,
|
||||
download_media=False # Skip downloads
|
||||
)
|
||||
|
||||
logger.info(f"✓ NFO updated: {updated_path}")
|
||||
|
||||
# Read updated content
|
||||
updated_content = updated_path.read_text(encoding="utf-8")
|
||||
logger.info(f"Updated NFO size: {len(updated_content)} bytes")
|
||||
|
||||
# Verify TMDB ID is still in the file
|
||||
if str(test_tmdb_id) not in updated_content:
|
||||
logger.error(f"TMDB ID {test_tmdb_id} not found after update!")
|
||||
return False
|
||||
logger.info(f"✓ TMDB ID {test_tmdb_id} still present after update")
|
||||
|
||||
# Step 3: Test update on non-existent NFO (should fail)
|
||||
logger.info("")
|
||||
logger.info("=" * 60)
|
||||
logger.info("STEP 3: Testing error handling")
|
||||
logger.info("=" * 60)
|
||||
|
||||
try:
|
||||
await nfo_service.update_tvshow_nfo(
|
||||
serie_folder="non_existent_folder",
|
||||
download_media=False
|
||||
)
|
||||
logger.error("✗ Should have raised FileNotFoundError!")
|
||||
return False
|
||||
except FileNotFoundError as e:
|
||||
logger.info(f"✓ Correctly raised FileNotFoundError: {e}")
|
||||
|
||||
# Step 4: Test update on NFO without TMDB ID
|
||||
logger.info("")
|
||||
logger.info("STEP 4: Testing NFO without TMDB ID")
|
||||
|
||||
# Create a minimal NFO without TMDB ID
|
||||
no_id_folder = test_folder.parent / "test_no_tmdb_id"
|
||||
no_id_folder.mkdir(parents=True, exist_ok=True)
|
||||
no_id_nfo = no_id_folder / "tvshow.nfo"
|
||||
no_id_nfo.write_text(
|
||||
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
'<tvshow>\n'
|
||||
' <title>Test Show</title>\n'
|
||||
'</tvshow>',
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
try:
|
||||
await nfo_service.update_tvshow_nfo(
|
||||
serie_folder="test_no_tmdb_id",
|
||||
download_media=False
|
||||
)
|
||||
logger.error("✗ Should have raised TMDBAPIError!")
|
||||
return False
|
||||
except TMDBAPIError as e:
|
||||
logger.info(f"✓ Correctly raised TMDBAPIError: {e}")
|
||||
|
||||
logger.info("")
|
||||
logger.info("=" * 60)
|
||||
logger.info("✓ ALL TESTS PASSED")
|
||||
logger.info("=" * 60)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"✗ Test failed with error: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
finally:
|
||||
await nfo_service.close()
|
||||
|
||||
# Cleanup test folders
|
||||
logger.info("\nCleaning up test folders...")
|
||||
import shutil
|
||||
try:
|
||||
if test_folder.exists():
|
||||
shutil.rmtree(test_folder)
|
||||
logger.info(f"Removed: {test_folder}")
|
||||
|
||||
no_id_folder = test_folder.parent / "test_no_tmdb_id"
|
||||
if no_id_folder.exists():
|
||||
shutil.rmtree(no_id_folder)
|
||||
logger.info(f"Removed: {no_id_folder}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Cleanup error: {e}")
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main entry point."""
|
||||
success = await test_update_nfo()
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -12,6 +12,8 @@ import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from lxml import etree
|
||||
|
||||
from src.core.entities.nfo_models import (
|
||||
ActorInfo,
|
||||
ImageInfo,
|
||||
@@ -158,21 +160,76 @@ class NFOService:
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If NFO file doesn't exist
|
||||
TMDBAPIError: If TMDB API fails
|
||||
TMDBAPIError: If TMDB API fails or no TMDB ID found in NFO
|
||||
"""
|
||||
nfo_path = self.anime_directory / serie_folder / "tvshow.nfo"
|
||||
folder_path = self.anime_directory / serie_folder
|
||||
nfo_path = folder_path / "tvshow.nfo"
|
||||
|
||||
if not nfo_path.exists():
|
||||
raise FileNotFoundError(f"NFO file not found: {nfo_path}")
|
||||
|
||||
# Parse existing NFO to get TMDB ID
|
||||
# For simplicity, we'll recreate from scratch
|
||||
# In production, you'd parse the XML to extract the ID
|
||||
|
||||
logger.info(f"Updating NFO for {serie_folder}")
|
||||
# Implementation would extract serie name and call create_tvshow_nfo
|
||||
# This is a simplified version
|
||||
raise NotImplementedError("Update NFO not yet implemented")
|
||||
|
||||
# Parse existing NFO to extract TMDB ID
|
||||
try:
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Try to find TMDB ID from uniqueid elements
|
||||
tmdb_id = None
|
||||
for uniqueid in root.findall(".//uniqueid"):
|
||||
if uniqueid.get("type") == "tmdb":
|
||||
tmdb_id = int(uniqueid.text)
|
||||
break
|
||||
|
||||
# Fallback: check for tmdbid element
|
||||
if tmdb_id is None:
|
||||
tmdbid_elem = root.find(".//tmdbid")
|
||||
if tmdbid_elem is not None and tmdbid_elem.text:
|
||||
tmdb_id = int(tmdbid_elem.text)
|
||||
|
||||
if tmdb_id is None:
|
||||
raise TMDBAPIError(
|
||||
f"No TMDB ID found in existing NFO. "
|
||||
f"Delete the NFO and create a new one instead."
|
||||
)
|
||||
|
||||
logger.debug(f"Found TMDB ID: {tmdb_id}")
|
||||
|
||||
except etree.XMLSyntaxError as e:
|
||||
raise TMDBAPIError(f"Invalid XML in NFO file: {e}")
|
||||
except ValueError as e:
|
||||
raise TMDBAPIError(f"Invalid TMDB ID format in NFO: {e}")
|
||||
|
||||
# Fetch fresh data from TMDB
|
||||
async with self.tmdb_client:
|
||||
logger.debug(f"Fetching fresh data for TMDB ID: {tmdb_id}")
|
||||
details = await self.tmdb_client.get_tv_show_details(
|
||||
tmdb_id,
|
||||
append_to_response="credits,external_ids,images"
|
||||
)
|
||||
|
||||
# Convert TMDB data to TVShowNFO model
|
||||
nfo_model = self._tmdb_to_nfo_model(details)
|
||||
|
||||
# Generate XML
|
||||
nfo_xml = generate_tvshow_nfo(nfo_model)
|
||||
|
||||
# Save updated NFO file
|
||||
nfo_path.write_text(nfo_xml, encoding="utf-8")
|
||||
logger.info(f"Updated NFO: {nfo_path}")
|
||||
|
||||
# Re-download media files if requested
|
||||
if download_media:
|
||||
await self._download_media_files(
|
||||
details,
|
||||
folder_path,
|
||||
download_poster=True,
|
||||
download_logo=True,
|
||||
download_fanart=True
|
||||
)
|
||||
|
||||
return nfo_path
|
||||
|
||||
def _find_best_match(
|
||||
self,
|
||||
|
||||
178
tests/unit/test_nfo_update_parsing.py
Normal file
178
tests/unit/test_nfo_update_parsing.py
Normal file
@@ -0,0 +1,178 @@
|
||||
"""Unit test for NFOService.update_tvshow_nfo() - tests XML parsing logic."""
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import shutil
|
||||
|
||||
from lxml import etree
|
||||
import pytest
|
||||
|
||||
from src.core.services.nfo_service import NFOService
|
||||
from src.core.services.tmdb_client import TMDBAPIError
|
||||
|
||||
|
||||
def create_sample_nfo(tmdb_id: int = 1429) -> str:
|
||||
"""Create a sample NFO XML with TMDB ID."""
|
||||
return f'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tvshow>
|
||||
<title>Attack on Titan</title>
|
||||
<originaltitle>Shingeki no Kyojin</originaltitle>
|
||||
<year>2013</year>
|
||||
<plot>Several hundred years ago, humans were nearly exterminated by Titans.</plot>
|
||||
<uniqueid type="tmdb" default="false">{tmdb_id}</uniqueid>
|
||||
<uniqueid type="tvdb" default="true">267440</uniqueid>
|
||||
<tmdbid>{tmdb_id}</tmdbid>
|
||||
<tvdbid>267440</tvdbid>
|
||||
</tvshow>'''
|
||||
|
||||
|
||||
def test_parse_nfo_with_uniqueid():
|
||||
"""Test parsing NFO with uniqueid elements."""
|
||||
# Create temporary directory structure
|
||||
temp_dir = Path(tempfile.mkdtemp())
|
||||
serie_folder = temp_dir / "test_series"
|
||||
serie_folder.mkdir()
|
||||
nfo_path = serie_folder / "tvshow.nfo"
|
||||
|
||||
try:
|
||||
# Write sample NFO
|
||||
nfo_path.write_text(create_sample_nfo(1429), encoding="utf-8")
|
||||
|
||||
# Parse it (same logic as in update_tvshow_nfo)
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Extract TMDB ID
|
||||
tmdb_id = None
|
||||
for uniqueid in root.findall(".//uniqueid"):
|
||||
if uniqueid.get("type") == "tmdb":
|
||||
tmdb_id = int(uniqueid.text)
|
||||
break
|
||||
|
||||
assert tmdb_id == 1429, f"Expected TMDB ID 1429, got {tmdb_id}"
|
||||
print(f"✓ Successfully parsed TMDB ID from uniqueid: {tmdb_id}")
|
||||
|
||||
finally:
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
|
||||
def test_parse_nfo_with_tmdbid_element():
|
||||
"""Test parsing NFO with tmdbid element (fallback)."""
|
||||
# Create NFO without uniqueid but with tmdbid element
|
||||
nfo_content = '''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tvshow>
|
||||
<title>Test Show</title>
|
||||
<tmdbid>12345</tmdbid>
|
||||
</tvshow>'''
|
||||
|
||||
temp_dir = Path(tempfile.mkdtemp())
|
||||
serie_folder = temp_dir / "test_series"
|
||||
serie_folder.mkdir()
|
||||
nfo_path = serie_folder / "tvshow.nfo"
|
||||
|
||||
try:
|
||||
nfo_path.write_text(nfo_content, encoding="utf-8")
|
||||
|
||||
# Parse it
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Try uniqueid first (should fail)
|
||||
tmdb_id = None
|
||||
for uniqueid in root.findall(".//uniqueid"):
|
||||
if uniqueid.get("type") == "tmdb":
|
||||
tmdb_id = int(uniqueid.text)
|
||||
break
|
||||
|
||||
# Fallback to tmdbid element
|
||||
if tmdb_id is None:
|
||||
tmdbid_elem = root.find(".//tmdbid")
|
||||
if tmdbid_elem is not None and tmdbid_elem.text:
|
||||
tmdb_id = int(tmdbid_elem.text)
|
||||
|
||||
assert tmdb_id == 12345, f"Expected TMDB ID 12345, got {tmdb_id}"
|
||||
print(f"✓ Successfully parsed TMDB ID from tmdbid element: {tmdb_id}")
|
||||
|
||||
finally:
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
|
||||
def test_parse_nfo_without_tmdb_id():
|
||||
"""Test parsing NFO without TMDB ID raises appropriate error."""
|
||||
# Create NFO without any TMDB ID
|
||||
nfo_content = '''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tvshow>
|
||||
<title>Test Show</title>
|
||||
</tvshow>'''
|
||||
|
||||
temp_dir = Path(tempfile.mkdtemp())
|
||||
serie_folder = temp_dir / "test_series"
|
||||
serie_folder.mkdir()
|
||||
nfo_path = serie_folder / "tvshow.nfo"
|
||||
|
||||
try:
|
||||
nfo_path.write_text(nfo_content, encoding="utf-8")
|
||||
|
||||
# Parse it
|
||||
tree = etree.parse(str(nfo_path))
|
||||
root = tree.getroot()
|
||||
|
||||
# Try to extract TMDB ID
|
||||
tmdb_id = None
|
||||
for uniqueid in root.findall(".//uniqueid"):
|
||||
if uniqueid.get("type") == "tmdb":
|
||||
tmdb_id = int(uniqueid.text)
|
||||
break
|
||||
|
||||
if tmdb_id is None:
|
||||
tmdbid_elem = root.find(".//tmdbid")
|
||||
if tmdbid_elem is not None and tmdbid_elem.text:
|
||||
tmdb_id = int(tmdbid_elem.text)
|
||||
|
||||
assert tmdb_id is None, "Should not have found TMDB ID"
|
||||
print("✓ Correctly identified NFO without TMDB ID")
|
||||
|
||||
finally:
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
|
||||
def test_parse_invalid_xml():
|
||||
"""Test parsing invalid XML raises appropriate error."""
|
||||
nfo_content = '''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<tvshow>
|
||||
<title>Unclosed tag
|
||||
</tvshow>'''
|
||||
|
||||
temp_dir = Path(tempfile.mkdtemp())
|
||||
serie_folder = temp_dir / "test_series"
|
||||
serie_folder.mkdir()
|
||||
nfo_path = serie_folder / "tvshow.nfo"
|
||||
|
||||
try:
|
||||
nfo_path.write_text(nfo_content, encoding="utf-8")
|
||||
|
||||
# Try to parse - should raise XMLSyntaxError
|
||||
try:
|
||||
tree = etree.parse(str(nfo_path))
|
||||
assert False, "Should have raised XMLSyntaxError"
|
||||
except etree.XMLSyntaxError:
|
||||
print("✓ Correctly raised XMLSyntaxError for invalid XML")
|
||||
|
||||
finally:
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Testing NFO XML parsing logic...")
|
||||
print()
|
||||
|
||||
test_parse_nfo_with_uniqueid()
|
||||
test_parse_nfo_with_tmdbid_element()
|
||||
test_parse_nfo_without_tmdb_id()
|
||||
test_parse_invalid_xml()
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("✓ ALL TESTS PASSED")
|
||||
print("=" * 60)
|
||||
Reference in New Issue
Block a user