st2_opsdroid_matcher.py

from opsdroid.helper import add_skill_attributes
from opsdroid.matchers import *
from opsdroid.skill import Skill


class StackStormSkill(Skill):
    _action_alias_matchers: list
        
    def match_action_alias(self, func):
        def matcher(func):
            func = add_skill_attributes(func)
            self._action_alias_matchers = func.matchers
            return func

        return matcher

    # stackstorm-hubot like periodic refresh
    # @match_crontab('*/2 * * * *')
    # or webhook that StackStorm can use to inform the bot whenever the aliases need to be refreshed
    @match_webhook("refresh_action_aliases")
    # could also have a cron that does a full refresh every hour, and then st2 only sends changed aliases.
    def refresh_action_alias_matchers(self, func):
        current_matchers = self._action_alias_matchers[:]
        new_matchers = self.get_matchers_from_st2()
        if current_matchers != new_matchers:
            self._action_alias_matchers.clear()
            self._action_alias_matchers.extend(new_matchers)
    
    @match_action_alias
    async def action_alias(self, message):
        pass
    
    nlp_intent_types = {
        "dialogflow": match_dialogflow_intent,
        "luisai": match_luisai_intent,
        "rasanlu": match_rasanlu,
        # other nlp matchers
        "watson": match_watson,
    }
    
    def get_matchers_from_st2(self):
        alias_formats = get_formats_from_st2()
        regex_matchers = [match_regex(fmt) for fmt in alias_formats]
        nlp_intents = get_intents_from_st2()
        nlp_matchers = [
            self.nlp_intents[intent_type](intent)
            for itent_type, intent in nlp_intents.items()
        ]
        return regex_matchers + nlp_matchers