Adds a comic format registry

There is the limitation that you still need to manually import all modules that register a new comic format, but this does give us a lot of flexibility
This commit is contained in:
2025-07-30 14:40:00 -06:00
parent a0d7638598
commit a35ca0dcab
4 changed files with 236 additions and 24 deletions

View File

@@ -5,19 +5,22 @@ import argparse
import datetime
import threading
import time
from urllib.parse import urlparse
from pathlib import Path
from comic_download.xkcd import XKCDComic, XKCDComicStrip
from comic_download import Comic, find_comic_class
import comic_download.xkcd
def setup_args():
parser = argparse.ArgumentParser(
prog='xkcd',
description='Downloads the entire XKCD collection')
parser.add_argument('url', action='extend', nargs='+', type=urlparse, help="The URLs to load")
parser.add_argument('-o', '--output', type=Path, default=Path(), help="The directory to dump the files to.")
parser.add_argument('-w', '--wait', type=int, default=0, help="How many seconds to wait between each request")
parser.add_argument('-t', '--threads', type=int, default=10, help="How many download threads will run at once")
parser.add_argument('-s', '--start', type=int, default=0, help="The comic index to start at. A zero will be interpreted as using up to the first comic.")
parser.add_argument('-e', '--end', type=int, default=0, help="The comic index to end at. A zero will be interpreted as using up to the last comic.")
parser.add_argument('-s', '--start', type=int, default=0, help="The comic index to start at. A zero will be interpreted as using up to the first comic. This option is only valid if there is a single URL.")
parser.add_argument('-e', '--end', type=int, default=0, help="The comic index to end at. A zero will be interpreted as using up to the last comic. This option is only valid if there is a single URL.")
parser.add_argument('-l', '--latest', action='store_true', help="If set, only the latest comic will be downloaded")
parser.add_argument('-p', '--plain', action='store_true', help="If set, only the raw image will be downloaded, and titles, caption, etc. will not be added.")
@@ -40,7 +43,7 @@ def setup_logging():
ch.doRollover()
logger.addHandler(ch)
def load_strip(base_path, index, plain, thread_limit):
def load_strip(base_path, comic, identifier, plain, thread_limit):
"""
Threading function for downloading XKCD strips.
@@ -48,7 +51,7 @@ def load_strip(base_path, index, plain, thread_limit):
"""
with thread_limit:
strip = XKCDComicStrip(index)
strip = comic.create_strip(identifier)
strip.await_load()
if not strip.get_package_path(base_path).exists():
strip.await_download()
@@ -65,23 +68,27 @@ if __name__ == "__main__":
base_path.mkdir(parents=True)
logging.info("Beginning parsing")
comic = XKCDComic()
comic.await_load()
logging.info("There are %d comics", comic.latest_identifier)
if args.latest:
r = range(comic.latest_identifier, comic.latest_identifier + 1)
else:
start = args.start or 1
end = args.end or comic.latest_identifier
r = range(end, start - 1, -1)
threads = []
thread_limit = threading.BoundedSemaphore(value=args.threads)
for index in r:
t = threading.Thread(name=str(index), target=load_strip, args=(base_path, index, args.plain, thread_limit))
threads.append(t)
t.start()
for url in args.url:
comic_type = find_comic_class(url.netloc + url.path)
comic = comic_type.create_from_url(url)
comic.await_load()
logging.info("There are %d comics", comic.latest_identifier)
if args.latest:
r = [comic.get_latest_strip()]
else:
r = comic.get_all_strips()
start = args.start or 0
end = args.end or (len(r) - 1)
r = r[start:end + 1]
for index in reversed(r):
t = threading.Thread(name=str(comic) + str(index), target=load_strip, args=(base_path, comic, index, args.plain, thread_limit))
threads.append(t)
t.start()
for t in threads:
t.join()

View File

@@ -1,2 +1,2 @@
# Copyright (c) 2025 Markil 3
from comic_download.comic_strip import SequenceException, ImageRepo, Comic, ComicStrip, file_safe_string, get_identifier_string
from comic_download.comic_strip import SequenceException, ImageRepo, Comic, ComicStrip, file_safe_string, get_identifier_string, register_comic_class, find_comic_class

View File

@@ -6,6 +6,7 @@ import argparse
from abc import ABC, abstractmethod
from datetime import datetime
from threading import Lock
import bisect
import time
import re
import textwrap
@@ -252,6 +253,9 @@ class Comic(ImageRepo, ABC):
"""
A generic representation of a comic collection. Comic-wide resources are stored here.
"""
strip_cls = None
"""A reference to the class used for this comic's strips."""
def __init__(self, identifier):
"""
@@ -265,6 +269,14 @@ class Comic(ImageRepo, ABC):
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
@classmethod
def create_from_url(cls, url):
"""
Creates a Comic instance from a given URL
"""
pass
@abstractmethod
def _load_data(self):
"""
@@ -277,7 +289,35 @@ class Comic(ImageRepo, ABC):
relevant images, loaded into `image_urls`.
"""
pass
@abstractmethod
def get_first_strip(self):
"""
Obtains the identifier of the first strip.
"""
pass
@abstractmethod
def get_latest_strip(self):
"""
Obtains the identifier of the latest strip.
"""
pass
@abstractmethod
def get_all_strips(self) -> list:
"""
Obtains a list of all strips.
"""
pass
def create_strip(self, identifier):
"""
Creates a new ComicStrip object.
The class variable #strip_cls must be set for this to work.
"""
return self.strip_cls(self, identifier)
class ComicStrip(ImageRepo, ABC):
@@ -427,3 +467,149 @@ class ComicStrip(ImageRepo, ABC):
image_path = Path(image.name)
if image_path.exists():
image_path.unlink()
class URLFormat:
"""An internal class used to help sort URLs based on subdomains and paths"""
def __init__(self, url: str):
self.raw_url = url
if not re.match(r'^\w+://', url):
url = 'http://' + url
self.url = urlparse(url)
self.domains = self.url.netloc.split('.')
self.paths = self.url.path.split('/')
def __lt__(self, other):
if type(other) == str:
other = URLFormat(other)
if len(self.domains) != len(other.domains):
return len(self.domains) < len(other.domains)
if self.domains != other.domains:
return self.domains < other.domains
if len(self.paths) != len(other.paths):
return len(self.paths) < len(other.paths)
if self.paths != other.paths:
return self.paths < other.paths
return False
def __le__(self, other):
if type(other) == str:
other = URLFormat(other)
if self == other:
return True
return self < other
def __gt__(self, other):
if type(other) == str:
other = URLFormat(other)
if len(self.domains) != len(other.domains):
return len(self.domains) > len(other.domains)
if self.domains != other.domains:
return self.domains > other.domains
if len(self.paths) != len(other.paths):
return len(self.paths) > len(other.paths)
if self.paths != other.paths:
return self.paths > other.paths
return False
def __ge__(self, other):
if type(other) == str:
other = URLFormat(other)
if self == other:
return True
return self > other
def __eq__(self, other):
if type(other) == str:
other = URLFormat(other)
return self.domains == other.domains and self.paths == other.paths
def __ne__(self, other):
if type(other) == str:
other = URLFormat(other)
return self.domains != other.domains or self.paths != other.paths
def __str__(self):
return f"{self.url.netloc}{self.url.path}"
def __repr__(self):
return f"\"{self.__str__()}\""
def __hash__(self):
return hash(tuple(self.domains + self.paths))
comic_registry = {}
"""A registry that maps URL Domains to Comic classes."""
_registry_subdomains = []
"""An internal list keeping track of the order we should use to find the right parser (priority is last)"""
def register_comic_class(url: str, comic_cls: type[Comic]):
"""
Registers a comic type.
The URL format provided here is used to help determine
which class should be used for a given strip. There are
a few rules for the URL parameter provided here:
1. The url should not include the schema (i.e. "http://").
For example, "xkcd.com" is okay, but "https://xkcd.com" is
not.
2. The URL provided here should usually just express the
domain name. Paths should only be expressed if absolutely
necessary, such as in cases where different comic formats
are published to the same site (i.e. Tumblr). For example,
"bsky.app" works, but "bsky.app/profile" is discourages
(although "bsky.app/profile/strange_user.bsky.social"
is acceptable if strange_user does something unusual
with the way he publishes his comics and needs a custom
parser).
3. If the subdomain is "www", it should be ommited.
Use "tumblr.com" instead of "www.tumblr.com" (although
"john_doe.tumblr.com" is valid).
:param url: The URL format to register.
:param comic_cls: The class to use for parsing this comic.
"""
url_f = URLFormat(url)
if url_f in comic_registry:
raise ValueError("URL %s is already registered" % url)
comic_registry[url_f] = comic_cls
bisect.insort(_registry_subdomains, url_f)
def find_comic_class(url: str) -> type[Comic]:
"""
Finds the appropriate comic for a given URL.
:param url: The URL of the comic to parse. This must be a proper URL (including schema, etc.)
:returns: A parser
"""
url_f = URLFormat(url)
logging.info(_registry_subdomains)
while len(url_f.domains) > 0:
i = bisect.bisect(_registry_subdomains, url_f)
if i == 0:
raise ValueError("Could not find a parser for url \"%s\"" % url)
if url_f.domains == _registry_subdomains[i - 1].domains:
# We have found a series of parsers with a matching subdomain.
# Now we check the paths.
while len(url_f.paths) > 0:
i = bisect.bisect(_registry_subdomains, url_f)
if i == 0:
raise ValueError("Could not find a parser for url \"%s\"" % url)
if url_f.paths == _registry_subdomains[i - 1].paths:
# We have found a registered URL pattern that matches the URL
# provided.
return comic_registry[url_f]
else:
# Drill down to find an exact matching path
url_f = URLFormat('.'.join(url_f.domains) + '/'.join(url_f.paths[:-1]))
else:
# The result we found has a different subdomain. Look for the root subdomain
url_f = URLFormat('.'.join(url_f.domains[1:]) + '/'.join(url_f.paths))
raise ValueError("Could not find a parse for url \"%s\"" % url)

View File

@@ -16,15 +16,20 @@ import requests
from PIL import Image, ImageDraw, ImageFont
from bs4 import BeautifulSoup
from comic_download.comic_strip import Comic, ComicStrip
from comic_download.comic_strip import Comic, ComicStrip, register_comic_class
class XKCDComic(Comic):
"""
The XKCD Comic.
"""
def __init__(self):
super().__init__("xkcd")
@classmethod
def create_from_url(cls, url):
return cls()
def _load_data(self):
self.title = "XKCD"
self.author = "Randal Munroe"
@@ -37,6 +42,15 @@ class XKCDComic(Comic):
self.data = result.json()
self.latest_identifier = self.data["num"]
def get_first_strip(self):
return 1
def get_latest_strip(self):
return self.latest_identifier
def get_all_strips(self) -> list:
return list(range(self.get_first_strip(), self.get_latest_strip() + 1))
def __new__(cls):
if not hasattr(cls, 'instance'):
cls.instance = super(XKCDComic, cls).__new__(cls)
@@ -46,13 +60,14 @@ class XKCDComicStrip(ComicStrip):
"""
A comic strip object represents a single XKCD strip.
"""
def __init__(self, index: int):
def __init__(self, comic, index: int):
"""
Creates a comic strip object.
:param comic: The XKCD comic.
:param index: The index of the strip to load.
"""
super().__init__(XKCDComic(), index)
super().__init__(comic, index)
self.url = f"https://xkcd.com/{index}"
self.data_url = f"{self.url}/info.0.json"
self.transcript = None
@@ -120,3 +135,7 @@ class XKCDComicStrip(ComicStrip):
f_img.save(f_image)
f_image.close()
self.transformed_images.append(f_image)
XKCDComic.strip_cls = XKCDComicStrip
register_comic_class("xkcd.com", XKCDComic)