mirrors/deluge
0
0
Fork 0
mirror of https://git.deluge-torrent.org/deluge synced 2025-08-02 22:48:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

[rencode] Update module to v1.0.4

Browse source
This commit is contained in:
Calum Lind 2016-04-26 18:41:13 +01:00
commit 616523c732
1 changed files with 97 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

184
deluge/rencode.py
View file

@ -59,11 +59,22 @@ same rencode version throughout your project.
""" """
import struct import struct
import sys
from threading import Lock from threading import Lock
from types import DictType, FloatType, IntType, ListType, LongType, NoneType, StringType, TupleType, UnicodeType
__version__ = '1.0.2' __version__ = ("Python", 1, 0, 4)
__all__ = ('dumps', 'loads') __all__ = ['dumps', 'loads']
py3 = sys.version_info[0] >= 3
if py3:
long = int
unicode = str
def int2byte(c):
return bytes([c])
else:
def int2byte(c):
return chr(c)
# Default number of bits for serialized floats, either 32 or 64 (also a parameter for dumps()). # Default number of bits for serialized floats, either 32 or 64 (also a parameter for dumps()).
DEFAULT_FLOAT_BITS = 32 DEFAULT_FLOAT_BITS = 32
@ -73,19 +84,19 @@ MAX_INT_LENGTH = 64
# The bencode 'typecodes' such as i, d, etc have been extended and # The bencode 'typecodes' such as i, d, etc have been extended and
# relocated on the base-256 character set. # relocated on the base-256 character set.
CHR_LIST = chr(59) CHR_LIST = int2byte(59)
CHR_DICT = chr(60) CHR_DICT = int2byte(60)
CHR_INT = chr(61) CHR_INT = int2byte(61)
CHR_INT1 = chr(62) CHR_INT1 = int2byte(62)
CHR_INT2 = chr(63) CHR_INT2 = int2byte(63)
CHR_INT4 = chr(64) CHR_INT4 = int2byte(64)
CHR_INT8 = chr(65) CHR_INT8 = int2byte(65)
CHR_FLOAT32 = chr(66) CHR_FLOAT32 = int2byte(66)
CHR_FLOAT64 = chr(44) CHR_FLOAT64 = int2byte(44)
CHR_TRUE = chr(67) CHR_TRUE = int2byte(67)
CHR_FALSE = chr(68) CHR_FALSE = int2byte(68)
CHR_NONE = chr(69) CHR_NONE = int2byte(69)
CHR_TERM = chr(127) CHR_TERM = int2byte(127)
# Positive integers with value embedded in typecode. # Positive integers with value embedded in typecode.
INT_POS_FIXED_START = 0 INT_POS_FIXED_START = 0
@ -120,10 +131,10 @@ def decode_int(x, f):
n = int(x[f:newf]) n = int(x[f:newf])
except (OverflowError, ValueError): except (OverflowError, ValueError):
n = long(x[f:newf]) n = long(x[f:newf])
if x[f] == '-': if x[f:f + 1] == '-':
if x[f + 1] == '0': if x[f + 1:f + 2] == '0':
raise ValueError raise ValueError
elif x[f] == '0' and newf != f + 1: elif x[f:f + 1] == '0' and newf != f + 1:
raise ValueError raise ValueError
return (n, newf + 1) return (n, newf + 1)
@ -140,6 +151,7 @@ def decode_inth(x, f):
def decode_intl(x, f): def decode_intl(x, f):
f += 1 f += 1
return (struct.unpack('!l', x[f:f + 4])[0], f + 4) return (struct.unpack('!l', x[f:f + 4])[0], f + 4)
@ -161,7 +173,7 @@ def decode_float64(x, f):
def decode_string(x, f): def decode_string(x, f):
colon = x.index(':', f) colon = x.index(b':', f)
try: try:
n = int(x[f:colon]) n = int(x[f:colon])
except (OverflowError, ValueError): except (OverflowError, ValueError):
@ -177,17 +189,17 @@ def decode_string(x, f):
def decode_list(x, f): def decode_list(x, f):
r, f = [], f + 1 r, f = [], f + 1
while x[f] != CHR_TERM: while x[f:f + 1] != CHR_TERM:
v, f = decode_func[x[f]](x, f) v, f = decode_func[x[f:f + 1]](x, f)
r.append(v) r.append(v)
return (tuple(r), f + 1) return (tuple(r), f + 1)
def decode_dict(x, f): def decode_dict(x, f):
r, f = {}, f + 1 r, f = {}, f + 1
while x[f] != CHR_TERM: while x[f:f + 1] != CHR_TERM:
k, f = decode_func[x[f]](x, f) k, f = decode_func[x[f:f + 1]](x, f)
r[k], f = decode_func[x[f]](x, f) r[k], f = decode_func[x[f:f + 1]](x, f)
return (r, f + 1) return (r, f + 1)
@ -203,16 +215,16 @@ def decode_none(x, f):
return (None, f + 1) return (None, f + 1)
decode_func = {} decode_func = {}
decode_func['0'] = decode_string decode_func[b'0'] = decode_string
decode_func['1'] = decode_string decode_func[b'1'] = decode_string
decode_func['2'] = decode_string decode_func[b'2'] = decode_string
decode_func['3'] = decode_string decode_func[b'3'] = decode_string
decode_func['4'] = decode_string decode_func[b'4'] = decode_string
decode_func['5'] = decode_string decode_func[b'5'] = decode_string
decode_func['6'] = decode_string decode_func[b'6'] = decode_string
decode_func['7'] = decode_string decode_func[b'7'] = decode_string
decode_func['8'] = decode_string decode_func[b'8'] = decode_string
decode_func['9'] = decode_string decode_func[b'9'] = decode_string
decode_func[CHR_LIST] = decode_list decode_func[CHR_LIST] = decode_list
decode_func[CHR_DICT] = decode_dict decode_func[CHR_DICT] = decode_dict
decode_func[CHR_INT] = decode_int decode_func[CHR_INT] = decode_int
@ -236,7 +248,7 @@ def make_fixed_length_string_decoders():
return (s, f + 1 + slen) return (s, f + 1 + slen)
return f return f
for i in range(STR_FIXED_COUNT): for i in range(STR_FIXED_COUNT):
decode_func[chr(STR_FIXED_START + i)] = make_decoder(i) decode_func[int2byte(STR_FIXED_START + i)] = make_decoder(i)
make_fixed_length_string_decoders() make_fixed_length_string_decoders()
@ -245,13 +257,13 @@ def make_fixed_length_list_decoders():
def make_decoder(slen): def make_decoder(slen):
def f(x, f): def f(x, f):
r, f = [], f + 1 r, f = [], f + 1
for i in range(slen): for _ in range(slen):
v, f = decode_func[x[f]](x, f) v, f = decode_func[x[f:f + 1]](x, f)
r.append(v) r.append(v)
return (tuple(r), f) return (tuple(r), f)
return f return f
for i in range(LIST_FIXED_COUNT): for i in range(LIST_FIXED_COUNT):
decode_func[chr(LIST_FIXED_START + i)] = make_decoder(i) decode_func[int2byte(LIST_FIXED_START + i)] = make_decoder(i)
make_fixed_length_list_decoders() make_fixed_length_list_decoders()
@ -262,9 +274,9 @@ def make_fixed_length_int_decoders():
return (j, f + 1) return (j, f + 1)
return f return f
for i in range(INT_POS_FIXED_COUNT): for i in range(INT_POS_FIXED_COUNT):
decode_func[chr(INT_POS_FIXED_START + i)] = make_decoder(i) decode_func[int2byte(INT_POS_FIXED_START + i)] = make_decoder(i)
for i in range(INT_NEG_FIXED_COUNT): for i in range(INT_NEG_FIXED_COUNT):
decode_func[chr(INT_NEG_FIXED_START + i)] = make_decoder(-1 - i) decode_func[int2byte(INT_NEG_FIXED_START + i)] = make_decoder(-1 - i)
make_fixed_length_int_decoders() make_fixed_length_int_decoders()
@ -273,13 +285,13 @@ def make_fixed_length_dict_decoders():
def make_decoder(slen): def make_decoder(slen):
def f(x, f): def f(x, f):
r, f = {}, f + 1 r, f = {}, f + 1
for j in range(slen): for _ in range(slen):
k, f = decode_func[x[f]](x, f) k, f = decode_func[x[f:f + 1]](x, f)
r[k], f = decode_func[x[f]](x, f) r[k], f = decode_func[x[f:f + 1]](x, f)
return (r, f) return (r, f)
return f return f
for i in range(DICT_FIXED_COUNT): for i in range(DICT_FIXED_COUNT):
decode_func[chr(DICT_FIXED_START + i)] = make_decoder(i) decode_func[int2byte(DICT_FIXED_START + i)] = make_decoder(i)
make_fixed_length_dict_decoders() make_fixed_length_dict_decoders()
@ -288,7 +300,7 @@ def loads(x, decode_utf8=False):
global _decode_utf8 global _decode_utf8
_decode_utf8 = decode_utf8 _decode_utf8 = decode_utf8
try: try:
r, l = decode_func[x[0]](x, 0) r, l = decode_func[x[0:1]](x, 0)
except (IndexError, KeyError): except (IndexError, KeyError):
raise ValueError raise ValueError
if l != len(x): if l != len(x):
@ -298,9 +310,9 @@ def loads(x, decode_utf8=False):
def encode_int(x, r): def encode_int(x, r):
if 0 <= x < INT_POS_FIXED_COUNT: if 0 <= x < INT_POS_FIXED_COUNT:
r.append(chr(INT_POS_FIXED_START + x)) r.append(int2byte(INT_POS_FIXED_START + x))
elif -INT_NEG_FIXED_COUNT <= x < 0: elif -INT_NEG_FIXED_COUNT <= x < 0:
r.append(chr(INT_NEG_FIXED_START - 1 - x)) r.append(int2byte(INT_NEG_FIXED_START - 1 - x))
elif -128 <= x < 128: elif -128 <= x < 128:
r.extend((CHR_INT1, struct.pack('!b', x))) r.extend((CHR_INT1, struct.pack('!b', x)))
elif -32768 <= x < 32768: elif -32768 <= x < 32768:
@ -311,6 +323,9 @@ def encode_int(x, r):
r.extend((CHR_INT8, struct.pack('!q', x))) r.extend((CHR_INT8, struct.pack('!q', x)))
else: else:
s = str(x) s = str(x)
if py3:
s = bytes(s, "ascii")
if len(s) >= MAX_INT_LENGTH: if len(s) >= MAX_INT_LENGTH:
raise ValueError('overflow') raise ValueError('overflow')
r.extend((CHR_INT, s, CHR_TERM)) r.extend((CHR_INT, s, CHR_TERM))
@ -325,18 +340,21 @@ def encode_float64(x, r):
def encode_bool(x, r): def encode_bool(x, r):
r.extend({False: CHR_FALSE, True: CHR_TRUE}[bool(x)]) r.append({False: CHR_FALSE, True: CHR_TRUE}[bool(x)])
def encode_none(x, r): def encode_none(x, r):
r.extend(CHR_NONE) r.append(CHR_NONE)
def encode_string(x, r): def encode_string(x, r):
if len(x) < STR_FIXED_COUNT: if len(x) < STR_FIXED_COUNT:
r.extend((chr(STR_FIXED_START + len(x)), x)) r.extend((int2byte(STR_FIXED_START + len(x)), x))
else: else:
r.extend((str(len(x)), ':', x)) s = str(len(x))
if py3:
s = bytes(s, "ascii")
r.extend((s, b':', x))
def encode_unicode(x, r): def encode_unicode(x, r):
@ -345,7 +363,7 @@ def encode_unicode(x, r):
def encode_list(x, r): def encode_list(x, r):
if len(x) < LIST_FIXED_COUNT: if len(x) < LIST_FIXED_COUNT:
r.append(chr(LIST_FIXED_START + len(x))) r.append(int2byte(LIST_FIXED_START + len(x)))
for i in x: for i in x:
encode_func[type(i)](i, r) encode_func[type(i)](i, r)
else: else:
@ -357,7 +375,7 @@ def encode_list(x, r):
def encode_dict(x, r): def encode_dict(x, r):
if len(x) < DICT_FIXED_COUNT: if len(x) < DICT_FIXED_COUNT:
r.append(chr(DICT_FIXED_START + len(x))) r.append(int2byte(DICT_FIXED_START + len(x)))
for k, v in x.items(): for k, v in x.items():
encode_func[type(k)](k, r) encode_func[type(k)](k, r)
encode_func[type(v)](v, r) encode_func[type(v)](v, r)
@ -369,23 +387,18 @@ def encode_dict(x, r):
r.append(CHR_TERM) r.append(CHR_TERM)
encode_func = {} encode_func = {}
encode_func[IntType] = encode_int encode_func[int] = encode_int
encode_func[LongType] = encode_int encode_func[long] = encode_int
encode_func[StringType] = encode_string encode_func[bytes] = encode_string
encode_func[ListType] = encode_list encode_func[list] = encode_list
encode_func[TupleType] = encode_list encode_func[tuple] = encode_list
encode_func[DictType] = encode_dict encode_func[dict] = encode_dict
encode_func[NoneType] = encode_none encode_func[type(None)] = encode_none
encode_func[UnicodeType] = encode_unicode encode_func[unicode] = encode_unicode
encode_func[bool] = encode_bool
lock = Lock() lock = Lock()
try:
from types import BooleanType
encode_func[BooleanType] = encode_bool
except ImportError:
pass
def dumps(x, float_bits=DEFAULT_FLOAT_BITS): def dumps(x, float_bits=DEFAULT_FLOAT_BITS):
""" """
@ -393,51 +406,48 @@ def dumps(x, float_bits=DEFAULT_FLOAT_BITS):
Here float_bits is either 32 or 64. Here float_bits is either 32 or 64.
""" """
lock.acquire() with lock:
try:
if float_bits == 32: if float_bits == 32:
encode_func[FloatType] = encode_float32 encode_func[float] = encode_float32
elif float_bits == 64: elif float_bits == 64:
encode_func[FloatType] = encode_float64 encode_func[float] = encode_float64
else: else:
raise ValueError('Float bits (%d) is not 32 or 64' % float_bits) raise ValueError('Float bits (%d) is not 32 or 64' % float_bits)
r = [] r = []
encode_func[type(x)](x, r) encode_func[type(x)](x, r)
finally: return b''.join(r)
lock.release()
return ''.join(r)
def test(): def test():
f1 = struct.unpack('!f', struct.pack('!f', 25.5))[0] f1 = struct.unpack('!f', struct.pack('!f', 25.5))[0]
f2 = struct.unpack('!f', struct.pack('!f', 29.3))[0] f2 = struct.unpack('!f', struct.pack('!f', 29.3))[0]
f3 = struct.unpack('!f', struct.pack('!f', -0.6))[0] f3 = struct.unpack('!f', struct.pack('!f', -0.6))[0]
ld = (({'a': 15, 'bb': f1, 'ccc': f2, '': (f3, (), False, True, '')}, ('a', 10 ** 20), ld = (({b'a': 15, b'bb': f1, b'ccc': f2, b'': (f3, (), False, True, b'')}, (b'a', 10**20),
tuple(range(-100000, 100000)), 'b' * 31, 'b' * 62, 'b' * 64, 2 ** 30, 2 ** 33, 2 ** 62, 2 ** 64, tuple(range(-100000, 100000)), b'b' * 31, b'b' * 62, b'b' * 64, 2**30, 2**33, 2**62,
2 ** 30, 2 ** 33, 2 ** 62, 2 ** 64, False, False, True, -1, 2, 0),) 2**64, 2**30, 2**33, 2**62, 2**64, False, False, True, -1, 2, 0),)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
d = dict(zip(range(-100000, 100000), range(-100000, 100000))) d = dict(zip(range(-100000, 100000), range(-100000, 100000)))
d.update({'a': 20, 20: 40, 40: 41, f1: f2, f2: f3, f3: False, False: True, True: False}) d.update({b'a': 20, 20: 40, 40: 41, f1: f2, f2: f3, f3: False, False: True, True: False})
ld = (d, {}, {5: 6}, {7: 7, True: 8}, {9: 10, 22: 39, 49: 50, 44: ''}) ld = (d, {}, {5: 6}, {7: 7, True: 8}, {9: 10, 22: 39, 49: 50, 44: b''})
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = ('', 'a' * 10, 'a' * 100, 'a' * 1000, 'a' * 10000, 'a' * 100000, 'a' * 1000000, 'a' * 10000000) ld = (b'', b'a' * 10, b'a' * 100, b'a' * 1000, b'a' * 10000, b'a' * 100000, b'a' * 1000000, b'a' * 10000000)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = tuple([dict(zip(range(n), range(n))) for n in range(100)]) + ('b',) ld = tuple([dict(zip(range(n), range(n))) for n in range(100)]) + (b'b',)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = tuple([dict(zip(range(n), range(-n, 0))) for n in range(100)]) + ('b',) ld = tuple([dict(zip(range(n), range(-n, 0))) for n in range(100)]) + (b'b',)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = tuple([tuple(range(n)) for n in range(100)]) + ('b',) ld = tuple([tuple(range(n)) for n in range(100)]) + (b'b',)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = tuple(['a' * n for n in range(1000)]) + ('b',) ld = tuple([b'a' * n for n in range(1000)]) + (b'b',)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
ld = tuple(['a' * n for n in range(1000)]) + (None, True, None) ld = tuple([b'a' * n for n in range(1000)]) + (None, True, None)
assert loads(dumps(ld)) == ld assert loads(dumps(ld)) == ld
assert loads(dumps(None)) is None assert loads(dumps(None)) is None
assert loads(dumps({None: None})) == {None: None} assert loads(dumps({None: None})) == {None: None}
assert 1e-10 < abs(loads(dumps(1.1)) - 1.1) < 1e-6 assert 1e-10 < abs(loads(dumps(1.1)) - 1.1) < 1e-6
assert 1e-10 < abs(loads(dumps(1.1, 32)) - 1.1) < 1e-6 assert 1e-10 < abs(loads(dumps(1.1, 32)) - 1.1) < 1e-6
assert abs(loads(dumps(1.1, 64)) - 1.1) < 1e-12 assert abs(loads(dumps(1.1, 64)) - 1.1) < 1e-12
assert loads(dumps(u"Hello World!!")) assert loads(dumps("Hello World!!"), decode_utf8=True)
try: try:
import psyco import psyco
psyco.bind(dumps) psyco.bind(dumps)