HEX
Server: LiteSpeed
System: Linux CentOS-79-64-minimal 3.10.0-1160.119.1.el7.x86_64 #1 SMP Tue Jun 4 14:43:51 UTC 2024 x86_64
User: vishn3436 (5293)
PHP: 8.0.15
Disabled: NONE
Upload Files
File: //scripts/script-server/src/model/external_model.py
from datetime import timezone

from utils import date_utils


class ExecutionInfo(object):
    def __init__(self):
        self.param_values = {}
        self.script = None


def config_to_external(config, id, external_id=None):
    parameters = []
    for parameter in config.parameters:
        external_param = parameter_to_external(parameter)

        if external_param is None:
            continue

        parameters.append(external_param)

    return {
        'id': id,
        'clientModelId': external_id,
        'name': config.name,
        'description': config.description,
        'schedulable': config.schedulable,
        'parameters': parameters,
        'outputFormat': config.output_format
    }


def parameter_to_external(parameter):
    if parameter.constant:
        return None

    return {
        'name': parameter.name,
        'description': parameter.description,
        'withoutValue': parameter.no_value,
        'required': parameter.required,
        'default': parameter.default,
        'type': parameter.type,
        'min': parameter.min,
        'max': parameter.max,
        'max_length': parameter.max_length,
        'values': parameter.values,
        'secure': parameter.secure,
        'fileRecursive': parameter.file_recursive,
        'fileType': parameter.file_type,
        'requiredParameters': parameter.get_required_parameters()
    }


def to_short_execution_log(history_entries, running_script_ids=None):
    if running_script_ids is None:
        running_script_ids = []

    result = []
    for entry in history_entries:
        running = entry.id in running_script_ids
        external_entry = _translate_history_entry(entry, running)
        result.append(external_entry)

    return result


def to_long_execution_log(entry, log, running):
    external_entry = _translate_history_entry(entry, running)
    external_entry['command'] = entry.command
    external_entry['log'] = log
    external_entry['outputFormat'] = entry.output_format

    return external_entry


def _translate_history_entry(entry, running):
    if entry.start_time:
        start_time = date_utils.astimezone(entry.start_time, timezone.utc).isoformat()
    else:
        start_time = None

    return {
        'id': entry.id,
        'startTime': start_time,
        'user': entry.user_name,
        'script': entry.script_name,
        'status': running_flag_to_status(running),
        'exitCode': entry.exit_code
    }


def running_flag_to_status(running):
    return 'running' if running else 'finished'


def to_execution_info(request_parameters):
    NAME_KEY = '__script_name'

    script_name = request_parameters.get(NAME_KEY)

    param_values = {}
    for name, value in request_parameters.items():
        if name == NAME_KEY:
            continue
        param_values[name] = value

    info = ExecutionInfo()
    info.script = script_name
    info.param_values = param_values

    return info


def server_conf_to_external(server_config, server_version):
    return {
        'title': server_config.title,
        'enableScriptTitles': server_config.enable_script_titles,
        'version': server_version
    }


def parse_external_schedule(external_schedule):
    return {
        'repeatable': external_schedule.get('repeatable'),
        'start_datetime': external_schedule.get('startDatetime'),
        'repeat_unit': external_schedule.get('repeatUnit'),
        'repeat_period': external_schedule.get('repeatPeriod'),
        'weekdays': external_schedule.get('weekDays')
    }