File size: 1,766 Bytes
3d3d712 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
from __future__ import annotations
import dataclasses
import json
import os
import secrets
from datetime import datetime
from typing import Any, Dict
def create_id(length: int = 4) -> str:
date_str = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
ran_str = secrets.token_hex(length)
return f"{date_str}-{ran_str}"
def read_yaml(path: str) -> Dict[str, Any]:
import yaml
try:
with open(path, "r") as file:
return yaml.safe_load(file)
except Exception as e:
raise ValueError(f"Yaml loading failed due to: {e}")
def validate_yaml(content: Any, schema: str) -> bool:
import jsonschema
# plugin_dir = PLUGIN.BASE_PATH
# plugin_schema_path = os.path.join(plugin_dir, plugin_name + ".yaml")
# content = read_yaml(plugin_schema_path)
assert schema in ["example_schema", "plugin_schema"]
if schema == "example_schema":
schema_path = os.path.join(os.path.dirname(__file__), "../plugin/taskweaver.conversation-v1.schema.json")
else:
schema_path = os.path.join(os.path.dirname(__file__), "../plugin/taskweaver.plugin-v1.schema.json")
with open(schema_path) as file:
schema_object: Any = json.load(file)
try:
jsonschema.validate(content, schema=schema_object)
return True
except jsonschema.ValidationError as e:
raise ValueError(f"Yaml validation failed due to: {e}")
class EnhancedJSONEncoder(json.JSONEncoder):
def default(self, o: Any):
if dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
return super().default(o)
def json_dumps(obj: Any) -> str:
return json.dumps(obj, cls=EnhancedJSONEncoder)
def json_dump(obj: Any, fp: Any):
json.dump(obj, fp, cls=EnhancedJSONEncoder)
|