pysp.py

# pysteampacker by puyodead1, based on supersteampacker https://github.com/Masquerade64/SuperSteamPacker
# pip install vdf requests
# this is fucking garbage, but it works
# omit --password to use cached credentials (after logging in with it at least once)
import argparse
import configparser
import datetime
import os
import platform
import re
import shutil
import subprocess
from pathlib import Path

import requests
import vdf

STEAMCMD_LINUX = "https://steamcdn-a.akamaihd.net/client/installer/steamcmd_linux.tar.gz"
STEAMCMD_WINDOWS = "https://steamcdn-a.akamaihd.net/client/installer/steamcmd.zip"
STEAMCMD_MAC = "https://steamcdn-a.akamaihd.net/client/installer/steamcmd_osx.tar.gz"
DEPOT_NAMES = "https://raw.githubusercontent.com/Masquerade64/SteamDepotNames/main/depots.ini"

CURRENT_SYSTEM = platform.system().lower()
CURRENT_SYSTEM = "win" if CURRENT_SYSTEM == "windows" else CURRENT_SYSTEM

CURRENT_ARCH = platform.architecture()[0][:2]

CURRENT_PLATFORM_CODE = f"{CURRENT_SYSTEM}{CURRENT_ARCH}"

ROOT_DIR = Path(__file__).parent
WORK_DIR = Path(ROOT_DIR, "work")
TEMP_DIR = Path(WORK_DIR, "temp")
COMPLETED_DIR = Path(ROOT_DIR, "completed")
SCRIPT_PATH = Path(WORK_DIR, "script.job")
STEAMCMD_DIR = Path(ROOT_DIR, "SteamCMD")
HOME_DIR = Path.home()
# steamcmd isnt relative on linux and probably mac, fuck you steam
STEAM_DIR = STEAMCMD_DIR if CURRENT_SYSTEM == "win" else Path(HOME_DIR, "Steam")
STEAMAPPS_DIR = Path(STEAM_DIR, "steamapps")
DEPOTCACHE_DIR = Path(STEAM_DIR, "depotcache")
LOGS_DIR = Path(STEAM_DIR, "logs")
STEAMCMD = Path(STEAMCMD_DIR, "steamcmd.exe" if CURRENT_SYSTEM == "win" else "steamcmd.sh")
STEAMCMD_CONTENT_LOG_FILE = Path(LOGS_DIR, "content_log.txt")
STEAMCMD_CONNECTION_LOG_FILE = Path(LOGS_DIR, "connection_log.txt")
STEAMCMD_SITELICENSE_FILE = Path(LOGS_DIR, "sitelicense_steamcmd.txt")
STEAMCMD_COMPAT_LOG_FILE = Path(LOGS_DIR, "compat_log.txt")


def download_steamcmd_windows() -> int:
    return subprocess.run(
        f'curl -sqL "{STEAMCMD_WINDOWS}" -o "{TEMP_DIR}/steamcmd.zip" && "C:\Windows\System32\\tar.exe" -xf "{TEMP_DIR}/steamcmd.zip" -C "{STEAMCMD_DIR}" && "{STEAMCMD}" +quit',
        shell=True,
    ).returncode


def download_steamcmd_linux() -> int:
    return subprocess.run(
        f'sudo apt-get install lib32gcc-s1 -y; curl -sqL "https://steamcdn-a.akamaihd.net/client/installer/steamcmd_linux.tar.gz" | tar zxvf - -C "{STEAMCMD_DIR}"; "{STEAMCMD}" +quit',
        shell=True,
    ).returncode


def download_steamcmd_mac() -> int:
    return subprocess.run(
        f'curl -sqL "https://steamcdn-a.akamaihd.net/client/installer/steamcmd_linux.tar.gz" | tar zxvf - -C "{STEAMCMD_DIR}"; "{STEAMCMD}" +quit',
        shell=True,
    ).returncode


def ensure_steamcmd() -> None:
    print("Checking for SteamCMD...")
    if not STEAMCMD.exists():
        print("SteamCMD not found, downloading...")
        if CURRENT_SYSTEM == "win":
            r = download_steamcmd_windows()
        elif CURRENT_SYSTEM == "linux":
            r = download_steamcmd_linux()
        elif CURRENT_SYSTEM == "darwin":
            r = download_steamcmd_mac()
        else:
            print("Unknown system")
            exit(1)

        if r != 0:
            print("Something went wrong while trying to install SteamCMD, non zero return code")
            exit(r)
    else:
        print("SteamCMD found")


def ensure_directories() -> None:
    STEAMCMD_DIR.mkdir(parents=True, exist_ok=True)

    print("Clearing any old data...")
    try:
        shutil.rmtree(WORK_DIR)
    except FileNotFoundError:
        pass

    COMPLETED_DIR.mkdir(exist_ok=True)

    print("Creating working folders...")
    TEMP_DIR.mkdir(parents=True, exist_ok=False)


def get_steam_game_data(appid: str) -> dict:
    r = requests.get(f"https://api.steamcmd.net/v1/info/{appid}")
    r.raise_for_status()
    data = r.json()
    if data["status"] != "success":
        raise Exception("Failed to get app data")

    return data


def run_script() -> int:
    return subprocess.run(f'"{STEAMCMD}" +runscript "{SCRIPT_PATH}"', shell=True).returncode


def cleanup(steamonly=False, completed=False) -> None:
    if STEAMAPPS_DIR.exists():
        shutil.rmtree(STEAMAPPS_DIR)

    if DEPOTCACHE_DIR.exists():
        shutil.rmtree(DEPOTCACHE_DIR)

    if steamonly:
        return

    if LOGS_DIR.exists():
        shutil.rmtree(LOGS_DIR)

    if WORK_DIR.exists():
        shutil.rmtree(WORK_DIR)

    if completed and COMPLETED_DIR.exists():
        shutil.rmtree(COMPLETED_DIR)


def edit_vdf(path: str, key: str, value: str):
    with open(path, "r") as file:
        file_contents = file.read()

    regex = re.compile(r'"{}"\s+"([^"]*)"'.format(key))
    match = regex.search(file_contents)

    if match:
        old_value = match.group(1)
        file_contents = file_contents.replace('"{}"'.format(old_value), '"{}"'.format(value))

        with open(path, "w") as file:
            file.write(file_contents)


def read_vdf(file_path, key_to_read):
    with open(file_path, "r") as file:
        file_contents = file.read()

    regex = re.compile(r'"{}"\s+"([^"]*)"'.format(key_to_read))
    match = regex.search(file_contents)

    if match:
        return match.group(1)

    return None


def sanitize_filename(filename):
    invalid_chars = '<>:"/\\|?*'
    return re.sub(f"[{re.escape(invalid_chars)}]", "", filename)


def load_depot_names() -> configparser.ConfigParser:
    r = requests.get(DEPOT_NAMES)
    r.raise_for_status()
    c = configparser.ConfigParser()
    c.read_string(r.text)

    return c


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Steam Packer")
    parser.add_argument("appid", type=int)
    parser.add_argument("--username", type=str)
    parser.add_argument("--password", type=str)
    parser.add_argument("--anonymous", action="store_true")
    parser.add_argument(
        "--arch", type=str, default=CURRENT_PLATFORM_CODE, choices=["win32", "win64", "linux32", "linux64", "macos"]
    )
    parser.add_argument("--branch", type=str, default="public")
    parser.add_argument("--branch_password", type=str)
    parser.add_argument("--clean", action="store_true")

    args = parser.parse_args()
    appid = str(args.appid)
    anonymous = args.anonymous
    username = args.username if not anonymous else "anonymous"
    password = args.password if not anonymous else ""
    arch = args.arch
    branch = args.branch
    branch_password = args.branch_password
    clean = args.clean

    if clean:
        cleanup(False, True)
        exit(0)

    ensure_directories()
    ensure_steamcmd()

    depot_names = load_depot_names()

    # assemble steamcmd script
    if arch == "win32":
        os_ = "\n@sSteamCmdForcePlatformType windows\n@sSteamCmdForcePlatformBitness 32"
    elif arch == "win64":
        os_ = "\n@sSteamCmdForcePlatformType windows\n@sSteamCmdForcePlatformBitness 64"
    elif arch == "linux32":
        os_ = "\n@sSteamCmdForcePlatformType linux\n@sSteamCmdForcePlatformBitness 32"
    elif arch == "linux64":
        os_ = "\n@sSteamCmdForcePlatformType linux\n@sSteamCmdForcePlatformBitness 64"
    elif arch == "macos":
        os_ = "\n@sSteamCmdForcePlatformType macos"

    script = f"""login {username}{f' {password}' if password else ''}
{os_}
app_update {appid} {(f'-beta {branch} ' if not branch_password else f'-beta {branch} -betapassword {branch_password} ') if branch != 'public' else ' '}validate
quit"""

    gamedata = get_steam_game_data(appid)

    if arch == "win32":
        safe_os = "Win32"
    elif arch == "win64":
        safe_os = "Win64"
    elif arch == "linux32":
        safe_os = "Linux32"
    elif arch == "linux64":
        safe_os = "Linux64"
    elif arch == "macos":
        safe_os = "Mac"

    if gamedata != None:
        # check if branch is valid
        appdata = gamedata["data"][appid]
        branches = appdata["depots"]["branches"]

        if branch not in branches:
            print(
                f"The branch '{branch}' does not exist for this app! Valid branches are: {', '.join(branches.keys())}"
            )
            exit(1)

        # get build id
        try:
            build_id = branches[branch]["buildid"]
        except Exception:
            build_id = "Unknown"

        # get app name
        appname = appdata["common"]["name"].replace(" ", "_")

        # get build time
        try:
            build_time = branches[branch]["timeupdated"]
            date_time = datetime.datetime.fromtimestamp(int(build_time))
            build_time = date_time.strftime("%B %d, %Y - %H:%M:%S UTC")
        except Exception:
            build_time = ""

        archive_path = Path(COMPLETED_DIR, f"{appname}.Build.{build_id}.{safe_os}.7z")
        if archive_path.exists():
            print("Archive for this build already exists, skipping!")
            exit(0)

        # write script
        with open(SCRIPT_PATH, "w") as f:
            f.write(script)

        print("Running script...")
        r = run_script()

        failed_sub = False
        rate_limited = False
        invalid_password = False
        steamguard_fail = False

        if STEAMCMD_CONTENT_LOG_FILE.exists():
            if "No subscription" in STEAMCMD_CONTENT_LOG_FILE.read_text():
                failed_sub = True

        if STEAMCMD_CONNECTION_LOG_FILE.exists():
            if "Rate Limit Exceeded" in STEAMCMD_CONNECTION_LOG_FILE.read_text():
                rate_limited = True

            if "Invalid Password" in STEAMCMD_CONNECTION_LOG_FILE.read_text():
                invalid_password = True

        if not STEAMCMD_SITELICENSE_FILE.exists() and not STEAMCMD_COMPAT_LOG_FILE.exists():
            steamguard_fail = True

        if r != 0 or rate_limited or failed_sub or steamguard_fail or invalid_password:
            if rate_limited and not failed_sub:
                print("Rate limited")
            else:
                if steamguard_fail or invalid_password:
                    print("Bad login")
                else:
                    print("Failure")

            cleanup()
            exit(1)

        temp_depotcache = Path(TEMP_DIR, "depotcache")
        temp_steamapps = Path(TEMP_DIR, "steamapps")
        temp_steamapps_workshop = Path(temp_steamapps, "workshop")
        temp_steamapps_downloading = Path(temp_steamapps, "downloading")
        temp_steamapps_temp = Path(temp_steamapps, "temp")

        shutil.move(DEPOTCACHE_DIR, temp_depotcache)
        shutil.move(STEAMAPPS_DIR, temp_steamapps)

        lib_folders_vdf = Path(TEMP_DIR, "steamapps", "libraryfolders.vdf")
        if lib_folders_vdf.exists():
            lib_folders_vdf.unlink()

        if temp_steamapps_downloading.exists():
            shutil.rmtree(temp_steamapps_downloading)

        if temp_steamapps_temp.exists():
            shutil.rmtree(temp_steamapps_temp)

        if temp_steamapps_workshop.exists():
            shutil.rmtree(temp_steamapps_workshop)

        acfs = list(temp_steamapps.glob("*.acf"))

        depot_manifest_list = []

        for acf in acfs:
            edit_vdf(acf, "LastOwner", "0")
            edit_vdf(acf, "LauncherPath", "0")

            d = vdf.loads(acf.read_text())
            installed_depots = d["AppState"]["InstalledDepots"]
            for id, depot in installed_depots.items():
                print(f"Depot {id}")
                manifest = depot["manifest"]
                try:
                    depot_name = depot_names.get("depots", id)
                    print(depot_name)
                except configparser.NoOptionError:
                    depot_name = "DepotName"

                depot_manifest_list.append(f"{id} - {depot_name} [Manifest {manifest}]")

        app_manifest = Path(temp_steamapps, f"appmanifest_{appid}.acf")

        game_name = read_vdf(app_manifest, "name").replace(" ", ".")
        game_name = sanitize_filename(game_name)

        build_number = read_vdf(app_manifest, "buildid")

        print("Compressing...")

        os.chdir(TEMP_DIR)

        try:
            subprocess.run(f'7z a -mx9 -sdel -pcs.rin.ru -v5g "{archive_path}" *', shell=True)
            print("Compresison complete")
        except Exception as e:
            print("Failed to compress!", e)
            cleanup(False, True)
            exit(1)

        os.chdir(ROOT_DIR)

        archive_02 = Path(str(archive_path.absolute()) + ".002")
        if not archive_02.exists():
            # only one archive
            archive_01 = Path(str(archive_path.absolute()) + ".001")
            if archive_01.exists():
                archive_01.rename(Path(archive_01.parent, archive_path.name))

        shutil.rmtree(TEMP_DIR)

        app_info_path = Path(COMPLETED_DIR, f"[CS.RIN.RU Info] {game_name}.Build.{build_number}.{safe_os}.{branch}.txt")
        with open(app_info_path, "w") as f:
            game_name = game_name.replace("_", " ").replace(".", " ")

            d = f"[url=][color=white][b]{game_name} [{safe_os}] [Branch: {branch}] (Clean Steam Files)[/b][/color][/url]\n"
            d += f"[size=85][color=white][b]Version:[/b] [i]{build_time} [Build {build_number}][/i][/color][/size]\n\n"
            d += f'[spoiler="[color=white]Depots & Manifests[/color]"][code=text]\n'

            for i, item in enumerate(depot_manifest_list):
                if i == len(depot_manifest_list) - 1:
                    d += item
                else:
                    d += item + "\n"

            d += f"[/code][/spoiler]\n"
            d += f"[color=white][b]Uploaded version:[/b] [i]{build_time} [Build {build_number}][/i][/color]"

            f.write(d)

        cleanup()

    else:
        print("Failed to get app data!")
        exit(1)

    print("All checks passed.")