Lightning speed, multithreaded, Lolicon downloader.

loli.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

# Possession of materials alluding to Loli or Lolicon vary upon jurisdiction.
# Henceforth in no event shall the author ("xsbee") be liable for any damage,
# financial, legal, reputational, mental or physical. Use at your own risk.

# ロリまたはロリコンをほのめかす資料の所有は、司法管轄区によって異なります。
# 今後いかなる場合でも、著者 (「xsbee」) は、金銭的、法律的、評判、精神的または身
# 体的損害について責任を負わないものとします。 自己責任。

import urllib3
import json
import logging
import os
import os.path
import shutil

from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlparse


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("lolidl")

http = urllib3.PoolManager(num_pools=2, maxsize=min(32, os.cpu_count() + 4))


def download(url, filepath):
    r = http.request("get", url, preload_content=False)
    out = open(filepath, "wb", buffering=0)

    shutil.copyfileobj(r, out)

    out.close()
    r.release_conn()


with ThreadPoolExecutor() as executor:
    while True:
        r = http.request(
            "GET", "https://api.lolicon.app/setu/v2", fields={"r18": True, "num": 20}
        )

        imglist = json.loads(r.data)

        if imglist["error"]:
            logger.log(logging.WARN, "API says %s", imglist["error"])
            continue

        dl_count = 0

        for ent in imglist["data"]:
            # Original URL is perhaps guranteed to be present (?)
            img_url = ent["urls"]["original"]

            # Basename of the pathname (is there a Content-Disposition present?)
            img_filename = os.path.basename(urlparse(img_url)[2])

            # If it already exists, don't waste time downloading it.
            if os.path.exists(img_filename):
                continue

            logger.log(logging.INFO, "Downloading %s to filename %s", ent["title"], img_filename)

            executor.submit(download, img_url, img_filename)

            dl_count += 1
        
        logger.log(logging.INFO, "Queued %d images", dl_count)