|
| 1 | +""" |
| 2 | + Parse INI files. |
| 3 | +
|
| 4 | +""" |
| 5 | +from __future__ import unicode_literals, print_function, absolute_import |
| 6 | + |
| 7 | +import io |
| 8 | +import re |
| 9 | +import logging |
| 10 | +from collections import MutableMapping |
| 11 | +try: |
| 12 | + from collections import OrderedDict |
| 13 | +except ImportError as e: |
| 14 | + from ordereddict import OrderedDict |
| 15 | + |
| 16 | + |
| 17 | +__version__ = '0.2.9' |
| 18 | +__project__ = 'Inirama' |
| 19 | +__author__ = "Kirill Klenov <horneds@gmail.com>" |
| 20 | +__license__ = "BSD" |
| 21 | + |
| 22 | + |
| 23 | +NS_LOGGER = logging.getLogger('inirama') |
| 24 | + |
| 25 | + |
| 26 | +class Scanner(object): |
| 27 | + |
| 28 | + def __init__(self, source, ignore=None, patterns=None): |
| 29 | + """ Init Scanner instance. |
| 30 | +
|
| 31 | + :param patterns: List of token patterns [(token, regexp)] |
| 32 | + :param ignore: List of ignored tokens |
| 33 | + """ |
| 34 | + self.reset(source) |
| 35 | + if patterns: |
| 36 | + self.patterns = [] |
| 37 | + for k, r in patterns: |
| 38 | + self.patterns.append((k, re.compile(r))) |
| 39 | + |
| 40 | + if ignore: |
| 41 | + self.ignore = ignore |
| 42 | + |
| 43 | + def reset(self, source): |
| 44 | + """ Reset scanner. |
| 45 | +
|
| 46 | + :param source: Source for parsing |
| 47 | + """ |
| 48 | + self.tokens = [] |
| 49 | + self.source = source |
| 50 | + self.pos = 0 |
| 51 | + |
| 52 | + def scan(self): |
| 53 | + """ Scan source and grab tokens. |
| 54 | + """ |
| 55 | + self.pre_scan() |
| 56 | + |
| 57 | + token = None |
| 58 | + end = len(self.source) |
| 59 | + |
| 60 | + while self.pos < end: |
| 61 | + |
| 62 | + best_pat = None |
| 63 | + best_pat_len = 0 |
| 64 | + |
| 65 | + # Check patterns |
| 66 | + for p, regexp in self.patterns: |
| 67 | + m = regexp.match(self.source, self.pos) |
| 68 | + if m: |
| 69 | + best_pat = p |
| 70 | + best_pat_len = len(m.group(0)) |
| 71 | + break |
| 72 | + |
| 73 | + if best_pat is None: |
| 74 | + raise SyntaxError( |
| 75 | + "SyntaxError[@char {0}: {1}]".format( |
| 76 | + self.pos, "Bad token.")) |
| 77 | + |
| 78 | + # Ignore patterns |
| 79 | + if best_pat in self.ignore: |
| 80 | + self.pos += best_pat_len |
| 81 | + continue |
| 82 | + |
| 83 | + # Create token |
| 84 | + token = ( |
| 85 | + best_pat, |
| 86 | + self.source[self.pos:self.pos + best_pat_len], |
| 87 | + self.pos, |
| 88 | + self.pos + best_pat_len, |
| 89 | + ) |
| 90 | + |
| 91 | + self.pos = token[-1] |
| 92 | + self.tokens.append(token) |
| 93 | + |
| 94 | + def pre_scan(self): |
| 95 | + """ Prepare source. |
| 96 | + """ |
| 97 | + pass |
| 98 | + |
| 99 | + def __repr__(self): |
| 100 | + """ Print the last 5 tokens that have been scanned in |
| 101 | + """ |
| 102 | + return '<Scanner: ' + ','.join( |
| 103 | + "{0}({2}:{3})".format(*t) for t in self.tokens[-5:]) + ">" |
| 104 | + |
| 105 | + |
| 106 | +class INIScanner(Scanner): |
| 107 | + patterns = [ |
| 108 | + ('SECTION', re.compile(r'\[[^]]+\]')), |
| 109 | + ('IGNORE', re.compile(r'[ \r\t\n]+')), |
| 110 | + ('COMMENT', re.compile(r'[;#].*')), |
| 111 | + ('KEY', re.compile(r'[\w_]+\s*[:=].*'))] |
| 112 | + |
| 113 | + ignore = ['IGNORE'] |
| 114 | + |
| 115 | + def pre_scan(self): |
| 116 | + escape_re = re.compile(r'\\\n[\t ]+') |
| 117 | + self.source = escape_re.sub('', self.source) |
| 118 | + |
| 119 | + |
| 120 | +undefined = object() |
| 121 | + |
| 122 | + |
| 123 | +class Section(MutableMapping): |
| 124 | + |
| 125 | + def __init__(self, namespace, *args, **kwargs): |
| 126 | + super(Section, self).__init__(*args, **kwargs) |
| 127 | + self.namespace = namespace |
| 128 | + self.__storage__ = dict() |
| 129 | + |
| 130 | + def __setitem__(self, name, value): |
| 131 | + self.__storage__[name] = str(value) |
| 132 | + |
| 133 | + def __getitem__(self, name): |
| 134 | + return self.__storage__[name] |
| 135 | + |
| 136 | + def __delitem__(self, name): |
| 137 | + del self.__storage__[name] |
| 138 | + |
| 139 | + def __len__(self): |
| 140 | + return len(self.__storage__) |
| 141 | + |
| 142 | + def __iter__(self): |
| 143 | + return iter(self.__storage__) |
| 144 | + |
| 145 | + def __repr__(self): |
| 146 | + return "<{0} {1}>".format(self.__class__.__name__, str(dict(self))) |
| 147 | + |
| 148 | + def iteritems(self): |
| 149 | + for key in self.__storage__.keys(): |
| 150 | + yield key, self[key] |
| 151 | + |
| 152 | + items = lambda s: list(s.iteritems()) |
| 153 | + |
| 154 | + |
| 155 | +class InterpolationSection(Section): |
| 156 | + |
| 157 | + var_re = re.compile('{([^}]+)}') |
| 158 | + |
| 159 | + def get(self, name, default=None): |
| 160 | + if name in self: |
| 161 | + return self[name] |
| 162 | + return default |
| 163 | + |
| 164 | + def __interpolate__(self, math): |
| 165 | + try: |
| 166 | + key = math.group(1).strip() |
| 167 | + return self.namespace.default.get(key) or self[key] |
| 168 | + except KeyError: |
| 169 | + return '' |
| 170 | + |
| 171 | + def __getitem__(self, name): |
| 172 | + value = super(InterpolationSection, self).__getitem__(name) |
| 173 | + sample = undefined |
| 174 | + while sample != value: |
| 175 | + try: |
| 176 | + sample, value = value, self.var_re.sub( |
| 177 | + self.__interpolate__, value) |
| 178 | + except RuntimeError: |
| 179 | + message = "Interpolation failed: {0}".format(name) |
| 180 | + NS_LOGGER.error(message) |
| 181 | + raise ValueError(message) |
| 182 | + return value |
| 183 | + |
| 184 | + |
| 185 | +class Namespace(object): |
| 186 | + |
| 187 | + default_section = 'DEFAULT' |
| 188 | + silent_read = True |
| 189 | + section_type = Section |
| 190 | + |
| 191 | + def __init__(self, **default_items): |
| 192 | + self.sections = OrderedDict() |
| 193 | + for k, v in default_items.items(): |
| 194 | + self[self.default_section][k] = v |
| 195 | + |
| 196 | + @property |
| 197 | + def default(self): |
| 198 | + """ Return default section or empty dict. |
| 199 | + """ |
| 200 | + return self.sections.get(self.default_section, dict()) |
| 201 | + |
| 202 | + def read(self, *files, **params): |
| 203 | + """ Read and parse INI files. |
| 204 | + """ |
| 205 | + for f in files: |
| 206 | + try: |
| 207 | + with io.open(f, encoding='utf-8') as ff: |
| 208 | + NS_LOGGER.info('Read from `{0}`'.format(ff.name)) |
| 209 | + self.parse(ff.read(), **params) |
| 210 | + except (IOError, TypeError, SyntaxError, io.UnsupportedOperation): |
| 211 | + if not self.silent_read: |
| 212 | + NS_LOGGER.error('Reading error `{0}`'.format(ff.name)) |
| 213 | + raise |
| 214 | + |
| 215 | + def write(self, f): |
| 216 | + """ |
| 217 | + Write self as INI file. |
| 218 | +
|
| 219 | + :param f: File object or path to file. |
| 220 | + """ |
| 221 | + if isinstance(f, str): |
| 222 | + f = io.open(f, 'w', encoding='utf-8') |
| 223 | + |
| 224 | + if not hasattr(f, 'read'): |
| 225 | + raise AttributeError("Wrong type of file: {0}".format(type(f))) |
| 226 | + |
| 227 | + NS_LOGGER.info('Write to `{0}`'.format(f.name)) |
| 228 | + for section in self.sections.keys(): |
| 229 | + f.write('[{0}]\n'.format(section)) |
| 230 | + for k, v in self[section].items(): |
| 231 | + f.write('{0:15}= {1}\n'.format(k, v)) |
| 232 | + f.write('\n') |
| 233 | + f.close() |
| 234 | + |
| 235 | + def parse(self, source, update=True, **params): |
| 236 | + """ Parse INI source. |
| 237 | + """ |
| 238 | + scanner = INIScanner(source) |
| 239 | + scanner.scan() |
| 240 | + |
| 241 | + section = self.default_section |
| 242 | + |
| 243 | + for token in scanner.tokens: |
| 244 | + if token[0] == 'KEY': |
| 245 | + name, value = re.split('[=:]', token[1], 1) |
| 246 | + name, value = name.strip(), value.strip() |
| 247 | + if not update and name in self[section]: |
| 248 | + continue |
| 249 | + self[section][name] = value |
| 250 | + |
| 251 | + elif token[0] == 'SECTION': |
| 252 | + section = token[1].strip('[]') |
| 253 | + |
| 254 | + def __getitem__(self, name): |
| 255 | + """ Look name in self sections. |
| 256 | + """ |
| 257 | + if not name in self.sections: |
| 258 | + self.sections[name] = self.section_type(self) |
| 259 | + return self.sections[name] |
| 260 | + |
| 261 | + def __repr__(self): |
| 262 | + return "<Namespace: {0}>".format(self.sections) |
| 263 | + |
| 264 | + |
| 265 | +class InterpolationNamespace(Namespace): |
| 266 | + |
| 267 | + section_type = InterpolationSection |
| 268 | + |
| 269 | +# lint_ignore=W0201,R0924,F0401 |
0 commit comments