"""CLI parsing + entry point dispatch. Subcommands: pull / push Sync. No-op if ``/.cloud-sync/sync.json`` is missing. All settings (url, instance_id) read from sync.json — flags are for runtime tweaks only. setup Interactive opt-in: prompt URL + token, mint instance_id, write sync.json + token. Player runs this once per instance. init Non-interactive setup (scripted / CI). disable Remove sync.json. Re-noops the instance. Cloud data untouched. Global Prism wiring (paste-once into Settings → Custom commands): PreLaunchCommand=python /opt/cloud-sync.pyz pull --pack-folder=$INST_MC_DIR PostExitCommand=python /opt/cloud-sync.pyz push --pack-folder=$INST_MC_DIR Instance opt-in is then per-instance via ``setup``; non-enabled instances launch normally without any sync work happening. """ from __future__ import annotations import argparse import sys from dataclasses import dataclass from pathlib import Path @dataclass(frozen=True) class Args: """Parsed CLI args. ``url`` comes from sync.json at runtime for pull/push; the CLI ``--url`` flag is only consulted by ``setup`` / ``init`` (recorded once into sync.json).""" pack_folder: Path token_file: Path restic_binary: Path | None # None → auto-discover allow_download: bool headless: bool instance_label: str | None # override; usually unset (env fallback) def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="instance-sync", description="Per-user Minecraft instance sync via the Timemachine Network.", ) p.add_argument("--version", action="version", version="instance-sync 0.1.0") sub = p.add_subparsers(dest="cmd", required=True) # --- pull / push: shared flags, no required --url --- for name in ("pull", "push"): sp = sub.add_parser(name, help=f"{name} instance state") _add_runtime_args(sp) # --- setup: interactive opt-in --- sp = sub.add_parser( "setup", help="enable sync on this instance (interactive: prompt URL + token)", ) _add_runtime_args(sp) sp.add_argument( "--url", default=None, help="Timemachine Network endpoint (skips the prompt if given)", ) # --- init: non-interactive setup --- sp = sub.add_parser( "init", help="non-interactive equivalent of setup (for scripted / CI use)", ) _add_runtime_args(sp) sp.add_argument( "--url", required=True, help="Timemachine Network endpoint (required)", ) sp.add_argument( "--token", default=None, help="discord_id:password token (read from stdin if omitted)", ) # --- disable: rename sync.json → sync.json.disabled --- sp = sub.add_parser( "disable", help="pause sync on this instance (renames sync.json -> sync.json.disabled). " "Re-enable with `enable`. Cloud data untouched.", ) sp.add_argument( "--pack-folder", default=".", type=Path, help="Minecraft instance directory (default: cwd)", ) # --- enable: rename sync.json.disabled → sync.json --- sp = sub.add_parser( "enable", help="resume sync on this instance (renames sync.json.disabled -> sync.json). " "Preserves the original instance_id + url so snapshots continue.", ) sp.add_argument( "--pack-folder", default=".", type=Path, help="Minecraft instance directory (default: cwd)", ) return p def _add_runtime_args(sp: argparse.ArgumentParser) -> None: sp.add_argument( "--pack-folder", default=".", type=Path, help="Minecraft instance directory (default: cwd)", ) sp.add_argument( "--token-file", default=None, type=Path, help="Token file path (default: /.cloud-sync/token)", ) sp.add_argument( "--restic-binary", default=None, type=Path, help="Path to a restic binary; overrides auto-discovery", ) sp.add_argument( "--no-download", action="store_true", help="Don't auto-fetch restic from upstream; fail if not found locally", ) sp.add_argument( "-g", "--no-gui", action="store_true", help="Headless mode (no Qt windows; status to stdout only)", ) sp.add_argument( "--instance-label", default=None, help="Override the instance display name. Default: $INST_NAME from " "Prism's env, then $INST_ID, then first 8 chars of instance_id.", ) def parse(argv: list[str]) -> tuple[str, argparse.Namespace]: ns = build_parser().parse_args(argv) return ns.cmd, ns def args_from(ns: argparse.Namespace) -> Args: """Convert argparse Namespace → Args dataclass. Centralizes path normalization so subcommands don't each repeat the recipe.""" pack = Path(ns.pack_folder).absolute().resolve() token = ( Path(ns.token_file).absolute() if getattr(ns, "token_file", None) is not None else pack / ".cloud-sync" / "token" ) return Args( pack_folder=pack, token_file=token, restic_binary=Path(ns.restic_binary).absolute() if getattr(ns, "restic_binary", None) else None, allow_download=not getattr(ns, "no_download", False), headless=getattr(ns, "no_gui", False), instance_label=getattr(ns, "instance_label", None), ) def main(argv: list[str] | None = None) -> int: """CLI entrypoint. Returns exit code (0=ok, 1=user cancel, 2=error).""" # Lazy imports keep tests of parse() isolated from UI deps. from . import sync, ui try: cmd, ns = parse(sys.argv[1:] if argv is None else argv) except SystemExit as e: return int(e.code) if isinstance(e.code, int) else 2 if cmd == "disable": return _run_disable(ns) if cmd == "enable": return _run_enable(ns) if cmd in ("setup", "init"): return _run_setup(ns, cmd) # pull / push — opt-in gated on sync.json. We re-check here BEFORE # spinning up the UI / printing any banner so non-sync-enabled # instances launch in true silence (clean Prism log). args = args_from(ns) from . import config as cfgmod if not cfgmod.exists(args.pack_folder): return 0 action = {"pull": sync.pull, "push": sync.push}[cmd] progress = ui.make_progress(headless=args.headless) title = "Instance sync — pulling" if cmd == "pull" else "Instance sync — pushing" try: return progress.run_with(lambda: action(args, progress), title) except KeyboardInterrupt: print("instance-sync: cancelled", file=sys.stderr) return 1 except Exception as e: # noqa: BLE001 print(f"instance-sync {cmd}: {e}", file=sys.stderr) return 2 def _run_disable(ns: argparse.Namespace) -> int: from . import config as cfgmod pack = Path(ns.pack_folder).absolute().resolve() if cfgmod.disable(pack): print( f"instance-sync: disabled — moved {cfgmod.config_path(pack).name} " f"to {cfgmod.disabled_path(pack).name} in {cfgmod.config_path(pack).parent}. " "Re-enable with `instance-sync enable`." ) return 0 if cfgmod.disabled_path(pack).exists(): print(f"instance-sync: already disabled ({cfgmod.disabled_path(pack)} present)") return 0 print(f"instance-sync: not enabled — no sync.json at {cfgmod.config_path(pack)}") return 0 def _run_enable(ns: argparse.Namespace) -> int: from . import config as cfgmod pack = Path(ns.pack_folder).absolute().resolve() try: renamed = cfgmod.enable(pack) except FileExistsError as e: print(f"instance-sync: {e}", file=sys.stderr) return 2 if renamed: print(f"instance-sync: enabled — restored {cfgmod.config_path(pack)}") return 0 if cfgmod.config_path(pack).exists(): print(f"instance-sync: already enabled ({cfgmod.config_path(pack)} present)") return 0 print( f"instance-sync: no sync.json.disabled at {cfgmod.disabled_path(pack)}. " "Run `instance-sync setup` to opt in this instance fresh.", file=sys.stderr, ) return 2 def _run_setup(ns: argparse.Namespace, cmd: str) -> int: """Wire `setup` (interactive) and `init` (flags-only) through the same flow module so both share the dialog/CLI logic.""" from . import setup_flow args = args_from(ns) url = getattr(ns, "url", None) token_override = getattr(ns, "token", None) return setup_flow.run( args=args, url=url, token=token_override, interactive=(cmd == "setup"), )