Общие сведения

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

Регистрация скрипта

Открыть на редактирование файл 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
```

В результате будет получен ответ с указанием количества скриптов и списком скриптов. В списке для каждого скрипта будут указаны его название, идентификатор, создатель, дата и время создания и обновления, тип, путь до скрипта и путь до файла конфигурации, а также статус файла конфигурации.

[2025-07-30 14:25:57.167 +03:00][    info][util] Found 'linked_scenario' python scripts: 3
[2025-07-30 14:25:57.167 +03:00][    info][util] List of found python scripts:
[2025-07-30 14:25:57.167 +03:00][    info][util] >>
[2025-07-30 14:25:57.167 +03:00][    info][util] -- name: Linked scenario copy
[2025-07-30 14:25:57.167 +03:00][    info][util] -- id: 41f836e5
[2025-07-30 14:25:57.167 +03:00][    info][util] -- creator: 
[2025-07-30 14:25:57.167 +03:00][    info][util] -- created at: Fri, 25 Apr 2025 10:11:52 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- updated at: Fri, 25 Apr 2025 10:13:38 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- type: linked_scenario
[2025-07-30 14:25:57.167 +03:00][    info][util] -- python script path: "/home/dmitrij/python-linked-scenario/link.py" (exists: true)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- script config path: "" (exists: false)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- config: valid
[2025-07-30 14:25:57.167 +03:00][    info][util] <<
[2025-07-30 14:25:57.167 +03:00][    info][util] >>
[2025-07-30 14:25:57.167 +03:00][    info][util] -- name: Linked scenario
[2025-07-30 14:25:57.167 +03:00][    info][util] -- id: ff09b4ce
[2025-07-30 14:25:57.167 +03:00][    info][util] -- creator: 
[2025-07-30 14:25:57.167 +03:00][    info][util] -- created at: Thu, 24 Apr 2025 11:53:53 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- updated at: Thu, 24 Apr 2025 11:53:53 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- type: linked_scenario
[2025-07-30 14:25:57.167 +03:00][    info][util] -- python script path: "/home/dmitrij/python-linked-scenario/link.py" (exists: true)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- script config path: "" (exists: false)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- config: valid
[2025-07-30 14:25:57.167 +03:00][    info][util] <<
[2025-07-30 14:25:57.167 +03:00][    info][util] >>
[2025-07-30 14:25:57.167 +03:00][    info][util] -- name: Gitlab linked script
[2025-07-30 14:25:57.167 +03:00][    info][util] -- id: 67b88a98
[2025-07-30 14:25:57.167 +03:00][    info][util] -- creator: 
[2025-07-30 14:25:57.167 +03:00][    info][util] -- created at: Fri, 04 Jul 2025 09:36:03 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- updated at: Fri, 04 Jul 2025 09:40:05 +0000
[2025-07-30 14:25:57.167 +03:00][    info][util] -- type: linked_scenario
[2025-07-30 14:25:57.167 +03:00][    info][util] -- python script path: "/home/dmitrij/python-linked-scenario/link.py" (exists: true)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- script config path: "/home/dmitrij/python-linked-scenario/config.json" (exists: true)
[2025-07-30 14:25:57.167 +03:00][    info][util] -- config: valid
[2025-07-30 14:25:57.167 +03:00][    info][util] <<

Для обратной совместимости сохранена поддержка скриптов, файлы конфигурации которых хранятся в директории /var/plmrepo/pyscripts/директория_зарегистрированного_python-скрипта. Так как такие скрипты были зарегистрированы без указания пути до файла конфигурации, при вызове списка зарегистрированных скриптов у такого скрипта соответствующая строка будет иметь следующий вид:

[2025-07-30 14:25:57.167 +03:00][    info][util] -- script config path: "" (exists: false)

Быстрые ссылки

Дерево страниц

Общие сведения

Подготовка к регистрации

Регистрация скрипта через конфигурационный файл plm-util

Регистрация скрипта

Обновление скрипта

Удаление скрипта

Вывод списка зарегистрированных скриптов

Быстрые ссылки

Дерево страниц

Управление Python-скриптом для детализации Drill-through

Общие сведения

Подготовка к регистрации

Регистрация скрипта через конфигурационный файл plm-util

Регистрация скрипта

Обновление скрипта

Удаление скрипта

Вывод списка зарегистрированных скриптов