47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
class IntentRegistry:
|
|
def __init__(self, registry_path: Path) -> None:
|
|
self._registry_path = registry_path
|
|
self._factories: dict[tuple[str, str], Callable[..., Any]] = {}
|
|
|
|
def register(self, domain_id: str, process_id: str, factory: Callable[..., Any]) -> None:
|
|
self._factories[(domain_id, process_id)] = factory
|
|
|
|
def get_factory(self, domain_id: str, process_id: str) -> Callable[..., Any] | None:
|
|
return self._factories.get((domain_id, process_id))
|
|
|
|
def is_valid(self, domain_id: str, process_id: str) -> bool:
|
|
return self.get_factory(domain_id, process_id) is not None
|
|
|
|
def load_intents(self) -> list[dict[str, Any]]:
|
|
if not self._registry_path.is_file():
|
|
return []
|
|
with self._registry_path.open("r", encoding="utf-8") as fh:
|
|
payload = yaml.safe_load(fh) or {}
|
|
intents = payload.get("intents")
|
|
if not isinstance(intents, list):
|
|
return []
|
|
output: list[dict[str, Any]] = []
|
|
for item in intents:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
domain_id = item.get("domain_id")
|
|
process_id = item.get("process_id")
|
|
if not isinstance(domain_id, str) or not isinstance(process_id, str):
|
|
continue
|
|
output.append(
|
|
{
|
|
"domain_id": domain_id,
|
|
"process_id": process_id,
|
|
"description": str(item.get("description") or ""),
|
|
"priority": int(item.get("priority") or 0),
|
|
}
|
|
)
|
|
return output
|