Improve docs and security defaults
This commit is contained in:
@@ -1,56 +1,99 @@
|
||||
import os
|
||||
import json
|
||||
"""Utilities for loading and managing stored anime series metadata."""
|
||||
|
||||
import logging
|
||||
from .series import Serie
|
||||
import os
|
||||
from json import JSONDecodeError
|
||||
from typing import Dict, Iterable, List
|
||||
|
||||
from src.core.entities.series import Serie
|
||||
|
||||
|
||||
class SerieList:
|
||||
def __init__(self, basePath: str):
|
||||
self.directory = basePath
|
||||
self.folderDict: dict[str, Serie] = {} # Proper initialization
|
||||
"""Represents the collection of cached series stored on disk."""
|
||||
|
||||
def __init__(self, base_path: str) -> None:
|
||||
self.directory: str = base_path
|
||||
self.folderDict: Dict[str, Serie] = {}
|
||||
self.load_series()
|
||||
|
||||
def add(self, serie: Serie):
|
||||
if (not self.contains(serie.key)):
|
||||
dataPath = os.path.join(self.directory, serie.folder, "data")
|
||||
animePath = os.path.join(self.directory, serie.folder)
|
||||
os.makedirs(animePath, exist_ok=True)
|
||||
if not os.path.isfile(dataPath):
|
||||
serie.save_to_file(dataPath)
|
||||
self.folderDict[serie.folder] = serie;
|
||||
def add(self, serie: Serie) -> None:
|
||||
"""Persist a new series if it is not already present."""
|
||||
|
||||
if self.contains(serie.key):
|
||||
return
|
||||
|
||||
data_path = os.path.join(self.directory, serie.folder, "data")
|
||||
anime_path = os.path.join(self.directory, serie.folder)
|
||||
os.makedirs(anime_path, exist_ok=True)
|
||||
if not os.path.isfile(data_path):
|
||||
serie.save_to_file(data_path)
|
||||
self.folderDict[serie.folder] = serie
|
||||
|
||||
def contains(self, key: str) -> bool:
|
||||
for k, value in self.folderDict.items():
|
||||
if value.key == key:
|
||||
return True
|
||||
return False
|
||||
"""Return True when a series identified by ``key`` already exists."""
|
||||
|
||||
def load_series(self):
|
||||
""" Scan folders and load data files """
|
||||
logging.info(f"Scanning anime folders in: {self.directory}")
|
||||
for anime_folder in os.listdir(self.directory):
|
||||
return any(value.key == key for value in self.folderDict.values())
|
||||
|
||||
def load_series(self) -> None:
|
||||
"""Populate the in-memory map with metadata discovered on disk."""
|
||||
|
||||
logging.info("Scanning anime folders in %s", self.directory)
|
||||
try:
|
||||
entries: Iterable[str] = os.listdir(self.directory)
|
||||
except OSError as error:
|
||||
logging.error(
|
||||
"Unable to scan directory %s: %s",
|
||||
self.directory,
|
||||
error,
|
||||
)
|
||||
return
|
||||
|
||||
for anime_folder in entries:
|
||||
anime_path = os.path.join(self.directory, anime_folder, "data")
|
||||
if os.path.isfile(anime_path):
|
||||
logging.debug(f"Found data folder: {anime_path}")
|
||||
self.load_data(anime_folder, anime_path)
|
||||
else:
|
||||
logging.warning(f"Skipping {anime_folder} - No data folder found")
|
||||
logging.debug("Found data file for folder %s", anime_folder)
|
||||
self._load_data(anime_folder, anime_path)
|
||||
continue
|
||||
|
||||
logging.warning(
|
||||
"Skipping folder %s because no metadata file was found",
|
||||
anime_folder,
|
||||
)
|
||||
|
||||
def _load_data(self, anime_folder: str, data_path: str) -> None:
|
||||
"""Load a single series metadata file into the in-memory collection."""
|
||||
|
||||
def load_data(self, anime_folder, data_path):
|
||||
""" Load pickle files from the data folder """
|
||||
try:
|
||||
self.folderDict[anime_folder] = Serie.load_from_file(data_path)
|
||||
logging.debug(f"Successfully loaded {data_path} for {anime_folder}")
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to load {data_path} in {anime_folder}: {e}")
|
||||
logging.debug("Successfully loaded metadata for %s", anime_folder)
|
||||
except (OSError, JSONDecodeError, KeyError, ValueError) as error:
|
||||
logging.error(
|
||||
"Failed to load metadata for folder %s from %s: %s",
|
||||
anime_folder,
|
||||
data_path,
|
||||
error,
|
||||
)
|
||||
|
||||
def GetMissingEpisode(self) -> List[Serie]:
|
||||
"""Return all series that still contain missing episodes."""
|
||||
|
||||
return [
|
||||
serie
|
||||
for serie in self.folderDict.values()
|
||||
if serie.episodeDict
|
||||
]
|
||||
|
||||
def get_missing_episodes(self) -> List[Serie]:
|
||||
"""PEP8-friendly alias for :meth:`GetMissingEpisode`."""
|
||||
|
||||
return self.GetMissingEpisode()
|
||||
|
||||
def GetList(self) -> List[Serie]:
|
||||
"""Return all series instances stored in the list."""
|
||||
|
||||
def GetMissingEpisode(self):
|
||||
"""Find all series with a non-empty episodeDict"""
|
||||
return [serie for serie in self.folderDict.values() if len(serie.episodeDict) > 0]
|
||||
|
||||
def GetList(self):
|
||||
"""Get all series in the list"""
|
||||
return list(self.folderDict.values())
|
||||
|
||||
def get_all(self) -> List[Serie]:
|
||||
"""PEP8-friendly alias for :meth:`GetList`."""
|
||||
|
||||
#k = AnimeList("\\\\sshfs.r\\ubuntu@192.168.178.43\\media\\serien\\Serien")
|
||||
#bbabab = k.GetMissingEpisode()
|
||||
#print(bbabab)
|
||||
return self.GetList()
|
||||
|
||||
Reference in New Issue
Block a user