NextClient/installer/cli.py at main · CS-NextClient/NextClient · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import argparse
import configparser
import json
from pathlib import Path
import sys

from installer import import_game as core
from installer.localization import Languages, Message, Text, Translate


def Main(arguments=None):
    parser = argparse.ArgumentParser(description="Install, update, or remove a standalone NextClient copy.")
    parser.add_argument("action", choices=("discover", "inspect", "install", "update", "commit", "rollback", "recover", "uninstall"))
    parser.add_argument("--bundle", type=Path)
    parser.add_argument("--source", type=Path)
    parser.add_argument("--steam-root", action="append", type=Path)
    parser.add_argument("--destination", type=Path)
    parser.add_argument("--operation-id")
    parser.add_argument("--owner-pid", type=int)
    parser.add_argument("--defer-commit", action="store_true")
    parser.add_argument("--keep-installer", action="store_true")
    parser.add_argument("--removal-mode", default="preserve")
    parser.add_argument("--progress", action="store_true")
    parser.add_argument("--cancel-file", type=Path)
    parser.add_argument("--status-file", type=Path)
    parser.add_argument("--report", type=Path)
    parser.add_argument("--language", choices=Languages(), default="english")
    args = parser.parse_args(arguments)

    def OutputProgress(percent, message, detail):
        # Machine tokens keep pipe decoding and progress state independent of the UI language.
        phase = ";".join((message.key, *map(str, message.arguments)))
        progress_detail = ";".join((detail.key, *map(str, detail.arguments)))
        print(f"NEXTCLIENT_PROGRESS|{percent}|{phase}|{progress_detail}", flush=True)

    progress = core.Progress(OutputProgress if args.progress else None,
                             args.cancel_file.exists if args.cancel_file else None)
    try:
        if args.action == "discover":
            result = {"state": "passed", **core.Discover(args.steam_root)}
        elif args.action == "inspect":
            if args.bundle is None or args.source is None:
                raise ValueError(Message("ErrorInspectionArguments"))
            result = core.InspectNew(args.bundle, args.source)
        else:
            if args.destination is None:
                raise ValueError(Message("ErrorDestinationRequired"))
            progress.Phase(Message("ProgressChecking"), 0, 5)
            if args.action in ("install", "update"):
                if args.bundle is None:
                    raise ValueError(Message("ErrorBundleRequired"))
                if args.action == "install":
                    source = args.source
                    if source is None:
                        discovered = core.Discover(args.steam_root)
                        if not discovered["source"]:
                            raise ValueError(Message("ErrorSteamRequired"))
                        source = Path(discovered["source"])
                    result = core.InstallNew(args.bundle, source, args.destination, args.defer_commit, progress, args.owner_pid)
                else:
                    result = core.Update(args.bundle, args.destination, args.defer_commit, progress, args.owner_pid)
            elif args.action in ("commit", "rollback"):
                result = core.Complete(args.destination, args.operation_id, args.action == "commit")
            elif args.action == "recover":
                result = core.Recover(args.destination)
            else:
                result = core.Uninstall(args.destination, args.keep_installer, args.removal_mode, progress)
            progress.cancel_requested = None
            progress.Phase(Message("ProgressCompleted"), 100, 100)
        code = 0
    except core.InstallationCanceled as error:
        result = {"state": "canceled", "error": Translate(error, args.language)}
        code = 2
    except Exception as error:
        message = (Translate(error, args.language) if len(error.args) == 1 and isinstance(error.args[0], Message)
                   else Text(args.language, "ErrorUnexpected", error))
        result = {"state": "failed", "error": message}
        code = 1
    if args.report:
        core.WriteJson(args.report, result)
    if args.status_file:
        status = configparser.ConfigParser(interpolation=None)
        status["result"] = {key: str(result.get(key) or "").replace("\r", " ").replace("\n", " ")
                            for key in ("state", "source", "error", "operation_id", "required_bytes", "source_bytes", "package_bytes")}
        with args.status_file.open("w", encoding="utf-16") as stream:
            status.write(stream)
    print(json.dumps(result, ensure_ascii=True), flush=True)
    return code


if __name__ == "__main__":
    sys.exit(Main())