Code formatting and updated flamecomic to use changed api

This commit is contained in:
2026-03-12 14:18:31 +01:00 Verified
parent 7591639177
commit 9cd15e33fb
7 changed files with 600 additions and 564 deletions
+31 -25
View File
@@ -1,31 +1,30 @@
import requests
from pathlib import Path
import argparse
import configparser
import json
from pathlib import Path
import questionary
from rich import print
from sites import parse_provider
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!"
epilog="Happy Reading!",
)
parser.add_argument(
"--auto",
default=False,
action=argparse.BooleanOptionalAction,
help="Run using auto mode (default: %(default)s)"
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"
help="Set temporary save dir instead of using config value",
)
args = parser.parse_args()
@@ -41,7 +40,9 @@ def get_config() -> configparser.ConfigParser:
config.read(config_file)
if not config.has_option("DEFAULT", "savedir"):
path = questionary.path("Enter default save path for downloaded Manga: ", qmark="").ask()
path = questionary.path(
"Enter default save path for downloaded Manga: ", qmark=""
).ask()
config["DEFAULT"] = {"savedir": path}
with open(config_file, "w") as cf:
@@ -49,26 +50,26 @@ def get_config() -> configparser.ConfigParser:
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:
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
cached_chapters = cache.get(series, {}).get("cached", []) + manga.cached
cache[series] = {
"url": manga.site.get_url(),
"cached": list(set(cached_chapters))
}
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)
json.dump(cache, fp, indent=4)
def parse_cache(cache:dict) -> list[Manga]:
def parse_cache(cache: dict) -> list[Manga]:
if not cache:
print("Cache is empty!")
exit()
@@ -80,25 +81,25 @@ def parse_cache(cache:dict) -> list[Manga]:
)
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 = ""
manga.append(questionary.text("Input Manga url:", qmark="").ask())
add_more = questionary.confirm(
"Do you want to add more Manga", default=False
).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]:
def parse_manga(manga: list[str]) -> list[Manga]:
parsed_manga = list()
for entry in manga:
provider = parse_provider(s, entry)
@@ -108,8 +109,11 @@ def parse_manga(manga:list[str]) -> list[Manga]:
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()
use_cached = questionary.confirm(
"Do you want to choose from cached Manga", default=True
).ask()
if use_cached:
return get_cached_manga()
@@ -117,6 +121,7 @@ def choose_manga() -> list[Manga]:
manga = input_manga()
return parse_manga(manga)
def main():
config = get_config()
@@ -136,5 +141,6 @@ def main():
manga.download(save_dir)
write_to_cache(manga)
if __name__ == "__main__":
main()
+37 -24
View File
@@ -1,27 +1,27 @@
import tempfile
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
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:
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:
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):
def _generateComicInfo(self, chapter: dict, dir: Path):
filename = dir.joinpath("ComicInfo.xml")
ComicInfo = XML.Element("ComicInfo")
@@ -34,7 +34,9 @@ class Manga():
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")
XML.SubElement(ComicInfo, "ScanInformation").text = self.info.get(
"scanInformation"
)
tree = XML.ElementTree(ComicInfo)
@@ -44,37 +46,48 @@ class Manga():
return filename
@staticmethod
def printMangaInfo(info:dict):
def printMangaInfo(info: dict):
print(info)
def choose_chapters(self, auto):
choices = [
questionary.Choice(
f"{chapter.get("nr"):02d}: {chapter.get("title")}",
f"{chapter.get('nr'):02d}: {chapter.get('title')}",
chapter,
checked = chapter.get("nr") not in self.cached
) for chapter in self.chapters
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(
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()
[
chapter
for chapter in self.chapters
if chapter.get("nr") not in self.cached
],
)
.ask()
)
def download(self, save_dir:Path):
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()
SpinnerColumn(),
) as p:
task = p.add_task(f"[bold]Processing [orange3]{self.info.get("series")}[/]", total=len(self.chapters))
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}[/]...")
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)
+18 -10
View File
@@ -1,11 +1,15 @@
from pathlib import Path
import asyncio
import aiohttp
import aiofiles
from aiohttp import ClientError
import zipfile
from pathlib import Path
async def download_image(session:aiohttp.ClientSession, url:str, temp_dir:Path, retries=3, timeout=10):
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)
@@ -20,15 +24,17 @@ async def download_image(session:aiohttp.ClientSession, url:str, temp_dir:Path,
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)
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):
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)
@@ -36,12 +42,14 @@ async def _download_images(urls, temp_dir:Path, concurrency=5):
tasks = [bounded_download(url) for url in urls]
return await asyncio.gather(*tasks)
def download_images(images:list[str], temp_dir:Path):
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):
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
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)
+1 -1
View File
@@ -5,8 +5,8 @@ PROVIDERS = [
FlameComics,
]
def parse_provider(session, url) -> Provider | None:
for site in PROVIDERS:
if site.domain() in url:
return site(session, url)
+36 -27
View File
@@ -1,17 +1,21 @@
import requests
import json
from bs4 import BeautifulSoup as bs
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:
def __init__(self, session: requests.Session, url) -> None:
self.s = session
self.id = self.parse_url(url)
@@ -20,54 +24,61 @@ class FlameComics(Provider):
return parse.group("id") # type: ignore
@staticmethod
def parse_info(payload:dict) -> dict:
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,
"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]:
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"))
IMAGE_URL.format(id=self.id, token=token, file=raw.get("name"))
)
return images[1:]
def parse_chapters(self, payload:list[dict]) -> list[dict]:
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
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
}
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")
@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
@@ -76,10 +87,8 @@ class FlameComics(Provider):
return self.get_info()
def get_url(self) -> str:
return INFO_URL.format(id = self.id)
return INFO_URL.format(id=self.id)
@staticmethod
def domain() -> str:
return "flamecomics.xyz"
+1 -1
View File
@@ -1,5 +1,6 @@
from abc import ABC, abstractmethod
class Provider(ABC):
@staticmethod
@@ -14,4 +15,3 @@ class Provider(ABC):
@abstractmethod
def get_url(self) -> str:
pass