forked from LBRYCommunity/lbry-sdk
134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
from error import DecodeError
|
|
|
|
|
|
class Encoding(object):
|
|
""" Interface for RPC message encoders/decoders
|
|
|
|
All encoding implementations used with this library should inherit and
|
|
implement this.
|
|
"""
|
|
|
|
def encode(self, data):
|
|
""" Encode the specified data
|
|
|
|
@param data: The data to encode
|
|
This method has to support encoding of the following
|
|
types: C{str}, C{int} and C{long}
|
|
Any additional data types may be supported as long as the
|
|
implementing class's C{decode()} method can successfully
|
|
decode them.
|
|
|
|
@return: The encoded data
|
|
@rtype: str
|
|
"""
|
|
|
|
def decode(self, data):
|
|
""" Decode the specified data string
|
|
|
|
@param data: The data (byte string) to decode.
|
|
@type data: str
|
|
|
|
@return: The decoded data (in its correct type)
|
|
"""
|
|
|
|
|
|
class Bencode(Encoding):
|
|
""" Implementation of a Bencode-based algorithm (Bencode is the encoding
|
|
algorithm used by Bittorrent).
|
|
|
|
@note: This algorithm differs from the "official" Bencode algorithm in
|
|
that it can encode/decode floating point values in addition to
|
|
integers.
|
|
"""
|
|
|
|
def encode(self, data):
|
|
""" Encoder implementation of the Bencode algorithm
|
|
|
|
@param data: The data to encode
|
|
@type data: int, long, tuple, list, dict or str
|
|
|
|
@return: The encoded data
|
|
@rtype: str
|
|
"""
|
|
if isinstance(data, (int, long)):
|
|
return 'i%de' % data
|
|
elif isinstance(data, str):
|
|
return '%d:%s' % (len(data), data)
|
|
elif isinstance(data, (list, tuple)):
|
|
encodedListItems = ''
|
|
for item in data:
|
|
encodedListItems += self.encode(item)
|
|
return 'l%se' % encodedListItems
|
|
elif isinstance(data, dict):
|
|
encodedDictItems = ''
|
|
keys = data.keys()
|
|
keys.sort()
|
|
for key in keys:
|
|
encodedDictItems += self.encode(key) # TODO: keys should always be bytestrings
|
|
encodedDictItems += self.encode(data[key])
|
|
return 'd%se' % encodedDictItems
|
|
else:
|
|
print data
|
|
raise TypeError("Cannot bencode '%s' object" % type(data))
|
|
|
|
def decode(self, data):
|
|
""" Decoder implementation of the Bencode algorithm
|
|
|
|
@param data: The encoded data
|
|
@type data: str
|
|
|
|
@note: This is a convenience wrapper for the recursive decoding
|
|
algorithm, C{_decodeRecursive}
|
|
|
|
@return: The decoded data, as a native Python type
|
|
@rtype: int, list, dict or str
|
|
"""
|
|
if len(data) == 0:
|
|
raise DecodeError('Cannot decode empty string')
|
|
try:
|
|
return self._decodeRecursive(data)[0]
|
|
except ValueError as e:
|
|
raise DecodeError(e.message)
|
|
|
|
@staticmethod
|
|
def _decodeRecursive(data, startIndex=0):
|
|
""" Actual implementation of the recursive Bencode algorithm
|
|
|
|
Do not call this; use C{decode()} instead
|
|
"""
|
|
if data[startIndex] == 'i':
|
|
endPos = data[startIndex:].find('e') + startIndex
|
|
return int(data[startIndex + 1:endPos]), endPos + 1
|
|
elif data[startIndex] == 'l':
|
|
startIndex += 1
|
|
decodedList = []
|
|
while data[startIndex] != 'e':
|
|
listData, startIndex = Bencode._decodeRecursive(data, startIndex)
|
|
decodedList.append(listData)
|
|
return decodedList, startIndex + 1
|
|
elif data[startIndex] == 'd':
|
|
startIndex += 1
|
|
decodedDict = {}
|
|
while data[startIndex] != 'e':
|
|
key, startIndex = Bencode._decodeRecursive(data, startIndex)
|
|
value, startIndex = Bencode._decodeRecursive(data, startIndex)
|
|
decodedDict[key] = value
|
|
return decodedDict, startIndex
|
|
elif data[startIndex] == 'f':
|
|
# This (float data type) is a non-standard extension to the original Bencode algorithm
|
|
endPos = data[startIndex:].find('e') + startIndex
|
|
return float(data[startIndex + 1:endPos]), endPos + 1
|
|
elif data[startIndex] == 'n':
|
|
# This (None/NULL data type) is a non-standard extension
|
|
# to the original Bencode algorithm
|
|
return None, startIndex + 1
|
|
else:
|
|
splitPos = data[startIndex:].find(':') + startIndex
|
|
try:
|
|
length = int(data[startIndex:splitPos])
|
|
except ValueError, e:
|
|
raise DecodeError, e
|
|
startIndex = splitPos + 1
|
|
endPos = startIndex + length
|
|
bytes = data[startIndex:endPos]
|
|
return bytes, endPos
|