Maas-Agent/apps/trigger/handler/impl/task/application_task.py

# coding=utf-8
"""
    @project: MaxKB
    @Author：虎虎
    @file： application_task.py
    @date：2026/1/14 19:14
    @desc:
"""
import json
import time
import traceback

import uuid_utils.compat as uuid
from django.db.models import QuerySet

from application.models import ChatUserType, Chat, ChatRecord, ChatSourceChoices, Application
from chat.serializers.chat import ChatSerializers
from common.utils.logger import maxkb_logger
from knowledge.models.knowledge_action import State
from trigger.handler.base_task import BaseTriggerTask
from trigger.models import TaskRecord, TriggerTask


def get_reference(fields, obj):
    for field in fields:
        value = obj.get(field)
        if value is None:
            return None
        else:
            obj = value
    return obj


def conversion_custom_value(value, _type):
    if ['array', 'dict', 'float', 'int', 'boolean', 'any'].__contains__(_type):
        try:
            return json.loads(value)
        except Exception as e:
            pass
    return value


def valid_value_type(value, _type):
    if _type == 'array':
        return isinstance(value, list)
    if _type == 'dict':
        return isinstance(value, dict)
    if _type == 'float':
        return isinstance(value, float)
    if _type == 'int':
        return isinstance(value, int)
    if _type == 'boolean':
        return isinstance(value, bool)
    return isinstance(value, str)


def get_field_value(value, kwargs, _type, required, default_value, field):
    source = value.get('source')
    if source == 'custom':
        _value = value.get('value')
        if _value:
            _value = conversion_custom_value(_value, _type)
        else:
            if default_value:
                return default_value
            if required:
                raise Exception(f'{field} is required')
            else:
                return None
    else:
        _value = get_reference(value.get('value'), kwargs)
    valid = valid_value_type(_value, _type)
    if not valid:
        raise Exception(f'{field} type error')
    return _value


def get_application_execute_parameters(parameter_setting, application_parameters_setting, kwargs):
    many_field = ['api_input_field_list', 'user_input_field_list']
    parameters = {'form_data': {}}
    for key, value in application_parameters_setting.items():
        setting = parameter_setting.get(key)
        if setting:
            if many_field.__contains__(key):
                for ck, cv in value.items():
                    _setting = setting.get(ck)
                    if _setting:
                        _value = get_field_value(_setting, kwargs, cv.get('type'), cv.get('required'),
                                                 cv.get('default_value'), ck)
                        parameters['form_data'][ck] = _value
                    else:
                        if cv.get('default_value'):
                            parameters['form_data'][ck] = cv.get('default_value')
                        else:
                            if cv.get('required'):
                                raise Exception(f'{ck} is required')
            else:
                value = get_field_value(setting, kwargs, value.get('type'), value.get('required'),
                                        value.get('default_value'), key)
                parameters['message' if key == 'question' else key] = value
        else:
            if value.get('default_value'):
                parameters['message' if key == 'question' else key] = value.get('default_value')
            else:
                if value.get('required'):
                    raise Exception(f'{"message" if key == "question" else key} is required')

    return parameters


def get_loop_workflow_node(node_list):
    result = []
    for item in node_list:
        if item.get('type') == 'loop-node':
            for loop_item in item.get('loop_node_data') or []:
                for inner_item in loop_item.values():
                    result.append(inner_item)
    return result


def get_workflow_state(details):
    node_list = details.values()
    all_node = [*node_list, *get_loop_workflow_node(node_list)]
    err = any([True for value in all_node if value.get('status') == 500 and not value.get('enableException')])
    if err:
        return State.FAILURE
    return State.SUCCESS


def get_user_field_component_input_type(input_type):
    if input_type == "MultiRow":
        return 'array'
    if input_type == "SwitchInput":
        return 'boolean'
    return 'string'


def get_application_parameters_setting(application):
    application_parameter_setting = {'question': {
        'required': True,
        'type': 'string'
    }}
    if application.type == 'SIMPLE':
        return application_parameter_setting
    else:
        base_node_list = [n for n in application.work_flow.get('nodes') if n.get('type') == "base-node"]
        if len(base_node_list) == 0:
            raise Exception('Incorrect application workflow information')
        base_node = base_node_list[0]
        api_input_field_list = base_node.get('properties').get('api_input_field_list') or []
        api_input_field_list = {user_field.get('variable'): {
            'required': user_field.get('is_required'),
            'default_value': user_field.get('default_value'),
            'type': 'string'
        } for user_field in api_input_field_list}
        user_input_field_list = base_node.get('properties').get('user_input_field_list') or []
        user_input_field_list = {user_field.get('field'): {
            'required': user_field.get('required'),
            'default_value': user_field.get('default_value'),
            'type': get_user_field_component_input_type(user_field.get('input_type'))
        } for user_field in user_input_field_list}
        application_parameter_setting['api_input_field_list'] = api_input_field_list
        application_parameter_setting['user_input_field_list'] = user_input_field_list
        node_data = base_node.get('properties').get('node_data') or {}
        file_upload_enable = node_data.get('file_upload_enable')
        if file_upload_enable:
            file_upload_setting = node_data.get('file_upload_setting') or {}
            for field in ['audio', 'document', 'image', 'other', 'video']:
                v = file_upload_setting.get(field)
                if v:
                    application_parameter_setting[field + '_list'] = {'required': False, 'default_value': [],
                                                                     'type': 'array'}
        return application_parameter_setting


class ApplicationTask(BaseTriggerTask):
    def support(self, trigger_task, **kwargs):
        return trigger_task.get('source_type') == 'APPLICATION'

    def execute(self, trigger_task, **kwargs):
        parameter_setting = trigger_task.get('parameter')
        task_record_id = uuid.uuid7()
        start_time = time.time()
        try:
            application = QuerySet(Application).filter(id=trigger_task.get('source_id')).only('type',
                                                                                              'work_flow').first()
            if application is None:
                QuerySet(TriggerTask).filter(id=trigger_task.get('id')).delete()
                return
            application_id = trigger_task.get('source_id')
            chat_id = uuid.uuid7()
            chat_user_id = str(uuid.uuid7())
            chat_record_id = str(uuid.uuid7())
            TaskRecord(id=task_record_id, trigger_id=trigger_task.get('trigger'),
                       trigger_task_id=trigger_task.get('id'),
                       source_type="APPLICATION",
                       source_id=application_id,
                       task_record_id=chat_record_id,
                       meta={'chat_id': chat_id},
                       state=State.STARTED).save()
            application_parameters_setting = get_application_parameters_setting(application)
            parameters = get_application_execute_parameters(parameter_setting, application_parameters_setting, kwargs)
            parameters['re_chat'] = False
            parameters['stream'] = True
            parameters['chat_record_id'] = chat_record_id
            message = parameters.get('message')
            ip_address = '-'
            if kwargs.get('body') is not None:
                ip_address = kwargs.get('body').get('ip_address')
            Chat.objects.get_or_create(id=chat_id, defaults={
                'application_id': application_id,
                'abstract': message,
                'chat_user_id': chat_user_id,
                'chat_user_type': ChatUserType.ANONYMOUS_USER.value,
                'asker': {'username': "游客"},
                'ip_address': ip_address,
                'source': {
                    'type': ChatSourceChoices.TRIGGER.value
                },
            })

            list(ChatSerializers(data={
                "chat_id": chat_id,
                "chat_user_id": chat_user_id,
                'chat_user_type': ChatUserType.ANONYMOUS_USER.value,
                'application_id': application_id,
                'ip_address': ip_address,
                'source': {
                    'type': ChatSourceChoices.TRIGGER.value
                },
                'debug': False
            }).chat(instance=parameters))
            chat_record = QuerySet(ChatRecord).filter(id=chat_record_id).first()
            if chat_record:
                state = get_workflow_state(chat_record.details)
                QuerySet(TaskRecord).filter(id=task_record_id).update(state=state, run_time=chat_record.run_time,
                                                                      meta={'parameter_setting': parameter_setting,
                                                                            'input': parameters, 'output': None})
            else:
                QuerySet(TaskRecord).filter(id=task_record_id).update(state=State.FAILURE,
                                                                      run_time=time.time() - start_time,
                                                                      meta={'parameter_setting': parameter_setting,
                                                                            'input': parameters, 'output': None,
                                                                            'err_message': 'Error: An unknown error occurred during the execution of the conversation'})
        except Exception as e:
            maxkb_logger.error(f"Application execution error: {traceback.format_exc()}")
            QuerySet(TaskRecord).filter(id=task_record_id).update(
                state=State.FAILURE,
                run_time=time.time() - start_time,
                meta={'input': {'parameter_setting': parameter_setting, **kwargs}, 'output': None,
                      'err_message': 'Error: ' + str(e)}
            )