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
from configparser import ConfigParser

import requests


def load_config():
    """
    Load configuarion from pack and local configuration files
    Fill in reasonable defaults where applicable.
    """
    config_p = ConfigParser()
    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(",")
    
    config["pack"]["game_version"] = game_version_from_string(config["pack"]["game_version"])

    # return the whole config file, pack configuration and modlist
    return 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"
    raise RuntimeError("Unable to detect an installed JRE.  Please install Java in order to use modpackman.")


# 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 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 get_version_from_file(fil):
    with open(fil) as f:
        for line in f:
            if line.strip().split()[0] == "#VERSION":
                return int(line.strip().split()[1])
    return 0

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=20):
    """
    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 
        ffx.get(url + '/files/all')
        mod_versions = ffx.find_elements_by_class_name("listing")[0].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:
                game_version = (0, 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)
        best_row = next(x for x in rows if x.game_version <= version)

        # 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()
        return None


def firefox():
    """
    Start a headless Firefox instance and return the Selenium refrence to it.
    """
    #print("Starting Selenium...")
    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 this folder
    if(os.path.exists("./geckodriver.exe")):
        return Firefox(executable_path='./geckodriver')
    return Firefox()


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