import re import sys import os import importlib import importlib.util import site from calculate.variables.old_vars.datavars import Variable, Namespace,\ HashVariable, TableVariable,\ IniCreated, DefaultValue from calculate.utils.gentoo import ProfileWalker from calculate.utils.fs import readFile from calculate.utils.files import list_directory from pyparsing import Literal, Word, ZeroOrMore, Group, Dict, Optional,\ restOfLine, empty, printables, OneOrMore, oneOf, nums,\ lineno, line, col, Keyword, SkipTo, LineEnd, Combine from enum import Enum class Define(Enum): assign = 0 append = 1 remove = 2 class CalculateIniParser: '''Класс парсера calculate.ini файлов.''' def __init__(self): self._errors = [] self.operations = {"=": Define.assign, "+=": Define.append, "-=": Define.remove} lbrack = Literal("[") rbrack = Literal("]") # comma = Literal(",").suppress() comment_symbol = Literal(';') | Literal('#') # Define = self.Define value_operation = (Literal("=") | Combine(Literal("+") + Literal("=")) | Combine(Literal("-") + Literal("="))) comment = comment_symbol + Optional(restOfLine) section_name = Word(printables+'\t', excludeChars='[]') value_name = Word(printables+'\t', excludeChars='=-+') # non_comma = Word(printables+'\t', excludeChars=',') clear_section = lbrack.suppress() + Group(empty) + rbrack.suppress() section_start = Group(OneOrMore(lbrack.suppress() + section_name + rbrack.suppress()) + (clear_section | ~lbrack()) + LineEnd().suppress()) # Если содержимое ini-файла не предваряется заголовком секции, # значит эта строка ошибочна. unexpected = Group(~section_start + SkipTo(LineEnd(), include=True))("error") unexpected.setParseAction(self._unexpected_token) key_value = (~lbrack + value_name + value_operation + empty + restOfLine + LineEnd().suppress()) def strip_key_value(tokens): tokens[0] = tokens[0].strip() tokens[1] = tokens[1].strip() key_value.setParseAction(strip_key_value) self.ini_section_parser = (section_start + Group(ZeroOrMore( Group(key_value | unexpected))) | unexpected) self.ini_section_parser.ignore(comment) def _unexpected_token(self, string, location, tokens): '''Метод вызываемый парсером, если обнаружена некорректная строка, предназначен для получения некорректной строки и ее дальнейшего разбора.''' print('UNEXPECTED TOKEN') error_line = line(location, string).strip() if error_line: self._errors.append((error_line, lineno(location, string), col(location, string))) @property def errors(self): errors = self._errors self._errors = [] return errors def parse(self, data): for tokens, start, end in self.ini_section_parser.scanString(data): if tokens.getName() == "error": continue section, defkeys = tokens section_list = section.asList() if section_list[-1] == []: yield {'clear_section': (section_list[:-1], )} else: yield {'start_section': (section.asList(), )} for defkey in defkeys: if defkey.getName() == "error": continue yield {'define_key': (section.asList(), defkey[0], defkey[2], self.operations[defkey[1]])} class NamespaceIniFiller: '''Класс, предназначенный для наполнения Namespace объекта переменными из calculate.ini файла.''' def __init__(self): self.ini_parser = CalculateIniParser() def error(self, lineno, error_message): self.errors.append(lineno, error_message) def fill(self, namespace, ini_file_text): self.namespace = namespace self.current_namespace = self.namespace self.errors = [] for parsed_line in self.ini_parser.parse(ini_file_text): self._line_processor(**parsed_line) def _line_processor(self, start_section=None, clear_section=None, define_key=None, error=None, **kwargs): if start_section is not None: self.start_section(*start_section) elif clear_section is not None: self.clear_section(*clear_section) elif define_key is not None: self.define_key(*define_key) for error in self.ini_parser.errors: self.set_error(*error) def start_section(self, sections): self.current_namespace = self.namespace for section in sections: if section not in self.current_namespace.childs: self.current_namespace.add_namespace(Namespace(section)) self.current_namespace = self.current_namespace[section] def clear_section(self, sections): current_namespace = self.namespace for section in sections: if section not in current_namespace.childs: return current_namespace = current_namespace[section] current_namespace.clear_childs() def change_value(self, key, value): self.current_namespace[key].set_value(value) def define_variable(self, key, value): self.current_namespace.add_string_variable(key, value) def append_value(self, key, value): line = self.current_namespace[key].get_value().split(",") vlist = value.split(",") for v in vlist: if v not in line: line.append(v) self.change_value(key, ",".join(line)) def remove_value(self, key, value): line = self.current_namespace[key].get_value().split(",") vlist = value.split(",") for v in vlist: if v in line: line.remove(v) self.change_value(key, ",".join(line)) def define_key(self, section, key, value, optype): if optype == Define.assign: if key not in self.current_namespace: self.define_variable(key, value) else: self.change_value(key, value) elif optype == Define.append: if key not in self.current_namespace: self.define_variable(key, value) else: self.append_value(key, value) elif optype == Define.remove: if key not in self.current_namespace: self.define_variable(key, value) else: self.remove_value(key, value) def set_error(self, line, lineno, col): self.error(lineno, "Syntax error: %s".format(line)) class NamespaceIniFillerStrict(NamespaceIniFiller): """ Объект используемый для наполения Namespace объекта переменными из calculate.ini файла, с возможность определять только """ availableSection = ["custom"] def fill(self, namespace, data): self.canCreate = False for newns in self.availableSection: if newns not in namespace: namespace.add_namespace(Namespace(newns)) super().fill(namespace, data) def start_section(self, sections): self.current_namespace = self.namespace self.canCreate = sections[0] in self.availableSection for section in sections: if section not in self.current_namespace.childs: if isinstance(self.current_namespace, TableVariable) or self.canCreate: self.current_namespace.add_namespace(Namespace(section)) else: self.current_namespace = None self.current_namespace = self.current_namespace[section] # def clear_section(self, sections): # current_namespace = self.namespace # self.canCreate = sections[0] in self.availableSection # for section in sections: # if section not in current_namespace.childs: # return # current_namespace = current_namespace[section] # current_namespace.clear_childs() def define_variable(self, key, value): if not self.canCreate: pass var = Variable(key) var.add_property(IniCreated(value)) self.current_namespace.add_variable(var) def change_value(self, key, value): var = self.current_namespace[key] var.set_value(value) if isinstance(var, HashVariable.HashValue): var = var.master_variable value = var.get_value() prop = var.find_property(DefaultValue) if prop: prop.value = value else: var.add_property(DefaultValue(value)) var.invalidate() class VariableLoader: '''Класс, используемый для загрузки переменных из python модуля.''' re_upper = re.compile("(.)([A-Z])") def _get_varlike_attrs(self, obj): '''Метод для получения списка аттрибутов похожих на переменные.''' for attrname in (x for x in dir(obj) if x[:1].isupper()): yield self.re_upper.sub(r"\1_\2", attrname).lower(), \ getattr(obj, attrname) def fill(self, namespace, dirpath, package): '''Загрузить в namespace переменные из указанных модулей.''' for fullfn in list_directory(dirpath, fullpath=True): dn, fn = os.path.split(fullfn) if os.path.isdir(fullfn): newns = namespace.add_namespace(Namespace(fn)) self.fill(newns, fullfn, "{}.{}".format(package, fn)) elif fn.endswith(".py"): module = self._load_module_source(package, fn, fullfn) for varname, cls in self._get_varlike_attrs(module): if Variable.is_implementation(cls): namespace.add_variable(cls(varname)) elif HashVariable.is_implementation(cls) or \ TableVariable.is_implementation(cls) or \ Namespace.is_implementation(cls): _newns = namespace.add_namespace(cls(varname, namespace)) for _varname, _cls in self._get_varlike_attrs(cls): if Variable.is_implementation(_cls): _newns.add_variable(_cls(_varname)) @classmethod def default(cls): site_dirs = [os.path.normpath(x) for x in site.getsitepackages()] for site_dir in site_dirs: calculate_dir = os.path.join(site_dir, "calculate/vars") if os.path.exists(calculate_dir): return (calculate_dir, "calculate.vars") def _load_module_source(self, package, name, path): if name.startswith('calculate.vars.'): full_name = name else: full_name = '.'.join([package, name]) if full_name in sys.modules: return sys.modules[full_name] spec = importlib.util.spec_from_file_location(full_name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) sys.modules[full_name] = module return module class ProfileFiller: """ Заполнитель значений переменных из файлов calculate.ini в профилях """ basename = "calculate.ini" def get_repository_map(self, namespace): return { x.name.get_value(): x.path.get_value() for x in namespace.os.gentoo.repositories } def fill(self, namespace, profile_path): nif = NamespaceIniFillerStrict() pw = ProfileWalker(self.basename, self.get_repository_map(namespace)) for fn in pw.find(profile_path): nif.fill(namespace, readFile(fn))