import re import shutil import sys from pathlib import Path from typing import Any from mangadlp import downloader, utils from mangadlp.api.mangadex import Mangadex from mangadlp.hooks import Hooks from mangadlp.logger import Logger # prepare logger log = Logger(__name__) class MangaDLP: """Download Mangas from supported sites. After initialization, start the script with the function get_manga(). Args: url_uuid (str): URL or UUID of the manga language (str): Manga language with country codes. "en" --> english chapters (str): Chapters to download, "all" for every chapter available list_chapters (bool): List all available chapters and exit file_format (str): Archive format to create. An empty string means don't archive the folder forcevol (bool): Force naming of volumes. Useful for mangas where chapters reset each volume download_path (str): Download path. Defaults to '/downloads' download_wait (float): Time to wait for each picture to download in seconds """ def __init__( self, url_uuid: str, language: str = "en", chapters: str = "", list_chapters: bool = False, file_format: str = "cbz", forcevol: bool = False, download_path: str = "downloads", download_wait: float = 0.5, manga_pre_hook_cmd: str = "", manga_post_hook_cmd: str = "", chapter_pre_hook_cmd: str = "", chapter_post_hook_cmd: str = "", ) -> None: # init parameters self.url_uuid: str = url_uuid self.language: str = language self.chapters: str = chapters self.list_chapters: bool = list_chapters self.file_format: str = file_format self.forcevol: bool = forcevol self.download_path: str = download_path self.download_wait: float = download_wait # hooks self.hook: Hooks = Hooks( manga_pre_hook_cmd, manga_post_hook_cmd, chapter_pre_hook_cmd, chapter_post_hook_cmd, ) self.hook_infos: dict = {} # prepare everything self._prepare() def _prepare(self) -> None: # set manga format suffix if self.file_format and "." not in self.file_format: self.file_format = f".{self.file_format}" # start prechecks self.pre_checks() # init api self.api_used = self.check_api(self.url_uuid) self.api = self.api_used(self.url_uuid, self.language, self.forcevol) # get manga title and uuid self.manga_uuid = self.api.manga_uuid self.manga_title = self.api.manga_title # get chapter list self.manga_chapter_list = self.api.chapter_list self.manga_total_chapters = len(self.manga_chapter_list) self.manga_path = Path(f"{self.download_path}/{self.manga_title}") def pre_checks(self) -> None: # prechecks userinput/options # no url and no readin list given if not self.url_uuid: log.error( 'You need to specify a manga url/uuid with "-u" or a list with "--read"' ) sys.exit(1) # checks if --list is not used if not self.list_chapters: if not self.chapters: # no chapters to download were given log.error( 'You need to specify one or more chapters to download. To see all chapters use "--list"' ) sys.exit(1) # if forcevol is used, but didn't specify a volume in the chapters selected if self.forcevol and ":" not in self.chapters: log.error("You need to specify the volume if you use --forcevol") sys.exit(1) # if forcevol is not used, but a volume is specified if not self.forcevol and ":" in self.chapters: log.error("Don't specify the volume without --forcevol") sys.exit(1) # check the api which needs to be used def check_api(self, url_uuid: str) -> type: # apis to check api_mangadex = re.compile("mangadex.org") api_mangadex2 = re.compile( "[a-z0-9]{8}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{12}" ) api_test = re.compile("test.test") # check url for match if api_mangadex.search(url_uuid) or api_mangadex2.search(url_uuid): return Mangadex # this is only for testing multiple apis if api_test.search(url_uuid): log.critical("Not supported yet") sys.exit(1) # no supported api found log.error(f"No supported api in link/uuid found: {url_uuid}") sys.exit(1) # once called per manga def get_manga(self) -> None: # create empty skipped chapters list skipped_chapters: list[Any] = [] error_chapters: list[Any] = [] print_divider = "=========================================" # show infos log.info(f"{print_divider}") log.lean(f"Manga Name: {self.manga_title}") log.info(f"Manga UUID: {self.manga_uuid}") log.info(f"Total chapters: {self.manga_total_chapters}") # list chapters if list_chapters is true if self.list_chapters: log.info(f"Available Chapters: {', '.join(self.manga_chapter_list)}") log.info(f"{print_divider}\n") sys.exit(0) # check chapters to download if not all if self.chapters.lower() == "all": chapters_to_download = self.manga_chapter_list else: chapters_to_download = utils.get_chapter_list( self.chapters, self.manga_chapter_list ) # show chapters to download log.lean(f"Chapters selected: {', '.join(chapters_to_download)}") log.info(f"{print_divider}") # create manga folder self.manga_path.mkdir(parents=True, exist_ok=True) # create dict with all variables for the hooks self.hook_infos.update( { "api": self.api.api_name, "manga_url_uuid": self.url_uuid, "manga_uuid": self.manga_uuid, "manga_title": self.manga_title, "language": self.language, "total_chapters": self.manga_total_chapters, "chapters_to_download": chapters_to_download, "file_format": self.file_format, "forcevol": self.forcevol, "download_path": self.download_path, "manga_path": self.manga_path, } ) # start manga pre hook self.hook.run("manga_pre", {"status": "starting"}, self.hook_infos) # get chapters for chapter in chapters_to_download: return_infos = self.get_chapter(chapter) error_chapters.append(return_infos.get("error")) skipped_chapters.append(return_infos.get("skipped")) if self.file_format and return_infos["chapter_path"]: return_infos = self.archive_chapter(return_infos["chapter_path"]) error_chapters.append(return_infos.get("error")) skipped_chapters.append(return_infos.get("skipped")) # check if chapter was skipped try: return_infos["skipped"] # chapter was not skipped except KeyError: # done with chapter log.info(f"Done with chapter '{chapter}'\n") # start chapter post hook self.hook.run("chapter_post", {"status": "successful"}, self.hook_infos) # done with manga log.info(f"{print_divider}") log.lean(f"Done with manga: {self.manga_title}") # filter skipped list skipped_chapters = list(filter(None, skipped_chapters)) if len(skipped_chapters) >= 1: log.lean(f"Skipped chapters: {', '.join(skipped_chapters)}") # filter error list error_chapters = list(filter(None, error_chapters)) if len(error_chapters) >= 1: log.lean(f"Chapters with errors: {', '.join(error_chapters)}") # start manga post hook self.hook.run("manga_post", {"status": "successful"}, self.hook_infos) log.info(f"{print_divider}\n") # once called per chapter def get_chapter(self, chapter: str) -> dict: # get chapter infos chapter_infos = self.api.get_chapter_infos(chapter) # get image urls for chapter try: chapter_image_urls = self.api.get_chapter_images( chapter, self.download_wait ) except KeyboardInterrupt: log.critical("Stopping") sys.exit(1) # check if the image urls are empty. if yes skip this chapter (for mass downloads) if not chapter_image_urls: log.error( f"No images: Skipping Vol. {chapter_infos['volume']} Ch.{chapter_infos['chapter']}" ) self.hook.run( "chapter_pre", {"status": "skipped", "reason": "No images"}, {} ) # add to skipped chapters list return ( { "error": f"{chapter_infos['volume']}:{chapter_infos['chapter']}", "chapter_path": None, } if self.forcevol else {"error": f"{chapter_infos['chapter']}", "chapter_path": None} ) # get filename for chapter (without suffix) chapter_filename = utils.get_filename( chapter_infos["name"], chapter_infos["volume"], chapter, self.forcevol ) # set download path for chapter (image folder) chapter_path = self.manga_path / chapter_filename # set archive path with file format chapter_archive_path = Path(f"{chapter_path}{self.file_format}") # check if chapter already exists # check for folder, if file format is an empty string if chapter_archive_path.exists(): log.warning(f"'{chapter_archive_path}' already exists. Skipping") self.hook.run( "chapter_pre", {"status": "skipped", "reason": "Existing"}, {} ) # add to skipped chapters list return ( { "skipped": f"{chapter_infos['volume']}:{chapter_infos['chapter']}", "chapter_path": None, } if self.forcevol else {"skipped": f"{chapter_infos['chapter']}", "chapter_path": None} ) # create chapter folder (skips it if it already exists) chapter_path.mkdir(parents=True, exist_ok=True) # verbose log log.verbose(f"Chapter UUID: {chapter_infos['uuid']}") log.verbose(f"Filename: '{chapter_archive_path.name}'") log.verbose(f"File path: '{chapter_archive_path}'") log.verbose(f"Image URLS:\n{chapter_image_urls}") # create dict with all variables for the hooks self.hook_infos.update( { "chapter_filename": chapter_filename, "chapter_path": chapter_path, "chapter_archive_path": chapter_archive_path, "chapter_uuid": chapter_infos["uuid"], "chapter_volume": chapter_infos["volume"], "chapter_number": chapter_infos["chapter"], "chapter_name": chapter_infos["name"], } ) # start chapter pre hook self.hook.run("chapter_pre", {"status": "starting"}, self.hook_infos) # log log.lean(f"Downloading: '{chapter_filename}'") # download images try: downloader.download_chapter( chapter_image_urls, chapter_path, self.download_wait ) except KeyboardInterrupt: log.critical("Stopping") sys.exit(1) except Exception: log.error(f"Cant download: '{chapter_filename}'. Skipping") self.hook.run( "chapter_post", {"status": "error", "reason": "Download error"}, {} ) # add to skipped chapters list return ( { "error": f"{chapter_infos['volume']}:{chapter_infos['chapter']}", "chapter_path": None, } if self.forcevol else {"error": f"{chapter_infos['chapter']}", "chapter_path": None} ) else: # Done with chapter log.lean(f"Successfully downloaded: '{chapter_filename}'") return {"chapter_path": chapter_path} # create an archive of the chapter if needed def archive_chapter(self, chapter_path: Path) -> dict: log.lean(f"Creating archive '{chapter_path}{self.file_format}'") try: # check if image folder is existing if not chapter_path.exists(): log.error(f"Image folder: {chapter_path} does not exist") raise IOError if self.file_format == ".pdf": utils.make_pdf(chapter_path) else: utils.make_archive(chapter_path, self.file_format) except Exception: log.error("Archive error. Skipping chapter") # add to skipped chapters list return { "error": chapter_path, } else: # remove image folder shutil.rmtree(chapter_path) return {}