430 lines
16 KiB
Python
430 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2025 Markil 3
|
|
import logging
|
|
import logging.handlers
|
|
import argparse
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime
|
|
from threading import Lock
|
|
import time
|
|
import re
|
|
import textwrap
|
|
import json
|
|
from zipfile import ZipFile
|
|
from urllib.parse import urlparse
|
|
|
|
from pathlib import Path
|
|
from tempfile import NamedTemporaryFile
|
|
|
|
import requests
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
def file_safe_string(string):
|
|
"""
|
|
Makes a string safe for saving to a file by replacing characters.
|
|
|
|
The characters "/", "&", "\\", "|", and "=" are replaced by " - ".
|
|
|
|
The ":" character is replaced by " -".
|
|
|
|
The "{" and "<" characters are replaced by "(", along with the corresponding closed brackets.
|
|
|
|
All quote-like characters are replaced by a single quote.
|
|
|
|
The "?", "*", "%", and "$" characters are removed entirely.
|
|
|
|
Note that file lengths are not considered here.
|
|
|
|
:param string: The string to convert.
|
|
:returns: The file-safe form of the string.
|
|
"""
|
|
new_string = re.sub(r'[/&\\|=]', " - ", string)
|
|
new_string = re.sub(r':', " -", new_string)
|
|
new_string = re.sub(r'[{<]', '(', new_string)
|
|
new_string = re.sub(r'[}>]', ')', new_string)
|
|
new_string = re.sub(r'["“‘`]', '\'', new_string)
|
|
new_string = re.sub(r'[?*%$]', '', new_string)
|
|
new_string = re.sub(r'\s+', ' ', new_string)
|
|
|
|
return new_string
|
|
|
|
|
|
|
|
def get_identifier_string(identifier):
|
|
"""
|
|
Obtains a default file-safe string representation of a comic or strip identifer. If the identifier is a list or tuple, the
|
|
elements will be joined by a dash character.
|
|
|
|
:param identifier: The identifier of the comic or strip.
|
|
|
|
:returns: The identifier in string form.
|
|
"""
|
|
if type(identifier) == str:
|
|
return identifier
|
|
elif type(identifier) == tuple or type(identifier) == list:
|
|
# Check for spaces. Their presence influenced the delimiter string.
|
|
if any(re.search(r'\s', id) for id in identifier):
|
|
delimiter = ' - '
|
|
else:
|
|
delimiter = '-'
|
|
return file_safe_string(delimiter.join([str(id) for id in identifier]))
|
|
else:
|
|
return str(identifier)
|
|
|
|
class SequenceException(Exception):
|
|
"""
|
|
This exception is thrown when a repository object is processed out of order (i.e. downloading images before we have loaded the images).
|
|
"""
|
|
pass
|
|
|
|
class ImageRepo(ABC):
|
|
"""
|
|
A generic representation of an online repository of data. This class allows for the
|
|
setup of downloading the main information in one pass, and secondary information (images)
|
|
seperately.
|
|
"""
|
|
|
|
def __init__(self, identifier, image_type):
|
|
"""
|
|
Creates an information repository
|
|
|
|
:param identifier: The globally-unique identifier used to find this repository.
|
|
:param image_type: The type of collection used for images (i.e. "dict" or "list"). This needs to be an indexed collection.
|
|
"""
|
|
self.identifier = identifier
|
|
self.image_urls = image_type()
|
|
"""A named collection of image URLs. The value is overridden by implementing classes. This can be any sort of indexed collection (named dictionary, ordered list, etc.)
|
|
|
|
Note that despite the name, this does not have to necessarily link to images. Any data that is accessible with GET HTTP requests are valid."""
|
|
self.images = image_type()
|
|
"""A collection of temporary files for the relevent images in #image_urls. This should be the same collection type as #image_urls."""
|
|
|
|
self._data_loaded = False
|
|
"""A flag for when the data has been successfully loaded."""
|
|
self._load_lock = Lock()
|
|
"""A lock for enforcing thread safety when loading this repository."""
|
|
self._download_lock = Lock()
|
|
"""A lock for enforcing thread safety when downloading secondary resources."""
|
|
|
|
def get_identifier_string(self):
|
|
"""
|
|
Obtains a file-safe string representation of the identifer. If the identifier is a list or tuple, the
|
|
elements will be joined by a dash character.
|
|
|
|
:returns: The identifier in string form.
|
|
"""
|
|
return get_identifier_string(self.identifier)
|
|
|
|
def is_loaded(self):
|
|
"""
|
|
Checks to see if the repostiory data is fully loaded.
|
|
|
|
:returns: True if the data has been fully loaded, false otherwise.
|
|
"""
|
|
return not self._load_lock.locked() and self._data_loaded
|
|
|
|
def is_downloaded(self):
|
|
"""
|
|
Checks to see if all images (if any are present) are downloaded.
|
|
|
|
:returns: True if the images have been fully downloaded, false otherwise.
|
|
"""
|
|
return not self.image_urls or (not self._download_lock.locked() and self.images)
|
|
|
|
def load_data(self):
|
|
"""
|
|
Queries information on the comic in a thread-safe way.
|
|
|
|
This load is only performed once, and subsequent calls
|
|
will be ignored. Calls made while an existing load is
|
|
being performed are also ignored. If you need to obtain
|
|
the data directly after triggering a load, #await_load
|
|
should be called.
|
|
|
|
Implementing classes should overload the #_load_data method.
|
|
|
|
:see: #await_load
|
|
:see: #_load_data
|
|
"""
|
|
if not self._load_lock.locked() and not self._data_loaded:
|
|
if self._load_lock.acquire():
|
|
self._load_data()
|
|
logging.info("Completed loading of %s", self.get_identifier_string())
|
|
self._data_loaded = True
|
|
self._load_lock.release()
|
|
|
|
def await_load(self):
|
|
"""
|
|
Locks the current thread until all data has been loaded, starting
|
|
a load as needed.
|
|
|
|
:see: #load_data
|
|
"""
|
|
|
|
if self._load_lock.locked():
|
|
self._load_lock.acquire()
|
|
self._load_lock.release()
|
|
else:
|
|
# Loads this ourselves
|
|
self.load_data()
|
|
|
|
|
|
@abstractmethod
|
|
def _load_data(self):
|
|
"""
|
|
Queries information on the repository. To be implemented by successor classes.
|
|
|
|
Successor classes should not attempt any thread safety or be concerned about
|
|
checking for previous/concurrent calls to this method, as this logic is handled
|
|
by #load_data.
|
|
"""
|
|
pass
|
|
|
|
def download_data(self):
|
|
"""
|
|
Downloads all images for the comic.
|
|
|
|
This download is only performed once, and subsequent calls
|
|
will be ignored. Calls made while an existing download is
|
|
being performed are also ignored. If you need to obtain
|
|
the images directly after triggering a download, #await_download
|
|
should be called.
|
|
|
|
Implementing classes should overload the #_download_data method.
|
|
|
|
:see: #await_download
|
|
:see: #_download_data
|
|
"""
|
|
|
|
if not self.is_loaded():
|
|
raise SequenceException("Cannot download images before first loading performed.")
|
|
|
|
# Checks to see if we need to download and that one is not already in progress
|
|
if not self._download_lock.locked() and self.image_urls and not self.images:
|
|
if self._download_lock.acquire():
|
|
self._download_data()
|
|
logging.info("Completed downloading of %s", self.get_identifier_string())
|
|
self._download_lock.release()
|
|
|
|
def await_download(self):
|
|
"""
|
|
Locks the current thread until all images have been downloaded, starting
|
|
the download as needed.
|
|
|
|
:see: #download_data
|
|
"""
|
|
|
|
if self._download_lock.locked():
|
|
self._download_lock.acquire()
|
|
self._download_lock.release()
|
|
else:
|
|
# Downloads this ourselves
|
|
self.download_data()
|
|
|
|
def _download_data(self):
|
|
"""
|
|
Downloads all images and saves them to temporary files in the #images field.
|
|
|
|
The default implementation should be satisfactory for most use cases, but
|
|
can be overridden if necessary. Overriders should not attempt any thread
|
|
safety or be concerned about checking for previous/concurrent calls to
|
|
this method, as this logic is handled by #download_data.
|
|
"""
|
|
if self.image_urls:
|
|
if type(self.image_urls) == dict:
|
|
iterator = self.image_urls.keys()
|
|
elif type(self.image_urls) == list:
|
|
iterator = range(len(self.image_urls))
|
|
for name in iterator:
|
|
logging.info(" Downloading repository %s resource %s", self.get_identifier_string(), name)
|
|
image_path = NamedTemporaryFile(mode='wb', suffix=str(name), prefix=self.get_identifier_string(), delete = False)
|
|
result = requests.get(self.image_urls[name])
|
|
result.raise_for_status()
|
|
image_path.write(result.content)
|
|
image_path.close()
|
|
if type(self.images) == list:
|
|
self.images.append(image_path)
|
|
else:
|
|
self.images[name] = image_path
|
|
|
|
|
|
class Comic(ImageRepo, ABC):
|
|
"""
|
|
A generic representation of a comic collection. Comic-wide resources are stored here.
|
|
"""
|
|
|
|
def __init__(self, identifier):
|
|
"""
|
|
Creates a comic collection.
|
|
|
|
:param identifier: The globally-unique identifier used to find the comic.
|
|
"""
|
|
super().__init__(identifier, dict)
|
|
self.title = None
|
|
"""The title of the comic."""
|
|
self.author = None
|
|
"""The author(s) of the comic. Can be a string for a name, a list of names, or a list of objects containing the names under "name" and role under "role"."""
|
|
|
|
@abstractmethod
|
|
def _load_data(self):
|
|
"""
|
|
Queries the source to obtain information on the comic.
|
|
|
|
Implementing methods need to find the following information:
|
|
* The title of the comic, loaded to `title`.
|
|
* The author of the comic, loaded to `author`.
|
|
* A collection of banners, author avatars, and other
|
|
relevant images, loaded into `image_urls`.
|
|
"""
|
|
pass
|
|
|
|
|
|
|
|
class ComicStrip(ImageRepo, ABC):
|
|
"""
|
|
A generic representation of a single comic strip or issue.
|
|
"""
|
|
|
|
def __init__(self, comic, identifier):
|
|
"""
|
|
Creates a comic strip
|
|
|
|
:param comic: The `Comic` object that this strip is part of.
|
|
:param indentifier: The globally unique identifier used to find the strip on the website. Can be a tuple.
|
|
"""
|
|
super().__init__(identifier, list)
|
|
|
|
self.comic = comic
|
|
"""The comic that this strip is part of."""
|
|
self.title = None
|
|
"""The title of the comic. Empty until the comic data is loaded."""
|
|
self.image_urls = []
|
|
"""A list of URLs we can download the comic panels from, in order. Empty until the comic data is loaded."""
|
|
self._image_paths = []
|
|
"""A list of temporary files for the raw downloaded panels. Used internally."""
|
|
self.captions = []
|
|
"""A list of captions for each panel."""
|
|
self.date = None
|
|
"""The date that the comic was published."""
|
|
self._transform_lock = Lock()
|
|
"""A lock for enforcing thread safety when transforming resources."""
|
|
self.transformed_images = []
|
|
"""A list of NamedTemporaryFiles linking to the transformed versions of the image downloads."""
|
|
self._transformed = False
|
|
"""A flag keeping track of whether the strip transformation took place."""
|
|
|
|
def get_filename(self):
|
|
"""
|
|
Obtains the name of the final export (not including the stem). Calling this method requires the data be loaded first.
|
|
|
|
:returns: A human-friendly file, like "05 - Riddles in the Dark" or "2025-06-01 - Chapter 5."
|
|
"""
|
|
prefix_num_form = "{:02}"
|
|
if type(self.identifier) == int:
|
|
prefix = prefix_num_form.format(self.identifier)
|
|
elif (type(self.identifier) == list or type(self.identifier) == tuple) and type(self.identifier[-1]) == int:
|
|
prefix = prefix_num_form.format(self.identifier[-1])
|
|
elif type(self.date) == datetime.date:
|
|
prefix = self.date.isoformat()
|
|
elif type(self.date) == datetime.datetime:
|
|
prefix = self.date.date().isoformat()
|
|
else:
|
|
prefix = self.get_identifier_string()
|
|
if self.title:
|
|
return (prefix + " - " + file_safe_string(self.title))[0:250]
|
|
else:
|
|
return prefix
|
|
|
|
def get_package_path(self, base_path: Path):
|
|
"""
|
|
Obtains the path to save the comic archive to.
|
|
|
|
:param base_path: The folder to save comics to.
|
|
:returns: The path to save to.
|
|
"""
|
|
return Path(base_path, f"{self.get_filename()}.cbz")
|
|
|
|
def load_data(self):
|
|
super().load_data()
|
|
self.comic.load_data()
|
|
|
|
def download_data(self):
|
|
super().download_data()
|
|
self.comic.download_data()
|
|
|
|
def await_load(self):
|
|
super().await_load()
|
|
self.comic.await_load()
|
|
|
|
def await_download(self):
|
|
super().await_download()
|
|
self.comic.await_download()
|
|
|
|
@abstractmethod
|
|
def _load_data(self):
|
|
"""
|
|
Queries the source to obtain information on the comic.
|
|
|
|
Implementing methods need to find the following information:
|
|
The title of the comic, loaded to `title`.
|
|
The publish date (or datetime) of the comic, loaded to `date`.
|
|
A list of panel images, loaded into `image_urls`.
|
|
A list of captions, loaded into `captions`. Any captions beyond the list of images can be added to new pages without comics.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def _transform_images(self):
|
|
"""
|
|
Called by the packaging method to create a series of locally-transformed
|
|
images and returning the ordered list of them.
|
|
|
|
This method is meant to do things like add alt text, descriptions, title pages, etc.
|
|
to the final comic. Simpler implementations can simply return a list of the
|
|
unmodified images with the comic headers from #comic at the beginning.
|
|
|
|
All implementations should assign the ordered list to #transformed_images. If this
|
|
field is not set, then packaging will default to just package the comic strip
|
|
images.
|
|
"""
|
|
self.transformed_images = self.images
|
|
|
|
def package_data(self, base_path: Path):
|
|
"""
|
|
Compresses all the data into a cbz file.
|
|
"""
|
|
if not self.is_loaded():
|
|
raise SequenceException("Cannot package comic strip before strip data has been fully loaded.")
|
|
if not self.is_downloaded():
|
|
raise SequenceException("Cannot package comic strip before strip data has been fully downloaded.")
|
|
if not self.comic.is_loaded():
|
|
raise SequenceException("Cannot package comic before data has been fully loaded.")
|
|
if not self.comic.is_downloaded():
|
|
raise SequenceException("Cannot package comic before data has been fully downloaded.")
|
|
if not self.transformed_images:
|
|
if not self._transformed:
|
|
if self._transform_lock.locked():
|
|
# Await for the existing transform to complete
|
|
self._transform_lock.acquire()
|
|
self._transform_lock.release()
|
|
else:
|
|
# Run the transformation outselves
|
|
self._transform_lock.acquire()
|
|
self._transform_images()
|
|
self._transformed = True
|
|
self._transform_lock.release()
|
|
with ZipFile(self.get_package_path(base_path), 'w') as comic_zip:
|
|
for i in range(len(self.transformed_images)):
|
|
image = self.transformed_images[i]
|
|
image_path = Path(image.name)
|
|
with open(image.name, 'rb') as image_stream, comic_zip.open(str(i + 1) + image_path.suffix, 'w') as zip_stream:
|
|
zip_stream.write(image_stream.read())
|
|
# Delete the base images, as they are no longer needed.
|
|
# We cannot touch the transformed images, since some
|
|
# of them may come directly from the base comic, and
|
|
# we have no way of telling.
|
|
for image in self.images:
|
|
image_path = Path(image.name)
|
|
if image_path.exists():
|
|
image_path.unlink()
|