summaryrefslogtreecommitdiff
path: root/libmat2/torrent.py
diff options
context:
space:
mode:
Diffstat (limited to 'libmat2/torrent.py')
-rw-r--r--libmat2/torrent.py23
1 files changed, 12 insertions, 11 deletions
diff --git a/libmat2/torrent.py b/libmat2/torrent.py
index 3a819fb..f5935e6 100644
--- a/libmat2/torrent.py
+++ b/libmat2/torrent.py
@@ -1,11 +1,12 @@
1from typing import Union, Tuple, Dict
1from . import abstract 2from . import abstract
2 3
3 4
4class TorrentParser(abstract.AbstractParser): 5class TorrentParser(abstract.AbstractParser):
5 mimetypes = {b'application/x-bittorrent', } 6 mimetypes = {'application/x-bittorrent', }
6 whitelist = {b'announce', b'announce-list', b'info'} 7 whitelist = {b'announce', b'announce-list', b'info'}
7 8
8 def get_meta(self) -> dict: 9 def get_meta(self) -> Dict[str, str]:
9 metadata = {} 10 metadata = {}
10 with open(self.filename, 'rb') as f: 11 with open(self.filename, 'rb') as f:
11 d = _BencodeHandler().bdecode(f.read()) 12 d = _BencodeHandler().bdecode(f.read())
@@ -54,7 +55,7 @@ class _BencodeHandler(object):
54 } 55 }
55 56
56 @staticmethod 57 @staticmethod
57 def __decode_int(s: str) -> (int, str): 58 def __decode_int(s: bytes) -> Tuple[int, bytes]:
58 s = s[1:] 59 s = s[1:]
59 next_idx = s.index(b'e') 60 next_idx = s.index(b'e')
60 if s.startswith(b'-0'): 61 if s.startswith(b'-0'):
@@ -64,7 +65,7 @@ class _BencodeHandler(object):
64 return int(s[:next_idx]), s[next_idx+1:] 65 return int(s[:next_idx]), s[next_idx+1:]
65 66
66 @staticmethod 67 @staticmethod
67 def __decode_string(s: str) -> (str, str): 68 def __decode_string(s: bytes) -> Tuple[bytes, bytes]:
68 sep = s.index(b':') 69 sep = s.index(b':')
69 str_len = int(s[:sep]) 70 str_len = int(s[:sep])
70 if str_len < 0: 71 if str_len < 0:
@@ -74,7 +75,7 @@ class _BencodeHandler(object):
74 s = s[1:] 75 s = s[1:]
75 return s[sep:sep+str_len], s[sep+str_len:] 76 return s[sep:sep+str_len], s[sep+str_len:]
76 77
77 def __decode_list(self, s: str) -> (list, str): 78 def __decode_list(self, s: bytes) -> Tuple[list, bytes]:
78 r = list() 79 r = list()
79 s = s[1:] # skip leading `l` 80 s = s[1:] # skip leading `l`
80 while s[0] != ord('e'): 81 while s[0] != ord('e'):
@@ -82,7 +83,7 @@ class _BencodeHandler(object):
82 r.append(v) 83 r.append(v)
83 return r, s[1:] 84 return r, s[1:]
84 85
85 def __decode_dict(self, s: str) -> (dict, str): 86 def __decode_dict(self, s: bytes) -> Tuple[dict, bytes]:
86 r = dict() 87 r = dict()
87 s = s[1:] # skip leading `d` 88 s = s[1:] # skip leading `d`
88 while s[0] != ord(b'e'): 89 while s[0] != ord(b'e'):
@@ -91,11 +92,11 @@ class _BencodeHandler(object):
91 return r, s[1:] 92 return r, s[1:]
92 93
93 @staticmethod 94 @staticmethod
94 def __encode_int(x: str) -> bytes: 95 def __encode_int(x: bytes) -> bytes:
95 return b'i' + bytes(str(x), 'utf-8') + b'e' 96 return b'i' + bytes(str(x), 'utf-8') + b'e'
96 97
97 @staticmethod 98 @staticmethod
98 def __encode_string(x: str) -> bytes: 99 def __encode_string(x: bytes) -> bytes:
99 return bytes((str(len(x))), 'utf-8') + b':' + x 100 return bytes((str(len(x))), 'utf-8') + b':' + x
100 101
101 def __encode_list(self, x: str) -> bytes: 102 def __encode_list(self, x: str) -> bytes:
@@ -104,17 +105,17 @@ class _BencodeHandler(object):
104 ret += self.__encode_func[type(i)](i) 105 ret += self.__encode_func[type(i)](i)
105 return b'l' + ret + b'e' 106 return b'l' + ret + b'e'
106 107
107 def __encode_dict(self, x: str) -> bytes: 108 def __encode_dict(self, x: dict) -> bytes:
108 ret = b'' 109 ret = b''
109 for k, v in sorted(x.items()): 110 for k, v in sorted(x.items()):
110 ret += self.__encode_func[type(k)](k) 111 ret += self.__encode_func[type(k)](k)
111 ret += self.__encode_func[type(v)](v) 112 ret += self.__encode_func[type(v)](v)
112 return b'd' + ret + b'e' 113 return b'd' + ret + b'e'
113 114
114 def bencode(self, s: str) -> bytes: 115 def bencode(self, s: Union[dict, list, bytes, int]) -> bytes:
115 return self.__encode_func[type(s)](s) 116 return self.__encode_func[type(s)](s)
116 117
117 def bdecode(self, s: str): 118 def bdecode(self, s: bytes) -> Union[dict, None]:
118 try: 119 try:
119 r, l = self.__decode_func[s[0]](s) 120 r, l = self.__decode_func[s[0]](s)
120 except (IndexError, KeyError, ValueError) as e: 121 except (IndexError, KeyError, ValueError) as e: