def process()

in processors/slack.py [0:0]

• 109 lines of code
• 36 McCabe index (conditional complexity)

    def process(self, output_var='slack'):
        if 'api' not in self.config and 'mode' not in self.config:
            raise NotConfiguredException('No Slack API call specified.')
        if 'token' not in self.config:
            raise NotConfiguredException('No Slack token specified.')
        if 'request' not in self.config:
            self.config['request'] = {}

        mode = self._jinja_expand_string(
            self.config['mode'], 'mode') if 'mode' in self.config else 'api'

        token = self._jinja_expand_string(self.config['token'], 'token')
        slack_api = None
        if mode == 'api':
            slack_api = self._jinja_expand_string(self.config['api'], 'api')
        request_params = self._jinja_expand_dict_all(self.config['request'],
                                                     'request')

        mode = self._jinja_expand_string(
            self.config['mode'], 'mode') if 'mode' in self.config else 'api'

        if mode == 'processMessages' or mode == 'lastImage':
            if 'messages' not in self.config:
                raise NotConfiguredException('No Slack messages specified.')
            if 'appId' not in self.config:
                raise NotConfiguredException('No Slack app ID specified.')
            app_id = self._jinja_expand_string(self.config['appId'], 'app_id')
            multi_modal = self._jinja_expand_bool(
                self.config['multimodal'],
                'multimodal') if 'multimodal' in self.config else False
            messages = self._jinja_expand_expr(self.config['messages'],
                                               'messages')

            no_question = self._jinja_expand_string(
                self.config['noQuestionPrompt'], 'no_question_prompt'
            ) if 'noQuestionPrompt' in self.config else "Answer the question in this audio clip or image."

            processed = []
            if 'messages' in messages:
                messages = messages['messages']

            for message in messages:
                new_message = None
                if 'app_id' in message:
                    if message['app_id'] == app_id:
                        parts = self._slack_message_to_parts(
                            message, token, multi_modal, no_question)
                        if len(parts) > 0:
                            new_message = {'role': 'MODEL', 'parts': parts}
                else:
                    parts = self._slack_message_to_parts(
                        message, token, multi_modal, no_question)
                    if len(parts) > 0:
                        new_message = {'role': 'USER', 'parts': parts}
                if new_message:
                    processed.append(new_message)

            # Prepend an initial prompt that can be instructions or such
            if 'prompt' in self.config:
                prompt_added = False
                for msg_idx, msg in enumerate(processed):
                    if msg['role'] == 'USER':
                        for part_idx, part in enumerate(msg['parts']):
                            if 'text' in part:
                                processed[msg_idx]['parts'][part_idx][
                                    'text'] = self._jinja_expand_string(
                                        self.config['prompt'],
                                        'prompt') + " " + processed[msg_idx][
                                            'parts'][part_idx]['text']
                                prompt_added = True
                                break
                    if prompt_added:
                        break

            if mode == 'lastImage':
                if len(processed) > 0:
                    last_message = processed[len(processed) - 1]
                    for part in last_message['parts']:
                        if 'inlineData' in part:
                            if part['inlineData']['mimeType'][0:6] == 'image/':
                                return {output_var: part['inlineData']['data']}
                return {output_var: None}
            return {output_var: processed}

        if slack_api == 'conversations.list':
            slack_response = self.call_slack(self.config['api'], token,
                                             request_params, True)
            self.logger.debug('Slack API %s responded.' % (slack_api),
                              extra={
                                  'slack_request': request_params,
                                  'slack_response': slack_response
                              })
            return {
                output_var: slack_response,
            }

        if slack_api == 'conversations.history':
            slack_response = self.call_slack(self.config['api'], token,
                                             request_params)
            self.logger.debug('Slack API %s responded.' % (slack_api),
                              extra={
                                  'slack_request': request_params,
                                  'slack_response': slack_response
                              })
            return {
                output_var: slack_response,
            }

        if slack_api == 'conversations.replies':
            slack_response = self.call_slack(self.config['api'], token,
                                             request_params, True)
            self.logger.debug('Slack API %s responded.' % (slack_api),
                              extra={
                                  'slack_request': request_params,
                                  'slack_response': slack_response
                              })
            return {
                output_var: slack_response,
            }

        return {
            output_var: None,
        }