bin/ntools.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import division
   3
   4 import os
   5 import calendar
   6
   7 __all__ = [
   8     'IPV4_IN_IPV6_PREFIX',
   9     'datetime_to_timestamp',
  10     'read_cfg',
  11     'clean_ais_charset',
  12     'clean_alnum',
  13     'clean_alnum_unicode',
  14     'open_with_mkdirs',
  15     'logliner',
  16     'dumpsource',
  17     'xml_escape',
  18     'alarm',
  19     'str_split_column_ipv6',
  20     'formataddr',
  21     ]
  22
  23 IPV4_IN_IPV6_PREFIX = '::ffff:'
  24
  25 def datetime_to_timestamp(dt):
  26     return calendar.timegm(dt.utctimetuple())
  27
  28 def read_cfg(filename):
  29     '''
  30     Function that reads a file in the form
  31     key=value
  32     and returns the resulting dictionary
  33     '''
  34     cfg = {}
  35     for line in file(filename).readlines():
  36         line = line.rstrip('\r\n\0')
  37         line = unicode(line, 'utf-8')
  38         if line.startswith(u'#'):
  39             continue # skip comments
  40         spl = line.split(u'=', 1)
  41         if len(spl) == 2:
  42             cfg[spl[0]] = spl[1]
  43         else:
  44             cfg[spl[0]] = None
  45     return cfg
  46
  47
  48 def clean_ais_charset(txt):
  49     assert isinstance(txt, str)
  50     result = ''
  51     for c in txt:
  52         oc = ord(c)
  53         if oc < 32 or oc > 95:
  54             result += ''
  55         else:
  56             result += c
  57     return result
  58
  59 def clean_alnum(txt):
  60     assert isinstance(txt, str)
  61     result = ''
  62     for c in txt:
  63         if ( c >= '0' and c <= '9' ) or ( c >= 'A' and c <= 'Z' ):
  64             result += c
  65     return result
  66
  67 def clean_alnum_unicode(txt):
  68     assert isinstance(txt, unicode)
  69     return unicode(clean_alnum(txt.encode('ascii7', 'replace')))
  70
  71
  72 def open_with_mkdirs(filename, mode):
  73     try:
  74         return file(filename, mode)
  75     except IOError, ioerr:
  76         # FIXME only if doesn't exists ...
  77         #print 'Creating directory', os.path.dirname(filename)
  78         os.makedirs(os.path.dirname(filename))
  79         return file(filename, mode)
  80
  81
  82 # log file source
  83 def logliner(filename):
  84     for line in file(filename).readlines():
  85         yield line
  86
  87
  88 # debug/display wraper source
  89 def dumpsource(source):
  90     for line in source:
  91         while line and line[-1] in '\r\n\0':
  92             line = line[:-1]
  93         print "INPUT", line
  94         yield line
  95
  96 def xml_escape(txt):
  97     return txt.replace(u'&', u'&amp;').replace(u'<', u'&lt;')
  98
  99 def alarm():
 100     os.system('touch /home/nirgal/kod/ais/alarm &')
 101
 102
 103 def str_split_column_ipv6(txt):
 104     '''
 105     Helper function that will split a column separated string of tokens.
 106     It will take care not to ignore : in [] for ipv6 support.
 107     '''
 108     res = []
 109     in_bracket = False
 110     iprev = None
 111     for i in range(len(txt)):
 112         if not in_bracket:
 113             if txt[i] == ':':
 114                 res.append(txt[iprev:i])
 115                 iprev = i+1
 116             elif txt[i] == '[':
 117                 in_bracket = True
 118         else: # in bracket
 119             if txt[i] == ']':
 120                 in_bracket = False
 121     res.append(txt[iprev:])
 122     return res
 123
 124
 125 def formataddr(addr):
 126     if addr.startswith(IPV4_IN_IPV6_PREFIX):
 127         return addr[7:]
 128     elif ':' in addr:
 129         return '['+addr+']'
 130     else:
 131         return addr
 132
 133
 134 #if __name__ == '__main__':
 135 #    for test in ('12:34:56:78',
 136 #                 ':12::56:',
 137 #                 '1:2:3:abc:[bd:ef::]:45',
 138 #                 '1:2:3:abc:[bd:ef:]:]:45'):
 139 #        print test, str_split_column_ipv6(test)