calculate-utils-4-lib/calculate/templates/format/xml_xfce_format.py

# vim: fileencoding=utf-8
#
from .base_format import BaseFormat
from pyparsing import originalTextFor, Literal, Word, printables, OneOrMore,\
                      Optional
from collections import OrderedDict
try:
    from lxml.etree import Element, SubElement, ElementTree, tostring
except ImportError:
    from xml.etree.ElementTree import Element, SubElement, ElementTree, \
                                      tostring


class XMLXfceFormat(BaseFormat):
    FORMAT = 'xml_xfce'

    _initialized = False

    def __new__(cls, *args, **kwargs):
        if not cls._initialized:
            cls._initialize_parser()
        return super().__new__(cls)

    def __init__(self, document_text: str, ignore_comments=False):
        processing_methods = OrderedDict({'channel': self._channel,
                                          'property': self._property,
                                          'value': self._value,
                                          'unknown': self._unknown})
        super().__init__(processing_methods)
        self._initialize_parser()

        if document_text == '':
            self._document_dictionary = OrderedDict()
        else:
            self._parse_xml_to_dictionary(document_text)

    @classmethod
    def _initialize_parser(cls):
        '''Метод для инициализации парсерa.'''
        action_symbols = (Literal('!') | Literal('-'))

        name = originalTextFor(OneOrMore(Word(printables)))

        cls._node_name = Optional(action_symbols)('action') + name('name')

        # Кортежи с названиями атрибутов различных элементов.
        cls._ELEMENT_ATTRIBUTES = ('tag', 'name', 'type', 'value')
        cls._CHANNEL_ATTRIBUTES = ('tag', 'name', 'version')
        cls._VALUE_ATTRIBUTES = ('tag', 'type', 'value')

    def _property(self, xml_element):
        '''Метод для парсинга тега property.'''
        try:
            parsing_result = self._node_name.parseString(
                                                     xml_element.attrib['name']
                                                        )
            element_name = (parsing_result.action,
                            xml_element.tag,
                            parsing_result.name,
                            xml_element.attrib['type'])
        except Exception:
            # Какая-то обработка ошибки.
            return OrderedDict()

        if xml_element.attrib['type'] == 'empty':
            output = OrderedDict()
            for child in xml_element:
                child_value = self._processing_methods.get(child.tag,
                                                           self._unknown
                                                           )(child)
                output.update(child_value)

        elif xml_element.attrib['type'] == 'array':
            output = []
            for child in xml_element:
                child_value = self._processing_methods.get(child.tag,
                                                           self._unknown
                                                           )(child)
                output.append(child_value)
        else:
            try:
                output = xml_element.attrib['value']
            except KeyError:
                # Какая-то обработка ошибки.
                return OrderedDict()

        return OrderedDict({element_name: output})

    def _value(self, xml_element):
        '''Метод для парсинга тега value.'''
        try:
            value = (xml_element.tag,
                     xml_element.attrib['type'],
                     xml_element.attrib['value'])
        except KeyError:
            # Какая-то обработка ошибки.
            return ('',)
        return value

    def _channel(self, xml_element):
        '''Метод для парсинга тега channel.'''
        output_dictionary = OrderedDict()
        try:
            parsing_result = self._node_name.parseString(
                                                     xml_element.attrib['name']
                                                        )
            element_name = (parsing_result.action,
                            xml_element.tag,
                            parsing_result.name,
                            xml_element.attrib['version'])
        except Exception:
            # Какая-то обработка ошибки.
            return OrderedDict()

        for child in xml_element:
            item_to_add = self._processing_methods.get(child.tag,
                                                       self._unknown)(child)
            output_dictionary.update(item_to_add)

        return OrderedDict({element_name: output_dictionary})

    def _unknown(self, xml_element):
        # Действия если элемент неизвестен.
        element_name = ('', xml_element.tag)
        return OrderedDict({element_name: 'Unknown element'})

    @property
    def document_text(self):
        '''Метод для получения исходного текста документа. Использует
        рекурсивный метод _build_section.'''
        channel = next(iter(self._document_dictionary.keys()))
        channel_head = OrderedDict(
                                   {key: value for key, value in
                                    zip(self._CHANNEL_ATTRIBUTES, channel[1:])}
                                  )

        root = Element(channel_head.pop('tag'), **channel_head)

        self._build_section(root, self._document_dictionary[channel])
        document_tree = ElementTree(root)
        xml_document = tostring(document_tree,
                                encoding="UTF-8",
                                xml_declaration=True,
                                pretty_print=True).decode()

        return xml_document

    def _build_section(self, current_element, dictionary):
        '''Метод для перевода словаря xml-документа обратно в текст документа.
        Для этого рекурсивно строит дерево xml-документа пригодное работы lxml.
        '''
        for dict_element in dictionary.keys():
            element_head = OrderedDict({key: value for key, value in
                                        zip(self._ELEMENT_ATTRIBUTES,
                                            dict_element[1:])})

            if element_head['type'] == 'empty':
                include_element = SubElement(current_element,
                                             element_head.pop('tag'),
                                             **element_head)
                self._build_section(include_element,
                                    dictionary[dict_element])

            elif element_head['type'] == 'array':
                include_element = SubElement(current_element,
                                             element_head.pop('tag'),
                                             **element_head)

                for list_element in dictionary[dict_element]:
                    list_element_head = OrderedDict(
                                                  {key: value for key, value in
                                                   zip(self._VALUE_ATTRIBUTES,
                                                       list_element)}
                                                   )
                    SubElement(include_element,
                               list_element_head.pop('tag'),
                               **list_element_head)
            else:
                SubElement(current_element,
                           element_head.pop('tag'),
                           **element_head,
                           value=dictionary[dict_element])