Примеры скриптов

Примеры построены на основе демо-данных. Для полноценной работы необходимо сначала установить демо-данные.

В следующем примере реализовано:

Получение токена с сервера авторизации;

Перевод данных экземпляра форм, для которого вызван скрипт, в новый статус.

import json
import requests

identity_host = 'http://' + parameters['domain'] + '/idsrv/connect/token'
dc_host = 'http://' + parameters['domain'] + '/datacollection/api'  # API
login = parameters['userName']  # Логин администратора
password = '******'  # Пароль администратора

# Идентификатор экземпляра формы
form_instance_id = parameters['formInstance']['Id']

# Группа показателей 'Mediastore. План продаж'
# Идентификатор группы показателей
measure_group_id = 'measureGroup_Plan_prodazh'

# Адрес для работы с элементами группы показателей
measure_group_elements_url = dc_host + '/measuregroups/' + measure_group_id + '/elements'

# Адрес для работы со статусами экземпляра формы
form_states_url = dc_host + '/measuregroups/' + measure_group_id + '/forms/' + form_instance_id + '/states'

# Метод для получения header, которые необходимо добавлять в каждый запрос
def get_headers():
    #  Получение токена
    data = {
        'grant_type': 'password',
        'scope': 'openid profile email roles viewer_api core_logic_facade',
        'password': password,
        'username': login,
        'response_type': 'id_token token'
    }

    authorization_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Basic cm8uY2xpZW50OmFtV25Cc3B9dipvfTYkSQ=='
    }

    authorization_response = requests.post(
        identity_host,
        data=data,
        headers=authorization_headers,
        verify=False)

    authorization_json_response = authorization_response.json()

    authorization_token = authorization_json_response['access_token']

    authorization_headers = {
        'X-API-VERSION': '2.0',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + authorization_token
    }

    return authorization_headers


query_headers = get_headers()


# Получение статуса экземпляра формы
response = requests.get(form_states_url, headers=query_headers, verify=False)

json_response = response.json()

# Идентификатор текущего статуса = 4
# Бизнес-процесс 'Утверждение планов' Статус 'Черновик'
current_state_id = json_response['states'][0]['id']

# Идентификатор нового статуса, в который будет осуществлён перевод данных = 5
# Бизнес-процесс 'Утверждение планов' Статус 'Готово к проверке'
new_state_id = 5

# Идентификатор бизнес-процесса = 3
business_process_id = 3

# Пользовательский комментарий к переводу статуса
state_comment = 'Переведено из скрипта'

body = {
    'businessProcessId': business_process_id,
    'stateId': new_state_id,
    'comment': state_comment
}

# Отправка запроса на перевод данных экземпляра формы в новый статус
requests.put(form_states_url, headers=query_headers, data=json.dumps(body), verify=False)

В следующем скрипте реализовано:

Получение токена с сервера авторизации;
Получение элемента измерения по фильтру;

Создание элемента, в случае его отсутствия.

import json
import requests

identity_host = 'http://' + parameters['domain'] + '/idsrv/connect/token'
dc_host = 'http://' + parameters['domain'] + '/datacollection/api'  # API
login = parameters['userName']  # Логин администратора
password = '******'  # Пароль администратора

# Измерение 'Mediastore. Филиалы'
# Идентификатор измерения
dimension_id = 'dim_Filiali'
# Адрес для работы с элементами измерения
dimension_elements_url = dc_host + '/dimensions/' + dimension_id + '/elements'


# Метод для получения header, которые необходимо добавлять в каждый запрос
def get_headers():
    #  Получение токена
    data = {
        'grant_type': 'password',
        'scope': 'openid profile email roles viewer_api core_logic_facade',
        'password': password,
        'username': login,
        'response_type': 'id_token token'
    }

    authorization_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Basic cm8uY2xpZW50OmFtV25Cc3B9dipvfTYkSQ=='
    }

    authorization_response = requests.post(
        identity_host,
        data=data,
        headers=authorization_headers,
        verify=False)

    authorization_json_response = authorization_response.json()

    authorization_token = authorization_json_response['access_token']

    authorization_headers = {
        'X-API-VERSION': '2.0',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + authorization_token
    }

    return authorization_headers

query_headers = get_headers()

# Создание нового элемента измерения 'Mediastore. Филиалы'

# Фильтр для получения необходимых элементов измерения
# Необходимо получить элементы с наименованием 'Екатеринбург'
filter_for_element = {
    'type': 'name',  # Фильтрация осуществляется по наименованию элемента
    'value': 'Екатеринбург',  # Наименование элемента сравнивается со строкой 'Екатеринбург'
    'condition': 'equals'  # Наименование элемента должно полностью совпадать с переданной строкой ('Екатеринбург')
}

# Отправка запроса для получения элементов измерения
response = requests.get(dimension_elements_url, headers=query_headers, data=json.dumps(filter_for_element), verify=False)

# Получение тела запроса в JSON формате
json_response = response.json()

# Получение списка элементов измерения
elements = json_response['elements']

# Если элементов измерения с наименованием 'Екатеринбург' не найдено
if len(elements) == 0:
    # То необходимо создать новый элемент

    # Описание элемента
    new_elements = [
        {
            'Name': 'Екатеринбург',  # Наименование элемента
            'Path': [],  # Создать элемент в корне измерения, т.е. элемент не должен находится ни в каком каталоге
            'Attributes': [{ "AttributeId": "attr_Tip_filiala", "Value": 1 }]  # Атрибут "Тип филиала" соответствует значению справочника с идентификатором 1
        }
    ]

    # Отправка запроса на создание элемента
    response = requests.post(dimension_elements_url, headers=query_headers, data=json.dumps(new_elements), verify=False)

    # Получение тела запроса в JSON формате
    json_response = response.json()

    print('Количество созданных элементов: ' + str(json_response['added']))

В следующем скрипте реализовано:

Получение токена с сервера авторизации;
Получение элементов группы показателей по фильтру;

Создание новых элементов группы показателей из существующих, путём изменения даты (т.е. элементы за 1 квартал 2020 года копируются в элементы за 1 квартал 2021 года) и увеличением значения на 10%.

import json
import requests

identity_host = 'http://' + parameters['domain'] + '/idsrv/connect/token'
dc_host = 'http://' + parameters['domain'] + '/datacollection/api'  # API
login = parameters['userName']  # Логин администратора
password = '******'  # Пароль администратора

# Группа показателей 'Mediastore. План продаж'
# Идентификатор группы показателей
measure_group_id = 'measureGroup_Plan_prodazh'

# Адрес для работы с элементами группы показателей
measure_group_elements_url = dc_host + '/measuregroups/' + measure_group_id + '/elements'

# Метод для получения header, которые необходимо добавлять в каждый запрос
def get_headers():
    #  Получение токена
    data = {
        'grant_type': 'password',
        'scope': 'openid profile email roles viewer_api core_logic_facade',
        'password': password,
        'username': login,
        'response_type': 'id_token token'
    }

    authorization_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Basic cm8uY2xpZW50OmFtV25Cc3B9dipvfTYkSQ=='
    }

    authorization_response = requests.post(
        identity_host,
        data=data,
        headers=authorization_headers,
        verify=False)

    authorization_json_response = authorization_response.json()

    authorization_token = authorization_json_response['access_token']

    authorization_headers = {
        'X-API-VERSION': '2.0',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + authorization_token
    }

    return authorization_headers


query_headers = get_headers()

# Получение элементов группы показателей
# Фильтр для получения элементов группы показателей
# Осуществляется получение элементов за 1 квартал 2020 года
filter_for_elements = {
    'type': 'calendar',  # Фильтрация осуществляется по календарю
    'value': '2020-01-01',  # Дата элемента группы показателей сравнивается со значением '2020-01-01'
    'condition': 'equals'  # Дата элемента группы показателей должна совпадать с переданным значением
}

# Отправка запроса для получения элементов группы показателей по фильтру
response = requests.get(
    measure_group_elements_url,
    headers=query_headers,
    data=json.dumps(filter_for_elements),
    verify=False)

# Получение результата запроса в формате JSON
json_response = response.json()

# Получение списка элементов из результата
elements = json_response['elements']

# Список новых элементов группы показателей
new_elements = []

# Для каждого полученного элемента
for element in elements:
    # Создание нового элемента, но с новой датой (1 квартал 2021 года)
    new_element = {
        'value': element['value'] * 1.1,
        'dimensions': [],
        'calendars': [
            {
                'id': json_response['measureGroup']['calendar']['id'],
                'name': json_response['measureGroup']['calendar']['name'],
                'date': '2021-01-01'
            }
        ],
    }

    # Копирование элементов измерений в новый элемент группы показателей
    for coordinate in element['dimensionElements']:
        new_element['dimensions'].append({
            'id': coordinate['dimensionId'],
            'elementId': coordinate['elementId']
        })

    # Добавление нового элемента в список
    new_elements.append(new_element)

# Если есть хотя бы один элемент для создания
if len(new_elements) > 0:
    # Отправка новый элементов в DataCollection
    response = requests.post(measure_group_elements_url, headers=query_headers, data=json.dumps(new_elements), verify=False)

    json_response = response.json()

    print('Количество созданных элементов: ' + str(json_response['created']))

В следующем скрипте реализовано:

Получение токена с сервера авторизации;
Получение элементов группы показателей по фильтру из экземпляра формы;

Обновление комментария у полученных элементов группы показателей.

import json
import requests
import datetime

from datetime import datetime

identity_host = 'http://' + parameters['domain'] + '/idsrv/connect/token'
dc_host = 'http://' + parameters['domain'] + '/datacollection/api'  # API
login = parameters['userName']  # Логин администратора
password = '******'  # Пароль администратора

form_instance = parameters['formInstance']

dimensionUnqiueName_Filiali = form_instance['Circulation']['DimensionElements'][0]['Dimension']['UniqueName']
dimensionUniqueName_Versii = form_instance['Circulation']['DimensionElements'][1]['Dimension']['UniqueName']

calendarValue = datetime.strptime(form_instance['Circulation']['CalendarFolders'][0]['Folder']['AbsolutelyValue'], '%Y-%m-%dT%H:%M:%SZ')

dimensionValue_Filiali = form_instance['Circulation']['DimensionElements'][0]['Element']['Id']
dimensionValue_Versii = form_instance['Circulation']['DimensionElements'][1]['Element']['Id']

# Группа показателей 'Mediastore. План продаж'
# Идентификатор группы показателей
measure_group_id = 'measureGroup_Plan_prodazh'

# Адрес для работы с элементами группы показателей
measure_group_elements_url = dc_host + '/measuregroups/' + measure_group_id + '/elements'

# Метод для получения header, которые необходимо добавлять в каждый запрос
def get_headers():
    #  Получение токена
    data = {
        'grant_type': 'password',
        'scope': 'openid profile email roles viewer_api core_logic_facade',
        'password': password,
        'username': login,
        'response_type': 'id_token token'
    }

    authorization_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Basic cm8uY2xpZW50OmFtV25Cc3B9dipvfTYkSQ=='
    }

    authorization_response = requests.post(
        identity_host,
        data=data,
        headers=authorization_headers,
        verify=False)

    authorization_json_response = authorization_response.json()

    authorization_token = authorization_json_response['access_token']

    authorization_headers = {
        'X-API-VERSION': '2.0',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + authorization_token
    }

    return authorization_headers


query_headers = get_headers()

# Получение элементов группы показателей
# Фильтр для получения элементов группы показателей
filter_for_elements = {
    "operation": "and",
    "filters": [
        {
            "value": calendarValue.strftime('%Y-%m-%d'), "type": "calendar", "condition": "equals"
        },
        {
            'type': 'DimensionId',  # Фильтрация осуществляется по элементу измерению
            'value': dimensionValue_Filiali,  # Идентификатор элемента получается из экземпляра
            'condition': 'equals',  # Идентификатор элемента измерения должен совпадать с переданным значением,
            'name': dimensionUnqiueName_Filiali  # Идентификатор измерения, по которому осуществляется фильтрация
        },
        {
            'type': 'DimensionId',
            'value': dimensionValue_Versii,
            'condition': 'equals',
            'name': dimensionUniqueName_Versii
        }     
    ]
}

# Отправка запроса для получения элементов группы показателей по фильтру
response = requests.get(
    measure_group_elements_url,
    headers=query_headers,
    data=json.dumps(filter_for_elements),
    verify=False)

# Получение результата запроса в формате JSON
json_response = response.json()

# Получение списка элементов из результата
elements = json_response['elements']

# Список новых элементов группы показателей
new_elements = []

# Для каждого полученного элемента
for element in elements:
    # Создание нового элемента, но с пустым значением и новым комментарием
    new_element = {
        'value': element['value'],
        'comment': 'Обновлено из скрипта',
        'mustUpdateComment': True,
        'dimensions': [],
        'calendars': [
            {
                'id': json_response['measureGroup']['calendar']['id'],
                'name': json_response['measureGroup']['calendar']['name'],
                'date': element['calendar']['date'][:10]
            }
        ],
    }

    # Копирование элементов измерений в новый элемент группы показателей
    for coordinate in element['dimensionElements']:
        new_element['dimensions'].append({
            'id': coordinate['dimensionId'],
            'elementId': coordinate['elementId']
        })

    # Добавление нового элемента в список
    new_elements.append(new_element)

# Если есть хотя бы один элемент для создания
if len(new_elements) > 0:
    # Отправка новый элементов в DataCollection
    response = requests.post(measure_group_elements_url, headers=query_headers, data=json.dumps(new_elements), verify=False)

    json_response = response.json()

    print('Количество созданных элементов: ' + str(json_response['created']))

В следующем скрипте реализовано:

Получение токена с сервера авторизации;
Получение элементов группы показателей из экземпляра формы;

Копирование данных из формы в аналогичную форму другой группы показателей, при условии, что оригинальная форма в статусе "согласовано".

# Группа pythonMG_master имеет атрибут reconciliation(логическое) для статуса переноса
# Формы тиражируются по календарю и измерению python_dim_t
# Общее измерение показателей в обеих группах показателей 
# Бизнес-процесс с конечным статусом "согласовано"

import json
import requests

identity_host = 'http://' + parameters['domain'] + '/idsrv/connect/token'
dc_host = 'http://' + parameters['domain'] + '/datacollection/api'  # API
login = parameters['userName']  # Логин администратора
password = '*******'  # Пароль администратора
CIRCULATION_DIMENSION = 'dim_python_dim_t' #тиражное измерение 
FINISH_STATE = u'согласовано' #статус формы, для которого работает перенос

# Идентификатор группы показателей
measure_group_id = 'measureGroup_pythonMG_master' #откуда копируем
copy_measure_group_id = 'measureGroup_pythonMG_slave' #куда копируем
form_instance_id = parameters['formInstance']['Id'] #экземпляр текущей формы

# Адреса для работы с элементами группы показателей
measure_group_elements_url = dc_host + '/measuregroups/' + measure_group_id + '/elements'
measure_group_attrib_url = dc_host + '/measuregroups/' + measure_group_id + '/attributes'
copy_measure_group_elements_url = dc_host + '/measuregroups/' + copy_measure_group_id + '/elements'
form_states_url = dc_host + '/measuregroups/' + measure_group_id + '/forms/' + form_instance_id + '/states'

# Метод для получения header, которые необходимо добавлять в каждый запрос
def get_headers():
    #  Получение токена
    data = {
        'grant_type': 'password',
        'scope': 'openid profile email roles viewer_api core_logic_facade',
        'password': password,
        'username': login,
        'response_type': 'id_token token'
    }

    authorization_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Basic cm8uY2xpZW50OmFtV25Cc3B9dipvfTYkSQ=='
    }
    
    authorization_response = requests.post(identity_host,data=data,headers=authorization_headers,verify=False)
    authorization_json_response = authorization_response.json()
    authorization_token = authorization_json_response['access_token']
    
    authorization_headers = {
        'X-API-VERSION': '2.0',
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + authorization_token
    }

    return authorization_headers

query_headers = get_headers()

# Получение элементов группы показателей
# Отправка запроса для получения элементов группы показателей по фильтру

response = requests.get(measure_group_elements_url,headers=query_headers,verify=False)

# Получение результата запроса в формате JSON
json_response = response.json()

# Получение списка элементов и ГП из результата
elements = json_response['elements']
measure_group = json_response['measureGroup']

new_elements = []
old_elements = []

# Получие состояния текущей формы
response = requests.get(form_states_url, headers=query_headers, verify=False)
json_response = response.json() 

current_state = ''

#получаем статус формы
if json_response['states']:
    current_state = json_response['states'][0]['name']
#получаем дату тиража формы 
form_date = json_response['form']['coordinates'][0]['value']['id'][:10]
current_circulation_dim_id = -1

#получение дополнительной координаты из тиража кроме даты (python_dim_t)
for form_coordinate in json_response['form']['coordinates']:
    if form_coordinate['description']['uniqueName'] == CIRCULATION_DIMENSION :
        current_circulation_dim_id = form_coordinate['value']['id']

#копируем только согласованную форму
if current_state == FINISH_STATE :
    for element in elements:
        # Добавляем в список элемент, если дата формы и элемента совпали
        # копируются только данные с формы
        
        element_dim_id = -1 
               
        for dims in element['dimensionElements']:
            if dims['dimensionId'] == 'dim_python_dim_t':
                element_dim_id = dims['elementId']
                
        if element['calendar']['date'][:10] == form_date and current_circulation_dim_id == element_dim_id :
            new_element = {
                'value': element['value'],
                'dimensions': [],
                'calendars': [
                    {
                        'id': measure_group['calendar']['id'],
                        'name': measure_group['calendar']['name'],
                        'date': element['calendar']['date'][:10]
                    }
                ],
                'measure': {
                    'id': element['measureElements'][0]['measureId'],
                    'elementId': element['measureElements'][0]['elementId']
                }
            }
            
            # Копирование элементов измерений в новый элемент группы показателей
            for coordinate in element['dimensionElements']:
                new_element['dimensions'].append({
                    'id': coordinate['dimensionId'],
                    'elementId': coordinate['elementId']
                })

            
            old_element = dict(new_element)
            #устанавливаем атрибут - признак переноса в true
            old_element['Attributes'] = []
            old_element['Attributes'].append( 
                    {
                        'Id': 'attr_reconciliation',
                        'value':'true'
                    }
            )
            # Добавление нового элемента в список
            new_elements.append(new_element)
            old_elements.append(old_element)
            
    #корректируем старые элементы        
    if len(old_elements) > 0:
       response = requests.post(measure_group_elements_url, headers=query_headers, data=json.dumps(old_elements), verify=False)
    #добавляем новые элементы
    if len(new_elements) > 0:
       response = requests.post(copy_measure_group_elements_url, headers=query_headers, data=json.dumps(new_elements), verify=False)

Примечание – В скриптах отключена проверка HTTPS сертификата для демонстрации логики написания скриптов. На реальных задачах отключение проверки сертификата не рекомендуется.

Примечание – В конфигурации по умолчанию сервис выполнения скриптов будет показывать ошибку при попытке выполнить скрипт с отключенной проверкой HTTPS сертификата, не смотря на то, что скрипт будет выполнен успешно. Если на проекте или макете отключение допустимо, необходимо кастомизировать конфигурацию.