feat: add NfoRepairService for missing NFO tag detection

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 <tvshow> 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