Spaces:
Running
Running
File size: 3,184 Bytes
534b431 dfedf76 534b431 dfedf76 534b431 dfedf76 534b431 | 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 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Load transcript reaction configurations from profile."""
from __future__ import annotations
import logging
import importlib
from typing import Any, Callable, Optional
from pathlib import Path
import yaml
from .base import TriggerConfig, ReactionConfig
from lyon_chatbox.config import config
logger = logging.getLogger(__name__)
PROFILES_DIRECTORY = Path(__file__).parent.parent.parent / "profiles"
def get_profile_reactions() -> list[ReactionConfig] | None:
"""Load reactions from current profile's reactions.yaml.
Returns:
List of ReactionConfig, or None if no profile or no reactions.yaml
"""
profile = config.LYON_CHATBOX_CUSTOM_PROFILE
if not profile:
logger.debug("No profile set, transcript reactions disabled")
return None
reactions_file = PROFILES_DIRECTORY / profile / "reactions.yaml"
if not reactions_file.exists():
logger.debug(f"No reactions.yaml in profile '{profile}'")
return None
try:
with open(reactions_file) as f:
yaml_config = yaml.safe_load(f)
if not yaml_config or not isinstance(yaml_config, list):
return None
reactions: list[ReactionConfig] = []
for entry in yaml_config:
name = entry.get("name")
callback_name = entry.get("callback")
if not name or not callback_name:
logger.warning(f"Skipping reaction entry missing name or callback: {entry}")
continue
callback = _import_callback(profile, callback_name)
if not callback:
continue
trigger_raw = entry.get("trigger", {})
trigger = _parse_trigger(trigger_raw)
params = entry.get("params", {})
reactions.append(ReactionConfig(
name=name,
callback=callback,
trigger=trigger,
params=params,
repeatable=entry.get("repeatable", False),
))
logger.info(f"Loaded {len(reactions)} reactions from profile '{profile}'")
return reactions if reactions else None
except Exception as e:
logger.warning(f"Failed to load reactions from profile '{profile}': {e}")
return None
def _parse_trigger(raw: dict[str, Any]) -> TriggerConfig:
"""Parse a trigger dict from YAML into a TriggerConfig."""
all_groups = [
TriggerConfig(words=group.get("words", []), entities=group.get("entities", []))
for group in raw.get("all", [])
]
return TriggerConfig(
words=raw.get("words", []),
entities=raw.get("entities", []),
all=all_groups,
)
def _import_callback(profile: str, callback_name: str) -> Optional[Callable[..., Any]]:
"""Import a callback function from profile module."""
try:
module_path = f"lyon_chatbox.profiles.{profile}.{callback_name}"
module = importlib.import_module(module_path)
return getattr(module, callback_name, None)
except (ImportError, AttributeError) as e:
logger.warning(f"Failed to import callback '{callback_name}' from profile '{profile}': {e}")
return None
|