mirror of
https://github.com/hansjone/oclaw.git
synced 2026-10-09 07:20:44 +08:00
oliver
4a23b715a2
本次迁移将网关/通道/工具/技能/脚本与协议资源集中到新结构，统一路径常量与脚本转发机制，减少顶层噪音并保证运行与测试行为一致。 Made-with: Cursor
78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from oclaw.platform.config.runtime_paths import ws_protocol_schemas_root
|
|
|
|
_SCHEMA_DIR = ws_protocol_schemas_root()
|
|
|
|
|
|
def _load_schema(name: str) -> dict[str, Any]:
|
|
p = _SCHEMA_DIR / name
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WsSchemas:
|
|
frame: Draft202012Validator
|
|
connect_params: Draft202012Validator
|
|
hello_ok: Draft202012Validator
|
|
agent_params: Draft202012Validator
|
|
agent_wait_params: Draft202012Validator
|
|
sessions_schema: dict[str, Any]
|
|
chat_schema: dict[str, Any]
|
|
|
|
|
|
_CACHED: WsSchemas | None = None
|
|
|
|
|
|
def get_ws_schemas() -> WsSchemas:
|
|
global _CACHED
|
|
if _CACHED is not None:
|
|
return _CACHED
|
|
|
|
frame_schema = _load_schema("frames.json")
|
|
connect_schema = _load_schema("connect.json")
|
|
hello_ok_schema = _load_schema("hello_ok.json")
|
|
agent_schema = _load_schema("agent.json")
|
|
agent_wait_schema = _load_schema("agent_wait.json")
|
|
sessions_schema = _load_schema("sessions.json")
|
|
chat_schema = _load_schema("chat.json")
|
|
|
|
_CACHED = WsSchemas(
|
|
frame=Draft202012Validator(frame_schema),
|
|
connect_params=Draft202012Validator(connect_schema),
|
|
hello_ok=Draft202012Validator(hello_ok_schema),
|
|
agent_params=Draft202012Validator(agent_schema),
|
|
agent_wait_params=Draft202012Validator(agent_wait_schema),
|
|
sessions_schema=sessions_schema,
|
|
chat_schema=chat_schema,
|
|
)
|
|
return _CACHED
|
|
|
|
|
|
def format_validation_errors(errors: list[Any]) -> str:
|
|
out: list[str] = []
|
|
for e in errors[:6]:
|
|
try:
|
|
path = "/".join(str(x) for x in list(e.path))
|
|
except Exception:
|
|
path = ""
|
|
msg = str(getattr(e, "message", "") or "")
|
|
if path:
|
|
out.append(f"{path}: {msg}")
|
|
else:
|
|
out.append(msg)
|
|
return "; ".join([x for x in out if x]) or "invalid_payload"
|
|
|
|
|
|
def validate_or_errors(validator: Draft202012Validator, obj: Any) -> list[Any]:
|
|
return list(validator.iter_errors(obj))
|
|
|
|
|
|
__all__ = ["WsSchemas", "get_ws_schemas", "format_validation_errors", "validate_or_errors"]
|
|
|