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