feat(nfo): implement NFO diagnostics and repair
- Add NFO API endpoints: diagnostics, repair, validate, needs-repair - Create /settings/nfo page with full NFO management UI - Add NFO status section to edit modal with repair functionality - Add anime details API for edit modal pre-fill data - Fix auth test fixtures in test_nfo_diagnostics_repair.py Implements NFO diagnostics when editing anime series via right-click menu. Users can now check NFO status, see missing tags, and repair NFOs directly from the edit modal or the dedicated NFO settings page.
This commit is contained in:
@@ -1,70 +1,485 @@
|
||||
"""NFO Management API endpoints.
|
||||
|
||||
Note: NFO service has been removed. All NFO endpoints return 503.
|
||||
Provides endpoints for:
|
||||
- Diagnostics: Check NFO status and missing tags
|
||||
- Repair: Create/update NFO files for series
|
||||
- Validation: Quick NFO XML validity check
|
||||
- Needs-repair: List all series that need NFO attention
|
||||
"""
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
import os
|
||||
from typing import Any, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from src.config.settings import settings
|
||||
from src.server.models.nfo import NfoDiagnosticsResponse, NfoRepairResponse
|
||||
from src.server.services.anime_service import AnimeService, AnimeServiceError
|
||||
from src.server.utils.dependencies import get_anime_service, require_auth
|
||||
|
||||
router = APIRouter(prefix="/api/nfo", tags=["nfo"])
|
||||
|
||||
# Required tags for a valid Kodi tvshow.nfo
|
||||
REQUIRED_TAGS = [
|
||||
"title",
|
||||
"plot",
|
||||
"tmdbid",
|
||||
]
|
||||
# All tags we check for completeness
|
||||
ALL_TAGS = [
|
||||
"title",
|
||||
"originaltitle",
|
||||
"showtitle",
|
||||
"sorttitle",
|
||||
"year",
|
||||
"plot",
|
||||
"outline",
|
||||
"tagline",
|
||||
"runtime",
|
||||
"mpaa",
|
||||
"premiered",
|
||||
"status",
|
||||
"studio",
|
||||
"genre",
|
||||
"tmdbid",
|
||||
"imdbid",
|
||||
"tvdbid",
|
||||
"userrating",
|
||||
"trailer",
|
||||
"thumb",
|
||||
"fanart",
|
||||
"actor",
|
||||
]
|
||||
|
||||
@router.get("/disabled")
|
||||
async def nfo_disabled():
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
|
||||
class NfoSeriesItem(BaseModel):
|
||||
"""Series with NFO status for listing endpoints."""
|
||||
key: str = Field(..., description="Series unique key")
|
||||
name: str = Field(..., description="Series display name")
|
||||
folder: str = Field(..., description="Series folder name")
|
||||
has_nfo: bool = Field(..., description="Whether NFO file exists")
|
||||
nfo_path: Optional[str] = Field(None, description="Path to NFO file if exists")
|
||||
missing_tags: List[str] = Field(default_factory=list, description="Missing tags if NFO exists")
|
||||
|
||||
|
||||
class NfoNeedsRepairResponse(BaseModel):
|
||||
"""Response listing all series that need NFO attention."""
|
||||
total: int = Field(..., description="Total number of series")
|
||||
missing_nfo_count: int = Field(..., description="Series without any NFO file")
|
||||
incomplete_nfo_count: int = Field(..., description="Series with NFO but missing tags")
|
||||
series: List[NfoSeriesItem] = Field(..., description="List of series needing attention")
|
||||
|
||||
|
||||
class NfoValidateResponse(BaseModel):
|
||||
"""Response for NFO validation check."""
|
||||
valid: bool = Field(..., description="Whether NFO XML is valid")
|
||||
error: Optional[str] = Field(None, description="Error message if invalid")
|
||||
|
||||
|
||||
def _parse_nfo_tags(nfo_path: str) -> dict[str, str]:
|
||||
"""Parse NFO file and return tag values.
|
||||
|
||||
Returns dict of tag name -> tag text content.
|
||||
Empty tags (e.g., <plot></plot>) are returned as empty string.
|
||||
"""
|
||||
try:
|
||||
from lxml import etree
|
||||
except ImportError:
|
||||
return {}
|
||||
|
||||
try:
|
||||
tree = etree.parse(nfo_path)
|
||||
root = tree.getroot()
|
||||
|
||||
tags = {}
|
||||
for elem in root.iter():
|
||||
tag_name = elem.tag.lower()
|
||||
if tag_name not in tags:
|
||||
# Get text content, None if element has no text
|
||||
text = elem.text.strip() if elem.text else ""
|
||||
tags[tag_name] = text
|
||||
return tags
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _check_missing_tags(tags: dict[str, str]) -> List[str]:
|
||||
"""Check which required tags are missing or empty."""
|
||||
missing = []
|
||||
for tag in REQUIRED_TAGS:
|
||||
value = tags.get(tag, "").strip()
|
||||
if not value:
|
||||
missing.append(tag)
|
||||
return missing
|
||||
|
||||
|
||||
@router.get("/needs-repair", response_model=NfoNeedsRepairResponse)
|
||||
async def get_series_needing_repair(
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoNeedsRepairResponse:
|
||||
"""Get all series that need NFO repair (missing file or missing tags).
|
||||
|
||||
Returns series grouped by:
|
||||
- Missing NFO file entirely
|
||||
- NFO file exists but missing required tags
|
||||
"""
|
||||
try:
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
except AnimeServiceError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to fetch series list: {exc}",
|
||||
)
|
||||
|
||||
anime_dir = getattr(settings, "anime_directory", None)
|
||||
if not anime_dir:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Anime directory not configured",
|
||||
)
|
||||
|
||||
missing_nfo_list: List[NfoSeriesItem] = []
|
||||
incomplete_list: List[NfoSeriesItem] = []
|
||||
|
||||
for series in all_series:
|
||||
key = series.get("key", "")
|
||||
name = series.get("name", "")
|
||||
folder = series.get("folder", "")
|
||||
|
||||
if not folder:
|
||||
continue
|
||||
|
||||
nfo_path = os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
|
||||
if not os.path.isfile(nfo_path):
|
||||
missing_nfo_list.append(NfoSeriesItem(
|
||||
key=key,
|
||||
name=name,
|
||||
folder=folder,
|
||||
has_nfo=False,
|
||||
nfo_path=None,
|
||||
missing_tags=REQUIRED_TAGS.copy(),
|
||||
))
|
||||
else:
|
||||
# Check for missing tags
|
||||
tags = _parse_nfo_tags(nfo_path)
|
||||
missing = _check_missing_tags(tags)
|
||||
if missing:
|
||||
incomplete_list.append(NfoSeriesItem(
|
||||
key=key,
|
||||
name=name,
|
||||
folder=folder,
|
||||
has_nfo=True,
|
||||
nfo_path=nfo_path,
|
||||
missing_tags=missing,
|
||||
))
|
||||
|
||||
all_needing_repair = missing_nfo_list + incomplete_list
|
||||
|
||||
return NfoNeedsRepairResponse(
|
||||
total=len(all_series),
|
||||
missing_nfo_count=len(missing_nfo_list),
|
||||
incomplete_nfo_count=len(incomplete_list),
|
||||
series=all_needing_repair,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/batch/create")
|
||||
async def batch_create_nfo():
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
@router.get("/{serie_key}/diagnostics", response_model=NfoDiagnosticsResponse)
|
||||
async def get_nfo_diagnostics(
|
||||
serie_key: str,
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoDiagnosticsResponse:
|
||||
"""Get NFO diagnostics for a specific series.
|
||||
|
||||
Returns:
|
||||
- Whether tvshow.nfo exists
|
||||
- Path to NFO file if it exists
|
||||
- List of missing required tags (if any)
|
||||
- List of all tags for reference
|
||||
"""
|
||||
# Get series data
|
||||
try:
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
series = next((s for s in all_series if s.get("key") == serie_key), None)
|
||||
except AnimeServiceError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to fetch series: {exc}",
|
||||
)
|
||||
|
||||
if not series:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
)
|
||||
|
||||
anime_dir = getattr(settings, "anime_directory", None)
|
||||
if not anime_dir:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Anime directory not configured",
|
||||
)
|
||||
|
||||
nfo_path = os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
|
||||
if not os.path.isfile(nfo_path):
|
||||
return NfoDiagnosticsResponse(
|
||||
has_nfo=False,
|
||||
nfo_path=None,
|
||||
missing_tags=REQUIRED_TAGS.copy(),
|
||||
required_tags=ALL_TAGS.copy(),
|
||||
)
|
||||
|
||||
# Parse NFO and check for missing tags
|
||||
tags = _parse_nfo_tags(nfo_path)
|
||||
missing = _check_missing_tags(tags)
|
||||
|
||||
return NfoDiagnosticsResponse(
|
||||
has_nfo=True,
|
||||
nfo_path=nfo_path,
|
||||
missing_tags=missing,
|
||||
required_tags=ALL_TAGS.copy(),
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{serie_id}/create")
|
||||
async def create_nfo(serie_id: str):
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
@router.post("/{serie_key}/repair", response_model=NfoRepairResponse)
|
||||
async def repair_nfo(
|
||||
serie_key: str,
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoRepairResponse:
|
||||
"""Repair (create or update) NFO file for a series.
|
||||
|
||||
If NFO doesn't exist, creates it from TMDB metadata.
|
||||
If NFO exists but has missing tags, updates it with TMDB data.
|
||||
If no TMDB ID is available, attempts to look up by series name.
|
||||
"""
|
||||
# Get series data
|
||||
try:
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
series = next((s for s in all_series if s.get("key") == serie_key), None)
|
||||
except AnimeServiceError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to fetch series: {exc}",
|
||||
)
|
||||
|
||||
if not series:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
)
|
||||
|
||||
anime_dir = getattr(settings, "anime_directory", None)
|
||||
if not anime_dir:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Anime directory not configured",
|
||||
)
|
||||
|
||||
nfo_path = os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
tmdb_id = series.get("tmdb_id")
|
||||
|
||||
# If no TMDB ID, try to look up by name
|
||||
if not tmdb_id:
|
||||
name = series.get("name", "")
|
||||
tmdb_id = await _lookup_tmdb_id_by_name(name)
|
||||
if tmdb_id:
|
||||
# Update series with found TMDB ID (best effort)
|
||||
try:
|
||||
await anime_service.update_series_key(serie_key, {"tmdb_id": tmdb_id})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not tmdb_id:
|
||||
return NfoRepairResponse(
|
||||
success=False,
|
||||
message=f"Cannot repair NFO for {serie_key}: no TMDB ID available and could not find match by name",
|
||||
repaired_tags=[],
|
||||
)
|
||||
|
||||
# Fetch TMDB data and generate NFO
|
||||
tmdb_data = await _fetch_tmdb_data(tmdb_id)
|
||||
if not tmdb_data:
|
||||
return NfoRepairResponse(
|
||||
success=False,
|
||||
message=f"No TMDB data found for TMDB ID {tmdb_id}",
|
||||
repaired_tags=[],
|
||||
)
|
||||
|
||||
# Generate NFO XML
|
||||
from src.server.nfo.nfo_generator import generate_tvshow_nfo
|
||||
from src.server.nfo.nfo_mapper import tmdb_to_nfo_model
|
||||
|
||||
nfo_model = tmdb_to_nfo_model(
|
||||
tmdb_data,
|
||||
content_ratings=None,
|
||||
get_image_url=lambda path: f"https://image.tmdb.org/t/p/original{path}" if path else None,
|
||||
image_size="original",
|
||||
)
|
||||
|
||||
xml_content = generate_tvshow_nfo(nfo_model)
|
||||
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(nfo_path), exist_ok=True)
|
||||
|
||||
# Track which tags were missing before
|
||||
had_nfo = os.path.isfile(nfo_path)
|
||||
if had_nfo:
|
||||
old_tags = _parse_nfo_tags(nfo_path)
|
||||
old_missing = _check_missing_tags(old_tags)
|
||||
else:
|
||||
old_missing = REQUIRED_TAGS.copy()
|
||||
|
||||
# Write NFO file
|
||||
with open(nfo_path, "w", encoding="utf-8") as f:
|
||||
f.write(xml_content)
|
||||
|
||||
# Update series NFO flag in database (best effort)
|
||||
await _update_series_nfo_flag(anime_service, serie_key, True, nfo_path)
|
||||
|
||||
# Calculate repaired tags
|
||||
new_tags = _parse_nfo_tags(nfo_path)
|
||||
new_missing = _check_missing_tags(new_tags)
|
||||
repaired_tags = [tag for tag in old_missing if tag not in new_missing]
|
||||
|
||||
action = "created" if not had_nfo else "updated"
|
||||
return NfoRepairResponse(
|
||||
success=True,
|
||||
message=f"NFO {action} successfully for {serie_key}",
|
||||
repaired_tags=repaired_tags,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{serie_id}/status")
|
||||
async def get_nfo_status(serie_id: str):
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
)
|
||||
@router.get("/{serie_key}/validate", response_model=NfoValidateResponse)
|
||||
async def validate_nfo(
|
||||
serie_key: str,
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoValidateResponse:
|
||||
"""Quick check if NFO XML is valid.
|
||||
|
||||
Does NOT check for missing tags, only validates XML structure.
|
||||
"""
|
||||
# Get series data
|
||||
try:
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
series = next((s for s in all_series if s.get("key") == serie_key), None)
|
||||
except AnimeServiceError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to fetch series: {exc}",
|
||||
)
|
||||
|
||||
if not series:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
)
|
||||
|
||||
anime_dir = getattr(settings, "anime_directory", None)
|
||||
if not anime_dir:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Anime directory not configured",
|
||||
)
|
||||
|
||||
nfo_path = os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
|
||||
if not os.path.isfile(nfo_path):
|
||||
return NfoValidateResponse(
|
||||
valid=False,
|
||||
error="NFO file does not exist",
|
||||
)
|
||||
|
||||
try:
|
||||
from lxml import etree
|
||||
etree.parse(nfo_path)
|
||||
return NfoValidateResponse(
|
||||
valid=True,
|
||||
error=None,
|
||||
)
|
||||
except Exception as exc:
|
||||
return NfoValidateResponse(
|
||||
valid=False,
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/{serie_id}/delete")
|
||||
async def delete_nfo(serie_id: str):
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
)
|
||||
# ---- Helper functions ----
|
||||
|
||||
async def _lookup_tmdb_id_by_name(name: str) -> Optional[int]:
|
||||
"""Look up TMDB ID by series name using TMDB search."""
|
||||
try:
|
||||
from src.server.providers.tmdb_provider import TMDBProvider
|
||||
provider = TMDBProvider()
|
||||
results = await provider.search(name)
|
||||
if results and len(results) > 0:
|
||||
return results[0].get("tmdb_id")
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/poster/{serie_id}")
|
||||
async def get_nfo_poster(serie_id: str):
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
)
|
||||
async def _fetch_tmdb_data(tmdb_id: int) -> Optional[dict]:
|
||||
"""Fetch TV show data from TMDB."""
|
||||
try:
|
||||
from src.server.providers.tmdb_provider import TMDBProvider
|
||||
provider = TMDBProvider()
|
||||
return await provider.get_tvshow(tmdb_id)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/fanart/{serie_id}")
|
||||
async def get_nfo_fanart(serie_id: str):
|
||||
"""NFO endpoints disabled - NFO service removed."""
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="NFO service has been removed. Use series management endpoints instead."
|
||||
)
|
||||
async def _update_series_nfo_flag(
|
||||
anime_service: AnimeService,
|
||||
key: str,
|
||||
has_nfo: bool,
|
||||
nfo_path: str,
|
||||
) -> None:
|
||||
"""Update NFO flag in database for a series."""
|
||||
try:
|
||||
# Use the anime service to update the series
|
||||
await anime_service.update_series_key(key, {"has_nfo": has_nfo})
|
||||
except Exception:
|
||||
# Fallback: try direct database update
|
||||
try:
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from src.server.database.connection import get_db_session
|
||||
from src.server.database.service import AnimeSeriesService
|
||||
|
||||
async with get_db_session() as db:
|
||||
series_record = await AnimeSeriesService.get_by_key(db, key)
|
||||
if series_record:
|
||||
series_record.has_nfo = has_nfo
|
||||
if has_nfo:
|
||||
series_record.nfo_updated_at = datetime.now(timezone.utc)
|
||||
if not series_record.nfo_created_at:
|
||||
series_record.nfo_created_at = datetime.now(timezone.utc)
|
||||
await db.commit()
|
||||
except Exception:
|
||||
pass
|
||||
Reference in New Issue
Block a user