lukas.pupkalipinski/Aniworld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Code Duplication 4: Create media utilities module

Browse Source 
- Created src/server/utils/media.py with reusable media file functions
- Functions: check_media_files(), get_media_file_paths(), has_all_images(), count_video_files(), has_video_files()
- Defined standard filename constants: POSTER_FILENAME, LOGO_FILENAME, FANART_FILENAME, NFO_FILENAME
- Defined VIDEO_EXTENSIONS set for media player compatibility
- Refactored src/server/api/nfo.py (7 locations) to use utility functions
- Refactored src/server/services/background_loader_service.py to use utility
- Functions accept both str and Path for compatibility
- Marked Code Duplications 1, 3, 4 as RESOLVED in instructions.md
- Updated Further Considerations as RESOLVED (addressed in Issues 7, 9, 10)
This commit is contained in:
Lukas
2026-01-24 21:34:43 +01:00
parent 4abaf8def7
commit 46271a9845
6 changed files with 390 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
src/server/api/nfo.py
View File

@@ -30,6 +30,7 @@ from src.server.models.nfo import (
NFOMissingSeries,
)
from src.server.utils.dependencies import get_series_app, require_auth
from src.server.utils.media import check_media_files, get_media_file_paths
logger = logging.getLogger(__name__)
@@ -73,31 +74,6 @@ async def get_nfo_service() -> NFOService:
)
def check_media_files(serie_folder: str) -> MediaFilesStatus:
"""Check status of media files for a series.
Args:
serie_folder: Series folder name
Returns:
MediaFilesStatus with file existence info
"""
folder_path = Path(settings.anime_directory) / serie_folder
poster_path = folder_path / "poster.jpg"
logo_path = folder_path / "logo.png"
fanart_path = folder_path / "fanart.jpg"
return MediaFilesStatus(
has_poster=poster_path.exists(),
has_logo=logo_path.exists(),
has_fanart=fanart_path.exists(),
poster_path=str(poster_path) if poster_path.exists() else None,
logo_path=str(logo_path) if logo_path.exists() else None,
fanart_path=str(fanart_path) if fanart_path.exists() else None
)
@router.get("/{serie_id}/check", response_model=NFOCheckResponse)
async def check_nfo(
serie_id: str,
@@ -135,17 +111,35 @@ async def check_nfo(
# Ensure folder name includes year if available
serie_folder = serie.ensure_folder_with_year()
folder_path = Path(settings.anime_directory) / serie_folder
# Check NFO
has_nfo = await nfo_service.check_nfo_exists(serie_folder)
nfo_path = None
if has_nfo:
nfo_path = str(
Path(settings.anime_directory) / serie_folder / "tvshow.nfo"
)
nfo_path = str(folder_path / "tvshow.nfo")
# Check media files
media_files = check_media_files(serie_folder)
# Check media files using utility function
media_status = check_media_files(
folder_path,
check_poster=True,
check_logo=True,
check_fanart=True,
check_nfo=False # Already checked above
)
# Get file paths
file_paths = get_media_file_paths(folder_path)
# Build MediaFilesStatus model
media_files = MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths["poster"] else None,
logo_path=str(file_paths["logo"]) if file_paths["logo"] else None,
fanart_path=str(file_paths["fanart"]) if file_paths["fanart"] else None
)
return NFOCheckResponse(
serie_id=serie_id,
@@ -229,7 +223,18 @@ async def create_nfo(
)
# Check media files
media_files = check_media_files(serie_folder)
folder_path = Path(settings.anime_directory) / serie_folder
media_status = check_media_files(folder_path)
file_paths = get_media_file_paths(folder_path)
media_files = MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
)
return NFOCreateResponse(
serie_id=serie_id,
@@ -313,7 +318,18 @@ async def update_nfo(
)
# Check media files
media_files = check_media_files(serie_folder)
folder_path = Path(settings.anime_directory) / serie_folder
media_status = check_media_files(folder_path)
file_paths = get_media_file_paths(folder_path)
media_files = MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
)
return NFOCreateResponse(
serie_id=serie_id,
@@ -449,7 +465,19 @@ async def get_media_status(
detail=f"Series not found: {serie_id}"
)
return check_media_files(serie.folder)
# Build full path and check media files
folder_path = Path(settings.anime_directory) / serie.folder
media_status = check_media_files(folder_path)
file_paths = get_media_file_paths(folder_path)
return MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
)
except HTTPException:
raise
@@ -521,7 +549,19 @@ async def download_media(
download_media=True
)
return check_media_files(serie_folder)
# Build full path and check media files
folder_path = Path(settings.anime_directory) / serie_folder
media_status = check_media_files(folder_path)
file_paths = get_media_file_paths(folder_path)
return MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
)
except HTTPException:
raise
@@ -678,7 +718,20 @@ async def get_missing_nfo(
has_nfo = await nfo_service.check_nfo_exists(serie_folder)
if not has_nfo:
media_files = check_media_files(serie_folder)
# Build full path and check media files
folder_path = Path(settings.anime_directory) / serie_folder
media_status = check_media_files(folder_path)
file_paths = get_media_file_paths(folder_path)
media_files = MediaFilesStatus(
has_poster=media_status.get("poster", False),
has_logo=media_status.get("logo", False),
has_fanart=media_status.get("fanart", False),
poster_path=str(file_paths["poster"]) if file_paths.get("poster") else None,
logo_path=str(file_paths["logo"]) if file_paths.get("logo") else None,
fanart_path=str(file_paths["fanart"]) if file_paths.get("fanart") else None
)
has_media = (
media_files.has_poster
or media_files.has_logo

21
src/server/services/background_loader_service.py
View File

@@ -234,6 +234,7 @@ class BackgroundLoaderService:
# Check database for series info
from src.server.database.service import AnimeSeriesService
from src.server.utils.media import check_media_files
series_db = await AnimeSeriesService.get_by_key(db, key)
if not series_db:
@@ -244,19 +245,25 @@ class BackgroundLoaderService:
# Check episodes
missing["episodes"] = not series_db.episodes_loaded
# Check files using utility function
folder_path = Path(anime_directory) / folder
media_status = check_media_files(
folder_path,
check_poster=True,
check_logo=True,
check_fanart=True,
check_nfo=True
)
# Check NFO file
nfo_path = Path(anime_directory) / folder / "tvshow.nfo"
missing["nfo"] = not nfo_path.exists() or not series_db.has_nfo
missing["nfo"] = not media_status.get("nfo", False) or not series_db.has_nfo
# Check logo
logo_path = Path(anime_directory) / folder / "logo.png"
missing["logo"] = not logo_path.exists() or not series_db.logo_loaded
missing["logo"] = not media_status.get("logo", False) or not series_db.logo_loaded
# Check images (poster and fanart)
poster_path = Path(anime_directory) / folder / "poster.jpg"
fanart_path = Path(anime_directory) / folder / "fanart.jpg"
missing["images"] = (
not (poster_path.exists() and fanart_path.exists())
not (media_status.get("poster", False) and media_status.get("fanart", False))
or not series_db.images_loaded
)

229
src/server/utils/media.py Normal file
View File

@@ -0,0 +1,229 @@
"""Media file utilities for AniWorld.
This module provides utilities for checking and validating media files
(videos, images, NFO files) in the anime directory.
"""
from pathlib import Path
from typing import Dict, Optional, Union
# Standard media file names as defined by Kodi/Plex conventions
POSTER_FILENAME = "poster.jpg"
LOGO_FILENAME = "logo.png"
FANART_FILENAME = "fanart.jpg"
NFO_FILENAME = "tvshow.nfo"
# Video file extensions supported by most media players
VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".webm", ".mov", ".m4v", ".flv", ".wmv"}
def check_media_files(
series_folder: Union[str, Path],
check_poster: bool = True,
check_logo: bool = True,
check_fanart: bool = True,
check_nfo: bool = True
) -> Dict[str, bool]:
"""Check existence of standard media files for a series.
Checks for standard Kodi/Plex media files in the series folder:
- poster.jpg: Series poster image
- logo.png: Series logo/clearlogo
- fanart.jpg: Series fanart/background image
- tvshow.nfo: Series NFO metadata file
Args:
series_folder: Path to the series folder (string or Path object)
check_poster: Whether to check for poster.jpg
check_logo: Whether to check for logo.png
check_fanart: Whether to check for fanart.jpg
check_nfo: Whether to check for tvshow.nfo
Returns:
Dict mapping file types to existence status:
{
"poster": bool,
"logo": bool,
"fanart": bool,
"nfo": bool
}
Example:
>>> from pathlib import Path
>>> series_path = Path("/anime/Attack on Titan (2013)")
>>> status = check_media_files(series_path)
>>> print(status["poster"]) # True if poster.jpg exists
"""
# Convert to Path object if string
folder_path = Path(series_folder) if isinstance(series_folder, str) else series_folder
result = {}
if check_poster:
poster_path = folder_path / POSTER_FILENAME
result["poster"] = poster_path.exists()
if check_logo:
logo_path = folder_path / LOGO_FILENAME
result["logo"] = logo_path.exists()
if check_fanart:
fanart_path = folder_path / FANART_FILENAME
result["fanart"] = fanart_path.exists()
if check_nfo:
nfo_path = folder_path / NFO_FILENAME
result["nfo"] = nfo_path.exists()
return result
def get_media_file_paths(
series_folder: Union[str, Path],
include_poster: bool = True,
include_logo: bool = True,
include_fanart: bool = True,
include_nfo: bool = True
) -> Dict[str, Optional[Path]]:
"""Get paths to standard media files for a series.
Returns paths only if the files exist. Useful for operations that need
the actual file paths (e.g., reading, copying, moving).
Args:
series_folder: Path to the series folder (string or Path object)
include_poster: Whether to include poster.jpg path
include_logo: Whether to include logo.png path
include_fanart: Whether to include fanart.jpg path
include_nfo: Whether to include tvshow.nfo path
Returns:
Dict mapping file types to paths (None if file doesn't exist):
{
"poster": Optional[Path],
"logo": Optional[Path],
"fanart": Optional[Path],
"nfo": Optional[Path]
}
Example:
>>> from pathlib import Path
>>> series_path = Path("/anime/Attack on Titan (2013)")
>>> paths = get_media_file_paths(series_path)
>>> if paths["poster"]:
... print(f"Poster found at: {paths['poster']}")
"""
# Convert to Path object if string
folder_path = Path(series_folder) if isinstance(series_folder, str) else series_folder
result = {}
if include_poster:
poster_path = folder_path / POSTER_FILENAME
result["poster"] = poster_path if poster_path.exists() else None
if include_logo:
logo_path = folder_path / LOGO_FILENAME
result["logo"] = logo_path if logo_path.exists() else None
if include_fanart:
fanart_path = folder_path / FANART_FILENAME
result["fanart"] = fanart_path if fanart_path.exists() else None
if include_nfo:
nfo_path = folder_path / NFO_FILENAME
result["nfo"] = nfo_path if nfo_path.exists() else None
return result
def has_all_images(series_folder: Union[str, Path]) -> bool:
"""Check if series has all standard image files (poster, logo, fanart).
Args:
series_folder: Path to the series folder (string or Path object)
Returns:
True if all image files exist, False otherwise
Example:
>>> from pathlib import Path
>>> series_path = Path("/anime/Attack on Titan (2013)")
>>> if has_all_images(series_path):
... print("Series has complete image set")
"""
# Convert to Path object if string
folder_path = Path(series_folder) if isinstance(series_folder, str) else series_folder
poster_path = folder_path / POSTER_FILENAME
logo_path = folder_path / LOGO_FILENAME
fanart_path = folder_path / FANART_FILENAME
return (
poster_path.exists()
and logo_path.exists()
and fanart_path.exists()
)
def count_video_files(series_folder: Union[str, Path], recursive: bool = True) -> int:
"""Count video files in the series folder.
Counts files with standard video extensions (.mp4, .mkv, .avi, etc.).
Args:
series_folder: Path to the series folder (string or Path object)
recursive: Whether to search subdirectories (for season folders)
Returns:
Number of video files found
Example:
>>> from pathlib import Path
>>> series_path = Path("/anime/Attack on Titan (2013)")
>>> video_count = count_video_files(series_path)
>>> print(f"Found {video_count} episodes")
"""
# Convert to Path object if string
folder_path = Path(series_folder) if isinstance(series_folder, str) else series_folder
if not folder_path.exists():
return 0
count = 0
pattern = "**/*" if recursive else "*"
for file_path in folder_path.glob(pattern):
if file_path.is_file() and file_path.suffix.lower() in VIDEO_EXTENSIONS:
count += 1
return count
def has_video_files(series_folder: Union[str, Path]) -> bool:
"""Check if series folder contains any video files.
Args:
series_folder: Path to the series folder (string or Path object)
Returns:
True if at least one video file exists, False otherwise
Example:
>>> from pathlib import Path
>>> series_path = Path("/anime/Attack on Titan (2013)")
>>> if not has_video_files(series_path):
... print("No episodes found")
"""
# Convert to Path object if string
folder_path = Path(series_folder) if isinstance(series_folder, str) else series_folder
if not folder_path.exists():
return False
for file_path in folder_path.rglob("*"):
if file_path.is_file() and file_path.suffix.lower() in VIDEO_EXTENSIONS:
return True
return False