From 9cd15e33fb295af12ad926f111bfb7ec0df70ae7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?William=20S=C3=B6derberg?= Date: Tue, 2 Dec 2025 01:54:24 +0100 Subject: [PATCH] Code formatting and updated flamecomic to use changed api --- .gitignore | 352 +++++++++++++++++++++---------------------- main.py | 286 ++++++++++++++++++----------------- manga.py | 181 +++++++++++----------- packager.py | 108 +++++++------ sites/__init__.py | 24 +-- sites/flamecomics.py | 179 +++++++++++----------- sites/provider.py | 34 ++--- 7 files changed, 600 insertions(+), 564 deletions(-) diff --git a/.gitignore b/.gitignore index 8da0eb2..f73b00c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,177 +1,177 @@ -manga.json -config.ini - -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py,cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/latest/usage/project/#working-with-version-control -.pdm.toml -.pdm-python -.pdm-build/ - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Ruff stuff: -.ruff_cache/ - -# PyPI configuration file +manga.json +config.ini + +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +#uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/latest/usage/project/#working-with-version-control +.pdm.toml +.pdm-python +.pdm-build/ + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file .pypirc \ No newline at end of file diff --git a/main.py b/main.py index 7eee80e..62dc488 100644 --- a/main.py +++ b/main.py @@ -1,140 +1,146 @@ -import requests -from pathlib import Path -import argparse -import configparser -import json - -import questionary -from rich import print - -from sites import parse_provider -from manga import Manga - -parser = argparse.ArgumentParser( - prog="Manga Downloader", - description="Download all your favourite manga!", - epilog="Happy Reading!" -) -parser.add_argument( - "--auto", - default=False, - action=argparse.BooleanOptionalAction, - help="Run using auto mode (default: %(default)s)" -) -parser.add_argument( - "--save-dir", - dest="save_dir", - type=Path, - help="Set temporary save dir instead of using config value" -) - -args = parser.parse_args() - -s = requests.session() -cache_file = Path(__file__).parent.resolve().joinpath("manga.json") -config_file = Path(__file__).parent.resolve().joinpath("config.ini") - - -def get_config() -> configparser.ConfigParser: - config = configparser.ConfigParser() - if config_file.exists(): - config.read(config_file) - - if not config.has_option("DEFAULT", "savedir"): - path = questionary.path("Enter default save path for downloaded Manga: ", qmark="").ask() - config["DEFAULT"] = {"savedir": path} - - with open(config_file, "w") as cf: - config.write(cf) - - return config - -def read_from_cache() -> dict: - if cache_file.exists(): - return json.loads(cache_file.read_bytes()) - else: - return {} - -def write_to_cache(manga:Manga) -> None: - cache = read_from_cache() - series = manga.info.get("series") - cached_chapters = cache.get(series, {}).get("cached",[]) + manga.cached - - cache[series] = { - "url": manga.site.get_url(), - "cached": list(set(cached_chapters)) - } - - with open(cache_file, "w") as fp: - json.dump(cache,fp, indent=4) - -def parse_cache(cache:dict) -> list[Manga]: - if not cache: - print("Cache is empty!") - exit() - else: - manga = list() - for entry in cache.values(): - manga.append( - Manga(parse_provider(s, entry.get("url")), entry.get("cached")) # type: ignore - ) - return manga - -def get_cached_manga() -> list[Manga]: - cache = read_from_cache() - return parse_cache(cache) - -def input_manga() -> list[str]: - manga = list() - add_more = True - while add_more: - manga.append( - questionary.text( - "Input Manga url:", - qmark = "" - ).ask() - ) - add_more = questionary.confirm("Do you want to add more Manga", default=False).ask() - - return manga - -def parse_manga(manga:list[str]) -> list[Manga]: - parsed_manga = list() - for entry in manga: - provider = parse_provider(s, entry) - if not provider: - print(f"Failed to parse: {entry}") - else: - parsed_manga.append(Manga(provider)) - return parsed_manga - -def choose_manga() -> list[Manga]: - use_cached = questionary.confirm("Do you want to choose from cached Manga", default=True).ask() - - if use_cached: - return get_cached_manga() - else: - manga = input_manga() - return parse_manga(manga) - -def main(): - - config = get_config() - save_dir = args.save_dir if args.save_dir else Path(config["DEFAULT"]["savedir"]) - - if args.auto: - if not config_file.exists(): - print("No config file exists") - exit(1) - else: - list_of_manga = get_cached_manga() - else: - list_of_manga = choose_manga() - - for manga in list_of_manga: - manga.choose_chapters(args.auto) - manga.download(save_dir) - write_to_cache(manga) - -if __name__ == "__main__": - main() +import argparse +import configparser +import json +from pathlib import Path + +import questionary +import requests +from manga import Manga +from rich import print +from sites import parse_provider + +parser = argparse.ArgumentParser( + prog="Manga Downloader", + description="Download all your favourite manga!", + epilog="Happy Reading!", +) +parser.add_argument( + "--auto", + default=False, + action=argparse.BooleanOptionalAction, + help="Run using auto mode (default: %(default)s)", +) +parser.add_argument( + "--save-dir", + dest="save_dir", + type=Path, + help="Set temporary save dir instead of using config value", +) + +args = parser.parse_args() + +s = requests.session() +cache_file = Path(__file__).parent.resolve().joinpath("manga.json") +config_file = Path(__file__).parent.resolve().joinpath("config.ini") + + +def get_config() -> configparser.ConfigParser: + config = configparser.ConfigParser() + if config_file.exists(): + config.read(config_file) + + if not config.has_option("DEFAULT", "savedir"): + path = questionary.path( + "Enter default save path for downloaded Manga: ", qmark="" + ).ask() + config["DEFAULT"] = {"savedir": path} + + with open(config_file, "w") as cf: + config.write(cf) + + return config + + +def read_from_cache() -> dict: + if cache_file.exists(): + return json.loads(cache_file.read_bytes()) + else: + return {} + + +def write_to_cache(manga: Manga) -> None: + cache = read_from_cache() + series = manga.info.get("series") + cached_chapters = cache.get(series, {}).get("cached", []) + manga.cached + + cache[series] = {"url": manga.site.get_url(), "cached": list(set(cached_chapters))} + + with open(cache_file, "w") as fp: + json.dump(cache, fp, indent=4) + + +def parse_cache(cache: dict) -> list[Manga]: + if not cache: + print("Cache is empty!") + exit() + else: + manga = list() + for entry in cache.values(): + manga.append( + Manga(parse_provider(s, entry.get("url")), entry.get("cached")) # type: ignore + ) + return manga + + +def get_cached_manga() -> list[Manga]: + cache = read_from_cache() + return parse_cache(cache) + + +def input_manga() -> list[str]: + manga = list() + add_more = True + while add_more: + manga.append(questionary.text("Input Manga url:", qmark="").ask()) + add_more = questionary.confirm( + "Do you want to add more Manga", default=False + ).ask() + + return manga + + +def parse_manga(manga: list[str]) -> list[Manga]: + parsed_manga = list() + for entry in manga: + provider = parse_provider(s, entry) + if not provider: + print(f"Failed to parse: {entry}") + else: + parsed_manga.append(Manga(provider)) + return parsed_manga + + +def choose_manga() -> list[Manga]: + use_cached = questionary.confirm( + "Do you want to choose from cached Manga", default=True + ).ask() + + if use_cached: + return get_cached_manga() + else: + manga = input_manga() + return parse_manga(manga) + + +def main(): + + config = get_config() + save_dir = args.save_dir if args.save_dir else Path(config["DEFAULT"]["savedir"]) + + if args.auto: + if not config_file.exists(): + print("No config file exists") + exit(1) + else: + list_of_manga = get_cached_manga() + else: + list_of_manga = choose_manga() + + for manga in list_of_manga: + manga.choose_chapters(args.auto) + manga.download(save_dir) + write_to_cache(manga) + + +if __name__ == "__main__": + main() diff --git a/manga.py b/manga.py index aa58deb..fd90e4b 100644 --- a/manga.py +++ b/manga.py @@ -1,84 +1,97 @@ -import xml.etree.ElementTree as XML -from pathlib import Path -import tempfile - -import questionary -from rich import print -from rich.progress import Progress, SpinnerColumn, TextColumn, MofNCompleteColumn - -from sites import Provider -from packager import download_images, zip_files - -class Manga(): - def __init__(self, provider:Provider, cached:list[int] = []) -> None: - self.site = provider - self.cached = cached - self.info, self.chapters = self.site.get_mediainfo() - self.slug = self.slugify(self.info.get("series")) # type: ignore - - @staticmethod - def slugify(string:str) -> str: - string = "".join(x for x in string if not x in '<>:"/\\|?*') - return str(string) - - def _generateComicInfo(self, chapter:dict, dir:Path): - filename = dir.joinpath("ComicInfo.xml") - ComicInfo = XML.Element("ComicInfo") - - XML.SubElement(ComicInfo, "Series").text = self.info.get("series") - XML.SubElement(ComicInfo, "Writer").text = ",".join(self.info.get("writer")) # type: ignore - XML.SubElement(ComicInfo, "Penciller").text = ",".join(self.info.get("penciller")) # type: ignore - XML.SubElement(ComicInfo, "Genre").text = ",".join(self.info.get("genre")) # type: ignore - XML.SubElement(ComicInfo, "Summary").text = self.info.get("summary") - XML.SubElement(ComicInfo, "Number").text = str(chapter.get("nr")) - XML.SubElement(ComicInfo, "Title").text = chapter.get("title") - XML.SubElement(ComicInfo, "LanguageISO").text = self.info.get("languageISO") - XML.SubElement(ComicInfo, "PageCount").text = str(len(chapter.get("images"))) # type: ignore - XML.SubElement(ComicInfo, "ScanInformation").text = self.info.get("scanInformation") - - tree = XML.ElementTree(ComicInfo) - - XML.indent(tree, space="\t", level=0) - tree.write(str(filename), encoding="utf-8") - - return filename - - @staticmethod - def printMangaInfo(info:dict): - print(info) - - def choose_chapters(self, auto): - choices = [ - questionary.Choice( - f"{chapter.get("nr"):02d}: {chapter.get("title")}", - chapter, - checked = chapter.get("nr") not in self.cached - ) for chapter in self.chapters - ] - self.chapters = questionary.checkbox( - "Choose which chapters to download", - choices - ).skip_if( - auto, - [chapter for chapter in self.chapters if chapter.get("nr") not in self.cached] - ).ask() - - def download(self, save_dir:Path): - with tempfile.TemporaryDirectory(prefix="manga_") as temp_dir: - if self.chapters: - with Progress( - TextColumn("[progress.description]{task.description}"), - MofNCompleteColumn(), - SpinnerColumn() - ) as p: - task = p.add_task(f"[bold]Processing [orange3]{self.info.get("series")}[/]", total=len(self.chapters)) - while not p.finished: - for chapter in self.chapters: - print(f"Downloading [bold blue]Chapter {chapter.get("nr"):02d}[/]...") - files = [self._generateComicInfo(chapter, Path(temp_dir))] - files.extend(download_images(chapter.get("images"), Path(temp_dir))) # type: ignore - zip_files(self.slug, chapter, files, save_dir) - self.cached.append(chapter.get("nr")) - p.update(task, advance=1) - else: - print("No new [orange3]Chapter[/] were found...") +import tempfile +import xml.etree.ElementTree as XML +from pathlib import Path + +import questionary +from packager import download_images, zip_files +from rich import print +from rich.progress import MofNCompleteColumn, Progress, SpinnerColumn, TextColumn +from sites import Provider + + +class Manga: + def __init__(self, provider: Provider, cached: list[int] = []) -> None: + self.site = provider + self.cached = cached + self.info, self.chapters = self.site.get_mediainfo() + self.slug = self.slugify(self.info.get("series")) # type: ignore + + @staticmethod + def slugify(string: str) -> str: + string = "".join(x for x in string if not x in '<>:"/\\|?*') + return str(string) + + def _generateComicInfo(self, chapter: dict, dir: Path): + filename = dir.joinpath("ComicInfo.xml") + ComicInfo = XML.Element("ComicInfo") + + XML.SubElement(ComicInfo, "Series").text = self.info.get("series") + XML.SubElement(ComicInfo, "Writer").text = ",".join(self.info.get("writer")) # type: ignore + XML.SubElement(ComicInfo, "Penciller").text = ",".join(self.info.get("penciller")) # type: ignore + XML.SubElement(ComicInfo, "Genre").text = ",".join(self.info.get("genre")) # type: ignore + XML.SubElement(ComicInfo, "Summary").text = self.info.get("summary") + XML.SubElement(ComicInfo, "Number").text = str(chapter.get("nr")) + XML.SubElement(ComicInfo, "Title").text = chapter.get("title") + XML.SubElement(ComicInfo, "LanguageISO").text = self.info.get("languageISO") + XML.SubElement(ComicInfo, "PageCount").text = str(len(chapter.get("images"))) # type: ignore + XML.SubElement(ComicInfo, "ScanInformation").text = self.info.get( + "scanInformation" + ) + + tree = XML.ElementTree(ComicInfo) + + XML.indent(tree, space="\t", level=0) + tree.write(str(filename), encoding="utf-8") + + return filename + + @staticmethod + def printMangaInfo(info: dict): + print(info) + + def choose_chapters(self, auto): + choices = [ + questionary.Choice( + f"{chapter.get('nr'):02d}: {chapter.get('title')}", + chapter, + checked=chapter.get("nr") not in self.cached, + ) + for chapter in self.chapters + ] + self.chapters = ( + questionary.checkbox("Choose which chapters to download", choices) + .skip_if( + auto, + [ + chapter + for chapter in self.chapters + if chapter.get("nr") not in self.cached + ], + ) + .ask() + ) + + def download(self, save_dir: Path): + with tempfile.TemporaryDirectory(prefix="manga_") as temp_dir: + if self.chapters: + with Progress( + TextColumn("[progress.description]{task.description}"), + MofNCompleteColumn(), + SpinnerColumn(), + ) as p: + task = p.add_task( + f"[bold]Processing [orange3]{self.info.get('series')}[/]", + total=len(self.chapters), + ) + while not p.finished: + for chapter in self.chapters: + print( + f"Downloading [bold blue]Chapter {chapter.get('nr'):02d}[/]..." + ) + files = [self._generateComicInfo(chapter, Path(temp_dir))] + files.extend(download_images(chapter.get("images"), Path(temp_dir))) # type: ignore + zip_files(self.slug, chapter, files, save_dir) + self.cached.append(chapter.get("nr")) + p.update(task, advance=1) + else: + print("No new [orange3]Chapter[/] were found...") diff --git a/packager.py b/packager.py index fd6e77a..28b1c2a 100644 --- a/packager.py +++ b/packager.py @@ -1,50 +1,58 @@ -from pathlib import Path -import asyncio -import aiohttp -import aiofiles -from aiohttp import ClientError -import zipfile - -async def download_image(session:aiohttp.ClientSession, url:str, temp_dir:Path, retries=3, timeout=10): - - filename = temp_dir.joinpath(Path(url).name) - - for attempt in range(1, retries + 1): - try: - async with session.get(url, timeout=timeout) as response: # type: ignore - if response.status == 200: - async with aiofiles.open(filename, "wb") as f: - await f.write(await response.read()) - return filename - else: - raise ClientError(f"Bad status {response.status} for {url}") - except (asyncio.TimeoutError, ClientError, aiohttp.ClientConnectorError) as e: - if attempt < retries: - await asyncio.sleep(2 ** attempt) - else: - print(f"Failed to download {url}: {e}") - return None - -async def _download_images(urls, temp_dir:Path, concurrency=5): - semaphore = asyncio.Semaphore(concurrency) - - async with aiohttp.ClientSession() as session: - async def bounded_download(url): - async with semaphore: - return await download_image(session, url, temp_dir) - - tasks = [bounded_download(url) for url in urls] - return await asyncio.gather(*tasks) - -def download_images(images:list[str], temp_dir:Path): - return asyncio.run(_download_images(images, temp_dir)) - -def zip_files(slug, chapter:dict, files:list[Path], save_dir:Path): - save_dir.joinpath(slug).mkdir(exist_ok=True) - filename = save_dir.joinpath(slug).joinpath(f"{slug} - Ch. {chapter.get("nr")}.cbz") # type: ignore - with zipfile.ZipFile(filename, "w") as zf: - for file in files: - zf.write(file, file.name) - zf.close() - - return filename +import asyncio +import zipfile +from pathlib import Path + +import aiofiles +import aiohttp +from aiohttp import ClientError + + +async def download_image( + session: aiohttp.ClientSession, url: str, temp_dir: Path, retries=3, timeout=10 +): + + filename = temp_dir.joinpath(Path(url).name) + + for attempt in range(1, retries + 1): + try: + async with session.get(url, timeout=timeout) as response: # type: ignore + if response.status == 200: + async with aiofiles.open(filename, "wb") as f: + await f.write(await response.read()) + return filename + else: + raise ClientError(f"Bad status {response.status} for {url}") + except (asyncio.TimeoutError, ClientError, aiohttp.ClientConnectorError) as e: + if attempt < retries: + await asyncio.sleep(2**attempt) + else: + print(f"Failed to download {url}: {e}") + return None + + +async def _download_images(urls, temp_dir: Path, concurrency=5): + semaphore = asyncio.Semaphore(concurrency) + + async with aiohttp.ClientSession() as session: + + async def bounded_download(url): + async with semaphore: + return await download_image(session, url, temp_dir) + + tasks = [bounded_download(url) for url in urls] + return await asyncio.gather(*tasks) + + +def download_images(images: list[str], temp_dir: Path): + return asyncio.run(_download_images(images, temp_dir)) + + +def zip_files(slug, chapter: dict, files: list[Path], save_dir: Path): + save_dir.joinpath(slug).mkdir(exist_ok=True) + filename = save_dir.joinpath(slug).joinpath(f"{slug} - Ch. {chapter.get('nr')}.cbz") # type: ignore + with zipfile.ZipFile(filename, "w") as zf: + for file in files: + zf.write(file, file.name) + zf.close() + + return filename diff --git a/sites/__init__.py b/sites/__init__.py index 5a58225..87cf669 100644 --- a/sites/__init__.py +++ b/sites/__init__.py @@ -1,12 +1,12 @@ -from .flamecomics import FlameComics -from .provider import Provider - -PROVIDERS = [ - FlameComics, -] - -def parse_provider(session, url) -> Provider | None: - for site in PROVIDERS: - if site.domain() in url: - return site(session, url) - +from .flamecomics import FlameComics +from .provider import Provider + +PROVIDERS = [ + FlameComics, +] + + +def parse_provider(session, url) -> Provider | None: + for site in PROVIDERS: + if site.domain() in url: + return site(session, url) diff --git a/sites/flamecomics.py b/sites/flamecomics.py index 5e958be..71a8ac9 100644 --- a/sites/flamecomics.py +++ b/sites/flamecomics.py @@ -1,85 +1,94 @@ -import requests -import json -from bs4 import BeautifulSoup as bs -import re -import iso639 -from .provider import Provider - -CDN_URL = "https://cdn.flamecomics.xyz" -IMAGE_URL = CDN_URL + "/uploads/images/series/{id}/{token}/{file}" -INFO_URL = "https://flamecomics.xyz/series/{id}" - -class FlameComics(Provider): - - def __init__(self, session:requests.Session, url) -> None: - self.s = session - self.id = self.parse_url(url) - - def parse_url(self, url): - parse = re.search(r"series\/(?P\d+)", url) - return parse.group("id") # type: ignore - - @staticmethod - def parse_info(payload:dict) -> dict: - return { - "series": payload["series"].get("title"), - "writer": payload["series"].get("author"), - "penciller": payload["series"].get("artist"), - "genre": payload["series"].get("tags"), - "summary": bs(payload["series"].get("description"), "html.parser").get_text(), - "languageISO": iso639.Language.from_name(payload["series"].get("language")).part1, - "scanInformation": "Reaper_Scans & Flame Comics", - } - - def generate_image_urls(self, raw_images:dict[str,dict], token:str) -> list[str]: - images = list() - for raw in raw_images.values(): - images.append( - IMAGE_URL.format(id = self.id, token = token, file = raw.get("name")) - ) - return images[1:] - - def parse_chapters(self, payload:list[dict]) -> list[dict]: - chapters = list() - for c in payload: - nbr = int(float(c.get("chapter", 0.0))) - title = c.get("title") - if not title: - title = f"Chapter {nbr}" - images = self.generate_image_urls(c.get("images"), c.get("token")) # type: ignore - - data = { - "nr": nbr, - "title": title, - "images": images - } - - chapters.append(data) - - return chapters - - def get_info(self) -> tuple[dict, list]: - r = self.s.get(INFO_URL.format(id = self.id)) - series_page = bs(r.content, "html.parser") - payload = json.loads( - series_page.find( - "script", - attrs = {"id": "__NEXT_DATA__"} - ).text # type: ignore - ).get("props").get("pageProps") - - info = self.parse_info(payload) - chapters = self.parse_chapters(payload["chapters"]) - return info, chapters - - def get_mediainfo(self) -> tuple[dict, list]: - return self.get_info() - - def get_url(self) -> str: - return INFO_URL.format(id = self.id) - - @staticmethod - def domain() -> str: - return "flamecomics.xyz" - - +import json +import re + +import iso639 +import requests +from bs4 import BeautifulSoup as bs + +from .provider import Provider + +CDN_URL = "https://cdn.flamecomics.xyz" +IMAGE_URL = CDN_URL + "/uploads/images/series/{id}/{token}/{file}" +INFO_URL = "https://flamecomics.xyz/series/{id}" +CHAPTER_URL = "https://flamecomics.xyz/series/{id}/{token}" + + +class FlameComics(Provider): + + def __init__(self, session: requests.Session, url) -> None: + self.s = session + self.id = self.parse_url(url) + + def parse_url(self, url): + parse = re.search(r"series\/(?P\d+)", url) + return parse.group("id") # type: ignore + + @staticmethod + def parse_info(payload: dict) -> dict: + return { + "series": payload["series"].get("title"), + "writer": payload["series"].get("author"), + "penciller": payload["series"].get("artist"), + "genre": payload["series"].get("tags"), + "summary": bs( + payload["series"].get("description"), "html.parser" + ).get_text(), + "languageISO": iso639.Language.from_name( + payload["series"].get("language") + ).part1, + "scanInformation": "Reaper_Scans & Flame Comics", + } + + def generate_image_urls(self, raw_images: dict[str, dict], token: str) -> list[str]: + images = list() + for raw in raw_images.values(): + images.append( + IMAGE_URL.format(id=self.id, token=token, file=raw.get("name")) + ) + return images[1:] + + def parse_chapters(self, payload: list[dict]) -> list[dict]: + chapters = list() + for c in payload: + nbr = int(float(c.get("chapter", 0.0))) + title = c.get("title") + if not title: + title = f"Chapter {nbr}" + r = self.s.get(CHAPTER_URL.format(id=self.id, token=c.get("token"))) + chapter_page = bs(r.content, "html.parser") + chapter_payload = self.get_page_props(chapter_page).get("chapter") + images = self.generate_image_urls(chapter_payload.get("images"), chapter_payload.get("token")) # type: ignore + + data = {"nr": nbr, "title": title, "images": images} + + chapters.append(data) + + return chapters + + @staticmethod + def get_page_props(page: bs) -> dict: + return ( + json.loads( + page.find("script", attrs={"id": "__NEXT_DATA__"}).text # type: ignore + ) + .get("props") + .get("pageProps") + ) + + def get_info(self) -> tuple[dict, list]: + r = self.s.get(INFO_URL.format(id=self.id)) + series_page = bs(r.content, "html.parser") + payload = self.get_page_props(series_page) + info = self.parse_info(payload) + chapters = self.parse_chapters(payload["chapters"]) + return info, chapters + + def get_mediainfo(self) -> tuple[dict, list]: + return self.get_info() + + def get_url(self) -> str: + return INFO_URL.format(id=self.id) + + @staticmethod + def domain() -> str: + return "flamecomics.xyz" diff --git a/sites/provider.py b/sites/provider.py index edf9b70..d91da43 100644 --- a/sites/provider.py +++ b/sites/provider.py @@ -1,17 +1,17 @@ -from abc import ABC, abstractmethod - -class Provider(ABC): - - @staticmethod - @abstractmethod - def domain() -> str: - pass - - @abstractmethod - def get_mediainfo(self) -> tuple[dict, list]: - pass - - @abstractmethod - def get_url(self) -> str: - pass - \ No newline at end of file +from abc import ABC, abstractmethod + + +class Provider(ABC): + + @staticmethod + @abstractmethod + def domain() -> str: + pass + + @abstractmethod + def get_mediainfo(self) -> tuple[dict, list]: + pass + + @abstractmethod + def get_url(self) -> str: + pass