109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""World-state JSON schema validator (player/NPC stats, inventory, etc.)."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Tuple
|
|
|
|
from jsonschema import ValidationError, validate
|
|
|
|
from app.logging_setup import get_logger
|
|
|
|
log = get_logger("state_validator")
|
|
|
|
|
|
def validate_state(state: Dict[str, Any], schema: Dict[str, Any]) -> Tuple[bool, List[str]]:
|
|
"""Validate state against world's JSON Schema. Returns (ok, errors)."""
|
|
if not schema:
|
|
return True, []
|
|
try:
|
|
validate(instance=state, schema=schema)
|
|
return True, []
|
|
except ValidationError as e:
|
|
return False, [f"{e.message} at path {list(e.absolute_path)}"]
|
|
except Exception as e:
|
|
return False, [f"schema_error: {e}"]
|
|
|
|
|
|
def apply_patch(state: Dict[str, Any], patch: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Apply a JSON-patch-like update to state.
|
|
|
|
Patch format:
|
|
{"set": {"path.to.field": value, ...},
|
|
"unset": ["path.to.field", ...],
|
|
"append": {"path.to.list": value, ...},
|
|
"increment": {"path.to.number": delta, ...}}
|
|
|
|
Paths use dot notation. Creates intermediate dicts as needed.
|
|
"""
|
|
if not patch:
|
|
return state
|
|
new_state = _deep_copy(state)
|
|
|
|
for op, items in patch.items():
|
|
if op == "set":
|
|
for path, value in items.items():
|
|
_set_path(new_state, path, value)
|
|
elif op == "unset":
|
|
for path in items:
|
|
_unset_path(new_state, path)
|
|
elif op == "append":
|
|
for path, value in items.items():
|
|
lst = _get_path(new_state, path) or []
|
|
if not isinstance(lst, list):
|
|
lst = []
|
|
lst.append(value)
|
|
_set_path(new_state, path, lst)
|
|
elif op == "increment":
|
|
for path, delta in items.items():
|
|
cur = _get_path(new_state, path) or 0
|
|
try:
|
|
cur = float(cur)
|
|
except (TypeError, ValueError):
|
|
cur = 0
|
|
_set_path(new_state, path, cur + delta)
|
|
elif op == "remove":
|
|
for path, value in items.items():
|
|
lst = _get_path(new_state, path) or []
|
|
if isinstance(lst, list):
|
|
lst = [x for x in lst if x != value]
|
|
_set_path(new_state, path, lst)
|
|
return new_state
|
|
|
|
|
|
def _deep_copy(obj: Any) -> Any:
|
|
if isinstance(obj, dict):
|
|
return {k: _deep_copy(v) for k, v in obj.items()}
|
|
if isinstance(obj, list):
|
|
return [_deep_copy(v) for v in obj]
|
|
return obj
|
|
|
|
|
|
def _get_path(obj: Any, path: str) -> Any:
|
|
cur = obj
|
|
for part in path.split("."):
|
|
if isinstance(cur, dict) and part in cur:
|
|
cur = cur[part]
|
|
else:
|
|
return None
|
|
return cur
|
|
|
|
|
|
def _set_path(obj: Dict[str, Any], path: str, value: Any) -> None:
|
|
cur = obj
|
|
parts = path.split(".")
|
|
for part in parts[:-1]:
|
|
if part not in cur or not isinstance(cur[part], dict):
|
|
cur[part] = {}
|
|
cur = cur[part]
|
|
cur[parts[-1]] = value
|
|
|
|
|
|
def _unset_path(obj: Dict[str, Any], path: str) -> None:
|
|
cur = obj
|
|
parts = path.split(".")
|
|
for part in parts[:-1]:
|
|
if not isinstance(cur, dict) or part not in cur:
|
|
return
|
|
cur = cur[part]
|
|
if isinstance(cur, dict):
|
|
cur.pop(parts[-1], None)
|