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

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

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

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

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

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 сертификата, не смотря на то, что скрипт будет выполнен успешно. Если на проекте или макете отключение допустимо, необходимо кастомизировать конфигурацию.

Visiology Platform 2.31

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

Analytics

Related content