XFE.SpaceEngineers.AgentBridge

"""Copy an AMS world, authorize that copy, and record original file hashes. No game required."""
import argparse
import datetime as dt
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import sys
import xml.etree.ElementTree as ET

TYPE = "{http://www.w3.org/2001/XMLSchema-instance}type"


def sha(path):
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def programs(path):
    result = []
    # Keep an individual grid intact, then free it; never deserialize game code.
    for _, node in ET.iterparse(path, events=("end",)):
        if node.get(TYPE) != "MyObjectBuilder_CubeGrid":
            continue
        for block in node.findall("./CubeBlocks/*"):
            if "ProgrammableBlock" not in block.get(TYPE, ""):
                continue
            code = block.findtext("Program") or block.findtext("ProgramData") or ""
            name = block.findtext("CustomName") or ""
            if "XFE AMS" not in code and "XFEAMS" not in name:
                continue
            result.append({"entityId": block.findtext("EntityId"), "name": name,
                           "gridId": node.findtext("EntityId"), "gridName": node.findtext("DisplayName"),
                           "sourceSha256": hashlib.sha256(code.encode("utf-8")).hexdigest(),
                           "sourceLength": len(code), "header": code.splitlines()[0] if code else "",
                           "customData": block.findtext("CustomData") or ""})
        node.clear()
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source", type=Path)
    parser.add_argument("--out", type=Path, default=Path("artifacts/world-preparation.json"))
    args = parser.parse_args()
    saves = Path(os.environ["APPDATA"]) / "SpaceEngineers/Saves"
    if args.source:
        choices = [args.source.resolve()]
    else:
        choices = sorted((p.parent for p in saves.glob("*/*/SANDBOX_0_0_0_.sbs")
                          if not p.parent.name.startswith("XFE Agent Debug")),
                         key=lambda p: (p / "Sandbox.sbc").stat().st_mtime, reverse=True)
    for source in choices:
        blocks = programs(source / "SANDBOX_0_0_0_.sbs")
        if blocks:
            break
    else:
        raise RuntimeError("No saved world containing AMS programmable blocks was found.")
    if args.out.resolve().is_relative_to(source.resolve()):
        raise ValueError("The report output must not be inside the original world directory.")
    timestamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
    target = source.parent / ("XFE Agent Debug AMS " + timestamp)
    if target.exists():
        raise FileExistsError(target)
    files = [p for p in source.rglob("*") if p.is_file() and
             not any(part.lower() == "backup" for part in p.relative_to(source).parts)]
    originals = {str(p.relative_to(source)): sha(p) for p in files}
    shutil.copytree(source, target, ignore=shutil.ignore_patterns("Backup", "backup"))
    # Only these explicit session settings change, and only in the newly created copy.
    for filename in ("Sandbox.sbc", "Sandbox_config.sbc"):
        path = target / filename
        if not path.exists():
            continue
        text = path.read_text(encoding="utf-8-sig")
        text = re.sub(r"<SessionName>.*?</SessionName>", "<SessionName>" + target.name + "</SessionName>", text, count=1, flags=re.S)
        text = re.sub(r"<OnlineMode>.*?</OnlineMode>", "<OnlineMode>OFFLINE</OnlineMode>", text, count=1)
        text = re.sub(r"<AutoSaveInMinutes>.*?</AutoSaveInMinutes>", "<AutoSaveInMinutes>0</AutoSaveInMinutes>", text, count=1)
        path.write_text(text, encoding="utf-8")
    if any(sha(source / name) != digest for name, digest in originals.items()):
        raise RuntimeError("Original world changed while copying; do not load this snapshot.")
    config_path = Path(os.environ["LOCALAPPDATA"]) / "XFE/SpaceEngineersAgent/config.json"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config = json.loads(config_path.read_text(encoding="utf-8-sig")) if config_path.exists() else {}
    allowed = config.setdefault("allowedWorldPaths", [])
    if str(target) not in allowed:
        allowed.append(str(target))
    temp = config_path.with_suffix(".tmp")
    temp.write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8")
    os.replace(temp, config_path)
    report = {"original": str(source), "debugWorld": str(target), "created": timestamp,
              "originalFiles": originals, "programmableBlocks": blocks,
              "changes": ["copy session name", "offline multiplayer", "disable automatic saving"]}
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({"original": str(source), "debugWorld": str(target), "programCount": len(blocks),
                      "report": str(args.out.resolve())}, ensure_ascii=False))


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")
    main()