# This file is part of torf. # # torf is free software: you can redistribute it and/or modify it under the # terms of the GNU General Public License as published by the Free Software # Foundation, either version 3 of the License, or (at your option) any later # version. # # torf is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR # A PARTICULAR PURPOSE. See the GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with torf. If not, see . import base64 import errno import hashlib import inspect import io import itertools import math import os import pathlib import re from collections import abc from datetime import datetime from lib.utils.torrent import flatbencode as bencode from . import __version__ from . import _errors as error from . import _generate as generate from . import _reuse as reuse from . import _utils as utils _PACKAGE_NAME = __name__.split(".")[0] # os.sched_getaffinity() is only available on some Unix platforms. try: NCORES = len(os.sched_getaffinity(0)) except AttributeError: import multiprocessing NCORES = multiprocessing.cpu_count() DEFAULT_TORRENT_NAME = "UNNAMED TORRENT" class Torrent: """ Torrent metainfo representation Create a new Torrent instance: >>> from torf import Torrent >>> torrent = Torrent('path/to/My Torrent', ... trackers=['https://localhost:123/announce'], ... comment='This is my first torrent. Be gentle.') Convenient access to metainfo via properties: >>> torrent.comment 'This is my first torrent. Be gentle.' >>> torrent.comment = "This is my first torrent. Let's rock!" >>> torrent.private = True Full control over unencoded metainfo: >>> torrent.metainfo['info']['private'] True >>> torrent.metainfo['more stuff'] = {'foo': 12, ... 'bar': ('x', 'y', 'z')} Hash pieces and update progress once per second: >>> def cb(torrent, filepath, pieces_done, pieces_total): ... print(f'{pieces_done/pieces_total*100:3.0f} % done') >>> success = torrent.generate(callback=cb, interval=1) 1 % done 2 % done [...] 100 % done Write torrent file: >>> torrent.write('my_torrent.torrent') Create magnet link: >>> torrent.magnet() 'magnet:?xt=urn:btih:e167b1fbb42ea72f051f4f50432703308efb8fd1&dn=My+Torrent&xl=142631&tr=https%3A%2F%2Flocalhost%3A123%2Fannounce' Read torrent from file: >>> t = Torrent.read('my_torrent.torrent') >>> t.comment "This is my first torrent. Let's rock!" >>> t.metainfo['more stuff'] {'bar': ['x', 'y', 'z'], 'foo': 12} """ def __init__( self, path=None, name=None, exclude_globs=(), exclude_regexs=(), include_globs=(), include_regexs=(), trackers=None, webseeds=None, httpseeds=None, private=None, comment=None, source=None, creation_date=None, created_by=f"{_PACKAGE_NAME} {__version__}", piece_size=None, piece_size_min=None, piece_size_max=None, randomize_infohash=False, ): self._path = None self._metainfo = {} self._exclude = { "globs": utils.MonitoredList(callback=self._filters_changed, type=str), "regexs": utils.MonitoredList(callback=self._filters_changed, type=re.compile), } self._include = { "globs": utils.MonitoredList(callback=self._filters_changed, type=str), "regexs": utils.MonitoredList(callback=self._filters_changed, type=re.compile), } self.piece_size_min = piece_size_min self.piece_size_max = piece_size_max self.trackers = trackers self.webseeds = webseeds self.httpseeds = httpseeds self.private = private self.comment = comment self.creation_date = creation_date self.created_by = created_by self.source = source self.randomize_infohash = randomize_infohash self.exclude_globs = exclude_globs self.exclude_regexs = exclude_regexs self.include_globs = include_globs self.include_regexs = include_regexs self.path = path # Values that are implicitly changed by setting self.path if piece_size is not None: self.piece_size = piece_size if name is not None: self.name = name @property def metainfo(self): """ Unencoded torrent metainfo as mutable mapping You can put anything in here as long as keys are convertable to :class:`bytes` and values are convertable to :class:`bytes`, :class:`int`, :class:`list` or :class:`dict`. See also :meth:`convert` and :meth:`validate`. The ``info`` key is guaranteed to exist. """ if "info" not in self._metainfo: self._metainfo["info"] = {} return self._metainfo @property def path(self): """ File system path to torrent content Files are filtered according to :attr:`exclude_globs`, :attr:`exclude_regexs`, :attr:`include_globs` and :attr:`include_regexs`. Setting or manipulating this property updates :attr:`metainfo`\\ ``['info']``: - ``name``, ``piece length`` and ``files`` or ``length`` are set. - ``pieces`` and ``md5sum`` are removed if they exist. :raises ReadError: if :attr:`path` or any path underneath it is not readable """ return self._path @path.setter def path(self, value): if value is None: # Keep info about name and files, but forget where they are stored self._path = None self.metainfo["info"].pop("pieces", None) else: basepath = pathlib.Path(str(value)) filepaths = tuple( utils.File(fp, size=utils.real_size(fp)) for fp in utils.list_files(basepath) ) self._set_files(filepaths, basepath) @property def location(self): """ :attr:`path` without :attr:`name` or `None` if :attr:`path` is `None` """ if self.path: return self._path.parent @property def files(self): """ List of relative paths in this torrent Paths are :class:`File` objects and items are automatically deduplicated. Every path starts with :attr:`name`. Setting or manipulating this property updates :attr:`metainfo`\\ ``['info']``: - ``name``, ``piece length`` and ``files`` or ``length`` are set. - ``pieces`` and ``md5sum`` are removed if they exist. See :attr:`filepaths` for a list of file system paths. :raises PathError: if any path is absolute :raises CommonPathError: if not all files share a common parent directory :raises ValueError: if any file is not a :class:`File` object """ info = self.metainfo["info"] if self.mode == "singlefile": files = (utils.File(info.get("name", DEFAULT_TORRENT_NAME), size=self.size),) elif self.mode == "multifile": basedir = info.get("name", DEFAULT_TORRENT_NAME) files = ( utils.File(os.path.join(basedir, *fileinfo["path"]), size=fileinfo["length"]) for fileinfo in info["files"] ) else: files = () return utils.Files(files, callback=self._files_changed) def _files_changed(self, files): self.files = files @files.setter def files(self, files): if not isinstance(files, utils.Iterable): raise ValueError(f"Not an Iterable: {files}") for f in files: if not isinstance(f, utils.File): raise ValueError(f"Not a File object: {f}") elif f.is_absolute(): raise error.PathError(f, msg="Not a relative path") if not files: self._set_files(files=()) else: # os.path.commonpath() returns '' if there is no common path and # raises ValueError if there are absolute and relative paths. try: basepath = os.path.commonpath(files) except ValueError: basepath = "" if basepath == "": raise error.CommonPathError(files) self._set_files(files, pathlib.Path(basepath)) @property def filepaths(self): """ List of paths of existing files in :attr:`path` included in the torrent Paths are :class:`Filepath` objects and items are automatically deduplicated. Directories are resolved into a list of files. Setting or manipulating this property updates :attr:`metainfo`\\ ``['info']``: - ``name``, ``piece length`` and ``files`` or ``length`` are set. - ``pieces`` and ``md5sum`` are removed if they exist. :raises ReadError: if any file path is not readable """ filepaths = () if self.path is not None: if self.mode == "singlefile": filepaths = (self.path,) elif self.mode == "multifile": dirpath = self.path filepaths = ( os.path.join(dirpath, *fileinfo["path"]) for fileinfo in self.metainfo["info"]["files"] ) return utils.Filepaths(filepaths, callback=self._filepaths_changed) def _filepaths_changed(self, filepaths): self.filepaths = filepaths @filepaths.setter def filepaths(self, filepaths): if not isinstance(filepaths, utils.Iterable): raise ValueError(f"Not an Iterable: {filepaths}") filepaths = utils.Filepaths(filepaths) # Resolve directories if not filepaths: self._set_files(files=()) else: # Make all paths absolute so we can find the common path. Do not # resolve symlinks so the user isn't confronted with unexpected # paths in case of an error. cwd = pathlib.Path.cwd() filepaths_abs = tuple(fp if fp.is_absolute() else cwd / fp for fp in filepaths) try: basepath = pathlib.Path(os.path.commonpath(filepaths_abs)) except ValueError: raise error.CommonPathError(filepaths) filepaths = tuple(utils.File(fp, size=utils.real_size(fp)) for fp in filepaths) self._set_files(filepaths, basepath) def _set_files(self, files, basepath=None): """ Update ``name`` and ``files`` or ``length``, remove ``pieces`` and ``md5sum`` in :attr:`metainfo`\\ ``['info']`` :param files: Sequence of :class:`File` :param basepath: path-like that all paths in `files` start with; may be ``None`` if ``files`` is empty """ def abspath(p): # Absolute path without resolved symlinks if p.is_absolute(): return pathlib.Path(os.path.normpath(p)) else: return pathlib.Path.cwd() / os.path.normpath(p) def relpath_without_parent(p): # Relative path without common parent directory return pathlib.Path(abspath(p)).relative_to(abspath(basepath)) def relpath_with_parent(p): # Relative path with common parent directory return pathlib.Path(abspath(p)).relative_to(abspath(basepath).parent) # Apply filters to relative paths with torrent name as first segment exclude_globs = tuple(str(g) for g in self._exclude["globs"]) exclude_regexs = tuple(re.compile(r) for r in self._exclude["regexs"]) exclude = tuple(itertools.chain(exclude_globs, exclude_regexs)) include_globs = tuple(str(g) for g in self._include["globs"]) include_regexs = tuple(re.compile(r) for r in self._include["regexs"]) include = tuple(itertools.chain(include_globs, include_regexs)) files = utils.filter_files( files, getter=relpath_with_parent, exclude=exclude, include=include, hidden=False, empty=False, ) info = self.metainfo["info"] if not files or all(f.size <= 0 for f in files): info.pop("files", None) info.pop("length", None) info.pop("pieces", None) info.pop("md5sum", None) elif len(files) == 1 and files[0] == basepath: # There is only one file and it is not in a directory. # NOTE: A directory with a single file in it is a multifile torrent. info["length"] = files[0].size info["name"] = files[0].name info.pop("files", None) info.pop("pieces", None) info.pop("md5sum", None) else: if str(basepath) == os.curdir: # Name of current working directory name = pathlib.Path.cwd().name elif str(basepath) == os.pardir: # Name of logical parent directory # NOTE: Path.resolve() returns the physical parent directory; if # the parent directory is a symlink, we get an unexpected name name = os.path.basename(os.path.dirname(os.getcwd())) elif str(basepath).endswith(os.curdir) or str(basepath).endswith(os.pardir): # Name of current/parent directory (logical parent, see NOTE above) name = pathlib.Path(os.path.normpath(basepath)).name else: name = basepath.name files_info = [] for f in sorted(files): files_info.append({"length": f.size, "path": list(relpath_without_parent(f).parts)}) info["name"] = name info["files"] = files_info info.pop("length", None) info.pop("pieces", None) info.pop("md5sum", None) # Set new path attribute if basepath exists if basepath is not None and os.path.exists(basepath): self._path = basepath else: self._path = None # Calculate new piece size self.piece_size = None @property def exclude_globs(self): """ List of case-insensitive wildcard patterns to exclude Include patterns take precedence over exclude patterns to allow including files that match an exclude pattern. Patterns are matched against paths in :attr:`files`. ========== ================================ Wildcard Description ========== ================================ ``*`` matches everything ``?`` matches any single character ``[SEQ]`` matches any character in ``SEQ`` ``[!SEQ]`` matches any char not in ``SEQ`` ========== ================================ """ return self._exclude["globs"] @exclude_globs.setter def exclude_globs(self, value): if not isinstance(value, utils.Iterable): raise ValueError(f"Must be Iterable, not {type(value).__name__}: {value}") self._exclude["globs"][:] = value @property def include_globs(self): """ List of case-insensitive wildcard patterns to include See :attr:`exclude_globs`. """ return self._include["globs"] @include_globs.setter def include_globs(self, value): if not isinstance(value, utils.Iterable): raise ValueError(f"Must be Iterable, not {type(value).__name__}: {value}") self._include["globs"][:] = value @property def exclude_regexs(self): """ List of regular expression patterns to exclude Include patterns take precedence over exclude patterns to allow including files that match an exclude pattern. Patterns are matched against paths in :attr:`files`. :raises re.error: if any regular expression is invalid """ return self._exclude["regexs"] @exclude_regexs.setter def exclude_regexs(self, value): if not isinstance(value, utils.Iterable): raise ValueError(f"Must be Iterable, not {type(value).__name__}: {value}") self._exclude["regexs"][:] = value @property def include_regexs(self): """ List of regular expression patterns to include See :attr:`exclude_regexs`. """ return self._include["regexs"] @include_regexs.setter def include_regexs(self, value): if not isinstance(value, utils.Iterable): raise ValueError(f"Must be Iterable, not {type(value).__name__}: {value}") self._include["regexs"][:] = value def _filters_changed(self, _): """Callback for MonitoredLists in Torrent._exclude""" # Apply filters if self.path is not None: # Read file list from disk again self.path = self.path else: # There are no existing files specified so we can just remove files self.files = self.files @property def filetree(self): """ :attr:`files` as a dictionary tree Parent nodes are dictionaries and leaf nodes are :class:`File` objects. The top node is always a dictionary with the single key :attr:`name`. Example: .. code:: python {'Torrent': {'bar': {'baz.mp3': File('Torrent/bar/baz.mp3', size=543210), 'baz.pdf': File('Torrent/bar/baz.pdf', size=999)}, 'foo.txt': File('Torrent/foo.txt', size=123456)}} """ tree = {} # Complete directory tree paths = (tuple(f.parts) for f in self.files) for path in paths: dirpath = path[:-1] # Path without filename filename = path[-1] subtree = tree for item in dirpath: if item not in subtree: subtree[item] = {} subtree = subtree[item] subtree[filename] = utils.File(path, size=self.partial_size(path)) return tree @property def name(self): """ Name of the torrent Default to last item in :attr:`path` or ``None`` if :attr:`path` is ``None``. If this property is set to ``None`` and :attr:`path` is not ``None``, it is set to the default. Setting this property sets or removes ``name`` in :attr:`metainfo`\\ ``['info']``. """ if "name" not in self.metainfo["info"] and self.path is not None: self.metainfo["info"]["name"] = self.path.name return self.metainfo["info"].get("name", None) @name.setter def name(self, value): if value is None: self.metainfo["info"].pop("name", None) self.name # Set default name else: self.metainfo["info"]["name"] = str(value) @property def mode(self): """ ``singlefile`` if this torrent contains one file that is not in a directory, ``multifile`` if it contains one or more files in a directory, or ``None`` if no content is specified (i.e. :attr:`files` is empty). """ if "length" in self.metainfo["info"]: return "singlefile" elif "files" in self.metainfo["info"]: return "multifile" @property def size(self): """Total size of content in bytes""" if self.mode == "singlefile": return self.metainfo["info"]["length"] elif self.mode == "multifile": return sum(fileinfo["length"] for fileinfo in self.metainfo["info"]["files"]) else: return 0 def partial_size(self, path): """ Return size of one or more files as specified in :attr:`metainfo` :param path: Relative path within torrent, starting with :attr:`name`; may point to file or directory :type path: str, path-like or iterable :raises PathError: if `path` is not known """ if isinstance(path, str): path = tuple(path.split(os.sep)) elif isinstance(path, os.PathLike): path = tuple(path.parts) elif isinstance(path, abc.Iterable): path = tuple(str(part) for part in path) else: raise ValueError(f"Must be str, Path or Iterable, not {type(path).__name__}: {path}") if self.mode == "singlefile" and path == (self.name,): return self.metainfo["info"]["length"] elif self.mode == "multifile": file_sizes = [] for info in self.metainfo["info"]["files"]: this_path = (self.name,) + tuple(c for c in info["path"] if c) if this_path == path: # path points to file return info["length"] elif utils.iterable_startswith(this_path, path): # path points to directory file_sizes.append(info["length"]) if file_sizes: return sum(file_sizes) raise error.PathError(os.path.join(*path), msg="Unknown path") @property def piece_size(self): """ Length of each piece in bytes If set to ``None`` and :attr:`size` is larger than 0, use the return value of :attr:`calculate_piece_size`. If set to ``None`` and :attr:`size` is smaller than 1, remove :attr:`metainfo`\\ ``['info']``\\ ``['piece length']`` . Setting this property sets or removes ``piece length`` in :attr:`metainfo`\\ ``['info']``. """ return self.metainfo["info"].get("piece length", 0) @piece_size.setter def piece_size(self, value): if value is None: if self.size <= 0: self.metainfo["info"].pop("piece length", None) return else: value = self.calculate_piece_size( self.size, min_size=self.piece_size_min, max_size=self.piece_size_max, ) try: piece_length = int(value) except (TypeError, ValueError): raise ValueError(f"piece_size must be int, not {type(value).__name__}: {value!r}") else: if not utils.is_divisible_by_16_kib(piece_length): raise error.PieceSizeError(piece_length) elif not self.piece_size_min <= piece_length <= self.piece_size_max: raise error.PieceSizeError( piece_length, min=self.piece_size_min, max=self.piece_size_max ) self.metainfo["info"]["piece length"] = piece_length @property def piece_size_min(self): """ Smallest allowed piece size Setting :attr:`piece_size` to a smaller value raises :class:`PieceSizeError`. """ return self._piece_size_min @piece_size_min.setter def piece_size_min(self, piece_size_min): if piece_size_min is None: self._piece_size_min = type(self).piece_size_min_default elif not utils.is_divisible_by_16_kib(piece_size_min): raise error.PieceSizeError(piece_size_min) else: self._piece_size_min = int(piece_size_min) @property def piece_size_max(self): """ Smallest allowed piece size Setting :attr:`piece_size` to a smaller value raises :class:`PieceSizeError`. """ return self._piece_size_max @piece_size_max.setter def piece_size_max(self, piece_size_max): if piece_size_max is None: self._piece_size_max = type(self).piece_size_max_default elif not utils.is_divisible_by_16_kib(piece_size_max): raise error.PieceSizeError(piece_size_max) else: self._piece_size_max = int(piece_size_max) piece_size_min_default = 16 * 1024 # 16 KiB """ Smallest allowed piece size (default value) Setting :attr:`piece_size` to a smaller value raises :class:`PieceSizeError`. """ piece_size_max_default = 16 * 1024 * 1024 # 16 MiB """ Greatest allowed piece size (default value) Setting :attr:`piece_size` to a greater value raises :class:`PieceSizeError`. """ @classmethod def calculate_piece_size(cls, size, min_size=None, max_size=None): """ Return the piece size for a total torrent size of ``size`` bytes For torrents up to 1 GiB, the maximum number of pieces is 1024 which means the maximum piece size is 1 MiB. With increasing torrent size both the number of pieces and the maximum piece size are gradually increased up to 10,240 pieces of 8 MiB. For torrents larger than 80 GiB the piece size is :attr:`piece_size_max` with as many pieces as necessary. It is safe to override this method to implement a custom algorithm. :param int min_size: Minimum piece size; defaults to :attr:`Torrent.piece_size_min` :param int max_size: Maximum piece size; defaults to :attr:`Torrent.piece_size_max` :return: calculated piece size """ if size <= 2**30: # 1 GiB / 1024 pieces = 1 MiB max pieces = size / 1024 elif size <= 4 * 2**30: # 4 GiB / 2048 pieces = 2 MiB max pieces = size / 2048 elif size <= 6 * 2**30: # 6 GiB / 3072 pieces = 2 MiB max pieces = size / 3072 elif size <= 8 * 2**30 or size <= 16 * 2**30: # 8 GiB / 2048 pieces = 4 MiB max pieces = size / 2048 elif size <= 32 * 2**30: # 32 GiB / 4096 pieces = 8 MiB max pieces = size / 4096 elif size <= 64 * 2**30: # 64 GiB / 8192 pieces = 8 MiB max pieces = size / 8192 else: pieces = size / 10240 if min_size is None: min_size = cls.piece_size_min_default if max_size is None: max_size = cls.piece_size_max_default # Math is magic! piece_size = 1 << max(0, math.ceil(math.log(pieces, 2))) return int(min(max(piece_size, min_size), max_size)) @property def pieces(self): """Number of pieces the content is split into""" size, piece_size = self.size, self.piece_size if size and piece_size and size > 0 and piece_size > 0: return math.ceil(size / piece_size) else: return 0 @property def hashes(self): """Tuple of SHA1 piece hashes as :class:`bytes`""" hashes = self.metainfo["info"].get("pieces") if isinstance(hashes, (bytes, bytearray)): # Each hash is 20 bytes long return tuple(bytes(hashes[pos : pos + 20]) for pos in range(0, len(hashes), 20)) else: return () @property def trackers(self): """ List of tiers (lists) of announce URLs http://bittorrent.org/beps/bep_0012.html This is a smart list that ensures the proper list-of-lists structure, validation and deduplication. You can set this property to a URL, an iterable of URLs or an iterable of iterables of URLs (i.e. "tiers"). This property automatically sets :attr:`metainfo`\\ ``['announce']`` and :attr:`metainfo`\\ ``['announce-list']`` when it is manipulated or set according to these rules: - If it contains a single URL, :attr:`metainfo`\\ ``['announce']`` is set and :attr:`metainfo`\\ ``['announce-list']`` is removed if it exists. - If it contains an iterable of URLs, :attr:`metainfo`\\ ``['announce']`` is set to the first URL and :attr:`metainfo`\\ ``['announce-list']`` is set to a list of tiers, one tier for each URL. - If it contains an iterable of iterables of URLs, :attr:`metainfo`\\ ``['announce']`` is set to the first URL of the first iterable and :attr:`metainfo`\\ ``['announce-list']`` is set to a list of tiers, one tier for each iterable of URLs. :raises URLError: if any of the announce URLs is invalid :raises ValueError: if set to anything that isn't an iterable or a string """ tiers = list(self.metainfo.get("announce-list", ())) announce = self.metainfo.get("announce", None) flat_urls = tuple(url for tier in tiers for url in tier) if announce is not None and announce not in flat_urls: tiers.insert(0, [announce]) return utils.Trackers(tiers, callback=self._trackers_changed) @trackers.setter def trackers(self, value): if value is None: value = () if isinstance(value, abc.Iterable): self._trackers_changed(utils.Trackers(value)) else: raise ValueError(f"Must be Iterable, str or None, not {type(value).__name__}: {value}") def _trackers_changed(self, trackers): # Set "announce" to first tracker of first tier try: self.metainfo["announce"] = str(trackers[0][0]) except IndexError: self.metainfo.pop("announce", None) # Remove "announce-list" if there's only one tracker if len(trackers.flat) <= 1: self.metainfo.pop("announce-list", None) else: if "announce-list" not in self.metainfo: self.metainfo["announce-list"] = [] # Set announce-list without changing its identity self.metainfo["announce-list"][:] = ([str(url) for url in tier] for tier in trackers) @property def webseeds(self): """ List of webseed URLs http://bittorrent.org/beps/bep_0019.html The list returned by this property automatically updates :attr:`metainfo`\\ ``['url-list']`` when manipulated. Setting this property sets :attr:`metainfo`\\ ``['url-list']``. :raises URLError: if any URL is invalid :raises ValueError: if set to anything that isn't an iterable or a string """ return utils.URLs(self.metainfo.get("url-list", ()), callback=self._webseeds_changed) @webseeds.setter def webseeds(self, value): if isinstance(value, str): urls = utils.URLs((value,)) elif isinstance(value, abc.Iterable): urls = utils.URLs(value) elif value is None: urls = utils.URLs(()) else: raise ValueError(f"Must be Iterable, str or None, not {type(value).__name__}: {value}") self._webseeds_changed(urls) def _webseeds_changed(self, webseeds): if webseeds: self.metainfo["url-list"] = [str(url) for url in webseeds] else: self.metainfo.pop("url-list", None) @property def httpseeds(self): """ List of webseed URLs http://bittorrent.org/beps/bep_0017.html The list returned by this property automatically updates :attr:`metainfo`\\ ``['httpseeds']`` when manipulated. Setting this property sets :attr:`metainfo`\\ ``['httpseeds']``. :raises URLError: if any URL is invalid :raises ValueError: if set to anything that isn't an iterable or a string """ return utils.URLs(self.metainfo.get("httpseeds", ()), callback=self._httpseeds_changed) @httpseeds.setter def httpseeds(self, value): if isinstance(value, str): urls = utils.URLs((value,)) elif isinstance(value, abc.Iterable): urls = utils.URLs(value) elif value is None: urls = utils.URLs(()) else: raise ValueError(f"Must be Iterable, str or None, not {type(value).__name__}: {value}") self._httpseeds_changed(urls) def _httpseeds_changed(self, httpseeds): if httpseeds: self.metainfo["httpseeds"] = [str(url) for url in httpseeds] else: self.metainfo.pop("httpseeds", None) @property def private(self): """ Whether torrent should use trackers exclusively for peer discovery ``True`` or ``False`` if :attr:`metainfo`\\ ``['info']``\\ ``['private']`` exists, ``None`` otherwise. """ if "private" in self.metainfo["info"]: return bool(self.metainfo["info"]["private"]) else: return None @private.setter def private(self, value): if value is None: self.metainfo["info"].pop("private", None) else: self.metainfo["info"]["private"] = bool(value) @property def comment(self): """ Comment string or ``None`` for no comment Setting this property sets or removes :attr:`metainfo`\\ ``['comment']``. """ return self.metainfo.get("comment", None) @comment.setter def comment(self, value): if value is not None: self.metainfo["comment"] = str(value) else: self.metainfo.pop("comment", None) @property def creation_date(self): """ :class:`datetime.datetime` instance or ``None`` for no creation date :class:`int` and :class:`float` are also allowed and converted with :meth:`datetime.datetime.fromtimestamp`. Setting this property sets or removes :attr:`metainfo`\\ ``['creation date']``. """ return self.metainfo.get("creation date", None) @creation_date.setter def creation_date(self, value): if isinstance(value, (float, int)): self.metainfo["creation date"] = datetime.fromtimestamp(value) elif isinstance(value, datetime): self.metainfo["creation date"] = value elif not value: self.metainfo.pop("creation date", None) else: raise ValueError( f"Must be None, int or datetime object, not {type(value).__name__}: {value!r}" ) @property def created_by(self): """ Application name or ``None`` for no creator Setting this property sets or removes :attr:`metainfo`\\ ``['created by']``. """ return self.metainfo.get("created by", None) @created_by.setter def created_by(self, value): if value is not None: self.metainfo["created by"] = str(value) else: self.metainfo.pop("created by", None) @property def source(self): """ Source string or ``None`` for no source Setting this property sets or removes :attr:`metainfo`\\ ``['info']``\\ ``['created by']``. """ return self.metainfo["info"].get("source", None) @source.setter def source(self, value): if value is not None: self.metainfo["info"]["source"] = str(value) else: self.metainfo["info"].pop("source", None) @property def infohash(self): """ SHA1 info hash :raises MetainfoError: if :attr:`validate` fails or :attr:`metainfo` is not bencodable """ try: # Try to calculate infohash self.validate() try: info = utils.encode_dict(self.metainfo["info"]) except ValueError as e: raise error.MetainfoError(e) else: return hashlib.sha1(bencode.encode(info)).hexdigest() except error.MetainfoError as e: # If we can't calculate infohash, see if it was explicitly specifed. # This is necessary to create a Torrent from a Magnet URI. try: return self._infohash except AttributeError: raise e @property def infohash_base32(self): """Base 32 encoded SHA1 info hash""" return base64.b32encode(base64.b16decode(self.infohash.upper())) @property def randomize_infohash(self): """ Whether to ensure that :attr:`infohash` is always unique This allows for cross-seeding without changing :attr:`piece_size`. Setting this property to ``True`` sets :attr:`metainfo`\\ ``['info']``\\ ``['entropy']`` to a random integer. Setting it to ``False`` removes that field. """ return bool(self.metainfo["info"].get("entropy", False)) @randomize_infohash.setter def randomize_infohash(self, value): if value: # According to BEP0003 "Integers have no size limitation", but some # parsers seem to have problems with large numbers. import random self.metainfo["info"]["entropy"] = random.randint(int(-2e9), int(2e9)) else: self.metainfo["info"].pop("entropy", None) @property def is_ready(self): """Whether this torrent is ready to be exported to a file or magnet link""" try: self.validate() except error.MetainfoError: return False else: return True def generate(self, threads=None, callback=None, interval=0): """ Hash pieces and report progress to `callback` This method sets :attr:`metainfo`\\ ``['info']``\\ ``['pieces']`` after all pieces are hashed successfully. :param int threads: How many threads to use for hashing pieces or ``None`` to use one thread per available CPU core :param callable callback: Callable to report progress and/or abort `callback` must accept 4 positional arguments: 1. Torrent instance (:class:`Torrent`) 2. Path of the currently hashed file (:class:`str`) 3. Number of hashed pieces (:class:`int`) 4. Total number of pieces (:class:`int`) If `callback` returns anything that is not ``None``, hashing is stopped. :param float interval: Minimum number of seconds between calls to `callback`; if 0, `callback` is called once per hashed piece :raises PathError: if :attr:`path` contains only empty files/directories :raises ReadError: if :attr:`path` or any file beneath it is not readable :raises RuntimeError: if :attr:`path` is None :return: ``True`` if all pieces were successfully hashed, ``False`` otherwise """ if self.path is None: raise RuntimeError("generate() called with no path specified") elif sum(utils.real_size(fp) for fp in self.filepaths) < 1: raise error.PathError(self.path, msg="Empty or all files excluded") hasher_threads = threads or NCORES # Read piece_size'd chunks from disk and send them to HasherPool reader = generate.Reader(torrent=self, queue_size=hasher_threads * 3) # Multiple threads that get chunks from Reader, calculate the hashes, # and push them to a hash queue hashers = generate.HasherPool( hasher_threads=hasher_threads, piece_queue=reader.piece_queue, ) # Collect piece hashes from HasherPool and call `callback` for status # reporting/cancellation collector = generate.Collector( torrent=self, reader=reader, hashers=hashers, callback=generate.GenerateCallback( callback=callback, interval=interval, torrent=self, ), ) # Collect piece hashes piece_hashes = collector.collect() concatenated_piece_hashes = b"".join(piece_hashes) hashes_count = len(concatenated_piece_hashes) / 20 if hashes_count == self.pieces: self.metainfo["info"]["pieces"] = concatenated_piece_hashes return True elif hashes_count < self.pieces: # Hashing was cancelled return False else: raise RuntimeError( f"Unexpected number of hashes generated: {hashes_count} instead of {self.pieces}" ) def verify(self, path, threads=None, callback=None, interval=0): """ Check if `path` contains all the data specified in this torrent Generate hashes from file content and compare them to the ones stored in :attr:`metainfo`\\ ``['info']``\\ ``['pieces']``. :param str path: Directory or file to read from :param int threads: How many threads to use for hashing pieces or ``None`` to use one thread per available CPU core :param callable callback: Callable to report progress and/or abort `callback` must accept 7 positional arguments: 1. Torrent instance (:class:`Torrent`) 2. File path in file system (:class:`str`) 3. Number of checked pieces (:class:`int`) 4. Total number of pieces (:class:`int`) 5. Index of the current piece (:class:`int`) 6. SHA1 hash of the current piece (:class:`bytes` or ``None``) 7. Exception (:class:`TorfError` or ``None``) If `callback` returns anything that is not ``None``, verification is stopped. :param float interval: Minimum number of seconds between calls to `callback` (if 0, `callback` is called once per piece); this is ignored if an error is found If a callback is specified, exceptions are not raised but passed to `callback` instead. :raises VerifyContentError: if a file contains unexpected data :raises VerifyIsDirectoryError: if `path` is a directory and this torrent contains a single file :raises VerifyNotDirectoryError: if `path` is a file and this torrent contains a directory :raises ReadError: if a file is not readable :raises MetainfoError: if :meth:`validate` fails :return: ``True`` if `path` is verified successfully, ``False`` otherwise """ # First make sure we are a valid torrent self.validate() # Wrapper around callback function that compares hashes verify_callback = generate.VerifyCallback( callback=callback, interval=interval, torrent=self, path=path, ) def early_exception(exception): piece_index = 0 pieces_done = 0 pieces_total = self.pieces filepath = None piece_hash = None exceptions = (exception,) verify_callback( piece_index, pieces_done, pieces_total, filepath, piece_hash, exceptions ) if self.mode == "singlefile" and os.path.isdir(path): early_exception(error.VerifyIsDirectoryError(path)) return False elif self.mode == "multifile" and not os.path.isdir(path): early_exception(error.VerifyNotDirectoryError(path)) return False else: hasher_threads = threads or NCORES # Read piece_size'd chunks from disk and send them to HasherPool reader = generate.Reader( torrent=self, queue_size=hasher_threads * 3, path=path, ) # Multiple threads that get chunks from Reader, calculate the hashes, # and push them to a hash queue hashers = generate.HasherPool( hasher_threads=hasher_threads, piece_queue=reader.piece_queue, ) # Collect piece hashes from HasherPool and call `callback` for status # reporting/cancellation collector = generate.Collector( torrent=self, reader=reader, hashers=hashers, callback=verify_callback, ) piece_hashes = collector.collect() return piece_hashes == self.hashes def verify_filesize(self, path, callback=None): """ Check if `path` has the expected file size Walk through :attr:`files` and check if each file exists relative to `path`, is readable and has the correct size. Excess files in `path` are ignored. This is fast and should find most manipulations, but :meth:`verify` is necessary to detect corruptions (e.g. due to bit rot). :param str path: Directory or file to check :param callable callback: Callable to report progress and/or abort `callback` must accept 6 positional arguments: 1. Torrent instance (:class:`Torrent`) 2. File path in file system (:class:`str`) 3. File path in torrent (:class:`str`) 4. Number of checked files (:class:`int`) 5. Total number of files (:class:`int`) 6. Exception (:class:`TorfError`) or ``None`` If `callback` returns anything that is not ``None``, verification is stopped. If a callback is specified, exceptions are not raised but passed to `callback` instead. :raises VerifyFileSizeError: if a file has an unexpected size :raises VerifyIsDirectoryError: if `path` is a directory and this torrent contains a single file :raises ReadError: if any file's size can't be determined :raises MetainfoError: if :meth:`validate` fails :return: ``True`` if `path` is verified successfully, ``False`` otherwise """ # First make sure we are a valid torrent self.validate() # Generate an ordered list of file system paths and their corresponding # paths inside the torrent # NOTE: The last segment in `path` is supposed to be the torrent name so # we must remove the name from each `file`. This allows # verification of any renamed file/directory against a torrent. filepaths = tuple( ( utils.File([path, *file.parts[1:]], file.size), file, ) for file in self.files ) files_total = len(filepaths) def cancel(file_index, exception): if callback: fs_filepath = filepaths[file_index][0] torrent_filepath = filepaths[file_index][1] files_done = file_index + 1 return_value = callback( self, fs_filepath, torrent_filepath, files_done, files_total, exception, ) if return_value is not None: return True elif exception: # Raise exception if there is no callback to handle it raise exception else: return False exception = None # If we expect a file, check if path is a file. We don't need to check # for a directory if we expect one because we are iterating over files # (filepaths), so the path "foo/bar/baz" will result in a ReadError if # "foo" or "foo/bar" is a file. if self.mode == "singlefile" and os.path.isdir(path): exception = error.VerifyIsDirectoryError(path) cancel(file_index=0, exception=exception) return False for file_index, (fs_filepath, torrent_filepath) in enumerate(filepaths): # Check if path exists if not os.path.exists(fs_filepath): exception = error.ReadError(errno.ENOENT, fs_filepath) if cancel(file_index, exception): return False else: continue # Check file size fs_filepath_size = utils.real_size(fs_filepath) expected_size = self.partial_size(torrent_filepath) if fs_filepath_size != expected_size: exception = error.VerifyFileSizeError(fs_filepath, fs_filepath_size, expected_size) if cancel(file_index, exception): return False else: continue # Report no error for current file if cancel(file_index, exception=None): return False if exception: # `exception` is just an indicator of success/failure. At this point # it was already raised or passed to cancel(). return False else: return True def validate(self): """ Check if all mandatory keys exist in :attr:`metainfo` and all standard keys have correct types References: | http://bittorrent.org/beps/bep_0003.html | https://wiki.theory.org/index.php/BitTorrentSpecification#Metainfo_File_Structure :raises MetainfoError: if :attr:`metainfo` would not generate a valid torrent file or magnet link """ md = self.metainfo info = md["info"] # Check values shared by singlefile and multifile torrents utils.assert_type(md, ("info",), (dict,), must_exist=True) utils.assert_type(md, ("info", "name"), (str,), must_exist=True) utils.assert_type( md, ("info", "piece length"), (int,), must_exist=True, check=utils.is_divisible_by_16_kib, ) utils.assert_type(md, ("info", "pieces"), (abc.ByteString,), must_exist=True) utils.assert_type(md, ("info", "private"), (bool, int), must_exist=False) utils.assert_type(md, ("announce",), (str,), must_exist=False, check=utils.is_url) utils.assert_type(md, ("announce-list",), (utils.Iterable,), must_exist=False) for i, _ in enumerate(md.get("announce-list", ())): utils.assert_type(md, ("announce-list", i), (utils.Iterable,)) for j, _ in enumerate(md["announce-list"][i]): utils.assert_type(md, ("announce-list", i, j), (str,), check=utils.is_url) if len(info["pieces"]) == 0: raise error.MetainfoError("['info']['pieces'] is empty") elif len(info["pieces"]) % 20 != 0: raise error.MetainfoError("length of ['info']['pieces'] is not divisible by 20") elif "length" in info and "files" in info: raise error.MetainfoError("['info'] includes both 'length' and 'files'") elif "length" in info: # Validate info as singlefile torrent utils.assert_type(md, ("info", "length"), (int, float), must_exist=True) utils.assert_type( md, ("info", "md5sum"), (str,), must_exist=False, check=utils.is_md5sum ) # Validate expected number of pieces piece_count = int(len(info["pieces"]) / 20) exp_piece_count = math.ceil(info["length"] / info["piece length"]) if piece_count != exp_piece_count: raise error.MetainfoError( f"Expected {exp_piece_count} pieces but there are {piece_count}" ) if self.path is not None: # Check if filepath actually points to a file if not os.path.isfile(self.path): raise error.MetainfoError( f"Metainfo includes {self.path} as file, but it is not a file" ) # Check if size matches path_size = utils.real_size(self.path) if path_size != info["length"]: raise error.MetainfoError( f"Mismatching file sizes in metainfo ({info['length']})" f" and file system ({path_size}): {self.path}" ) elif "files" in info: # Validate info as multifile torrent utils.assert_type(md, ("info", "files"), (utils.Iterable,), must_exist=True) for i, fileinfo in enumerate(info["files"]): utils.assert_type(md, ("info", "files", i), (abc.Mapping,), must_exist=True) utils.assert_type(md, ("info", "files", i, "length"), (int, float), must_exist=True) utils.assert_type( md, ("info", "files", i, "path"), (utils.Iterable,), must_exist=True ) utils.assert_type( md, ("info", "files", i, "md5sum"), (str,), must_exist=False, check=utils.is_md5sum, ) for j, item in enumerate(fileinfo["path"]): utils.assert_type(md, ("info", "files", i, "path", j), (str,)) # - validate() should ensure that ['info']['pieces'] is math.ceil(self.size / # self.piece_size) bytes long. piece_count = int(len(info["pieces"]) / 20) exp_piece_count = math.ceil( sum(fileinfo["length"] for fileinfo in info["files"]) / info["piece length"] ) if piece_count != exp_piece_count: raise error.MetainfoError( f"Expected {exp_piece_count} pieces but there are {piece_count}" ) if self.path is not None: # Check if filepath actually points to a directory if not os.path.isdir(self.path): raise error.MetainfoError( f"Metainfo includes {self.path} as directory, but it is not a directory" ) for i, fileinfo in enumerate(info["files"]): filepath = os.path.join(self.path, os.path.join(*fileinfo["path"])) # Check if filepath exists and is a file if not os.path.exists(filepath): raise error.MetainfoError( f"Metainfo includes file that doesn't exist: {filepath}" ) if not os.path.isfile(filepath): raise error.MetainfoError( f"Metainfo includes file that isn't a file: {filepath}" ) # Check if sizes match filesize = utils.real_size(filepath) if filesize != fileinfo["length"]: raise error.MetainfoError( f"Mismatching file sizes in metainfo ({fileinfo['length']})" f" and file system ({filesize}): {filepath}" ) else: raise error.MetainfoError("Missing 'length' or 'files' in 'info'") def convert(self): """ Return :attr:`metainfo` with all keys encoded to :class:`bytes` and all values encoded to :class:`bytes`, :class:`int`, :class:`list` or :class:`OrderedDict` :raises MetainfoError: if a value cannot be converted properly """ try: return utils.encode_dict(self.metainfo) except ValueError as e: raise error.MetainfoError(e) def dump(self, validate=True): """ Create bencoded :attr:`metainfo` (i.e. the content of a torrent file) :param bool validate: Whether to run :meth:`validate` first :return: :attr:`metainfo` as bencoded :class:`bytes` """ if validate: self.validate() return bencode.encode(self.convert()) def write_stream(self, stream, validate=True): """ Write :attr:`metainfo` to a file-like object :param stream: Writable file-like object (e.g. :class:`io.BytesIO`) :param bool validate: Whether to run :meth:`validate` first :raises WriteError: if writing to `stream` fails :raises MetainfoError: if :attr:`metainfo` is invalid """ content = self.dump(validate=validate) try: # Remove existing data from stream *after* dump() didn't raise # anything so we don't destroy it prematurely. if stream.seekable(): stream.seek(0) stream.truncate(0) stream.write(content) except OSError as e: raise error.WriteError(e.errno) def write(self, filepath, validate=True, overwrite=False): """ Write :attr:`metainfo` to torrent file :param filepath: Path of the torrent file :param bool validate: Whether to run :meth:`validate` first :param bool overwrite: Whether to silently overwrite `filepath` (only after all pieces were hashed successfully) :raises WriteError: if writing to `filepath` fails :raises MetainfoError: if :attr:`metainfo` is invalid """ if not overwrite and os.path.exists(filepath): raise error.WriteError(errno.EEXIST, filepath) # Get file content before opening the file in case there are errors like # incomplete metainfo content = io.BytesIO() self.write_stream(content, validate=validate) content.seek(0) try: with open(filepath, "wb") as f: f.write(content.read()) except OSError as e: raise error.WriteError(e.errno, filepath) def magnet(self, name=True, size=True, trackers=True, tracker=False): """ :class:`Magnet` instance :param bool name: Whether to include the name :param bool size: Whether to include the size :param bool trackers: Whether to include all trackers :param bool tracker: Whether to include only the first tracker of the first tier (overrides `trackers`) :raises MetainfoError: if :attr:`metainfo` is invalid """ kwargs = {"xt": "urn:btih:" + self.infohash} if name: kwargs["dn"] = self.name if size: kwargs["xl"] = self.size if tracker: kwargs["tr"] = (self.trackers[0][0],) elif trackers: kwargs["tr"] = (url for tier in self.trackers for url in tier) if self.webseeds is not None: kwargs["ws"] = self.webseeds # Prevent circular import issues from ._magnet import Magnet return Magnet(**kwargs) # Maximum number of bytes that read() reads from torrent files. This limit # exists because we don't want to read gigabytes before raising an error. MAX_TORRENT_FILE_SIZE = int(10e6) # 10MB @classmethod def read_stream(cls, stream, validate=True): """ Read torrent metainfo from file-like object :param stream: Readable file-like object (e.g. :class:`io.BytesIO`) :param bool validate: Whether to run :meth:`validate` on the new Torrent instance NOTE: If the "info" field is not a dictionary, :class:`MetainfoError` is raised even if `validate` is set to False :raises ReadError: if reading from `stream` fails :raises BdecodeError: if `stream` does not produce a valid bencoded byte sequence :raises MetainfoError: if `validate` is `True` and the read metainfo is invalid :return: New :class:`Torrent` instance """ try: content = stream.read(cls.MAX_TORRENT_FILE_SIZE) except OSError as e: raise error.ReadError(e.errno) else: try: metainfo_enc = bencode.decode(content) except (bencode.DecodingError, ValueError): raise error.BdecodeError() else: if not isinstance(metainfo_enc, abc.Mapping): raise error.BdecodeError() # Extract 'pieces' from metainfo before decoding because it's the # only byte sequence that isn't supposed to be decoded to a string. if ( b"info" in metainfo_enc and isinstance(metainfo_enc[b"info"], dict) and b"pieces" in metainfo_enc[b"info"] ): pieces = metainfo_enc[b"info"].pop(b"pieces") metainfo = utils.decode_dict(metainfo_enc) metainfo["info"]["pieces"] = pieces else: metainfo = utils.decode_dict(metainfo_enc) # "info" must be a dictionary. If validation is not wanted, it's OK # if it doesn't exist because the "metainfo" property will add it # automatically. utils.assert_type(metainfo, ("info",), (dict,), must_exist=validate) torrent = cls() torrent._metainfo = metainfo # Convert "creation date" to datetime.datetime and "private" to # bool, but only if they exist if b"creation date" in metainfo_enc: torrent.creation_date = metainfo_enc[b"creation date"] if b"private" in metainfo_enc.get(b"info", {}): torrent.private = metainfo_enc[b"info"][b"private"] if validate: torrent.validate() return torrent @classmethod def read(cls, filepath, validate=True): """ Read torrent metainfo from file :param filepath: Path of the torrent file :param bool validate: Whether to run :meth:`validate` on the new Torrent instance :raises ReadError: if reading from `filepath` fails :raises BdecodeError: if `filepath` does not contain a valid bencoded byte sequence :raises MetainfoError: if `validate` is `True` and the read metainfo is invalid :return: New :class:`Torrent` instance """ try: with open(filepath, "rb") as f: return cls.read_stream(f, validate=validate) except (OSError, error.ReadError) as e: raise error.ReadError(e.errno, filepath) except error.BdecodeError: raise error.BdecodeError(filepath) def copy(self): """Create a new :class:`Torrent` instance with the same metainfo""" from copy import deepcopy cp = type(self)() cp._metainfo = deepcopy(self._metainfo) return cp def reuse(self, path, callback=None, interval=0): """ Copy ``pieces`` and ``piece length`` from existing torrent If `path` is a directory, it is searched recursively for a matching ``.torrent`` file. Otherwise, it is :meth:`read` as a ``.torrent`` file. `path` may also be an iterable, in which case each item is treated as described above until a match is found. A matching torrent is found by searching for a torrent with the same :attr:`name` and :attr:`files` (file size must also match). If a match is found, compare three piece hashes per file to reduce the risk of a false positive. .. warning:: This should almost always work in practice, but a false positive match is theoretically possible, and there is no way to avoid that. .. note:: This will likely change :attr:`infohash` and turn this instance into a new torrent. :param path: Path to torrent file or directory or iterable of file/directory paths :param callable callback: Callable to report progress and/or abort `callback` must accept 6 positional arguments: 1. Torrent instance (:class:`Torrent`) 2. Torrent file path (:class:`str`) or ``None`` 3. Number of checked torrent files (:class:`int`) 4. Total number of torrent files (:class:`int`) 5. Status indicator: ``False`` - File name or size does not match ``True`` - File name and size matches ``None`` - Hashing file and comparing piece hashes 6. Exception (:class:`TorfError`) or ``None`` If `callback` returns anything that is not ``None``, stop searching. :param float interval: Minimum number of seconds between calls to `callback`; if 0, `callback` is called for each torrent file; `callback` is always called if `exception` is not ``None`` :raises ReadError: if reading a torrent file fails :raises BdecodeError: if parsing a torrent file fails :raises MetainfoError: if a torrent file contains invalid or insufficient metadata :return: `True` if a matching torrent was found, `False` otherwise """ if not self.path: raise RuntimeError("reuse() called with no path specified") elif not self.files: raise RuntimeError("reuse() called while file list is empty") if isinstance(path, (str, pathlib.PurePath)): paths = [path] elif isinstance(path, abc.Iterable): paths = tuple(path) else: raise ValueError(f"Invalid path argument: {path!r}") torrent_file_items = reuse.find_torrent_files( *paths, max_file_size=self.MAX_TORRENT_FILE_SIZE ) maybe_call_callback = reuse.ReuseCallback( callback=callback, interval=interval, torrent=self, torrent_files_total=torrent_file_items.total, ) for candidate_path, files_done, exception in torrent_file_items: try: if candidate_path: candidate = Torrent.read(candidate_path) elif exception: raise exception else: raise RuntimeError("Both candidate_path and exception are None?!") except (error.ReadError, error.BdecodeError, error.MetainfoError) as e: cancelled = maybe_call_callback(candidate_path, files_done, False, e) if cancelled is not None: break else: assert exception is None if reuse.is_file_match(self, candidate): cancelled = maybe_call_callback(candidate_path, files_done, None, exception) if cancelled is not None: break if reuse.is_content_match(self, candidate): maybe_call_callback(candidate_path, files_done, True, exception) reuse.copy(candidate, self) return True else: cancelled = maybe_call_callback( candidate_path, files_done, False, exception ) if cancelled is not None: break else: cancelled = maybe_call_callback(candidate_path, files_done, False, exception) if cancelled is not None: break return False def __repr__(self): sig = inspect.signature(self.__init__) args = [] def get_class_default(name): if hasattr(type(self), f"{param.name}_default"): return getattr(type(self), f"{param.name}_default") elif hasattr(type(self), param.name): return getattr(type(self), param.name) for param in sig.parameters.values(): value = getattr(self, param.name) default = param.default if default is param.empty: # Positional argument args.append(repr(value)) elif ( value # Keyword argument value is different from default? and default != value # Keyword argument value is different from class default and value != get_class_default(param.name) ): args.append(f"{param.name}={value!r}") return type(self).__name__ + "(" + ", ".join(args) + ")" def __eq__(self, other): if isinstance(other, type(self)): return self._metainfo == other._metainfo else: return NotImplemented