refactor: overhaul NFO settings UI and backend
- Rework nfo-settings page with improved styling and layout - Update edit-modal and context-menu with enhanced functionality - Refactor NFO API endpoints and models - Remove deprecated test_nfo_diagnostics_repair.py - Clean up tasks.md documentation
This commit is contained in:
@@ -1,22 +1,25 @@
|
||||
"""NFO Management API endpoints.
|
||||
|
||||
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
|
||||
Provides endpoints for NFO diagnostics, repair, and validation for anime series.
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic import BaseModel
|
||||
|
||||
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.models.nfo import (
|
||||
NfoDiagnosticsResponse,
|
||||
NfoRepairResponse,
|
||||
NfoSeriesDiagnostics,
|
||||
)
|
||||
from src.server.services.anime_service import AnimeService
|
||||
from src.server.utils.dependencies import get_anime_service, require_auth
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/nfo", tags=["nfo"])
|
||||
|
||||
# Required tags for a valid Kodi tvshow.nfo
|
||||
@@ -25,403 +28,365 @@ REQUIRED_TAGS = [
|
||||
"plot",
|
||||
"tmdbid",
|
||||
]
|
||||
# All tags we check for completeness
|
||||
ALL_TAGS = [
|
||||
"title",
|
||||
"originaltitle",
|
||||
"showtitle",
|
||||
"sorttitle",
|
||||
OPTIONAL_TAGS = [
|
||||
"year",
|
||||
"plot",
|
||||
"premiered",
|
||||
"genre",
|
||||
"studio",
|
||||
"rating",
|
||||
"mpaa",
|
||||
"outline",
|
||||
"tagline",
|
||||
"runtime",
|
||||
"mpaa",
|
||||
"premiered",
|
||||
"status",
|
||||
"studio",
|
||||
"genre",
|
||||
"tmdbid",
|
||||
"imdbid",
|
||||
"id",
|
||||
"imdb_id",
|
||||
"tvdbid",
|
||||
"userrating",
|
||||
"trailer",
|
||||
"imdbid",
|
||||
"uniqueid",
|
||||
"thumb",
|
||||
"fanart",
|
||||
"actor",
|
||||
"trailer",
|
||||
]
|
||||
|
||||
|
||||
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")
|
||||
"""Response for NFO XML validation."""
|
||||
|
||||
valid: bool
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
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 {}
|
||||
class NfoNeedsRepairListResponse(BaseModel):
|
||||
"""Response listing series that need NFO repair."""
|
||||
|
||||
total: int
|
||||
series: List[NfoSeriesDiagnostics]
|
||||
|
||||
|
||||
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}",
|
||||
)
|
||||
|
||||
def _get_nfo_path(folder: str) -> str:
|
||||
"""Get the full path to a series' tvshow.nfo file."""
|
||||
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] = []
|
||||
|
||||
return os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
|
||||
|
||||
def _parse_nfo_file(nfo_path: str) -> tuple[Optional[Any], List[str]]:
|
||||
"""Parse an NFO file and return (xml_root, missing_tags).
|
||||
|
||||
Returns tuple of (xml_root element or None, list of missing required tags).
|
||||
If file cannot be read/parsed, returns (None, all_required_tags).
|
||||
"""
|
||||
from lxml import etree
|
||||
|
||||
missing: List[str] = []
|
||||
|
||||
if not os.path.isfile(nfo_path):
|
||||
return None, REQUIRED_TAGS.copy()
|
||||
|
||||
try:
|
||||
tree = etree.parse(nfo_path)
|
||||
root = tree.getroot()
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to parse NFO file %s: %s", nfo_path, exc)
|
||||
return None, REQUIRED_TAGS.copy()
|
||||
|
||||
# Check for required tags
|
||||
for tag in REQUIRED_TAGS:
|
||||
elements = root.findall(tag)
|
||||
# Check if tag exists and has non-empty text
|
||||
found = False
|
||||
for elem in elements:
|
||||
if elem.text and elem.text.strip():
|
||||
found = True
|
||||
break
|
||||
if not found:
|
||||
missing.append(tag)
|
||||
|
||||
return root, missing
|
||||
|
||||
|
||||
async def _get_series_data(
|
||||
anime_service: AnimeService, key: str
|
||||
) -> Optional[dict]:
|
||||
"""Get series data by key from anime_service."""
|
||||
# Get all series and find by key
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
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,
|
||||
)
|
||||
if series.get("key") == key:
|
||||
return series
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/{serie_key}/diagnostics", response_model=NfoDiagnosticsResponse)
|
||||
@router.get("/{key}/diagnostics", response_model=NfoDiagnosticsResponse)
|
||||
async def get_nfo_diagnostics(
|
||||
serie_key: str,
|
||||
key: str,
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoDiagnosticsResponse:
|
||||
"""Get NFO diagnostics for a specific series.
|
||||
|
||||
|
||||
Checks if tvshow.nfo exists in the series folder and validates
|
||||
that required tags are present.
|
||||
|
||||
Args:
|
||||
key: Series unique key (provider-assigned, URL-safe identifier)
|
||||
_auth: Authentication dependency
|
||||
anime_service: AnimeService dependency
|
||||
|
||||
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
|
||||
NfoDiagnosticsResponse with has_nfo, nfo_path, missing_tags, required_tags
|
||||
|
||||
Raises:
|
||||
HTTPException 404: If series not found
|
||||
HTTPException 503: If anime directory not configured
|
||||
"""
|
||||
# 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:
|
||||
series_data = await _get_series_data(anime_service, key)
|
||||
if not series_data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
detail=f"Series not found: {key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
|
||||
folder = series_data.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
detail=f"Series has no folder configured: {key}",
|
||||
)
|
||||
|
||||
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):
|
||||
|
||||
nfo_path = _get_nfo_path(folder)
|
||||
nfo_exists = os.path.isfile(nfo_path)
|
||||
|
||||
if not nfo_exists:
|
||||
return NfoDiagnosticsResponse(
|
||||
has_nfo=False,
|
||||
nfo_path=None,
|
||||
missing_tags=REQUIRED_TAGS.copy(),
|
||||
required_tags=ALL_TAGS.copy(),
|
||||
required_tags=REQUIRED_TAGS.copy(),
|
||||
)
|
||||
|
||||
# Parse NFO and check for missing tags
|
||||
tags = _parse_nfo_tags(nfo_path)
|
||||
missing = _check_missing_tags(tags)
|
||||
|
||||
|
||||
# Parse and check for missing tags
|
||||
_, missing = _parse_nfo_file(nfo_path)
|
||||
|
||||
return NfoDiagnosticsResponse(
|
||||
has_nfo=True,
|
||||
nfo_path=nfo_path,
|
||||
missing_tags=missing,
|
||||
required_tags=ALL_TAGS.copy(),
|
||||
required_tags=REQUIRED_TAGS.copy(),
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{serie_key}/repair", response_model=NfoRepairResponse)
|
||||
@router.post("/{key}/repair", response_model=NfoRepairResponse)
|
||||
async def repair_nfo(
|
||||
serie_key: str,
|
||||
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.
|
||||
"""Repair NFO for a specific series.
|
||||
|
||||
Creates or updates the tvshow.nfo file using TMDB metadata.
|
||||
|
||||
Args:
|
||||
key: Series unique key
|
||||
_auth: Authentication dependency
|
||||
anime_service: AnimeService dependency
|
||||
|
||||
Returns:
|
||||
NfoRepairResponse with success status, message, and repaired_tags
|
||||
|
||||
Raises:
|
||||
HTTPException 404: If series not found
|
||||
HTTPException 400: If no TMDB ID available and cannot lookup by name
|
||||
HTTPException 503: If anime directory not configured
|
||||
"""
|
||||
# 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:
|
||||
series_data = await _get_series_data(anime_service, key)
|
||||
if not series_data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
detail=f"Series not found: {key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
|
||||
folder = series_data.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
detail=f"Series has no folder configured: {key}",
|
||||
)
|
||||
|
||||
|
||||
tmdb_id = series_data.get("tmdb_id")
|
||||
name = series_data.get("name", "")
|
||||
|
||||
if not tmdb_id:
|
||||
logger.info("No TMDB ID for %s, attempting lookup by name: %s", key, name)
|
||||
# Try to lookup TMDB ID by series name
|
||||
tmdb_id = await _lookup_tmdb_id_by_name(anime_service, name)
|
||||
if not tmdb_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"No TMDB ID available for {key} and could not find match for '{name}'",
|
||||
)
|
||||
|
||||
# Fetch TMDB data and create NFO
|
||||
try:
|
||||
repaired_tags = await _create_or_update_nfo(
|
||||
key=key,
|
||||
folder=folder,
|
||||
tmdb_id=tmdb_id,
|
||||
series_data=series_data,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.error("Failed to repair NFO for %s: %s", key, exc)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to repair NFO: {str(exc)}",
|
||||
)
|
||||
|
||||
if repaired_tags:
|
||||
return NfoRepairResponse(
|
||||
success=True,
|
||||
message=f"NFO repaired successfully. {len(repaired_tags)} tags updated.",
|
||||
repaired_tags=repaired_tags,
|
||||
)
|
||||
else:
|
||||
return NfoRepairResponse(
|
||||
success=True,
|
||||
message="NFO is already complete, no changes needed.",
|
||||
repaired_tags=[],
|
||||
)
|
||||
|
||||
|
||||
async def _lookup_tmdb_id_by_name(anime_service: AnimeService, name: str) -> Optional[int]:
|
||||
"""Try to lookup a TMDB ID by series name using TMDB provider."""
|
||||
try:
|
||||
from src.server.providers.tmdb_provider import TMDBProvider
|
||||
|
||||
provider = TMDBProvider()
|
||||
results = await provider.search(name)
|
||||
if results:
|
||||
return results[0].get("tmdb_id")
|
||||
except Exception as exc:
|
||||
logger.warning("TMDB lookup failed for %s: %s", name, exc)
|
||||
return None
|
||||
|
||||
|
||||
async def _create_or_update_nfo(
|
||||
key: str,
|
||||
folder: str,
|
||||
tmdb_id: int,
|
||||
series_data: dict,
|
||||
) -> List[str]:
|
||||
"""Create or update NFO file for a series.
|
||||
|
||||
Returns list of tags that were repaired/added.
|
||||
"""
|
||||
from src.server.providers.tmdb_provider import TMDBProvider
|
||||
|
||||
from src.server.nfo.nfo_generator import generate_tvshow_nfo
|
||||
from src.server.nfo.nfo_mapper import tmdb_to_nfo_model
|
||||
|
||||
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)
|
||||
|
||||
series_path = os.path.join(anime_dir, folder)
|
||||
nfo_path = os.path.join(series_path, "tvshow.nfo")
|
||||
|
||||
# Fetch TMDB data
|
||||
provider = TMDBProvider()
|
||||
tmdb_data = await provider.get_series_info(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
|
||||
|
||||
raise Exception(f"No TMDB data returned for TMDB ID {tmdb_id}")
|
||||
|
||||
# Convert 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,
|
||||
get_image_url=provider.get_image_url,
|
||||
image_size="original",
|
||||
)
|
||||
|
||||
|
||||
# Generate XML
|
||||
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()
|
||||
|
||||
os.makedirs(series_path, exist_ok=True)
|
||||
|
||||
# Check existing NFO for missing tags before overwriting
|
||||
_, missing_before = _parse_nfo_file(nfo_path)
|
||||
|
||||
# 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,
|
||||
|
||||
logger.info("Created/updated tvshow.nfo for %s at %s", key, nfo_path)
|
||||
|
||||
# Update series NFO status in DB
|
||||
await anime_service.update_series_nfo_status(
|
||||
key=key,
|
||||
has_nfo=True,
|
||||
nfo_path=nfo_path,
|
||||
)
|
||||
|
||||
# Return list of repaired tags (all tags that were missing before)
|
||||
return missing_before
|
||||
|
||||
@router.get("/{serie_key}/validate", response_model=NfoValidateResponse)
|
||||
|
||||
@router.get("/{key}/validate", response_model=NfoValidateResponse)
|
||||
async def validate_nfo(
|
||||
serie_key: str,
|
||||
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.
|
||||
"""Validate NFO XML structure for a series.
|
||||
|
||||
Checks if the tvshow.nfo file is valid XML.
|
||||
|
||||
Args:
|
||||
key: Series unique key
|
||||
_auth: Authentication dependency
|
||||
anime_service: AnimeService dependency
|
||||
|
||||
Returns:
|
||||
NfoValidateResponse with valid=True/False and optional error message
|
||||
"""
|
||||
# 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:
|
||||
series_data = await _get_series_data(anime_service, key)
|
||||
if not series_data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Series not found: {serie_key}",
|
||||
detail=f"Series not found: {key}",
|
||||
)
|
||||
|
||||
folder = series.get("folder", "")
|
||||
|
||||
folder = series_data.get("folder", "")
|
||||
if not folder:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Series has no folder assigned",
|
||||
detail=f"Series has no folder configured: {key}",
|
||||
)
|
||||
|
||||
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")
|
||||
|
||||
|
||||
nfo_path = _get_nfo_path(folder)
|
||||
|
||||
if not os.path.isfile(nfo_path):
|
||||
return NfoValidateResponse(
|
||||
valid=False,
|
||||
error="NFO file does not exist",
|
||||
error="No NFO file found",
|
||||
)
|
||||
|
||||
|
||||
try:
|
||||
from lxml import etree
|
||||
|
||||
etree.parse(nfo_path)
|
||||
return NfoValidateResponse(
|
||||
valid=True,
|
||||
error=None,
|
||||
)
|
||||
return NfoValidateResponse(valid=True)
|
||||
except Exception as exc:
|
||||
return NfoValidateResponse(
|
||||
valid=False,
|
||||
@@ -429,57 +394,128 @@ async def validate_nfo(
|
||||
)
|
||||
|
||||
|
||||
# ---- Helper functions ----
|
||||
@router.get("/needs-repair", response_model=NfoNeedsRepairListResponse)
|
||||
async def get_series_needing_repair(
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> NfoNeedsRepairListResponse:
|
||||
"""Get list of all series that need NFO repair.
|
||||
|
||||
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
|
||||
Returns series that either have no NFO file or have missing required tags.
|
||||
|
||||
Args:
|
||||
_auth: Authentication dependency
|
||||
anime_service: AnimeService dependency
|
||||
|
||||
Returns:
|
||||
NfoNeedsRepairListResponse with total count and list of series diagnostics
|
||||
"""
|
||||
all_series = await anime_service.list_series_with_filters()
|
||||
series_needing_repair: List[NfoSeriesDiagnostics] = []
|
||||
|
||||
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",
|
||||
)
|
||||
|
||||
for series in all_series:
|
||||
key = series.get("key", "")
|
||||
folder = series.get("folder", "")
|
||||
name = series.get("name", "")
|
||||
|
||||
if not folder:
|
||||
continue
|
||||
|
||||
nfo_path = os.path.join(anime_dir, folder, "tvshow.nfo")
|
||||
nfo_exists = os.path.isfile(nfo_path)
|
||||
|
||||
if not nfo_exists:
|
||||
series_needing_repair.append(NfoSeriesDiagnostics(
|
||||
key=key,
|
||||
name=name,
|
||||
folder=folder,
|
||||
has_nfo=False,
|
||||
missing_tags=REQUIRED_TAGS.copy(),
|
||||
))
|
||||
continue
|
||||
|
||||
# Parse and check for missing tags
|
||||
_, missing = _parse_nfo_file(nfo_path)
|
||||
if missing:
|
||||
series_needing_repair.append(NfoSeriesDiagnostics(
|
||||
key=key,
|
||||
name=name,
|
||||
folder=folder,
|
||||
has_nfo=True,
|
||||
missing_tags=missing,
|
||||
))
|
||||
|
||||
return NfoNeedsRepairListResponse(
|
||||
total=len(series_needing_repair),
|
||||
series=series_needing_repair,
|
||||
)
|
||||
|
||||
|
||||
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.post("/batch/repair")
|
||||
async def batch_repair_nfo(
|
||||
keys: List[str],
|
||||
_auth: dict = Depends(require_auth),
|
||||
anime_service: AnimeService = Depends(get_anime_service),
|
||||
) -> dict:
|
||||
"""Repair NFO for multiple series at once.
|
||||
|
||||
Args:
|
||||
keys: List of series keys to repair
|
||||
_auth: Authentication dependency
|
||||
anime_service: AnimeService dependency
|
||||
|
||||
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
|
||||
Returns:
|
||||
Summary dict with success count, failure count, and errors
|
||||
"""
|
||||
results = {
|
||||
"total": len(keys),
|
||||
"success": 0,
|
||||
"failed": 0,
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
for key in keys:
|
||||
try:
|
||||
from datetime import datetime, timezone
|
||||
# Get series data
|
||||
series_data = await _get_series_data(anime_service, key)
|
||||
if not series_data:
|
||||
results["failed"] += 1
|
||||
results["errors"].append(f"{key}: Series not found")
|
||||
continue
|
||||
|
||||
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
|
||||
folder = series_data.get("folder", "")
|
||||
if not folder:
|
||||
results["failed"] += 1
|
||||
results["errors"].append(f"{key}: No folder configured")
|
||||
continue
|
||||
|
||||
tmdb_id = series_data.get("tmdb_id")
|
||||
name = series_data.get("name", "")
|
||||
|
||||
if not tmdb_id:
|
||||
tmdb_id = await _lookup_tmdb_id_by_name(anime_service, name)
|
||||
if not tmdb_id:
|
||||
results["failed"] += 1
|
||||
results["errors"].append(f"{key}: No TMDB ID and lookup failed")
|
||||
continue
|
||||
|
||||
await _create_or_update_nfo(
|
||||
key=key,
|
||||
folder=folder,
|
||||
tmdb_id=tmdb_id,
|
||||
series_data=series_data,
|
||||
)
|
||||
results["success"] += 1
|
||||
|
||||
except Exception as exc:
|
||||
results["failed"] += 1
|
||||
results["errors"].append(f"{key}: {str(exc)}")
|
||||
|
||||
return results
|
||||
|
||||
Reference in New Issue
Block a user