doomcc/dcc/fetch.py

import dcc.config
import io
import json
import pathlib
import shutil
import subprocess
import urllib.request
import zipfile


class Fetch(dcc.config.Base):
    def get_parser(self, prog_name):
        parser = super().get_parser(prog_name)
        parser.add_argument("id_or_name")
        return parser

    def take_action(self, parsed_args):
        idgames_id = parsed_args.id_or_name
        if not parsed_args.id_or_name.isdigit():
            idgames_id = self.search_idgames(parsed_args.id_or_name)

        reply = self.fetch_url(
            "https://www.doomworld.com/idgames/api/" +
            "api.php?action=get&id={}&out=json".format(idgames_id)
        )
        rpath = "/".join([
            self.fetch_mirror,
            reply["content"]["dir"],
            reply["content"]["filename"]
        ])
        wad = reply["content"]["filename"][0:-4]

        with urllib.request.urlopen(rpath) as response:
            z = zipfile.ZipFile(io.BytesIO(response.read()))
            z.extractall(path=self.pwads.joinpath(wad))

    # TODO: explicit error handling.  Let users choose when >1 result.
    def search_idgames(self, wad):
        reply = self.fetch_url(
            "https://www.doomworld.com/idgames/api/" +
            "api.php?action=search&query={}&out=json".format(wad)
        )
        files = reply["content"]["file"]
        if type(files) is dict:  # One result.
            return files["id"]
        else:  # More than one.  Zero will raise an error.
            return files[0]["id"]

    # Ideally this would just be urllib.request.open.  However, Cloudflare can
    # interfere with this.  Therefore we support using a curl_impersonate
    # binary instead.
    def fetch_url(self, url):
        fetcher = self._config.get("url_fetcher")
        if fetcher is None:
            with urllib.request.urlopen(url) as response:
                return json.loads(response.read())
        
        fetcher_path = shutil.which(fetcher)
        if fetcher_path is None:
            raise Exception(f"Fetch util {fetcher} not found on PATH.")

        proc = subprocess.run([fetcher_path, url], capture_output=True, check = True)
        return json.loads(proc.stdout)