| Python-скрипт (далее – скрипт) обеспечивает детализацию Drill-through, выполняя ее по условиям, описанным в файле конфигурации в формате JSON (см. Детализация Drill-through). Скрипт необходимо зарегистрировать для приложения при помощи утилиты plm-util, чтобы приложение могло с ним работать. |
При регистрации скрипта необходимо указывать его название, которое в приложении отображается в выпадающем списке варианта «Связанный сценарий» контекстного меню факта (см. Детализация Drill-through).
В приложении может быть зарегистрировано сколько угодно много скриптов. Для каждого зарегистрированного скрипта обязательно должен быть предоставлен файл конфигурации в формате JSON (далее – файл конфигурации). Если для какого-либо зарегистрированного скрипта файл конфигурации отсутствует или имеет недопустимую структуру и детализация включена параметром конфигурации приложения plm.user_interface.scenario_hyperlink_enabled, то в ЛЮБОМ воспроизведенном сценарии, для ЛЮБОЙ его мультисферы контекстное меню факта НЕ ОТКРЫВАЕТСЯ, а в лог приложения записывается ошибка вида, соответственно:
Failed to get list of registered python scripts: linked scenario script 'Имя сценария' config file not exists Failed to get list of registered python scripts: Rapidjson assertion error |
Путь к файлу конфигурации для конкретного зарегистрированного скрипта устанавливается при регистрации скрипта или при редактировании файла-ресурса через утилиту plm-util (см. разделы «Регистрация скрипта» и «Обновление скрипта» ниже). Это позволяет иметь единственный файл с исходным кодом скрипта и отдельные файлы конфигурации для каждого конкретного зарегистрированного скрипта.
Для обратной совместимости сохранена поддержка скриптов, файлы конфигурации которых хранятся в директории /var/plmrepo/pyscripts/директория_зарегистрированного_python-скрипта. |
Регистрировать, обновлять и удалять можно только по одному скрипту за раз.
Остановить работу приложения:
service polymatica stop |
Убедиться, что в файле конфигурации утилиты plm-util.conf указан параметр, определяющий путь до репозитория Polymatica Analytics, обычно это /var/plmrepo:
plm.manager.repo_directory = /var/plmrepo |
Создать в системе хоста приложения директорию /var/plmrepo/pyscripts/директория_python-скрипта, создать в ней файл с расширением .py и вставить в него код скрипта. Например:
/var/plmrepo/pyscripts/drill_through_on_population_ms/script.py |
version = "1.2.0"
import os
import sys
import json
import logging
from datetime import datetime
from typing import List
from collections import namedtuple
def setup_logging():
# Create logs directory if it doesn't exist
log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'logs')
os.makedirs(log_dir, exist_ok=True)
# Create logger
logger = logging.getLogger('linked_scenario')
logger.setLevel(logging.INFO)
# Clear existing handlers
logger.handlers.clear()
# File handler - one file per month
log_file = os.path.join(log_dir, f'linked_scenario_{datetime.now().strftime("%Y-%m")}.log')
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
# Add handler
logger.addHandler(file_handler)
return logger
def main():
import argparse
# Setup logging
logger = setup_logging()
logger.info("=" * 80)
logger.info("Starting linked scenario script")
logger.info(f"Script version: {version}")
logger.info("=" * 80)
def err(err_type: str, msg: str):
logger.error(f"Critical error of type '{err_type}': {msg}")
json.dump({"type": err_type, "message": msg}, sys.stderr, ensure_ascii=False)
sys.exit(1)
try:
logger.info("Importing polymatica modules...")
from polymatica import business_scenarios
from polymatica.exceptions import PolymaticaException, RightsError, ScenarioError
logger.info("Polymatica modules imported successfully")
except ImportError as e:
logger.error(f"Polymatica import error: {str(e)}")
err(err_type="polyapi", msg=f"PolyAPI import error: {str(e)}")
sys.stdout = open(1, "w", encoding='utf-8', closefd=False)
ROOT_PATH = os.path.dirname(os.path.abspath(__file__))
logger.info(f"Root directory: {ROOT_PATH}")
# Log command line arguments
logger.info(f"script args: {sys.argv[1:]}")
print(f"script args: {sys.argv[1:]}")
class KeyValue(argparse.Action):
def __call__(self, parser, namespace,
values, option_string=None):
setattr(namespace, self.dest, dict())
for value in values:
key, value = value.split('=')
getattr(namespace, self.dest)[key] = value
class LoggerDisabler:
"""
Контекстный менеджер, временно отключающий логгирование,
чтобы ошибки из полиапи не попадали напрямую в stderr, так как они
передаются в этот скрипт и через функцию err попадают в stderr внутри json.
"""
def __enter__(self):
logging.disable(logging.ERROR)
def __exit__(self, exc_type, exc_val, exc_tb):
logging.disable(logging.NOTSET)
parser = argparse.ArgumentParser()
parser.add_argument('--dimension-elements',
nargs='*',
action=KeyValue)
parser.add_argument('--connection_session', nargs=3)
parser.add_argument('--url')
parser.add_argument('--layer_id')
parser.add_argument('--scenario_id')
parser.add_argument('--cube_id')
parser.add_argument('--config_path', default=os.path.join(ROOT_PATH, 'config.json'))
args = parser.parse_args()
# Log all input parameters
logger.info("Input parameters:")
logger.info(f" - dimension_elements: {args.dimension_elements}")
logger.info(f" - connection_session: {args.connection_session}")
logger.info(f" - url: {args.url}")
logger.info(f" - layer_id: {args.layer_id}")
logger.info(f" - scenario_id: {args.scenario_id}")
logger.info(f" - cube_id: {args.cube_id}")
logger.info(f" - config_path: {args.config_path}")
class DimFilter:
def __init__(self, dim_id: str, filter_val: str):
self.dim_id = dim_id
self.filter_val = filter_val
class MapInfo:
def __init__(self, scenario_id: str, cube_id: str, filters: List[DimFilter]):
self.scenario_id = scenario_id # 1. запустить
self.cube_id = cube_id # 2. найти на слое мультисферы этого куба
self.filters = filters # 3. применить к мультисферам
def extract_map_info(cfg) -> MapInfo:
logger.info("Starting map info extraction from configuration")
logger.info(f"Looking for scenario ID: {args.scenario_id}")
logger.info(f"Looking for cube ID: {args.cube_id}")
for scenario_pair in cfg['scenarios']:
from_sc_id = scenario_pair['from_scenario_id']
to_sc_id = scenario_pair['to_scenario_id']
logger.info(f"Checking scenario pair: {from_sc_id} -> {to_sc_id}")
if args.scenario_id == from_sc_id:
logger.info(f"Found matching scenario pair: {from_sc_id} -> {to_sc_id}")
dims_to_filter = []
to_cube_id = 0
for cube_pair in scenario_pair['cubes']:
logger.info(f"Checking cube pair: {cube_pair['from_cube_id']} -> {cube_pair['to_cube_id']}")
if cube_pair['from_cube_id'] == args.cube_id:
to_cube_id = cube_pair['to_cube_id']
logger.info(f"Found matching cube pair: {args.cube_id} -> {to_cube_id}")
logger.info(f"Available dimensions in config: {cube_pair['dimensions']}")
for dim in cube_pair['dimensions']:
from_dim_id = dim['from_dimension_id']
to_dim_id = dim['to_dimension_id']
logger.info(f"Checking dimension: {from_dim_id} -> {to_dim_id}")
if from_dim_id in args.dimension_elements.keys():
dim_filter = DimFilter(dim_id=to_dim_id,
filter_val=args.dimension_elements[from_dim_id].encode('utf8',
'surrogateescape').decode(
'utf8'))
dims_to_filter.append(dim_filter)
logger.info(f"Added filter for dimension {from_dim_id} -> {to_dim_id} with value '{dim_filter.filter_val}'")
break
logger.info(f"Created MapInfo: scenario_id={to_sc_id}, cube_id={to_cube_id}, filters_count={len(dims_to_filter)}")
return MapInfo(filters=dims_to_filter, cube_id=to_cube_id, scenario_id=to_sc_id)
logger.error("No matching scenario pair found in configuration")
raise Exception("в конфиге не найдена пара сценариев")
# Load configuration
logger.info(f"Loading configuration from file: {args.config_path}")
try:
with open(args.config_path) as f:
json_config = json.load(f)
logger.info("Configuration loaded successfully")
except OSError as e:
logger.error(f"Error opening configuration file: {e}")
err(err_type="os", msg="cannot open {}".format(args.config_path))
except json.JSONDecodeError as e:
logger.error(f"JSON parsing error: {e}")
err(err_type="json", msg=str(e))
# Extract map info
logger.info("Extracting map info from configuration...")
try:
map_info = extract_map_info(json_config)
logger.info(f"Map info extracted successfully: scenario_id={map_info.scenario_id}, cube_id={map_info.cube_id}")
except Exception as e:
logger.error(f"Error extracting map info: {e}")
err(err_type="config", msg="Ошибка в конфиге: {}".format(str(e)))
# Check filter count match
if len(map_info.filters) != len(args.dimension_elements):
logger.error(f"Filter count mismatch: expected {len(args.dimension_elements)}, found {len(map_info.filters)}")
err(err_type="config",
msg="Пары для размерностей \"{}\" не заданы в конфиге".format(args.dimension_elements.keys()))
# Connect to Polymatica
logger.info("Connecting to Polymatica...")
try:
session_params = {"session_id": args.connection_session[0],
"manager_uuid": args.connection_session[1],
"full_polymatica_version": args.connection_session[2]}
logger.info(f"Session parameters: {session_params}")
logger.info(f"Connection URL: {json_config['connection']['url']}")
with LoggerDisabler():
bs = business_scenarios.BusinessLogic("", session_auth=session_params, url=json_config['connection']['url'])
logger.info("Successfully connected to Polymatica Analytics")
except Exception as e:
logger.error(f"Polymatica Analytics connection error: {e}")
err(err_type="connect", msg=str(e))
# Run scenario
logger.info(f"Running scenario '{map_info.scenario_id}' on layer '{args.layer_id}'")
try:
with LoggerDisabler():
logger.info("Checking scenario cubes permissions...")
bs._check_scenario_cubes_permission(scenario_id=map_info.scenario_id)
logger.info("Checking scenario dimensions and facts permissions...")
if bs.check_scenarios_dims_facts_permission(scenario_id=map_info.scenario_id):
logger.info("Running scenario on layer...")
bs.run_scenario_on_layer(scenario_id=map_info.scenario_id, layer_id=args.layer_id)
logger.info("Scenario started successfully on layer")
except RightsError as e:
logger.error(f"Rights error: {e.user_msg}")
err(err_type="rights", msg="Не удалось запустить связанный сценарий: " + e.user_msg)
except ScenarioError as e:
error_msg = e.user_msg
logger.error(f"Scenario error: {error_msg}")
if "No such RuntimeId in store" in error_msg:
error_msg = "не найден целевой слой."
err(err_type="scenario", msg="Не удалось запустить связанный сценарий: " + error_msg)
except Exception as e:
logger.error(f"Unexpected error running scenario: {e}")
err(err_type="scenario", msg="Не удалось запустить связанный сценарий")
logger.info(f"Scenario '{map_info.scenario_id}' completed successfully on layer '{args.layer_id}'")
modules_to_filter = []
ModuleInfo = namedtuple('ModuleInfo', ['id', 'name', 'type', 'cube_id'])
logger.info("Getting module list...")
modules = [ModuleInfo(*module) for module in bs.get_module_list()]
logger.info(f"Found modules: {len(modules)}")
for module in modules:
logger.info(f"Module: id={module.id}, name='{module.name}', type='{module.type}', cube_id={module.cube_id}")
if module.type == 'Мультисфера' and module.cube_id == map_info.cube_id:
modules_to_filter.append(module.id)
logger.info(f"Found matching multisphere module: {module.id} (cube: {module.cube_id})")
if len(modules_to_filter) == 0:
logger.error(f"No multisphere module found for cube ID '{map_info.cube_id}' in scenario '{map_info.scenario_id}'")
err(err_type="module",
msg="Не найден модуль, созданный из куба с id \"{}\" в сценарии \"{}\"".format(map_info.cube_id,
map_info.scenario_id))
logger.info(f"Found multisphere modules for filtering: {len(modules_to_filter)}")
def find_dim(dims_list, dim_id):
for dim in dims_list:
if dim.get('id', '') == dim_id:
return dim
return None
# Apply filters to modules
logger.info("Applying filters to multisphere modules...")
for module_id in modules_to_filter:
logger.info(f"Processing module: {module_id}")
bs.set_multisphere_module_id(module_id)
logger.info("Getting module dimensions list...")
dims = bs._get_dimensions_list()
logger.info(f"Found dimensions in module: {len(dims)}")
for filter_to_put in map_info.filters:
logger.info(f"Applying filter: dim_id={filter_to_put.dim_id}, value='{filter_to_put.filter_val}'")
dim = find_dim(dims, filter_to_put.dim_id)
if not dim:
logger.warning(f"Dimension with ID '{filter_to_put.dim_id}' not found in module {module_id}")
continue
try:
logger.info(f"Setting filter for dimension {filter_to_put.dim_id}...")
bs.put_dim_filter_by_value(dim_id=filter_to_put.dim_id, value=filter_to_put.filter_val,
clear_filter=True)
logger.info(f"Filter applied successfully: dim_id={filter_to_put.dim_id}, value='{filter_to_put.filter_val}'")
except PolymaticaException as e:
logger.warning(f"Filter skipped (excluded by another filter): dim_id={filter_to_put.dim_id}, value='{filter_to_put.filter_val}', error={e}")
logger.info("=" * 80)
logger.info("Linked scenario script completed successfully")
logger.info("=" * 80)
if __name__ == "__main__":
main()
|
Открыть на редактирование файл plm-util.conf:
nano /etc/polymatica/plm-util.conf |
Добавить в него строки:
# Команда регистрации скрипта create-pyscript # Название регистрируемого скрипта Python pyscripts.create.name = Python Linked Scenario # Тип регистрируемого скрипта Python [Возможные типы: "formatted_export", "linked_scenario". Для скрипта детализации используется тип "linked_scenario".] pyscripts.create.type = linked_scenario # Путь до регистрируемого скрипта Python на диске pyscripts.create.script = /var/plmscripts/linked_scenario/first/script.py # Путь до файла конфигурации регистрируемого скрипта pyscripts.create.config = /path/to/python-linked-scenario/config.json # Идентификатор факта, для которого задан скрипт типа "linked_scenario". Функциональность временно выключена. pyscripts.create.measure_id = 07141982 |
Сохранить изменения в plm-util.conf и выполнить команду:
plm-util --create-pyscript |
Открыть на редактирование файл plm-util.conf:
nano /etc/polymatica/plm-util.conf |
Добавить (раскомментировать) строки:
# Команда обновления скрипта update-pyscript # Внесение изменений в зарегистрированный скрипт Python по имени # pyscripts.update.by_name = Python Linked Scenario # либо # Внесение изменений в зарегистрированный скрипт Python по идентификатору # pyscripts.update.by_id = 1a71ec1d # Новое наименование для скрипта Python pyscripts.update.new_name = New PlmUtil Script # Новый тип скрипта Python pyscripts.update.new_type = linked_scenario # Путь до нового скрипта Python pyscripts.update.new_script = /var/plmscripts/linked_scenario/test_python_script_2.py # Путь до файла конфигурации нового скрипта pyscripts.update.new_config = /path/to/python-linked-scenario/config.json |
Сохранить изменения в plm-util.conf и выполнить команду:
plm-util --update-pyscript |
Открыть на редактирование файл plm-util.conf:
nano /etc/polymatica/plm-util.conf |
Добавить в него строки:
delete-pyscript # Удаление зарегистрированного скрипта Python по имени # pyscripts.delete.by_name = Python Linked Scenario либо # Удаление зарегистрированного скрипта Python по идентификатору # pyscripts.delete.by_id = e654a03f |
Выполните команду:
plm-util --delete-pyscript |
После успешного завершения процедуры можно запустить сервисы Polymatica командой:
systemctl start polymatica.service |
Открыть на редактирование файл plm-util.conf:
nano /etc/polymatica/plm-util.conf |
Добавить (раскомментировать) строки:
# Напечатать информацию по имеющимся скриптам list_pyscripts # Напечатать информацию только по скриптам детализации (тип "linked_scenario") pyscripts.list.type = linked_scenario |
Сохранить изменения в plm-util.conf и выполнить команду:
plm-util --list_pyscripts |
В результате будет получен ответ с указанием количества скриптов и списком скриптов. В списке для каждого скрипта будут указаны его название, идентификатор, создатель, дата и время создания и обновления, тип, путь до скрипта и путь до файла конфигурации, а также статус файла конфигурации.
|
Для обратной совместимости сохранена поддержка скриптов, файлы конфигурации которых хранятся в директории /var/plmrepo/pyscripts/директория_зарегистрированного_python-скрипта. Так как такие скрипты были зарегистрированы без указания пути до файла конфигурации, при вызове списка зарегистрированных скриптов у такого скрипта соответствующая строка будет иметь следующий вид: [2025-07-30 14:25:57.167 +03:00][ info][util] -- script config path: "" (exists: false) |