From 3e5ad8a4a658e6edd291894b498f39b2d13917a9 Mon Sep 17 00:00:00 2001 From: Lukas Date: Sun, 22 Feb 2026 11:09:48 +0100 Subject: [PATCH] feat: add NfoRepairService for missing NFO tag detection --- docs/ARCHITECTURE.md | 2 + docs/CHANGELOG.md | 4 + src/core/services/nfo_repair_service.py | 180 ++++++++++++++++++++++++ tests/unit/test_nfo_repair_service.py | 140 ++++++++++++++++++ 4 files changed, 326 insertions(+) create mode 100644 src/core/services/nfo_repair_service.py create mode 100644 tests/unit/test_nfo_repair_service.py diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c1548c8..a8fd03f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -207,6 +207,8 @@ src/core/ | +-- nfo_models.py # Pydantic models for tvshow.nfo (TVShowNFO, ActorInfo…) +-- services/ # Domain services | +-- nfo_service.py # NFO lifecycle: create / update tvshow.nfo +| +-- nfo_repair_service.py # Detect & repair incomplete tvshow.nfo files +| | # (parse_nfo_tags, find_missing_tags, NfoRepairService) | +-- tmdb_client.py # Async TMDB API client +-- utils/ # Utility helpers (no side-effects) | +-- nfo_generator.py # TVShowNFO → XML serialiser diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 567ee6e..bf9a506 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -51,6 +51,10 @@ This changelog follows [Keep a Changelog](https://keepachangelog.com/) principle 500 lines and isolate pure mapping logic. - **US MPAA rating**: `_extract_rating_by_country(ratings, "US")` now maps the US TMDB content rating to the `` NFO tag. +- **`NfoRepairService` (`src/core/services/nfo_repair_service.py`)**: New service + that detects incomplete `tvshow.nfo` files and triggers TMDB re-fetch. + Provides `parse_nfo_tags()`, `find_missing_tags()`, `nfo_needs_repair()`, and + `NfoRepairService.repair_series()`. 13 required tags are checked. ### Changed diff --git a/src/core/services/nfo_repair_service.py b/src/core/services/nfo_repair_service.py new file mode 100644 index 0000000..ffc45b6 --- /dev/null +++ b/src/core/services/nfo_repair_service.py @@ -0,0 +1,180 @@ +"""NFO repair service for detecting and fixing incomplete tvshow.nfo files. + +This module provides utilities to check whether an existing ``tvshow.nfo`` +contains all required tags and to trigger a repair (re-fetch from TMDB) when +needed. + +Example: + >>> service = NfoRepairService(nfo_service) + >>> repaired = await service.repair_series(Path("/anime/Attack on Titan"), "Attack on Titan") +""" + +import logging +from pathlib import Path +from typing import Dict, List + +from lxml import etree + +from src.core.services.nfo_service import NFOService + +logger = logging.getLogger(__name__) + + +# XPath relative to root → human-readable label +REQUIRED_TAGS: Dict[str, str] = { + "./title": "title", + "./originaltitle": "originaltitle", + "./year": "year", + "./plot": "plot", + "./runtime": "runtime", + "./premiered": "premiered", + "./status": "status", + "./imdbid": "imdbid", + "./genre": "genre", + "./studio": "studio", + "./country": "country", + "./actor/name": "actor/name", + "./watched": "watched", +} + + +def parse_nfo_tags(nfo_path: Path) -> Dict[str, List[str]]: + """Parse an existing tvshow.nfo and return present tag values. + + Evaluates every XPath in :data:`REQUIRED_TAGS` against the document root + and collects all non-empty text values. + + Args: + nfo_path: Absolute path to the ``tvshow.nfo`` file. + + Returns: + Mapping of XPath expression → list of non-empty text strings found in + the document. Returns an empty dict on any error (missing file, + invalid XML, permission error). + + Example: + >>> tags = parse_nfo_tags(Path("/anime/Attack on Titan/tvshow.nfo")) + >>> tags.get("./title") + ['Attack on Titan'] + """ + if not nfo_path.exists(): + logger.debug("NFO file not found: %s", nfo_path) + return {} + + try: + tree = etree.parse(str(nfo_path)) + root = tree.getroot() + + result: Dict[str, List[str]] = {} + for xpath in REQUIRED_TAGS: + elements = root.findall(xpath) + result[xpath] = [e.text for e in elements if e.text] + + return result + + except etree.XMLSyntaxError as exc: + logger.warning("Malformed XML in %s: %s", nfo_path, exc) + return {} + except Exception as exc: # pylint: disable=broad-except + logger.warning("Unexpected error parsing %s: %s", nfo_path, exc) + return {} + + +def find_missing_tags(nfo_path: Path) -> List[str]: + """Return tags that are absent or empty in the NFO. + + Args: + nfo_path: Absolute path to the ``tvshow.nfo`` file. + + Returns: + List of human-readable tag labels (values from :data:`REQUIRED_TAGS`) + whose XPath matched no elements or only elements with empty text. + An empty list means the NFO is complete. + + Example: + >>> missing = find_missing_tags(Path("/anime/series/tvshow.nfo")) + >>> if missing: + ... print("Missing:", missing) + """ + parsed = parse_nfo_tags(nfo_path) + missing: List[str] = [] + for xpath, label in REQUIRED_TAGS.items(): + if not parsed.get(xpath): + missing.append(label) + return missing + + +def nfo_needs_repair(nfo_path: Path) -> bool: + """Return ``True`` if the NFO is missing any required tag. + + Args: + nfo_path: Absolute path to the ``tvshow.nfo`` file. + + Returns: + True if :func:`find_missing_tags` returns a non-empty list. + + Example: + >>> if nfo_needs_repair(Path("/anime/series/tvshow.nfo")): + ... await service.repair_series(series_path, series_name) + """ + return bool(find_missing_tags(nfo_path)) + + +class NfoRepairService: + """Service that detects and repairs incomplete tvshow.nfo files. + + Wraps the module-level helpers with structured logging and delegates + the actual TMDB re-fetch to an injected :class:`NFOService` instance. + + Attributes: + _nfo_service: The underlying NFOService used to update NFOs. + """ + + def __init__(self, nfo_service: NFOService) -> None: + """Initialise the repair service. + + Args: + nfo_service: Configured :class:`NFOService` instance. + """ + self._nfo_service = nfo_service + + async def repair_series(self, series_path: Path, series_name: str) -> bool: + """Repair an NFO file if required tags are missing. + + Checks ``{series_path}/tvshow.nfo`` for completeness. If tags are + missing, logs them and calls + ``NFOService.update_tvshow_nfo(series_name)`` to re-fetch metadata + from TMDB. + + Args: + series_path: Absolute path to the series folder. + series_name: Series folder name used as the identifier for + :meth:`NFOService.update_tvshow_nfo`. + + Returns: + ``True`` if a repair was triggered, ``False`` if the NFO was + already complete (or did not exist). + """ + nfo_path = series_path / "tvshow.nfo" + missing = find_missing_tags(nfo_path) + + if not missing: + logger.info( + "NFO repair skipped — complete: %s", + series_name, + ) + return False + + logger.info( + "NFO repair triggered for %s — missing tags: %s", + series_name, + ", ".join(missing), + ) + + await self._nfo_service.update_tvshow_nfo( + series_name, + download_media=False, + ) + + logger.info("NFO repair completed: %s", series_name) + return True diff --git a/tests/unit/test_nfo_repair_service.py b/tests/unit/test_nfo_repair_service.py new file mode 100644 index 0000000..83de1de --- /dev/null +++ b/tests/unit/test_nfo_repair_service.py @@ -0,0 +1,140 @@ +"""Unit tests for NfoRepairService — Task 1.""" + +import shutil +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.core.services.nfo_repair_service import ( + NfoRepairService, + REQUIRED_TAGS, + find_missing_tags, + nfo_needs_repair, + parse_nfo_tags, +) + +REPO_ROOT = Path(__file__).parents[2] +BAD_NFO = REPO_ROOT / "tvshow.nfo.bad" +GOOD_NFO = REPO_ROOT / "tvshow.nfo.good" + +# Tags known to be absent/empty in tvshow.nfo.bad +EXPECTED_MISSING_FROM_BAD = { + "originaltitle", "year", "plot", "runtime", "premiered", + "status", "imdbid", "genre", "studio", "country", "actor/name", "watched", +} + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def bad_nfo(tmp_path: Path) -> Path: + """Copy tvshow.nfo.bad into a temp dir and return path to the copy.""" + dest = tmp_path / "tvshow.nfo" + shutil.copy(BAD_NFO, dest) + return dest + + +@pytest.fixture() +def good_nfo(tmp_path: Path) -> Path: + """Copy tvshow.nfo.good into a temp dir and return path to the copy.""" + dest = tmp_path / "tvshow.nfo" + shutil.copy(GOOD_NFO, dest) + return dest + + +@pytest.fixture() +def mock_nfo_service() -> MagicMock: + """Return a MagicMock NFOService with an async update_tvshow_nfo.""" + svc = MagicMock() + svc.update_tvshow_nfo = AsyncMock(return_value=Path("/fake/tvshow.nfo")) + return svc + + +# --------------------------------------------------------------------------- +# find_missing_tags +# --------------------------------------------------------------------------- + + +def test_find_missing_tags_with_bad_nfo(bad_nfo: Path) -> None: + """Bad NFO must report all 12 incomplete/missing tags.""" + missing = find_missing_tags(bad_nfo) + assert set(missing) == EXPECTED_MISSING_FROM_BAD, ( + f"Unexpected missing set: {set(missing)}" + ) + + +def test_find_missing_tags_with_good_nfo(good_nfo: Path) -> None: + """Good NFO must report no missing tags.""" + missing = find_missing_tags(good_nfo) + assert missing == [] + + +# --------------------------------------------------------------------------- +# nfo_needs_repair +# --------------------------------------------------------------------------- + + +def test_nfo_needs_repair_returns_true_for_bad_nfo(bad_nfo: Path) -> None: + assert nfo_needs_repair(bad_nfo) is True + + +def test_nfo_needs_repair_returns_false_for_good_nfo(good_nfo: Path) -> None: + assert nfo_needs_repair(good_nfo) is False + + +# --------------------------------------------------------------------------- +# NfoRepairService.repair_series +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_repair_series_calls_update_when_nfo_needs_repair( + tmp_path: Path, mock_nfo_service: MagicMock +) -> None: + """repair_series must call update_tvshow_nfo exactly once for a bad NFO.""" + shutil.copy(BAD_NFO, tmp_path / "tvshow.nfo") + service = NfoRepairService(mock_nfo_service) + + result = await service.repair_series(tmp_path, "Test Series") + + assert result is True + mock_nfo_service.update_tvshow_nfo.assert_called_once_with( + "Test Series", download_media=False + ) + + +@pytest.mark.asyncio +async def test_repair_series_skips_when_nfo_is_complete( + tmp_path: Path, mock_nfo_service: MagicMock +) -> None: + """repair_series must NOT call update_tvshow_nfo for a complete NFO.""" + shutil.copy(GOOD_NFO, tmp_path / "tvshow.nfo") + service = NfoRepairService(mock_nfo_service) + + result = await service.repair_series(tmp_path, "Test Series") + + assert result is False + mock_nfo_service.update_tvshow_nfo.assert_not_called() + + +# --------------------------------------------------------------------------- +# parse_nfo_tags edge cases +# --------------------------------------------------------------------------- + + +def test_parse_nfo_tags_handles_missing_file_gracefully() -> None: + """parse_nfo_tags must return empty dict for non-existent path.""" + result = parse_nfo_tags(Path("/nonexistent/dir/tvshow.nfo")) + assert result == {} + + +def test_parse_nfo_tags_handles_malformed_xml_gracefully(tmp_path: Path) -> None: + """parse_nfo_tags must return empty dict for malformed XML.""" + bad_xml = tmp_path / "tvshow.nfo" + bad_xml.write_text("<<< not valid xml >>>", encoding="utf-8") + result = parse_nfo_tags(bad_xml) + assert result == {}