code.delx.au - pymsnt/blob - src/utils.py

   1 # Copyright 2004 James Bunton <james@delx.cjb.net>
   2 # Licensed for distribution under the GPL version 2, check COPYING for details
   3
   4 def fudgestr(text, num):
   5         if(not (text.__class__ in [str, unicode])): return ""
   6         newtext = ""
   7         for c in text:
   8                 i = ord(c)
   9                 if(i >= num):
  10                         i = ord(' ')
  11                 newtext += chr(i)
  12         return newtext
  13
  14 def latin1(text):
  15         return fudgestr(text, 128)
  16
  17
  18 def copyDict(dic):
  19         """ Does a deep copy of a dictionary """
  20         out = {}
  21         for key in dic.keys():
  22                 out[key] = dic[key]
  23         return out
  24
  25 def copyList(lst):
  26         """ Does a deep copy of a list """
  27         out = []
  28         for i in lst:
  29                 out.append(i)
  30         return out
  31
  32 def mutilateMe(me):
  33         """ Mutilates a class :) """
  34 #       for key in dir(me):
  35 #               exec "me." + key + " = None"
  36
  37 def getLang(el):
  38         return el.getAttribute((u'http://www.w3.org/XML/1998/namespace', u'lang'))
  39
  40
  41 errorCodeMap = {
  42 "bad-request"                   :       400,
  43 "conflict"                      :       409,
  44 "feature-not-implemented"       :       501,
  45 "forbidden"                     :       403,
  46 "gone"                          :       302,
  47 "internal-server-error"         :       500,
  48 "item-not-found"                :       404,
  49 "jid-malformed"                 :       400,
  50 "not-acceptable"                :       406,
  51 "not-allowed"                   :       405,
  52 "not-authorized"                :       401,
  53 "payment-required"              :       402,
  54 "recipient-unavailable"         :       404,
  55 "redirect"                      :       302,
  56 "registration-required"         :       407,
  57 "remote-server-not-found"       :       404,
  58 "remote-server-timeout"         :       504,
  59 "resource-constraint"           :       500,
  60 "service-unavailable"           :       503,
  61 "subscription-required"         :       407,
  62 "undefined-condition"           :       500,
  63 "unexpected-request"            :       400
  64 }
  65
  66 def doPath(path):
  67         if(path and path[0] == "/"):
  68                 return path
  69         else:
  70                 return "../" + path
  71
  72
  73 def parseText(text):
  74         t = TextParser()
  75         t.parseString(text)
  76         return t.root
  77
  78 def parseFile(filename):
  79         t = TextParser()
  80         t.parseFile(filename)
  81         return t.root
  82
  83 class TextParser:
  84         """ Taken from http://xoomer.virgilio.it/dialtone/rsschannel.py """
  85
  86         def __init__(self):
  87                 self.root = None
  88
  89         def parseFile(self, filename):
  90                 return self.parseString(file(filename).read())
  91
  92         def parseString(self, data):
  93                 if(checkTwisted()):
  94                         from twisted.xish.domish import SuxElementStream
  95                 else:
  96                         from tlib.domish import SuxElementStream
  97                 es = SuxElementStream()
  98                 es.DocumentStartEvent = self.docStart
  99                 es.DocumentEndEvent = self.docEnd
 100                 es.ElementEvent = self.element
 101                 es.parse(data)
 102                 return self.root
 103
 104         def docStart(self, e):
 105                 self.root = e
 106
 107         def docEnd(self):
 108                 pass
 109
 110         def element(self, e):
 111                 self.root.addChild(e)
 112
 113
 114
 115 checkTwistedCached = None
 116 def checkTwisted():
 117         """ Returns False if we're using an old version that needs tlib, otherwise returns True """
 118         global checkTwistedCached
 119         if(checkTwistedCached == None):
 120                 import twisted.copyright
 121                 checkTwistedCached = (VersionNumber(twisted.copyright.version) >= VersionNumber("2.0.0"))
 122         return checkTwistedCached
 123
 124 class VersionNumber:
 125         def __init__(self, vstring):
 126                 self.varray = [0]
 127                 index = 0
 128                 flag = True
 129                 for c in vstring:
 130                         if(c == '.'):
 131                                 self.varray.append(0)
 132                                 index += 1
 133                                 flag = True
 134                         elif(c.isdigit() and flag):
 135                                 self.varray[index] *= 10
 136                                 self.varray[index] += int(c)
 137                         else:
 138                                 flag = False
 139
 140         def __cmp__(self, other):
 141                 i = 0
 142                 while(True):
 143                         if(i == len(other.varray)):
 144                                 if(i < len(self.varray)):
 145                                         return 1
 146                                 else:
 147                                         return 0
 148                         if(i == len(self.varray)):
 149                                 if(i < len(other.varray)):
 150                                         return -1
 151                                 else:
 152                                         return 0
 153
 154                         if(self.varray[i] > other.varray[i]):
 155                                 return 1
 156                         elif(self.varray[i] < other.varray[i]):
 157                                 return -1
 158
 159                         i += 1
 160
 161
 162
 163 class RollingStack:
 164         def __init__(self, size):
 165                 self.lst = []
 166                 self.size = size
 167
 168         def push(self, data):
 169                 self.lst.append(str(data))
 170                 if(len(self.lst) > self.size):
 171                         self.lst.remove(self.lst[0])
 172
 173         def grabAll(self):
 174                 return "".join(self.lst)
 175
 176         def flush(self):
 177                 self.lst = []
 178
 179