diff options
Diffstat (limited to 'libmat2/torrent.py')
| -rw-r--r-- | libmat2/torrent.py | 23 |
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 @@ | |||
| 1 | from typing import Union, Tuple, Dict | ||
| 1 | from . import abstract | 2 | from . import abstract |
| 2 | 3 | ||
| 3 | 4 | ||
| 4 | class TorrentParser(abstract.AbstractParser): | 5 | class 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: |
