mirror of
ssh://git.janware.com/srv/git/janware/proj/jw-python
synced 2026-01-15 18:03:31 +01:00
125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
|
|
import re
|
||
|
|
import json
|
||
|
|
from collections import OrderedDict
|
||
|
|
from . import log
|
||
|
|
import shlex
|
||
|
|
|
||
|
|
class Options: # export
|
||
|
|
|
||
|
|
def __parse_json(self, spec):
|
||
|
|
spec = spec.strip()
|
||
|
|
if len(spec) < 3:
|
||
|
|
return None
|
||
|
|
if spec[0] != '{':
|
||
|
|
spec = '{' + spec + '}'
|
||
|
|
try:
|
||
|
|
return json.loads(spec, object_pairs_hook=OrderedDict)
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
return None
|
||
|
|
|
||
|
|
def __parse(self, opts_str):
|
||
|
|
r = self.__parse_json(opts_str)
|
||
|
|
if r is not None:
|
||
|
|
return r
|
||
|
|
r = OrderedDict()
|
||
|
|
opt_strs = shlex.split(opts_str)
|
||
|
|
for opt_str in opt_strs:
|
||
|
|
opt_str = re.sub('\s*=\s*', '=', opt_str)
|
||
|
|
sides = opt_str.split('=')
|
||
|
|
lhs = sides[0].strip()
|
||
|
|
if not len(lhs):
|
||
|
|
continue
|
||
|
|
if self.__allowed_keys and not lhs in self.__allowed_keys:
|
||
|
|
raise Exception('Field "{}" not supported'.format(lhs))
|
||
|
|
rhs = ' '.join(sides[1:]).strip() if len(sides) > 1 else self.__true_val
|
||
|
|
r[lhs] = rhs
|
||
|
|
return r
|
||
|
|
|
||
|
|
def __recache(self):
|
||
|
|
self.__list.clear()
|
||
|
|
self.__list = list(self.__dict)
|
||
|
|
self.__str = str(dict(self.__dict))
|
||
|
|
|
||
|
|
def __getitem__(self, key):
|
||
|
|
if not key in self.__dict.keys():
|
||
|
|
return None
|
||
|
|
return self.__dict[key]
|
||
|
|
|
||
|
|
def __str__(self):
|
||
|
|
return self.__str
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
return self.__str
|
||
|
|
|
||
|
|
def __format__(self, fmt):
|
||
|
|
return self.__str
|
||
|
|
|
||
|
|
def __len__(self):
|
||
|
|
return len(self.__list)
|
||
|
|
|
||
|
|
def __contains__(self, keys):
|
||
|
|
if not type(keys) in [list, set]:
|
||
|
|
return keys in self.__dict.keys()
|
||
|
|
for key in keys:
|
||
|
|
if not key in self.__dict.keys():
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
def __iter__(self):
|
||
|
|
return iter(self.__list)
|
||
|
|
|
||
|
|
def __next__(self):
|
||
|
|
return next(self.__list)
|
||
|
|
|
||
|
|
def __init__(self, spec=None, delimiter=',', allowed_keys=None, true_val=True):
|
||
|
|
self.__true_val = true_val
|
||
|
|
self.__allowed_keys = None
|
||
|
|
self.__delimiter = delimiter
|
||
|
|
self.__spec = spec
|
||
|
|
self.__dict = OrderedDict() if spec is None else self.__parse(spec)
|
||
|
|
self.__list = []
|
||
|
|
self.__str = None
|
||
|
|
self.__recache()
|
||
|
|
|
||
|
|
def dump(self, prio):
|
||
|
|
caller = log.get_caller_pos()
|
||
|
|
for key, val in self.__dict.items():
|
||
|
|
log.slog(prio, "{}=\"{}\"".format(key, val))
|
||
|
|
|
||
|
|
def keys(self):
|
||
|
|
return self.__dict.keys()
|
||
|
|
|
||
|
|
def items(self):
|
||
|
|
return self.__dict.items()
|
||
|
|
|
||
|
|
def get(self, key, default=None, by_index=False):
|
||
|
|
if by_index:
|
||
|
|
if type(key) != int:
|
||
|
|
raise KeyError('Tried to get value from options string with ' +
|
||
|
|
'index {} of type "{}": {}'.format(key, type(key), self.__spec))
|
||
|
|
if key >= len(self.__list):
|
||
|
|
if default is not None:
|
||
|
|
return default
|
||
|
|
raise KeyError('Tried to get value from options string with ' +
|
||
|
|
'index {} of {}: {}'.format(key, len(self.__list), self.__spec))
|
||
|
|
return self.__list[key]
|
||
|
|
if key in self.__dict.keys():
|
||
|
|
return self.__dict[key]
|
||
|
|
if default is not None:
|
||
|
|
return default
|
||
|
|
raise KeyError('Key "{}" is not present in options string: {}'.format(key, self.__spec))
|
||
|
|
|
||
|
|
def update(self, rhs):
|
||
|
|
if hasattr(rhs, 'items'):
|
||
|
|
for key, val in rhs.items():
|
||
|
|
self.__dict[key] = val
|
||
|
|
return
|
||
|
|
if isinstance(rhs, str):
|
||
|
|
self.update(self.__parse(rhs))
|
||
|
|
return
|
||
|
|
raise Exception('Tried to update options with object of incompatible type {}'.format(type(rhs)))
|
||
|
|
|
||
|
|
def append_to(self, obj):
|
||
|
|
for opt in self.__list:
|
||
|
|
setattr(obj, opt[0], opt[1])
|