92 lines
2.2 KiB
Python
92 lines
2.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import io
|
||
|
|
import os
|
||
|
|
import shutil
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from ruamel.yaml import YAML
|
||
|
|
|
||
|
|
from .schema import CommandEntry, SchemaError, command_from_dict
|
||
|
|
|
||
|
|
|
||
|
|
DEFAULT_YAML_PATH = Path("C:/Jarvis/python/commands.yaml")
|
||
|
|
|
||
|
|
|
||
|
|
def _make_yaml() -> YAML:
|
||
|
|
y = YAML()
|
||
|
|
y.preserve_quotes = True
|
||
|
|
y.indent(mapping=2, sequence=4, offset=2)
|
||
|
|
y.width = 4096
|
||
|
|
return y
|
||
|
|
|
||
|
|
|
||
|
|
def load_yaml(path: Path = DEFAULT_YAML_PATH):
|
||
|
|
y = _make_yaml()
|
||
|
|
with open(path, "rt", encoding="utf-8") as f:
|
||
|
|
return y.load(f) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def list_command_ids(path: Path = DEFAULT_YAML_PATH) -> list:
|
||
|
|
try:
|
||
|
|
data = load_yaml(path)
|
||
|
|
except FileNotFoundError:
|
||
|
|
return []
|
||
|
|
return list(data.keys())
|
||
|
|
|
||
|
|
|
||
|
|
def _dump_to_string(data) -> str:
|
||
|
|
y = _make_yaml()
|
||
|
|
buf = io.StringIO()
|
||
|
|
y.dump(data, buf)
|
||
|
|
return buf.getvalue()
|
||
|
|
|
||
|
|
|
||
|
|
def _atomic_write(path: Path, text: str) -> None:
|
||
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
|
|
with open(tmp, "wt", encoding="utf-8", newline="\n") as f:
|
||
|
|
f.write(text)
|
||
|
|
os.replace(tmp, path)
|
||
|
|
|
||
|
|
|
||
|
|
def _backup(path: Path) -> Optional[Path]:
|
||
|
|
if not path.exists():
|
||
|
|
return None
|
||
|
|
bak = path.with_suffix(path.suffix + ".bak")
|
||
|
|
shutil.copy2(path, bak)
|
||
|
|
return bak
|
||
|
|
|
||
|
|
|
||
|
|
def append_command(entry: CommandEntry, path: Path = DEFAULT_YAML_PATH) -> None:
|
||
|
|
try:
|
||
|
|
data = load_yaml(path)
|
||
|
|
except FileNotFoundError:
|
||
|
|
data = {}
|
||
|
|
if entry.cmd_id in data:
|
||
|
|
raise SchemaError(f"command id already exists: {entry.cmd_id}")
|
||
|
|
data[entry.cmd_id] = entry.to_dict()
|
||
|
|
_backup(path)
|
||
|
|
text = _dump_to_string(data)
|
||
|
|
_atomic_write(path, text)
|
||
|
|
|
||
|
|
|
||
|
|
def upsert_command(entry: CommandEntry, path: Path = DEFAULT_YAML_PATH) -> None:
|
||
|
|
try:
|
||
|
|
data = load_yaml(path)
|
||
|
|
except FileNotFoundError:
|
||
|
|
data = {}
|
||
|
|
data[entry.cmd_id] = entry.to_dict()
|
||
|
|
_backup(path)
|
||
|
|
text = _dump_to_string(data)
|
||
|
|
_atomic_write(path, text)
|
||
|
|
|
||
|
|
|
||
|
|
def validate_file(path: Path = DEFAULT_YAML_PATH) -> list:
|
||
|
|
data = load_yaml(path)
|
||
|
|
out = []
|
||
|
|
for cmd_id, body in data.items():
|
||
|
|
plain_body = body
|
||
|
|
out.append(command_from_dict(cmd_id, dict(plain_body)))
|
||
|
|
return out
|