96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
# Copyright (c) 2025 Markil 3
|
|
import logging
|
|
import logging.handlers
|
|
import argparse
|
|
import datetime
|
|
import threading
|
|
import time
|
|
from urllib.parse import urlparse
|
|
from pathlib import Path
|
|
from comic_download import Comic, find_comic_class
|
|
import comic_download.xkcd
|
|
import comic_download.existential_comics
|
|
|
|
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. 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.")
|
|
|
|
return parser
|
|
|
|
def setup_logging():
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
ch = logging.StreamHandler()
|
|
ch.setLevel(logging.INFO)
|
|
formatter = logging.Formatter('%(levelname)s - %(message)s')
|
|
ch.setFormatter(formatter)
|
|
logger.addHandler(ch)
|
|
|
|
ch = logging.handlers.RotatingFileHandler("gutenberg_download.log", encoding='utf-8')
|
|
ch.setLevel(logging.DEBUG)
|
|
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
ch.setFormatter(formatter)
|
|
ch.doRollover()
|
|
logger.addHandler(ch)
|
|
|
|
def load_strip(base_path, comic, identifier, plain, thread_limit):
|
|
"""
|
|
Threading function for downloading XKCD strips.
|
|
|
|
Arguments:
|
|
|
|
"""
|
|
with thread_limit:
|
|
strip = comic.create_strip(identifier)
|
|
strip.await_load()
|
|
if not strip.get_package_path(base_path).exists():
|
|
strip.await_download()
|
|
strip.package_data(base_path)
|
|
|
|
if __name__ == "__main__":
|
|
setup_logging()
|
|
parser = setup_args()
|
|
|
|
args = parser.parse_args()
|
|
base_path = args.output
|
|
|
|
if not base_path.is_dir():
|
|
base_path.mkdir(parents=True)
|
|
|
|
logging.info("Beginning parsing")
|
|
|
|
threads = []
|
|
thread_limit = threading.BoundedSemaphore(value=args.threads)
|
|
|
|
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 comic.get_first_strip()
|
|
end = args.end or comic.get_latest_strip()
|
|
r = r[start - 1:end]
|
|
|
|
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()
|