15 Commits
Author SHA1 Message Date
roman 65796860b8 Первая версия модуля для kodi 2019-05-06 11:28:21 +03:00
7sDream 2a4eecb830 release version 0.3.0 2018-06-23 13:18:00 +08:00
7sDream 1bd845af2d land dev 2018-06-23 13:13:23 +08:00
7sDream 58b29be0a6 fix typo hash_fields -> hash_field 2018-06-23 13:01:08 +08:00
7sDream 177d1c7de9 hash_fields(finish #4) and hash raw:
- hash fields method and parameter allow user customize hash field list
- hash raw parameter allow the output of hash field to be raw bytes
- BEncoder now support encode raw bytes
2018-06-23 12:57:04 +08:00
7sDream ee3128b32b Many changes, see bellow:
- Added
  - BEncoder
  - BDecoder
  - encode
  - decode
  - errors argument for TorrentFileParser and parse_torrent_file (#4)
- Changed
  - TorrentFileCreator rename to BEncoder
  - TorrentFileParser, BEncoder don't need outmost level is dict
- Updated
  - README
  - CHANGELOG
  - LICENSE

and tests for above changes.
2018-06-23 01:44:10 +08:00
七秒不觉梦 94b8f1eee0 Merge pull request #5 from yasuotakei/patch-1
Fix Issue #4
2018-06-22 21:21:57 +08:00
Yasuo c1212c0203 Fix Issue #4 2018-06-21 13:56:46 +00:00
7sDream afd439e51f Merge branch 'master' into dev 2018-05-25 17:30:43 +08:00
7sDream d572794ac6 Bump new version: 0.2.0 2018-05-25 17:24:31 +08:00
7sDream 4eeceb975b Removed: unused test files in MANIFEST.in 2018-04-29 00:20:08 +08:00
7sDream 550c475bfc update to v0.1.5rc1 2018-04-29 00:14:54 +08:00
7sDream cb06fe00d8 Merge branch 'dev' 2018-04-29 00:09:44 +08:00
7sDream cae5919459 Added: TorrentFileCreator and create_torrent_file 2018-04-29 00:08:48 +08:00
7sDream 25aafaa58b fix changelog unreleased link url 2018-04-06 14:03:28 +08:00
13 changed files with 691 additions and 499 deletions

No files matched your search

-71
View File
@@ -1,71 +0,0 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
## [Unreleased]
## [0.1.4] - 2018-04-06
### Added
- `encoding` option can be `auto`, which will use `chardet` package to decide which encoding to use. If `chardet` is noe installed, will raise a warning and fallback to 'utf-8'. (Thanks to [@ltfychrise])
- Add changelog.
### Change
- Reorganize test codes/files.
### Fixed
- Fix integer filed can't be negative bug. (Thanks to [@ltfychrise])
- Fix `_seek_back` method not make `_pos` back bug. (Thanks to [@ltfychrise])
## [0.1.3] - 2017-06-21
### Added
- Now `UnicodeDecodeError` is wrapped in `InvalidTorrentDataException`.
### Fixed
- Use `IOError` instead of `FileNotFoundError` in Python 2.
### Changed
- `InvalidTorrentFileException` rename to `InvalidTorrentDataException`.
## [0.1.2] - 2017-06-21
### Changed
- Emm, I don't know, I just changed the version code...
## [0.1.1] - 2017-06-20
### Added
- CLI add coding `--coding/-c` option for file string filed encoding.
### Changed
- `ed2k` and `filehash` field now use same structure as 'pieces'.
## [0.1.0] - 2017-05-23
### Added
- Parse torrent from file and data into a dict.
- CLI provided.
- Simple tests.
- Available on pip.
[@ltfychrise]: https://github.com/ltfychrise
[Unreleased]: https://github.com/7sDream/torrent_parser/compare/v0.1.3...HEAD
[0.1.4]: https://github.com/7sDream/torrent_parser/compare/v0.1.3...v0.1.4
[0.1.3]: https://github.com/7sDream/torrent_parser/compare/v0.1.2...v0.1.3
[0.1.2]: https://github.com/7sDream/torrent_parser/compare/v0.1.1...v0.1.2
[0.1.1]: https://github.com/7sDream/torrent_parser/compare/v0.1.0...v0.1.1
[0.1.0]: https://github.com/7sDream/torrent_parser/tree/v0.1.0
+1 -1
View File
@@ -1,6 +1,6 @@
The MIT License (MIT) The MIT License (MIT)
Copyright (c) 2017 7sDream Copyright (c) 2017 - 2018 7sDream
Permission is hereby granted, free of charge, to any person obtaining a copy Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal of this software and associated documentation files (the "Software"), to deal
-3
View File
@@ -1,3 +0,0 @@
include README.md LICENSE CHANGELOG.md
include test.py
include test.torrent
+28 -3
View File
@@ -1,4 +1,16 @@
# Torrent file parser for Python # Torrent file parser and creator for Python
A simple parser for `.torrent` file.
Can also edit and write back to torrent format after version 0.2.0.
## Features
- Decoder and encoder for torrent files
- Auto decode bytes field to string with used specified encoding and error handler
- Auto detect encoding when use `auto` as encoding(need `chardet` installed)
- Auto decode hash value filed to hash blocks, also customizable
- CLI provided, with JSON output
## Install ## Install
@@ -30,8 +42,21 @@ $ cat test.torrent | pytp
```pycon ```pycon
>>> import torrent_parser as tp >>> import torrent_parser as tp
>>> data = tp.parse_torrent_file('test.torrent') >>> data = tp.parse_torrent_file('test.torrent')
>>> print(data['announce']) >>> data['announce']
http://tracker.trackerfix.com:80/announce http://tracker.trackerfix.com:80/announce
>>> data['announce'] = 'http://127.0.0.1:12345'
>>> tp.create_torrent_file('new.torrent', data)
```
or you don't operate with file, just raw bytes:
```pycon
>>> import torrent_parser as tp
>>> data = tp.decode(b'd3:negi-1ee')
>>> data['neg']
-1
>>> tp.encode(data)
b'd3:negi-1ee'
``` ```
## Test ## Test
@@ -52,4 +77,4 @@ See [License][LICENSE].
[screenshots-normal]: http://rikka-10066868.image.myqcloud.com/1492616d-9f14-4fe2-9146-9a3ac06c6868.png [screenshots-normal]: http://rikka-10066868.image.myqcloud.com/1492616d-9f14-4fe2-9146-9a3ac06c6868.png
[screenshots-indent]: http://rikka-10066868.image.myqcloud.com/eadc4184-6deb-42eb-bfd4-239da8f50c08.png [screenshots-indent]: http://rikka-10066868.image.myqcloud.com/eadc4184-6deb-42eb-bfd4-239da8f50c08.png
[LICENSE]: https://github.com/7sDream/torrent_parser/blob/master/LICENSE [LICENSE]: https://github.com/7sDream/torrent_parser/blob/master/LICENSE
[CHANGELOG]: https://github.com/7sDream/torrent_parser/blob/master/CHANGELOG.md [CHANGELOG]: https://github.com/7sDream/torrent_parser/blob/master/CHANGELOG.md
+17
View File
@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<addon id="script.module.torrent_parser" name="torrent_parser" version="0.3.0" provider-name="inpos">
<requires>
<import addon="xbmc.python" version="2.25.0"/>
<import addon="script.module.chardet" />
</requires>
<extension point="xbmc.python.module" library="lib"/>
<extension point="xbmc.addon.metadata">
<platform>all</platform>
<language>en_GB ru_RU</language>
<summary lang="en_GB">Torrent file parser and creator for Python</summary>
<summary lang="ru_RU">Обработка и создание файлов .torrent</summary>
<description lang="ru">Разработан 7sDream (https://github.com/7sDream/torrent_parser)</description>
<description lang="en">Developed by 7sDream (https://github.com/7sDream/torrent_parser)</description>
<email>roman@ukamnya.ru</email>
<source>https://git.ukamnya.ru/roman/script.module.torrent_parser</source></extension>
</addon>
+645
View File
@@ -0,0 +1,645 @@
#!/usr/bin/env python
# coding: utf-8
"""
A .torrent file parser for both Python 2 and 3
Usage:
data = parse_torrent_file(filename)
# or
with open(filename, 'rb') as f: # the binary mode 'b' is necessary
data = TorrentFileParser(f).parse()
# then you can edit the data
data['announce-list'].append(['http://127.0.0.1:8080'])
# and create a new torrent file from data
create_torrent_file('new.torrent', data)
# or
with open('new.torrent', 'wb') as f:
f.write(TorrentFileCreator(data).encode())
# or you don't deal with file, just object in memory
data = decode(b'i12345e') # data = 12345
content = encode(data) # content = b'i12345e'
"""
from __future__ import print_function, unicode_literals
import argparse
import binascii
import collections
import io
import json
import sys
import warnings
try:
FileNotFoundError
except NameError:
# Python 2 do not have FileNotFoundError, use IOError instead
# noinspection PyShadowingBuiltins
FileNotFoundError = IOError
try:
# noinspection PyPackageRequirements
from chardet import detect as _detect
except ImportError:
def _detect(_):
warnings.warn("No chardet module installed, encoding will be utf-8")
return {'encoding': 'utf-8', 'confidence': 1}
try:
# noinspection PyUnresolvedReferences
# For Python 2
str_type = unicode
except NameError:
# For Python 3
str_type = str
__all__ = [
'InvalidTorrentDataException',
'BEncoder',
'BDecoder',
'encode',
'decode',
'TorrentFileParser',
'create_torrent_file',
'parse_torrent_file',
]
__version__ = '0.3.0'
def detect(content):
return _detect(content)['encoding']
class InvalidTorrentDataException(Exception):
def __init__(self, pos, msg=None):
msg = msg or "Invalid torrent format when read at pos {pos}"
msg = msg.format(pos=pos)
super(InvalidTorrentDataException, self).__init__(msg)
class __EndCls(object):
pass
_END = __EndCls()
def _check_hash_field_params(name, value):
return isinstance(name, str_type) \
and isinstance(value, tuple) and len(value) == 2 \
and isinstance(value[0], int) and isinstance(value[1], bool)
class TorrentFileParser(object):
TYPE_LIST = 'list'
TYPE_DICT = 'dict'
TYPE_INT = 'int'
TYPE_STRING = 'string'
TYPE_END = 'end'
LIST_INDICATOR = b'l'
DICT_INDICATOR = b'd'
INT_INDICATOR = b'i'
END_INDICATOR = b'e'
STRING_INDICATOR = b''
STRING_DELIMITER = b':'
HASH_FIELD_PARAMS = {
# field length need_list
'pieces': (20, True),
'ed2k': (16, False),
'filehash': (20, False),
}
TYPES = [
(TYPE_LIST, LIST_INDICATOR),
(TYPE_DICT, DICT_INDICATOR),
(TYPE_INT, INT_INDICATOR),
(TYPE_END, END_INDICATOR),
(TYPE_STRING, STRING_INDICATOR),
]
def __init__(
self, fp, use_ordered_dict=False, encoding='utf-8', errors='strict',
hash_fields=None, hash_raw=False,
):
"""
:param fp: a **binary** file-like object to parse,
which means need 'b' mode when use built-in open function
:param bool use_ordered_dict: Use collections.OrderedDict as dict
container default False, which mean use built-in dict
:param str encoding: file content encoding, default utf-8, use 'auto'
to enable charset auto detection (need 'chardet' package installed)
:param str errors: how to deal with encoding error when try to parse
string from content with ``encoding``
:param Dict[str, Tuple[int, bool]] hash_fields: extra fields should
be treated as hash value. dict key is the field name, value is a
two-element tuple of (hash_block_length, as_a_list).
See :any:`hash_field` for detail
"""
if getattr(fp, 'read', ) is None \
or getattr(fp, 'seek') is None:
raise ValueError('Parameter fp needs a file like object')
self._pos = 0
self._encoding = encoding
self._content = fp
self._use_ordered_dict = use_ordered_dict
self._error_handler = errors
self._hash_fields = dict(TorrentFileParser.HASH_FIELD_PARAMS)
if hash_fields is not None:
for k, v in hash_fields.items():
if _check_hash_field_params(k, v):
self._hash_fields[k] = v
else:
raise ValueError(
"Invalid hash field parameter, it should be type of "
"Dict[str, Tuple[int, bool]]"
)
self._hash_raw = bool(hash_raw)
def hash_field(self, name, block_length=20, need_list=False):
"""
Let field with the `name` to be treated as hash value, don't decode it
as a string.
:param str name: field name
:param int block_length: hash block length for split
:param bool need_list: if True, when the field only has one block(
or even empty) its parse result will be a one-element list(
or empty list); If False, will be a string in 0 or 1 block condition
:return: return self, so you can chained call
"""
v = (block_length, need_list)
if _check_hash_field_params(name, v):
self._hash_fields[name] = v
else:
raise ValueError("Invalid hash field parameter")
return self
def parse(self):
"""
:rtype: dict|list|int|str|bytes
:raise: :any:`InvalidTorrentDataException` when parse failed or error
happened when decode string using specified encoding
"""
self._restart()
data = self._next_element()
try:
c = self._read_byte(1, True)
raise InvalidTorrentDataException(
0, 'Expect EOF, but get [{}] at pos {}'.format(c, self._pos)
)
except EOFError: # expect EOF
pass
return data
def _read_byte(self, count=1, raise_eof=False):
assert count >= 0
gotten = self._content.read(count)
if count != 0 and len(gotten) == 0:
if raise_eof:
raise EOFError()
raise InvalidTorrentDataException(
self._pos,
'Unexpected EOF when reading torrent file'
)
self._pos += count
return gotten
def _seek_back(self, count):
self._content.seek(-count, 1)
self._pos = self._pos - count
def _restart(self):
self._content.seek(0, 0)
self._pos = 0
def _dict_items_generator(self):
while True:
k = self._next_element()
if k is _END:
return
if not isinstance(k, str_type):
raise InvalidTorrentDataException(
self._pos, "Type of dict key can't be " + type(k).__name__
)
if k in self._hash_fields:
v = self._next_hash(*self._hash_fields[k])
else:
v = self._next_element(k)
if k == 'encoding':
self._encoding = v
yield k, v
def _next_dict(self):
data = collections.OrderedDict() if self._use_ordered_dict else dict()
for key, element in self._dict_items_generator():
data[key] = element
return data
def _list_items_generator(self):
while True:
element = self._next_element()
if element is _END:
return
yield element
def _next_list(self):
return [element for element in self._list_items_generator()]
def _next_int(self, end=END_INDICATOR):
value = 0
char = self._read_byte(1)
neg = False
while char != end:
if not neg and char == b'-':
neg = True
elif not b'0' <= char <= b'9':
raise InvalidTorrentDataException(self._pos - 1)
else:
value = value * 10 + int(char) - int(b'0')
char = self._read_byte(1)
return -value if neg else value
def _next_string(self, need_decode=True, field=None):
length = self._next_int(self.STRING_DELIMITER)
raw = self._read_byte(length)
if need_decode:
encoding = self._encoding
if encoding == 'auto':
self.encoding = encoding = detect(raw)
try:
string = raw.decode(encoding, self._error_handler)
except UnicodeDecodeError as e:
msg = [
"Fail to decode string at pos {pos} using encoding ",
e.encoding
]
if field:
msg.extend([
' when parser field "', field, '"'
', maybe it is an hash field. ',
'You can use self.hash_field("', field, '") ',
'to let it be treated as hash value, ',
'so this error may disappear'
])
raise InvalidTorrentDataException(
self._pos - length + e.start,
''.join(msg)
)
return string
return raw
def _next_hash(self, p_len, need_list):
raw = self._next_string(need_decode=False)
if len(raw) % p_len != 0:
raise InvalidTorrentDataException(
self._pos - len(raw), "Hash bit length not match at pos {pos}"
)
if self._hash_raw:
return raw
res = [
binascii.hexlify(chunk).decode('ascii')
for chunk in (raw[x:x+p_len] for x in range(0, len(raw), p_len))
]
if len(res) == 0 and not need_list:
return ''
if len(res) == 1 and not need_list:
return res[0]
return res
@staticmethod
def _next_end():
return _END
def _next_type(self):
for (element_type, indicator) in self.TYPES:
indicator_length = len(indicator)
char = self._read_byte(indicator_length)
if indicator == char:
return element_type
self._seek_back(indicator_length)
raise InvalidTorrentDataException(self._pos)
def _type_to_func(self, t):
return getattr(self, '_next_' + t)
def _next_element(self, field=None):
element_type = self._next_type()
if element_type is TorrentFileParser.TYPE_STRING and field is not None:
element = self._type_to_func(element_type)(field=field)
else:
element = self._type_to_func(element_type)()
return element
class BEncoder(object):
TYPES = {
(dict,): TorrentFileParser.TYPE_DICT,
(list,): TorrentFileParser.TYPE_LIST,
(int,): TorrentFileParser.TYPE_INT,
(str_type, bytes): TorrentFileParser.TYPE_STRING,
}
def __init__(self, data, encoding='utf-8', hash_fields=None):
"""
:param dict|list|int|str data: data will be encoded
:param str encoding: string field output encoding
:param List[str] hash_fields: see
:any:`TorrentFileParser.__init__`
"""
self._data = data
self._encoding = encoding
self._hash_fields = list(TorrentFileParser.HASH_FIELD_PARAMS.keys())
if hash_fields is not None:
self._hash_fields.extend(str_type(hash_fields))
def hash_field(self, name):
"""
see :any:`TorrentFileParser.hash_field`
:param str name:
:return: return self, so you can chained call
"""
return self._hash_fields.append(str_type(name))
def encode(self):
"""
Encode to bytes
:rtype: bytes
"""
return b''.join(self._output_element(self._data))
def encode_to_filelike(self):
"""
Encode to a file-like(BytesIO) object
:rtype: BytesIO
"""
return io.BytesIO(self.encode())
def _output_string(self, data):
if isinstance(data, str_type):
data = data.encode(self._encoding)
yield str(len(data)).encode('ascii')
yield TorrentFileParser.STRING_DELIMITER
yield data
@staticmethod
def _output_int(data):
yield TorrentFileParser.INT_INDICATOR
yield str(data).encode('ascii')
yield TorrentFileParser.END_INDICATOR
def _output_decode_hash(self, data):
if isinstance(data, str_type):
data = [data]
result = []
for hash_line in data:
if not isinstance(hash_line, str_type):
raise InvalidTorrentDataException(
None,
"Hash must be " + str_type.__name__ + " not " +
type(hash_line).__name__,
)
if len(hash_line) % 2 != 0:
raise InvalidTorrentDataException(
None,
"Hash(" + hash_line + ") length(" + str(len(hash_line)) +
") is a not even number",
)
try:
raw = binascii.unhexlify(hash_line)
except binascii.Error as e:
raise InvalidTorrentDataException(
None, str(e),
)
result.append(raw)
for x in self._output_string(b''.join(result)):
yield x
def _output_dict(self, data):
yield TorrentFileParser.DICT_INDICATOR
for k, v in data.items():
if not isinstance(k, str_type):
raise InvalidTorrentDataException(
None, "Dict key must be " + str_type.__name__,
)
for x in self._output_element(k):
yield x
if k in self._hash_fields:
for x in self._output_decode_hash(v):
yield x
else:
for x in self._output_element(v):
yield x
yield TorrentFileParser.END_INDICATOR
def _output_list(self, data):
yield TorrentFileParser.LIST_INDICATOR
for v in data:
for x in self._output_element(v):
yield x
yield TorrentFileParser.END_INDICATOR
def _type_to_func(self, t):
return getattr(self, '_output_' + t)
def _output_element(self, data):
for types, t in self.TYPES.items():
if isinstance(data, types):
# noinspection PyCallingNonCallable
return self._type_to_func(t)(data)
raise InvalidTorrentDataException(
None,
"Invalid type for torrent file: " + type(data).__name__,
)
class BDecoder(object):
def __init__(
self, data, use_ordered_dict=False, encoding='utf-8', errors='strict',
hash_fields=None, hash_raw=False,
):
"""
See :any:`TorrentFileParser.__init__` for parameter description.
:param bytes data: raw data to be decoded
:param bool use_ordered_dict:
:param str encoding:
:param str errors:
:param Dict[str, Tuple[int, bool]] hash_fields:
:param bool hash_raw:
"""
self._parser = TorrentFileParser(
io.BytesIO(bytes(data)),
use_ordered_dict,
encoding,
errors,
hash_fields,
hash_raw,
)
def hash_field(self, name, block_length=20, need_dict=False):
"""
See :any:`TorrentFileParser.hash_field` for parameter description
:param name:
:param block_length:
:param need_dict:
:return: return self, so you can chained call
"""
self._parser.hash_field(name, block_length, need_dict)
return self
def decode(self):
return self._parser.parse()
def encode(data, encoding='utf-8', hash_fields=None):
"""
Shortcut function for encode python object to torrent file format(bencode)
See :any:`BEncoder.__init__` for parameter description
:param dict|list|int|str|bytes data: data to be encoded
:param str encoding:
:param List[str] hash_fields:
:rtype: bytes
"""
return BEncoder(data, encoding, hash_fields).encode()
def decode(
data, use_ordered_dict=False, encoding='utf-8', errors='strict',
hash_fields=None, hash_raw=False,
):
"""
Shortcut function for decode bytes as torrent file format(bencode) to python
object
See :any:`BDecoder.__init__` for parameter description
:param bytes data: raw data to be decoded
:param bool use_ordered_dict:
:param str encoding:
:param str errors:
:param Dict[str, Tuple[int, bool]] hash_fields:
:param bool hash_raw:
:rtype: dict|list|int|str|bytes|bytes
"""
return BDecoder(
data, use_ordered_dict, encoding, errors, hash_fields, hash_raw,
).decode()
def parse_torrent_file(
filename, use_ordered_dict=False, encoding='utf-8', errors='strict',
hash_fields=None, hash_raw=False,
):
"""
Shortcut function for parse torrent object using TorrentFileParser
See :any:`TorrentFileParser.__init__` for parameter description
:param str filename: torrent filename
:param bool use_ordered_dict:
:param str encoding:
:param str errors:
:param Dict[str, Tuple[int, bool]] hash_fields:
:param bool hash_raw:
:rtype: dict|list|int|str|bytes
"""
with open(filename, 'rb') as f:
return TorrentFileParser(
f, use_ordered_dict, encoding, errors, hash_fields, hash_raw,
).parse()
def create_torrent_file(filename, data, encoding='utf-8', hash_fields=None):
"""
Shortcut function for create a torrent file using BEncoder
see :any:`BDecoder.__init__` for parameter description
:param str filename: output torrent filename
:param dict|list|int|str|bytes data:
:param str encoding:
:param List[str] hash_fields:
"""
with open(filename, 'wb') as f:
f.write(BEncoder(data, encoding, hash_fields).encode())
def __main():
parser = argparse.ArgumentParser()
parser.add_argument('file', nargs='?', default='',
help='input file, will read form stdin if empty')
parser.add_argument('--dict', '-d', action='store_true', default=False,
help='use built-in dict, default will be OrderedDict')
parser.add_argument('--sort', '-s', action='store_true', default=False,
help='sort output json item by key')
parser.add_argument('--indent', '-i', type=int, default=None,
help='json output indent for every inner level')
parser.add_argument('--ascii', '-a', action='store_true', default=False,
help='ensure output json use ascii char, '
'escape other char use \\u')
parser.add_argument('--coding', '-c', default='utf-8',
help='string encoding, default "utf-8"')
parser.add_argument('--errors', '-e', default='strict',
help='decoding error handler, default "strict", you can'
' use "ignore" or "replace" to avoid exception')
parser.add_argument('--version', '-v', action='store_true', default=False,
help='print version and exit')
args = parser.parse_args()
if args.version:
print(__version__)
exit(0)
try:
if args.file == '':
target_file = io.BytesIO(
getattr(sys.stdin, 'buffer', sys.stdin).read()
)
else:
target_file = open(args.file, 'rb')
except FileNotFoundError:
sys.stderr.write('File "{}" not exist\n'.format(args.file))
exit(1)
# noinspection PyUnboundLocalVariable
data = TorrentFileParser(
target_file, not args.dict, args.coding, args.errors
).parse()
data = json.dumps(
data, ensure_ascii=args.ascii,
sort_keys=args.sort, indent=args.indent
)
print(data)
if __name__ == '__main__':
__main()
-6
View File
@@ -1,6 +0,0 @@
[metadata]
description-file = README.md
license-file = LICENSE
[bdist_wheel]
universal = 1
-41
View File
@@ -1,41 +0,0 @@
#!/usr/bin/env python
# coding=utf-8
try:
from setuptools import setup
except ImportError:
from distutils.core import setup
import torrent_parser
setup(
name='torrent_parser',
keywords=['file', 'torrent', 'JSON', 'parser'],
version=torrent_parser.__version__,
py_modules=['torrent_parser'],
url='https://github.com/7sDream/torrent_parser',
license='MIT',
author='7sDream',
author_email='7seconddream@gmail.com',
description='A .torrent file parser for both Python 2 and 3',
install_requires=[],
entry_points={
'console_scripts': ['pytp=torrent_parser:__main']
},
classifiers=[
'Development Status :: 3 - Alpha',
'Environment :: Console',
'Intended Audience :: Developers',
'Intended Audience :: End Users/Desktop',
'License :: OSI Approved :: MIT License',
'Natural Language :: English',
'Operating System :: OS Independent',
'Programming Language :: Python',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Topic :: Multimedia',
'Topic :: Software Development :: Libraries :: Python Modules',
'Topic :: Utilities',
]
)
-1
View File
@@ -1 +0,0 @@
from .test_all import *
-56
View File
@@ -1,56 +0,0 @@
from __future__ import unicode_literals
import unittest
import collections
from torrent_parser import TorrentFileParser, parse_torrent_file
class Test(unittest.TestCase):
REAL_FILE = 'tests/testfiles/real.torrent'
NEG_FILE = 'tests/testfiles/neg.torrent'
def test_parse_torrent_file_use_shortcut(self):
parse_torrent_file(self.REAL_FILE)
def test_parse_torrent_file_use_class(self):
with open(self.REAL_FILE, 'rb') as fp:
TorrentFileParser(fp).parse()
def test_encoding_auto(self):
with open(self.REAL_FILE, 'rb') as fp:
TorrentFileParser(fp, encoding='auto').parse()
def test_parse_torrent_file_to_ordered_dict(self):
data = parse_torrent_file(self.REAL_FILE, True)
self.assertIsInstance(data, collections.OrderedDict)
with open(self.REAL_FILE, 'rb') as fp:
data = TorrentFileParser(fp, True).parse()
self.assertIsInstance(data, collections.OrderedDict)
def test_parse_correctness(self):
data = parse_torrent_file(self.REAL_FILE)
self.assertIn(['udp://tracker.publicbt.com:80/announce'],
data['announce-list'])
self.assertEqual(data['creation date'], 1409254242)
def test_parse_two_times(self):
with open(self.REAL_FILE, 'rb') as fp:
parser = TorrentFileParser(fp)
data = parser.parse()
self.assertIn(['udp://tracker.publicbt.com:80/announce'],
data['announce-list'])
self.assertEqual(data['creation date'], 1409254242)
data = parser.parse()
self.assertIn(['udp://tracker.publicbt.com:80/announce'],
data['announce-list'])
self.assertEqual(data['creation date'], 1409254242)
def test_int_is_negative(self):
data = parse_torrent_file(self.NEG_FILE)
self.assertEqual(data['neg'], -1)
if __name__ == '__main__':
unittest.main()
-1
View File
@@ -1 +0,0 @@
d3:negi-1ee
Binary file not shown.
-316
View File
@@ -1,316 +0,0 @@
#!/usr/bin/env python
# coding: utf-8
"""
A .torrent file parser for both Python 2 and 3
Usage:
data = parse_torrent_file(filename)
# or
with open(filename, 'rb') as f: # the binary mode 'b' is necessary
data = TorrentFileParser(f).parse()
"""
from __future__ import print_function, unicode_literals
import argparse
import collections
import io
import json
import sys
import warnings
try:
FileNotFoundError
except NameError:
# Python 2 do not have FileNotFoundError, use IOError instead
# noinspection PyShadowingBuiltins
FileNotFoundError = IOError
try:
# noinspection PyPackageRequirements
from chardet import detect as _detect
except ImportError:
def _detect(_):
warnings.warn("No chardet module installed, encoding will be utf-8")
return {'encoding': 'utf-8', 'confidence': 1}
__all__ = [
'InvalidTorrentDataException',
'parse_torrent_file',
'TorrentFileParser',
]
__version__ = '0.1.4'
def detect(content):
return _detect(content)['encoding']
class InvalidTorrentDataException(Exception):
def __init__(self, pos, msg=None):
msg = msg or "Invalid torrent format when read at pos {pos}"
msg = msg.format(pos=pos)
super(InvalidTorrentDataException, self).__init__(msg)
class __EndCls(object):
pass
_END = __EndCls()
class TorrentFileParser(object):
TYPE_LIST = 'list'
TYPE_DICT = 'dict'
TYPE_INT = 'int'
TYPE_STRING = 'string'
TYPE_END = 'end'
LIST_INDICATOR = b'l'
DICT_INDICATOR = b'd'
INT_INDICATOR = b'i'
END_INDICATOR = b'e'
STRING_INDICATOR = b''
TYPES = [
(TYPE_LIST, LIST_INDICATOR),
(TYPE_DICT, DICT_INDICATOR),
(TYPE_INT, INT_INDICATOR),
(TYPE_END, END_INDICATOR),
(TYPE_STRING, STRING_INDICATOR),
]
def __init__(self, fp, use_ordered_dict=False, encoding='utf-8'):
"""
:param fp: a **binary** file-like object to parse,
which means need 'b' mode when use built-in open function
:param encoding: file content encoding, default utf-8, use 'auto' to
enable charset auto detection ('chardet' package should be installed)
:param use_ordered_dict: Use collections.OrderedDict as dict container
default False, which mean use built-in dict
"""
if getattr(fp, 'read', ) is None \
or getattr(fp, 'seek') is None:
raise ValueError('Argument fp needs a file like object')
self._pos = 0
self._encoding = encoding
self._content = fp
self._use_ordered_dict = use_ordered_dict
def parse(self):
"""
:return: the parse result
:type: depends on ``use_ordered_dict`` option when init the parser
see :any:`TorrentFileParser.__init__`
"""
self._restart()
data = self._next_element()
try:
c = self._read_byte(1, True)
raise InvalidTorrentDataException(
0, 'Expect EOF, but get [{}] at pos {}'.format(c, self._pos)
)
except EOFError: # expect EOF
pass
if isinstance(data, dict):
return data
raise InvalidTorrentDataException('Outermost element is not a dict')
def _read_byte(self, count=1, raise_eof=False):
assert count >= 0
gotten = self._content.read(count)
if count != 0 and len(gotten) == 0:
if raise_eof:
raise EOFError()
raise InvalidTorrentDataException(
self._pos,
'Unexpected EOF when reading torrent file'
)
self._pos += count
return gotten
def _seek_back(self, count):
self._content.seek(-count, 1)
self._pos = self._pos - count
def _restart(self):
self._content.seek(0, 0)
self._pos = 0
def _dict_items_generator(self):
while True:
k = self._next_element()
if k is _END:
return
if k == 'pieces':
v = self._next_hash()
elif k == 'ed2k':
v = self._next_hash(16, False)
elif k == 'filehash':
v = self._next_hash(20, False)
else:
v = self._next_element()
if k == 'encoding':
self._encoding = v
yield k, v
def _next_dict(self):
data = collections.OrderedDict() if self._use_ordered_dict else dict()
for key, element in self._dict_items_generator():
data[key] = element
return data
def _list_items_generator(self):
while True:
element = self._next_element()
if element is _END:
return
yield element
def _next_list(self):
return [element for element in self._list_items_generator()]
def _next_int(self, end=END_INDICATOR):
value = 0
char = self._read_byte(1)
neg = False
while char != end:
if not neg and char == b'-':
neg = True
elif not b'0' <= char <= b'9':
raise InvalidTorrentDataException(self._pos - 1)
else:
value = value * 10 + int(char) - int(b'0')
char = self._read_byte(1)
return -value if neg else value
def _next_string(self, decode=True):
length = self._next_int(b':')
raw = self._read_byte(length)
if decode:
encoding = self._encoding
if encoding == 'auto':
encoding = detect(raw)
try:
string = raw.decode(encoding)
except UnicodeDecodeError as e:
raise InvalidTorrentDataException(
self._pos - length + e.start,
"Fail to decode string at pos {pos} using " + e.encoding
)
return string
return raw
@staticmethod
def __to_hex(v):
return hex(ord(v) if isinstance(v, str) else v)[2:].rjust(2, str(0))
def _next_hash(self, p_len=20, need_list=True):
raw = self._next_string(decode=False)
if len(raw) % p_len != 0:
raise InvalidTorrentDataException(
self._pos - len(raw), "Hash bit length not match at pos {pos}"
)
res = [
''.join([self.__to_hex(c) for c in h])
for h in (raw[x:x+p_len] for x in range(0, len(raw), p_len))
]
if len(res) == 0 and not need_list:
return ''
if len(res) == 1 and not need_list:
return res[0]
return res
@staticmethod
def _next_end():
return _END
def _next_type(self):
for (element_type, indicator) in self.TYPES:
indicator_length = len(indicator)
char = self._read_byte(indicator_length)
if indicator == char:
return element_type
self._seek_back(indicator_length)
raise InvalidTorrentDataException(self._pos)
def _type_to_func(self, t):
return getattr(self, '_next_' + t)
def _next_element(self):
element_type = self._next_type()
element = self._type_to_func(element_type)()
return element
def parse_torrent_file(filename, use_ordered_dict=False):
"""
Shortcut function for parse torrent object use TorrentFileParser
:param string filename: torrent filename
:param bool use_ordered_dict: see :any:`TorrentFileParser.__init__`
:rtype: dict if ``use_ordered_dict`` is false,
collections.OrderedDict otherwise
"""
with open(filename, 'rb') as f:
return TorrentFileParser(f, use_ordered_dict).parse()
def __main():
parser = argparse.ArgumentParser()
parser.add_argument('file', nargs='?', default='',
help='input file, will read form stdin if empty')
parser.add_argument('--dict', '-d', action='store_true', default=False,
help='use built-in dict, default will be OrderedDict')
parser.add_argument('--sort', '-s', action='store_true', default=False,
help='sort output json item by key')
parser.add_argument('--indent', '-i', type=int, default=None,
help='json output indent for every inner level')
parser.add_argument('--ascii', '-a', action='store_true', default=False,
help='ensure output json use ascii char, '
'escape other char use \\u')
parser.add_argument('--coding', '-c', default='utf-8',
help='string encoding, default utf-8')
parser.add_argument('--version', '-v', action='store_true', default=False,
help='print version and exit')
args = parser.parse_args()
if args.version:
print(__version__)
exit(0)
try:
if args.file == '':
target_file = io.BytesIO(
getattr(sys.stdin, 'buffer', sys.stdin).read()
)
else:
target_file = open(args.file, 'rb')
except FileNotFoundError:
sys.stderr.write('File "{}" not exist\n'.format(args.file))
exit(1)
# noinspection PyUnboundLocalVariable
data = TorrentFileParser(target_file, not args.dict, args.coding).parse()
data = json.dumps(
data, ensure_ascii=args.ascii,
sort_keys=args.sort, indent=args.indent
)
print(data)
if __name__ == '__main__':
__main()