DawnsorrowandCursor
0d6b0b2f80
Session-gated MinIO save sync with AppImage GUI, CLI edit/session flow, and Gitea release helper. Co-authored-by: Cursor <[email protected]>
273 lines
9.2 KiB
Python
273 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import tomllib
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from syncgames.errors import ConfigError
|
|
|
|
SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
s = name.lower().strip()
|
|
s = re.sub(r"[^a-z0-9]+", "-", s)
|
|
s = re.sub(r"-+", "-", s).strip("-")
|
|
if not s or not SLUG_RE.match(s):
|
|
raise ConfigError(f"Cannot slugify name into valid game-id: {name!r}")
|
|
return s
|
|
|
|
|
|
def expand_path(p: str) -> Path:
|
|
return Path(os.path.expanduser(os.path.expandvars(p))).resolve()
|
|
|
|
|
|
@dataclass
|
|
class AgentConfig:
|
|
device_id: str
|
|
endpoint_url: str
|
|
bucket: str = "syncgames"
|
|
region: str = "us-east-1"
|
|
path_style: bool = True
|
|
access_key: str = ""
|
|
secret_key: str = ""
|
|
store: str = "minio" # minio | filesystem
|
|
ssot_root: Path | None = None
|
|
lease_ttl_hours: int = 6
|
|
config_root: Path = field(default_factory=Path)
|
|
state_root: Path = field(default_factory=Path)
|
|
|
|
@property
|
|
def games_dir(self) -> Path:
|
|
return self.config_root / "games"
|
|
|
|
|
|
@dataclass
|
|
class GameConfig:
|
|
id: str
|
|
name: str
|
|
platform: str
|
|
paths: list[str]
|
|
versions_to_keep: int = 5
|
|
steam_appid: int | None = None
|
|
process_names: list[str] = field(default_factory=list)
|
|
|
|
def resolved_paths(self) -> list[Path]:
|
|
return [expand_path(p) for p in self.paths]
|
|
|
|
|
|
def default_config_root() -> Path:
|
|
env = os.environ.get("SYNCGAMES_CONFIG")
|
|
if env:
|
|
return Path(env).expanduser().resolve()
|
|
xdg = os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))
|
|
return Path(xdg) / "syncgames"
|
|
|
|
|
|
def default_state_root() -> Path:
|
|
env = os.environ.get("SYNCGAMES_STATE")
|
|
if env:
|
|
return Path(env).expanduser().resolve()
|
|
xdg = os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state"))
|
|
return Path(xdg) / "syncgames"
|
|
|
|
|
|
def repo_config_root() -> Path | None:
|
|
"""If running from the SyncGames checkout, prefer repo config/ when present."""
|
|
here = Path(__file__).resolve()
|
|
candidate = here.parents[2] / "config" # SyncGames/config
|
|
if (candidate / "agent.toml.example").exists() or (candidate / "agent.toml").exists():
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def load_agent_config(config_root: Path | None = None) -> AgentConfig:
|
|
root = config_root or default_config_root()
|
|
agent_path = root / "agent.toml"
|
|
# Bootstrap: allow repo example copy location
|
|
if not agent_path.exists():
|
|
repo = repo_config_root()
|
|
if repo and (repo / "agent.toml").exists():
|
|
root = repo
|
|
agent_path = root / "agent.toml"
|
|
elif repo and (repo / "agent.toml.example").exists():
|
|
raise ConfigError(
|
|
f"No agent.toml found. Copy {repo / 'agent.toml.example'} to "
|
|
f"{default_config_root() / 'agent.toml'} or {repo / 'agent.toml'}"
|
|
)
|
|
else:
|
|
raise ConfigError(f"Missing agent config: {agent_path}")
|
|
|
|
with agent_path.open("rb") as f:
|
|
data = tomllib.load(f)
|
|
|
|
store = data.get("store", "minio")
|
|
ssot = data.get("ssot_root")
|
|
return AgentConfig(
|
|
device_id=str(data["device_id"]),
|
|
endpoint_url=str(data.get("endpoint_url", "")),
|
|
bucket=str(data.get("bucket", "syncgames")),
|
|
region=str(data.get("region", "us-east-1")),
|
|
path_style=bool(data.get("path_style", True)),
|
|
access_key=str(data.get("access_key", "")),
|
|
secret_key=str(data.get("secret_key", "")),
|
|
store=store,
|
|
ssot_root=expand_path(ssot) if ssot else None,
|
|
lease_ttl_hours=int(data.get("lease_ttl_hours", 6)),
|
|
config_root=root,
|
|
state_root=default_state_root(),
|
|
)
|
|
|
|
|
|
def load_game(config: AgentConfig, game_id: str) -> GameConfig:
|
|
path = config.games_dir / f"{game_id}.toml"
|
|
if not path.exists():
|
|
raise ConfigError(f"Unknown game: {game_id} (expected {path})")
|
|
with path.open("rb") as f:
|
|
data = tomllib.load(f)
|
|
paths = data.get("paths") or []
|
|
if isinstance(paths, str):
|
|
paths = [paths]
|
|
procs = data.get("process_names") or data.get("process_name") or []
|
|
if isinstance(procs, str):
|
|
procs = [procs]
|
|
return GameConfig(
|
|
id=str(data.get("id", game_id)),
|
|
name=str(data.get("name", game_id)),
|
|
platform=str(data.get("platform", "other")),
|
|
paths=[str(p) for p in paths],
|
|
versions_to_keep=int(data.get("versions_to_keep", 5)),
|
|
steam_appid=int(data["steam_appid"]) if data.get("steam_appid") else None,
|
|
process_names=[str(p) for p in procs],
|
|
)
|
|
|
|
|
|
def list_games(config: AgentConfig) -> list[GameConfig]:
|
|
games_dir = config.games_dir
|
|
if not games_dir.exists():
|
|
return []
|
|
out: list[GameConfig] = []
|
|
for path in sorted(games_dir.glob("*.toml")):
|
|
if path.name.endswith(".removed.toml"):
|
|
continue
|
|
out.append(load_game(config, path.stem))
|
|
return out
|
|
|
|
|
|
def write_game_toml(config: AgentConfig, game: GameConfig) -> Path:
|
|
config.games_dir.mkdir(parents=True, exist_ok=True)
|
|
path = config.games_dir / f"{game.id}.toml"
|
|
lines = [
|
|
f'id = "{game.id}"',
|
|
f'name = "{game.name}"',
|
|
f'platform = "{game.platform}"',
|
|
f"versions_to_keep = {game.versions_to_keep}",
|
|
"paths = [",
|
|
]
|
|
for p in game.paths:
|
|
lines.append(f' "{p}",')
|
|
lines.append("]")
|
|
if game.steam_appid is not None:
|
|
lines.append(f"steam_appid = {game.steam_appid}")
|
|
if game.process_names:
|
|
lines.append("process_names = [")
|
|
for n in game.process_names:
|
|
lines.append(f' "{n}",')
|
|
lines.append("]")
|
|
lines.append("")
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _toml_str(value: str) -> str:
|
|
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
|
|
|
|
|
def save_agent_config(config: AgentConfig, *, path: Path | None = None) -> Path:
|
|
"""Write agent.toml (secrets included — file mode 0600)."""
|
|
root = config.config_root if config.config_root else default_config_root()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
out = path or (root / "agent.toml")
|
|
lines = [
|
|
f"device_id = {_toml_str(config.device_id)}",
|
|
f"store = {_toml_str(config.store)}",
|
|
f"endpoint_url = {_toml_str(config.endpoint_url)}",
|
|
f"bucket = {_toml_str(config.bucket)}",
|
|
f"region = {_toml_str(config.region)}",
|
|
f"path_style = {'true' if config.path_style else 'false'}",
|
|
f"access_key = {_toml_str(config.access_key)}",
|
|
f"secret_key = {_toml_str(config.secret_key)}",
|
|
f"lease_ttl_hours = {int(config.lease_ttl_hours)}",
|
|
]
|
|
if config.store == "filesystem" and config.ssot_root is not None:
|
|
lines.append(f"ssot_root = {_toml_str(str(config.ssot_root))}")
|
|
lines.append("")
|
|
out.write_text("\n".join(lines), encoding="utf-8")
|
|
try:
|
|
out.chmod(0o600)
|
|
except OSError:
|
|
pass
|
|
return out
|
|
|
|
|
|
def try_load_agent_config(config_root: Path | None = None) -> AgentConfig | None:
|
|
try:
|
|
return load_agent_config(config_root)
|
|
except ConfigError:
|
|
return None
|
|
|
|
|
|
def _bundled_root() -> Path | None:
|
|
"""PyInstaller _MEIPASS or package-adjacent bundled_* dirs."""
|
|
import sys
|
|
|
|
if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"):
|
|
return Path(sys._MEIPASS)
|
|
here = Path(__file__).resolve().parents[1]
|
|
if (here / "bundled_config").exists() or (here / "bundled_systemd").exists():
|
|
return here
|
|
return None
|
|
|
|
|
|
def seed_example_games(config: AgentConfig, source_games_dir: Path | None = None) -> int:
|
|
"""Copy missing example game tomls from the SyncGames checkout into config."""
|
|
if source_games_dir is None:
|
|
repo = repo_config_root()
|
|
if repo and (repo / "games").exists():
|
|
source_games_dir = repo / "games"
|
|
else:
|
|
bundled = _bundled_root()
|
|
if bundled and (bundled / "bundled_config" / "games").exists():
|
|
source_games_dir = bundled / "bundled_config" / "games"
|
|
if source_games_dir is None or not source_games_dir.exists():
|
|
return 0
|
|
config.games_dir.mkdir(parents=True, exist_ok=True)
|
|
copied = 0
|
|
for src in sorted(source_games_dir.glob("*.toml")):
|
|
dest = config.games_dir / src.name
|
|
if not dest.exists():
|
|
dest.write_text(src.read_text(encoding="utf-8"), encoding="utf-8")
|
|
copied += 1
|
|
return copied
|
|
|
|
|
|
def install_systemd_units(unit_source: Path | None = None) -> Path:
|
|
"""Install user systemd units; returns the unit directory."""
|
|
import shutil
|
|
|
|
xdg = os.environ.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))
|
|
unit_dir = Path(xdg) / "systemd" / "user"
|
|
unit_dir.mkdir(parents=True, exist_ok=True)
|
|
if unit_source is None:
|
|
candidate = Path(__file__).resolve().parents[2] / "systemd"
|
|
if not candidate.exists():
|
|
bundled = _bundled_root()
|
|
candidate = (bundled / "bundled_systemd") if bundled else candidate
|
|
unit_source = candidate
|
|
for name in ("syncgames-agent.service", "[email protected]"):
|
|
src = unit_source / name
|
|
if src.exists():
|
|
shutil.copy2(src, unit_dir / name)
|
|
return unit_dir |