path: root/util.py

#!/usr/bin/env python3
import os
import sys
import hashlib
import shutil
import re
import collections
import urllib.parse
import multiprocessing
import pathlib
import base64
from configparser import RawConfigParser

from selenium.webdriver.common.by import By

import requests


def load_config():
    """
    Load configuarion from pack and local configuration files
    Fill in reasonable defaults where applicable.
    """
    config_p = RawConfigParser()
    config_p.read(["pack.ini", "local-config.ini"])
    config = config_p._sections
    config["pack"]["sanitized_name"] = sanitize_text(config["pack"]["name"])
    
    if "location" not in config["pack"]:
        config['pack']['location'] = os.path.join(find_minecraft_directory(), config['pack']['sanitized_name'])
    
    if "whitelist" not in config["pack"]:
        config["pack"]["whitelist"] = []
    else:
        config["pack"]["whitelist"] = config["pack"]["whitelist"].split(",")

    if "blacklist" not in config["pack"]:
        config["pack"]["blacklist"] = []
    else:
        config["pack"]["blacklist"] = config["pack"]["blacklist"].split(",")
    
    config["pack"]["game_version"] = game_version_from_string(config["pack"]["game_version"])

    # return the whole config file, pack configuration and modlist
    return config


def update_self():
    """
    Try to download new versions of all of the pack configuration and data files
    in order to update to the latest version.  Will overwrite any existing pack.ini and other
    config files, so be careful!
    """
    global config

    base_url = config["pack"]["pack_base_url"].strip("/") + "/"
    base_url = base_url.replace('tree', 'plain')
    download_text_file(base_url + "pack.ini", "pack.ini")
    download_text_file(base_url + "pack-lock.ini", "pack-lock.ini")
    download_file(base_url + "icon.png", "icon.png")

    pack_lock = RawConfigParser()
    pack_lock.read(["pack-lock.ini"])
    for path in pack_lock["global"]["config_files"].split(","):
        if not path:
            continue
        download_text_file(f"{base_url}config/{path}", os.path.join("config", path))
    
    config = load_config()


def find_minecraft_directory():
    """
    Find the location of the user's .minecraft folder based on
    their operating system.
    :returns: the absolute path to the .minecraft directory
    """
    if sys.platform == "linux":
        return os.path.join(os.path.expanduser('~'), ".minecraft")
    elif sys.platform == "win32":
        return os.path.join(os.environ["APPDATA"], ".minecraft")
    elif sys.platform == "darwin":
        return os.path.join(os.path.expanduser('~'), "Library", "Application Support", "minecraft")
    else:
        raise RuntimeError(f"Unsupported operating system `{sys.platform}`. Please define a location for the pack in your `local-config.ini` file")


def find_jre():
    """
    Find a usable install of Java, either from a user-installed JRE or
    from the Minecraft Launcher's integrated JRE.

    :return: the absolute path of a working Java executable
    """
    if shutil.which("java") is not None:
        return shutil.which("java")
    if sys.platform == 'win32':  # We can try and use the Minecraft Launcher's integrated JRE on Windows
        if os.path.exists("C:\\Program Files (x86)\\Minecraft Launcher\\runtime\\jre-x64\\java.exe"):
            return "C:\\Program Files (x86)\\Minecraft Launcher\\runtime\\jre-x64\\java.exe"
        elif os.path.exists("C:\\Program Files (x86)\\Minecraft Launcher\\runtime\\jre-x64\\bin\\java.exe"):
            return "C:\\Program Files (x86)\\Minecraft Launcher\\runtime\\jre-x64\\bin\\java.exe"
    raise RuntimeError("Unable to detect an installed JRE.  Please install Java in order to use modpackman.")


def download_file(url, destination):
    """
    Given a url, performs a requests request to get the remote object
    and write it to destination.
    Note that this only works on binary files.
    """
    with open(destination, "wb") as f:
        with requests.get(url, stream=True) as dl:
            shutil.copyfileobj(dl.raw, f)

def download_text_file(url, destination):
    """
    Given the URL to a text file, download it to the file named
    by `destination`.  Note that this only works for text files, not binary files.
    """
    with open(destination, "w") as f:
        f.write(requests.get(url).text)


# take a string and only keep filename-friendly parts
def sanitize_text(text):
    sanitized = ""
    replacement_map = {" ": "-"}
    for char in text:
        if char.isalnum():
            sanitized += char.lower()
        elif char in replacement_map:
            sanitized += replacement_map[char]
    return sanitized


def generate_base64_icon(filename):
    with open(filename, "rb") as f:
        return "data:image/png;base64," + base64.b64encode(f.read()).decode("utf8")


def read_file(fil):
    """
    Given a filename, read its contents in as a list of tuples.
    This function strips out comment lines and whitespaces.
    """
    strings = []
    with open(fil) as f:
        for line in f:
            string = line.strip().split()
            if len(line) > 1 and line[0] != '#':
                    # run strip on each element
                    string = tuple(map(lambda x: x.strip(), string))
                    strings.append(string)

    return strings


def game_version_from_string(string):
    if string is not None:
        try:
            return tuple(int(x) for x in string.split('.'))
        except:
            pass
    return (2, 0, 0)



def threaded_find_url(homepage_url, game_version):
    """
    Helper function that finds a single mod URL based on the homepage.
    """
    if 'curseforge' in homepage_url:
        ffx = firefox()
        final_url = find_cdn(ffx, homepage_url, game_version)
        ffx.close()
    else:
        final_url = requests.get(homepage_url).url
    return final_url


def find_updated_urls(forge_urls, game_version, threads=8):
    """
    Given a list of mod homepage URLs, find all of their direct download links in parallel.
    """

    # First, check that we can successfully open a Firefox instance in the main thread.
    # This provides us with a much nicer error message and quicker feedback.
    f = firefox()
    f.close()

    with multiprocessing.Pool(threads) as pool:
        # No progress indicator possible
        # return pool.map(threaded_find_url, forge_urls)

        # Much longer, but allows us to do a nice progress indicator
        result_futures = []
        for url in forge_urls:
            result_futures.append(pool.apply_async(threaded_find_url, (url, game_version)))

        results = []
        for i,f in enumerate(result_futures):
            results.append(f.get())
            print(f'\r{i+1}/{len(result_futures)} URLs updated ({round((i+1)/len(result_futures)*100)}%)', end='')
        print()

        return results


def threaded_calc_sha1(direct_url):
    """
    Helper function that downloads and calculates a single SHA1 hash from a direct download URL.
    """
    resp = requests.get(direct_url)
    hsh = hashlib.sha1(resp.content).hexdigest()
    return hsh


def find_checksums(direct_urls, threads=8):
    """
    Given a list of direct download URLs, download them all and calculate the SHA1 checksum of the file at that location.
    """
    
    with multiprocessing.Pool(threads) as pool:
        # Much longer, but allows us to do a nice progress indicator
        result_futures = []
        for url in direct_urls:
            result_futures.append(pool.apply_async(threaded_calc_sha1, (url,)))

        results = []
        for i,f in enumerate(result_futures):
            results.append(f.get())
            print(f'\r{i+1}/{len(result_futures)} checksums calculated ({round((i+1)/len(result_futures)*100)}%)', end='')
        print()

        return results


def find_cdn(ffx, url, version):
    """
    Given a mod home URL, finds the most up-to-date mod version compatible with the given game version.
    Returns the direct Forge CDN download URL
    """
    try:
        # This goes to the "all files" page, where we get a table view of all
        page_index = 0;
        while True:
            ffx.get(url + f'/files/all?page={page_index}')
            mod_versions = ffx.find_element(By.CLASS_NAME, "listing").find_elements(By.XPATH, "tbody/tr") # extract the table of files from the page
            row_info = collections.namedtuple("row_info", ["type", "filename", "cdn_id", "game_version"]) # create a custom tuple because data
            rows = []
            for version_entry in mod_versions:
                # parse out the four fields that we use
                entry_cells = version_entry.find_elements(By.TAG_NAME, "td")
                release_type = entry_cells[0].text
                # Note that this is NOT the final filename - this is just the "release name".
                filename = urllib.parse.quote(entry_cells[1].find_elements(By.TAG_NAME, "a")[0].text)
                try:
                    game_version = tuple([int(x) for x in entry_cells[4].find_element(By.CLASS_NAME, "mr-2").text.split(".")]) # get game version and convert to tuple
                except:
                    if ".".join(map(str, version)) in filename:
                        game_version = version
                    else:
                        game_version = (2, 0, 0)
                cdn_id = entry_cells[1].find_element(By.TAG_NAME, "a").get_property("href").split("/")[-1]

                #TODO make this configurable
                if 'fabric' not in filename.lower() or 'forge' in filename.lower():
                    rows.append(row_info(release_type, filename, cdn_id, game_version))
            rows.sort(key=lambda x: x.game_version, reverse=True)
            try:
                best_row = next(x for x in rows if x.game_version <= version)
                break
            except StopIteration:
                if len(ffx.find_elements(By.CLASS_NAME, "pagination-next--inactive")) != 0:
                    raise
                page_index += 1


        # We need to find the real, ForgeCDN compatible filename now by going to the file page.
        ffx.get(f'{url}/files/{best_row.cdn_id}')
        # This will probably break in the future
        filename = ffx.find_elements(By.XPATH, "html/body/div/main/div/div/section/div/div/div/section/section/article/div/div/span")[1].text
        # URL escape the filename!
        filename = urllib.parse.quote(filename)

        # ForgeCDN requires that the leading zeroes are stripped from each portion of the CDN ID, hence the int() cast.
        return f'https://media.forgecdn.net/files/{int(best_row.cdn_id[:4])}/{int(best_row.cdn_id[4:])}/{filename}'

    except:
        import traceback; traceback.print_exc()
        print(f"\n[!] Failed to retrieve valid CDN URL for {url}")
        return None


def firefox():
    """
    Start a headless Firefox instance and return the Selenium refrence to it.
    """
    try:
        from selenium.webdriver import Firefox
        from selenium.webdriver.firefox.options import Options
    except:
        print("Applying updates requires the `selenium` package")
        exit(0)
    options = Options()
    options.add_argument('-headless')
    options.add_argument('--window-size 1920,1080')
    
    # for ~~cursed~~ windows people, put geckodriver in the folder next to modpackman.py
    if(os.path.exists("../../geckodriver.exe")):
        return Firefox(executable_path='../../geckodriver', options=options)
    return Firefox(options=options)

def get_version_from_file(filename="pack-lock.ini"):
    pack_lock = RawConfigParser()
    pack_lock.read('pack-lock.ini')
    return pack_lock['global']['pack_version']
 


# Configuration is automatically loaded from pack.ini and local-config.ini,
# and made accessible here as a global
config = load_config()