dzgui/dzgui/api/steam.py

import logging
import os
import psutil
import re
import requests
import subprocess
import vdf  # type: ignore

from pathlib import Path
from requests.exceptions import HTTPError
from typing import Any, Generator, Union
from warnings import deprecated

from dzgui.api.acf import ACF
from dzgui.const.endpoints import VALIDATE
from dzgui.init.prereqs import has_any_client
from dzgui.init.prereqs import has_flatpak_client, has_steam_client
from dzgui.const.constants import (
    APPID_DAYZ,
    APPID_DAYZ_EXP,
    APP_NAME,
    DAYZ_BINARY,
    DAYZ_LAUNCHER,
    DEBIAN_STEAM_PATH,
    DEFAULT_STEAM_PATH,
    DOWNLOADS_PATH,
    FLATPAK_STEAM_PATH,
    LIBRARYFOLDERS_PATH,
    REQUEST_TIMEOUT,
    UBUNTU_STEAM_PATH,
    VDF_PATH,
)
from dzgui.const.endpoints import (
    APP_DETAILS,
    SUB_ENDPOINT,
    STEAM_PUBLISHED_FILES,
    UNSUB_ENDPOINT,
)
from dzgui.strings import wizard
from dzgui.strings.errors import api_validation_error, workshop_endpoint_error
from dzgui.util.bash import concat_bash_args

logger = logging.getLogger(APP_NAME)


class AppNotInstalledError(Exception):
    """App not present in user's libraryfolders"""

    pass


class AppMovedError(Exception):
    """VDF points to a nonexistent location on disk"""

    pass


class VDFLoadError(Exception):
    """Malformed VDF or JSON conversion"""

    pass


def get_steam_paths() -> list[tuple[Path, str]]:
    paths = []
    if has_any_client():
        HOME = Path.home()
        env = os.environ.get("XDG_DATA_HOME")
        if env is not None:
            XDG_DATA_HOME = env
        else:
            XDG_DATA_HOME = DEFAULT_STEAM_PATH

        DEFAULT_PATH = HOME.joinpath(XDG_DATA_HOME)
        FLATPAK_PATH = HOME.joinpath(FLATPAK_STEAM_PATH)
        UBUNTU_PATH = HOME.joinpath(UBUNTU_STEAM_PATH)
        DEBIAN_PATH = HOME.joinpath(DEBIAN_STEAM_PATH)

        human = {
            DEFAULT_PATH: wizard.desc_default_path,
            FLATPAK_PATH: wizard.desc_flatpak_path,
            UBUNTU_PATH: wizard.desc_ubuntu_path,
            DEBIAN_PATH: wizard.desc_debian_path,
        }

        for path in DEFAULT_PATH, FLATPAK_PATH, UBUNTU_PATH, DEBIAN_PATH:
            if path is FLATPAK_PATH and has_flatpak_client() is False:
                continue
            if path is not FLATPAK_PATH and has_steam_client() is False:
                continue
            if path.joinpath(VDF_PATH).is_file():
                paths.append((path, human[path]))
    return paths


def concat_mods(mods: list[str]) -> str:
    # TODO: circular import workaround
    from dzgui.api.mods import _hash

    hashes = []
    for mod in mods:
        md5sum = _hash(mod)
        hashes.append(md5sum)
    return ";".join(hashes)


def get_local_signatures(version_file: Path) -> dict[str, int]:
    hashes: dict[str, int] = {}
    lines = version_file.read_text().splitlines()
    for line in lines:
        data = line.split(",")
        _id = data[0]
        mod_hash = int(data[1])
        hashes[_id] = mod_hash
    return hashes


def get_needs_update(
    version_file: Path, remote_hashes: list[tuple[str, str, int, int]]
) -> list[tuple[str, str, int, int]]:
    local_stamps = get_local_signatures(version_file)
    needs_update: list[tuple[str, str, int, int]] = []
    for title, _id, stamp, size in remote_hashes:
        if _id not in local_stamps:
            needs_update.append((title, _id, stamp, size))
        elif stamp != local_stamps[_id]:
            needs_update.append((title, _id, stamp, size))
        else:
            continue
    return needs_update


def get_remote_signatures(mods: list[str]) -> list[tuple[str, str, int, int]]:
    """
    Attempts to continue connecting even if signatures are empty
    """
    payload: dict[str, str | int] = {}
    payload["itemcount"] = len(mods)
    for i, mod in enumerate(mods):
        payload[f"publishedfileids[{i}]"] = mod
    try:
        r = requests.post(STEAM_PUBLISHED_FILES, payload)
    except Exception as e:
        logger.critical(e)
        return []
    if r.status_code != 200:
        return []

    hashes: list[tuple[str, str, int, int]] = []
    j = r.json()
    rows = j["response"]["publishedfiledetails"]
    for row in rows:
        # NOTE: not a valid published file
        if row["result"] == 9:
            continue
        title = str(row["title"])
        _id = str(row["publishedfileid"])
        time = int(row["time_updated"])
        size = int(row["file_size"])
        hashes.append((title, _id, time, size))
    return hashes


def connect_debug(
    client: str, addr: str, appid: int, name: str, mods: list[str]
) -> str:
    concat = concat_mods(mods)
    client_args = concat_bash_args(client)
    params = [
        "-applaunch",
        str(appid),
        f"-connect={addr}",
        "-nolauncher",
        "-nosplash",
        "-skipintro",
        f"-name={name}",
        f"-mod={concat}",
    ]
    client_args.extend(params)
    return " ".join(client_args)


# TODO: set config to name=user, use official server and no mods,
# ensure that formatted string is identical to fixture with same hash
def connect(client: str, addr: str, appid: int, name: str, mods: list[str]) -> int:
    concat = concat_mods(mods)
    client_args = concat_bash_args(client)
    # TODO: test flatpak arg expansion
    params = [
        "-applaunch",
        str(appid),
        f"-connect={addr}",
        "-nolauncher",
        "-nosplash",
        "-skipintro",
        f"-name={name}",
        f"-mod={concat}",
    ]
    proc = subprocess.run([*client_args, *params])
    return proc.returncode


def load_to_menu(client: str, appid: int, name: str, mods: list[str]) -> int:
    """Loads to the menu screen with the selected mods; used for AFK/pre-joining"""
    concat = concat_mods(mods)
    client_args = concat_bash_args(client)
    logger.debug(f"Setting client arguments to: {client_args}")
    params = [
        "-applaunch",
        str(appid),
        "-nolauncher",
        "-nosplash",
        "-skipintro",
        f"-name={name}",
        f"-mod={concat}",
    ]
    proc = subprocess.run([*client_args, *params])
    return proc.returncode


def launch_offline(
    client: str, appid: int, name: str, mods: list[str], mission: str
) -> int:
    """Launch offline with specific mods/missions"""
    symlinks = ";".join(mods)
    client_args = concat_bash_args(client)
    params = [
        "-applaunch",
        str(appid),
        "-nolauncher",
        "-nosplash",
        "-skipintro",
        f"-name={name}",
        f"-mod={symlinks}",
    ]
    if len(mission) > 0:
        arg = f"-mission={mission}"
        params.append(arg)
    proc = subprocess.run([*client_args, *params])
    return proc.returncode


def find_loginusers(path: Path) -> Path:
    return path / "config" / "loginusers.vdf"


def find_user_id(path: Path) -> str | None:
    resolved_path = find_loginusers(path)
    try:
        with open(resolved_path, "r") as f:
            v = vdf.load(f)
            users = v["users"]
            # NOTE: beta client: /package/beta
            last_user = str(next(iter(users)))
            if len(users) == 1:
                return last_user
            last_stamp = int(users[last_user]["timestamp"])
            for user in users:
                stamp = int(users[user]["timestamp"])
                if stamp > last_stamp:
                    last_stamp = stamp
                    last_user = user
            return last_user
    except Exception as e:
        logger.debug(e)
        return None


def find_user_id_32(path: Path) -> int:
    uid = find_user_id(path)
    if uid is None:
        raise ValueError("Failed to parse a valid Steam32 ID")
    return int(uid) & 0xFFFFFFFF


class SteamAPIKeyError(Exception):
    def __init__(self, msg: str | None = None) -> None:
        super().__init__(msg or api_validation_error)


class SteamWorkshopEndpointError(Exception):
    def __init__(self, msg: str | None = None) -> None:
        super().__init__(msg or workshop_endpoint_error)


def update_workshop(key: str, mod: int, endpoint: str) -> None:
    payload: dict[str, Union[int, str]] = {
        "publishedfileid": mod,
        "appid": APPID_DAYZ,
        "key": key,
        "list_type": 1,
        "notify_client": 1,
    }
    try:
        res = requests.post(endpoint, params=payload, timeout=REQUEST_TIMEOUT)
        res.raise_for_status()
    except HTTPError as e:
        logger.critical(e)
        match res.status_code:
            case 403:
                raise SteamAPIKeyError()
            case 404:
                raise SteamWorkshopEndpointError()
            case _:
                # TODO: apply redaction filter
                raise e
    except Exception as e:
        logger.critical(e)
        raise e


def subscribe(key: str, mod: int) -> None:
    update_workshop(key, mod, SUB_ENDPOINT)


def unsubscribe(key: str, mod: int) -> None:
    update_workshop(key, mod, UNSUB_ENDPOINT)


def validate_mod(client: str, mod: str) -> None:
    client_args = concat_bash_args(client)
    subprocess.Popen([*client_args, "+workshop_download_item", str(APPID_DAYZ), mod])


@deprecated("Cf. https://github.com/ValveSoftware/steam-for-linux/issues/9672")
def get_registry() -> Any | None:
    home = os.getenv("HOME")
    try:
        with open(f"{home}/.steam/registry.vdf") as f:
            registry = vdf.load(f)
        return registry
    except Exception as e:
        logger.critical(e)
        return None


def _is_dayz_running() -> bool:
    registry = get_registry()
    if registry is None:
        return False
    apps = registry["Registry"]["HKCU"]["Software"]["Valve"]["Steam"]["apps"].items()
    for app in apps:
        k, v = app
        if k in (APPID_DAYZ, APPID_DAYZ_EXP):
            try:
                state = v["Running"]
                # NOTE: 0 denotes False
                return bool(int(state))
            except Exception as e:
                logger.critical(e)
                return False
    return False


def _get_running_app() -> int | None:
    registry = get_registry()
    if registry is None:
        return None
    try:
        return int(
            registry["Registry"]["HKCU"]["Software"]["Valve"]["Steam"]["RunningAppID"]
        )
    except Exception:
        return None


def get_running_app() -> int | None:
    PROC_NAME = "steam"
    SUBPROC_NAME = "reaper"
    FLAG = "AppId"
    # NOTE: may cause conflicts if multiple apps are running
    try:
        for proc in psutil.process_iter():
            if proc.name() == PROC_NAME:
                subprocs = proc.children()
                filtered = (proc for proc in subprocs if proc.name() == SUBPROC_NAME)
                try:
                    proc = next(filtered)
                except StopIteration:
                    return None
                args = proc.cmdline()
                appid = (row for row in args if FLAG in row)
                try:
                    return int(next(appid).split("=")[1])
                except StopIteration:
                    return None
    except Exception as e:
        logger.debug(e)
    return None


def get_last_updated(path: Path, appid: int) -> str | None:
    root_path = get_app_path(path, appid)
    acf = root_path.joinpath(f"steamapps/appmanifest_{appid}.acf")
    try:
        return ACF(acf).get_last_updated()
    except Exception as e:
        logger.critical(e)
        return None


def get_last_played(path: Path, appid: int) -> str | None:
    root_path = get_app_path(path, appid)
    acf = root_path.joinpath(f"steamapps/appmanifest_{appid}.acf")
    try:
        return ACF(acf).get_last_played()
    except Exception as e:
        logger.critical(e)
        return None


def get_app_allows_downloads(path: Path, appid: int) -> bool:
    try:
        root_path = get_app_path(path, appid)
        acf = root_path.joinpath(f"steamapps/appmanifest_{appid}.acf")
        flag = ACF(acf).get_allows_downloads()
    except AppNotInstalledError:
        flag = 0
    match flag:
        # NOTE: adheres to global client setting
        case 0:
            return get_client_allows_downloads(path)
        # NOTE: always allow
        case 1:
            return True
        # NOTE: never allow
        case 2:
            return False
        # NOTE: no other known values at this time, fallback (assume permits)
        case _:
            return True


def get_config(path: Path) -> Path:
    return path.joinpath("config/config.vdf")


def get_client_allows_downloads(path: Path) -> bool:
    config = get_config(path)
    try:
        with open(config) as f:
            settings = vdf.load(f)
            # NOTE: "1" denotes "allow"
            allow = bool(
                int(
                    settings["InstallConfigStore"]["Software"]["Valve"]["Steam"][
                        "AllowDownloadsDuringGameplay"
                    ]
                )
            )
            return allow
    except Exception as e:
        logger.critical(e)
        return True


def is_dayz_running() -> bool:
    """Subprocesses spawned from Steam will not show up in regular process tree"""
    procs = []
    substring = [DAYZ_BINARY, DAYZ_LAUNCHER]
    for proc in psutil.process_iter():
        try:
            procs.append(proc.cmdline())
        except Exception as e:
            logger.warning(e)
            continue
    return any(s in item for sublist in procs for item in sublist for s in substring)


def get_pending_downloads(steam_path: Path, appid: int) -> Generator[Path, None, None]:
    root_path = get_app_path(steam_path, appid)
    downloads_dir = root_path / DOWNLOADS_PATH
    files = downloads_dir.iterdir()
    return files


def is_dayz_downloading(steam_path: Path, appid: int) -> bool:
    files = get_pending_downloads(steam_path, appid)
    reg = re.compile(rf"^state_(?:{APPID_DAYZ_EXP}|{APPID_DAYZ})_.*\.patch$")
    pending = [f for f in files if reg.match(f.name)]
    if len(pending) > 0:
        return True
    return False


def get_app_path(folders_path: Path, appid: int) -> Path:
    app_path = None

    try:
        with open(folders_path / LIBRARYFOLDERS_PATH) as f:
            folders = vdf.load(f)
    except Exception:
        raise VDFLoadError("Failed to parse libraryfolders")

    for obj in folders["libraryfolders"]:
        if str(appid) in folders["libraryfolders"][obj]["apps"]:
            app_path = folders["libraryfolders"][obj]["path"]
            if Path(app_path).exists():
                break

    if app_path is None:
        raise AppNotInstalledError(
            f"Failed to find a libraryfolder for the appid {appid}"
        )
    if Path(app_path).exists() is False:
        raise AppMovedError(
            f"The location '{app_path}' pointed to by '{appid}' no longer exists and may have been changed on the disk."
        )

    return Path(app_path)


def get_app_name(appid: int) -> str | None:
    payload = {"appids": [appid]}
    res = requests.get(APP_DETAILS, params=payload)
    if res.status_code != 200:
        return None
    try:
        return str(res.json()[str(appid)]["data"]["name"])
    except Exception as e:
        logger.debug(e)
        return None


@deprecated("Currently unused")
def validate_dayz(client: str) -> None:
    client_args = concat_bash_args(client)
    uri = VALIDATE + str(APPID_DAYZ)
    subprocess.Popen([*client_args, uri])