vcs.maemo.org Git - dbuscron/blob - dbuscron/parser.py

   1 # encoding: utf-8
   2 from __future__ import with_statement
   3 import re
   4 from dbuscron.bus import DbusBus
   5
   6 def unescape_():
   7     h = '[0-9A-Fa-f]'
   8     r = re.compile(r'\\x('+h+r'{2})|\\u('+h+'{4})')
   9     def unescape(value):
  10         if not (value and \
  11             (r'\x' in value or r'\u' in value)):
  12             return value
  13
  14         return r.sub(\
  15             lambda m: chr(int(m.group(1), 16)) \
  16                 if m.group(1) is not None else \
  17                     unichr(int(m.group(2), 16))\
  18                         .encode('utf-8'),\
  19             value)
  20     return unescape
  21 unescape = unescape_()
  22
  23 def product(*args):
  24     if args:
  25         head, tail = args[0], args[1:]
  26         for h in head:
  27             for t in product(*tail):
  28                 yield (h,) + t
  29
  30     else:
  31         yield ()
  32
  33 class CrontabParserError(SyntaxError):
  34     def __init__(self, message, lineno, expected=None):
  35         if expected:
  36             if isinstance(expected, (tuple, list)):
  37                 exp = ' (expected %s or %s)' % (', '.join(expected[:-1]), expected[-1])
  38         else:
  39             exp = ''
  40
  41         msg = '%s%s at line %d' % (message, exp, lineno)
  42
  43         SyntaxError.__init__(self, msg)
  44
  45 class CrontabParser(object):
  46     __fields_sep = re.compile(r'\s+')
  47     __envvar_sep = re.compile(r'\s*=\s*')
  48     __fields_chk = {
  49             'bus_'         : None,
  50             'type_'        : ('signal','method_call','method_return','error'),
  51             'sender_'      : None,
  52             'interface_'   : re.compile(r'^[a-zA-Z][a-zA-Z0-9_.]+$'),
  53             'path_'        : re.compile(r'^/[a-zA-Z0-9_/]+$'),
  54             'member_'      : re.compile(r'^[a-zA-Z][a-zA-Z0-9_]+$'),
  55             'destination_' : None,
  56             'args_'        : None,
  57             }
  58     __fields = [
  59             'bus_',
  60             'type_',
  61             'sender_',
  62             'interface_',
  63             'path_',
  64             'member_',
  65             'destination_',
  66             'args_',
  67             ]
  68
  69     def __init__(self, fname):
  70         self.__bus = DbusBus()
  71         self.__filename = fname
  72         self.__environ = dict()
  73
  74     @property
  75     def environ(self):
  76         return self.__environ
  77
  78     def __iter__(self):
  79         # bus type sender interface path member destination args command
  80         lineno = 0
  81         with open(self.__filename) as f:
  82             for line in f:
  83                 lineno += 1
  84                 line = line.strip()
  85
  86                 if not line or line.startswith('#'):
  87                     continue
  88
  89                 parts = self.__fields_sep.split(line, 8)
  90                 if len(parts) < 9:
  91                     parts = self.__envvar_sep.split(line, 1)
  92                     if len(parts) == 2:
  93                         self.__environ[parts[0]] = parts[1]
  94                         continue
  95
  96                     raise CrontabParserError('Unexpected number of records', lineno)
  97
  98                 rule = [('s','S'), self.__fields_chk['type_'], (None,), (None,), (None,), (None,), (None,), (None,)]
  99
 100                 for p in range(0, 8):
 101                     if parts[p] != '*':
 102                         rule[p] = parts[p].split(',')
 103
 104                 command = parts[8]
 105
 106                 for r in product(*rule):
 107                     r = list(r)
 108                     if r[0] == 'S':
 109                         r[0] = self.__bus.system
 110                     elif r[0] == 's':
 111                         r[0] = self.__bus.session
 112                     else:
 113                         raise CrontabParserError('Unexpected bus value', lineno, expected=('S', 's', '*'))
 114
 115                     if r[7]:
 116                         r[7] = map(unescape, r[7].split(';'))
 117
 118                     ruled = dict()
 119                     for i, f in enumerate(self.__fields):
 120                         if r[i] is not None and self.__fields_chk[f]:
 121                             if isinstance(self.__fields_chk[f], tuple):
 122                                 if r[i] not in self.__fields_chk[f]:
 123                                     raise CrontabParserError('Unexpected %s value' % (f.strip('_')), lineno,
 124                                             expected=self.__fields_chk[f])
 125                             else:
 126                                 if not self.__fields_chk[f].match(r[i]):
 127                                     raise CrontabParserError('Incorrect %s value' % (f.strip('_')), lineno)
 128                         ruled[f] = r[i]
 129
 130                     yield ruled, command
 131
 132 def OptionsParser(args=None, help=u'', **opts):
 133
 134     from optparse import OptionParser
 135     import dbuscron
 136     parser = OptionParser(usage=help, version="%prog "+dbuscron.__version__)
 137     for opt, desc in opts.iteritems():
 138         names = desc.pop('names')
 139         desc['dest'] = opt
 140         parser.add_option(*names, **desc)
 141
 142     return parser.parse_args(args)[0]
 143