134 lines
5.4 KiB
Python
134 lines
5.4 KiB
Python
import os
|
|
import re
|
|
import logging
|
|
from aniworld.models import Anime, Episode
|
|
from aniworld.common import get_season_episode_count, get_movie_episode_count
|
|
from aniworld.search import search_anime
|
|
from Loader import download
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
filename="loader.log",
|
|
level=logging.DEBUG,
|
|
format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
class MatchNotFoundError(Exception):
|
|
"""Custom exception raised when the pattern match is not found."""
|
|
pass
|
|
|
|
class Loader:
|
|
def __init__(self, basePath: str):
|
|
self.directory = basePath
|
|
logging.info(f"Initialized Loader with base path: {self.directory}")
|
|
|
|
def __find_mp4_files(self):
|
|
logging.info("Scanning for .mp4 files")
|
|
for root, dirs, files in os.walk(self.directory):
|
|
mp4_files = [file for file in files if file.endswith('.mp4')]
|
|
|
|
if mp4_files:
|
|
relative_path = os.path.relpath(root, self.directory)
|
|
root_folder_name = relative_path.split(os.sep)[0]
|
|
logging.debug(f"Found {len(mp4_files)} .mp4 files in {root_folder_name}")
|
|
yield root_folder_name, mp4_files
|
|
for dir in self.__find_empty_folders():
|
|
relative_path = os.path.relpath(dir, self.directory)
|
|
logging.debug(f"Found no .mp4 files in {relative_path}")
|
|
yield relative_path, []
|
|
for dir in self.__find_folders_without_mp4():
|
|
relative_path = os.path.relpath(dir, self.directory)
|
|
logging.debug(f"Found no .mp4 files in {relative_path}")
|
|
yield relative_path, []
|
|
|
|
def __find_empty_folders(self):
|
|
"""Finds and returns a list of empty folders in the given directory."""
|
|
empty_folders = []
|
|
|
|
for root, dirs, files in os.walk(self.directory):
|
|
if not files and not dirs: # If the folder contains no files or subdirectories
|
|
empty_folders.append(root)
|
|
|
|
return empty_folders
|
|
def __find_folders_without_mp4(self):
|
|
"""
|
|
Finds folders that are either empty or contain no .mp4 files.
|
|
|
|
:param root_dir: The directory to search within.
|
|
:return: List of folders that are empty or have no .mp4 files.
|
|
"""
|
|
result_folders = []
|
|
|
|
for foldername, subfolders, filenames in os.walk(self.directory):
|
|
# Check if the folder is empty or has no MP4 files
|
|
folder_name = os.path.relpath(foldername, self.directory)
|
|
if not any(file.lower().endswith('.mp4') for file in filenames):
|
|
if (folder_name != "."):
|
|
result_folders.append(foldername)
|
|
|
|
return result_folders
|
|
def __remove_year(self, input_string: str):
|
|
cleaned_string = re.sub(r'\(\d{4}\)', '', input_string).strip()
|
|
logging.debug(f"Removed year from '{input_string}' -> '{cleaned_string}'")
|
|
return cleaned_string
|
|
|
|
def __check_and_generate_key(self, folder_name: str):
|
|
folder_path = os.path.join(self.directory, folder_name)
|
|
key_file = os.path.join(folder_path, 'key')
|
|
|
|
if os.path.exists(key_file):
|
|
with open(key_file, 'r') as file:
|
|
key = file.read().strip()
|
|
logging.debug(f"Key found for folder '{folder_name}': {key}")
|
|
return key
|
|
else:
|
|
key = search_anime(self.__remove_year(folder_name))
|
|
with open(key_file, 'w') as file:
|
|
file.write(key)
|
|
logging.info(f"Generated new key for folder '{folder_name}': {key}")
|
|
return key
|
|
|
|
def __GetEpisodeAndSeason(self, filename: str):
|
|
pattern = r'S(\d{2})E(\d{2})'
|
|
match = re.search(pattern, filename)
|
|
if match:
|
|
season = match.group(1)
|
|
episode = match.group(2)
|
|
logging.debug(f"Extracted season {season}, episode {episode} from '{filename}'")
|
|
return season, episode
|
|
else:
|
|
logging.error(f"Failed to find season/episode pattern in '{filename}'")
|
|
raise MatchNotFoundError("Season and episode pattern not found in the filename.")
|
|
|
|
def LoadMissing(self):
|
|
logging.info("Starting process to load missing episodes")
|
|
result = self.__find_mp4_files()
|
|
|
|
for folder, mp4_files in result:
|
|
try:
|
|
key = self.__check_and_generate_key(folder)
|
|
missings = self.__GetMissingEpisodesAndSeason(key, mp4_files)
|
|
|
|
for season, missing_episodes in missings:
|
|
folder_path = os.path.join(self.directory, folder, f"Season {season}")
|
|
|
|
for episode in missing_episodes:
|
|
anime = Anime(
|
|
episode_list=[
|
|
Episode(slug=key, season=season, episode=episode)
|
|
],
|
|
language="German Dub",
|
|
output_directory=folder_path
|
|
)
|
|
logging.info(f"Downloading episode {episode} of season {season} for anime {key}")
|
|
download(anime)
|
|
except Exception as e:
|
|
logging.error(f"Error processing folder '{folder}': {e}")
|
|
continue
|
|
|
|
# Read the base directory from an environment variable
|
|
#directory_to_search = os.getenv("ANIME_DIRECTORY", "\\\\sshfs.r\\ubuntu@192.168.178.43\\media\\serien\\Serien")
|
|
directory_to_search = os.getenv("ANIME_DIRECTORY", "D:\sss")
|
|
|
|
loader = Loader(directory_to_search)
|
|
loader.LoadMissing() |