Spaces:
Running
Running
| """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 | |