Подключение к ExtSdk2 через OLE - Сценарии ExtSdk2

Для работы со СБИС3 Плагин из внешних систем, таких как 1С или SAP, требуется получить COM-объект, зарегистрированный Плагином в Вашей системе. Его название "Tensor.SbisPluginClientCOM". Все примеры кода указаны на языке Python.

Получение COM-объекта

class OleSbis3Plugin:

    def __init__(self, **kwargs):
        self.awaited_answers = {}
        self.ole = win32com.client.Dispatch("Tensor.SbisPluginClientCOM")
        self.version = None
        self._wait_connected()
        self.host = None
        self.account_id = None
        self.guid_module = None

    def _wait_connected(self):
        event = self.read_all_xml(['Event_connected'])
        self.version = event[0]['data']['Version']
        return

Из данного примера кода видно, что получение COM-объекта занимает одну строку "self.ole = win32com.client.Dispatch("Tensor.SbisPluginClientCOM")" с использованием библиотеки pywin32. Далее Плагин подготавливает необходимые внутренние инструменты для подключения и в случае успеха пробрасывает событие с именем "Event_connected".

Успешное подключение следует обработать в отдельной функции "_wait_connected()" как указано в примере. Она ожидает результат работы функции "read_all_xml()"

Пример вызовов методов плагина через COM-объект

    def get_module(self, name):
        self.guid_module = self.ole.GetModule(name)

    def call_method(self, ole_method, module_method, parameters_module_method):
        try:
            query_id = str(uuid.uuid4())
            param_list = [query_id, self.guid_module, module_method, parameters_module_method]
            ole_method_dict = {
                'CallMethodWithoutAuth': {"method": self.ole.CallMethodWithoutAuth, "params": param_list + [self.host]},
                'CallMethod': {"method": self.ole.CallMethod, "params": param_list + [self.account_id]},
                'CallMethodWithoutAuthXML': {"method": self.ole.CallMethodWithoutAuthXML,
                                             "params": param_list + [self.host]},
                'CallMethodXML': {"method": self.ole.CallMethodXML, "params": param_list + [self.account_id]}
            }

            ole_method_dict[ole_method]['method'](*ole_method_dict[ole_method]['params'])

            return query_id
        except Exception as err:
            raise err

Для корректной адресации наших запросов в плагин к модулю ExtSdk2 необходимо сначала получить его guid. Guid можно получить в функции get_module, которая вызывает непосредственно метод COM-объекта self.ole.GetModule(name). Вызов методов ExtSdk2 выполняется через вызов методов COM-объекта. Пример в функции "call_method". Методы "self.ole.CallMethodWithoutAuth" и "self.ole.CallMethodWithoutAuthXML" вызываются для авторизации в плагине и получить идентификатор сессии с которым можно вызывать все остальные методы модуля через "CallMethod" и "CallMethodXML". Методы на вход принимают query_id вызываемого метода, guid_module - guid полученного модуля, module_method - метод модуля, который хотим позвать, parameters_module_method - параметры вызываемого метода, а так же либо host на котором мы хотим работать( online.sbis.ru ), либо account_id - полученный идентификатор сессии с которым можно вызывать остальные методы модуля. На выходе получаем query_id вызываемого метода, который нужно передать в функцию read_all_xml для ожидания ответа.

Пример реализации получения ответов от COM-объекта

    @staticmethod
    def parse_Message(_event):
        _event['data'] = XmlJsonSerializer.decode(_event['data'])
        pass

    @staticmethod
    def parse_Error(_event):
        if _event['queryID'] == 'ReadAllXml':  # ошибка сериализатора
            raise Exception(_event['data'])
        else:
            _event['data'] = XmlJsonSerializer.decode(_event['data'])
            if _event['data']['detail'] == '#ASYNC':
                _event['queryID'] = -1
        pass

    @staticmethod
    def parse_Event(_event):
        _event['data'] = XmlJsonSerializer.decode(_event['data'])
        query_id = f'Event_{_event["data"]["eventName"]}'

        if query_id == 'Event_extsdk2.async.complete' \
                or query_id == 'Event_extsdk2.async.progress' \
                or query_id == 'Event_extsdk2.async.error':
            _event['data'] = _event['data']['data']
            _event['queryID'] = _event['data']['QueryId']

            if query_id == 'Event_extsdk2.async.error':
                _event['type'] = 'Error'
                _event['data'] = _event['data']['Error']
        else:
            _event['queryID'] = query_id
        pass

    @staticmethod
    def pars_event(_event):
        if _event['type'] == 'Message' or _event['type'] == 'Event':
            return _event['data']
        elif _event['type'] == 'Error':
            return _event['data']
            # raise ExtException(_event['data'])
        return _event

    def read_all_xml(self, requests):
        '''
        Чтение и обработка событий сконвертированных в json_xml
        :param requests: массив идентификаторов запросов, на которые мы ждем ответ.
        :return:
        '''

        supported_events = {
            'Message': self.parse_Message,
            'Error': self.parse_Error,
            'Event': self.parse_Event
        }
        result = []
        while len(result) == 0:
            xml_string = self.ole.ReadAllXml2()
            events = XmlJsonSerializer.decode(xml_string)
            for event in events:
                event_type = event.get('type')
                if event_type not in supported_events:
                    raise Exception(f'Warning not supported event {event_type}')
                supported_events[event_type](event)

                try:
                    query_id = requests.index(event['queryID'])
                    if query_id >= 0:
                        requests.pop(query_id)
                        result.append(self.pars_event(event))
                except ValueError:
                    pass
        return result

Выше представлен пример реализации получения ответов от COM-объекта плагина в функции "read_all_xml", которая постоянно опрашивает плагин на наличие ответов через вызов функции self.ole.ReadAllXml2(). "В качестве аргумента "requests" передается массив с query_id, вызываемых методов в плагине, на которые мы хотим получить ответ. ReadAllXml2" непосредственно функция COM-объекта, возвращающая массив событий в виде xml строки. Пришедшая строка сериализуется в массив Python по событиям "events = XmlJsonSerializer.decode(xml_string)" и разбирается по типам событий в функциях "parse_Message", "parse_Error", "parse_Event" соответственно. На выходе получаем массив ответов от плагина по интересующим нас методам.

Пример подключения к ExtSdk2 через OLE

    @classmethod
    def setUpClass(cls):
        cls.plugin = OleSbis3Plugin()
        cls.plugin.host = online.sbis.ru
        cls.plugin.get_module("ExtSdk2")

Создаем новое подключение при инициализации объекта OleSbis3Plugin(), в котором описаны все вышеупомянутые методы. Устанавливаем хост и получаем guid интересующего нас модуля. На этом подключение к ExtSdk2 завершается, можно переходить к авторизации.