from __future__ import annotations import hashlib import os from dataclasses import dataclass from typing import Any class BencodeError(ValueError): pass class BDecoder: def __init__(self, data: bytes): self.data = data self.info_span: tuple[int, int] | None = None def parse(self) -> Any: value, pos = self._value(0, top=True) if pos != len(self.data): raise BencodeError(f"trailing data at byte {pos}") return value def _value(self, pos: int, *, top: bool = False) -> tuple[Any, int]: if pos >= len(self.data): raise BencodeError("unexpected end of bencode") c = self.data[pos] if c == ord("i"): return self._int(pos) if c == ord("l"): return self._list(pos) if c == ord("d"): return self._dict(pos, top=top) if ord("0") <= c <= ord("9"): return self._bytes(pos) raise BencodeError(f"invalid bencode byte {c!r} at {pos}") def _int(self, pos: int) -> tuple[int, int]: end = self.data.find(b"e", pos) if end < 0: raise BencodeError("unterminated integer") raw = self.data[pos + 1:end] if not raw: raise BencodeError("empty integer") return int(raw), end + 1 def _bytes(self, pos: int) -> tuple[bytes, int]: colon = self.data.find(b":", pos) if colon < 0: raise BencodeError("unterminated byte string length") n = int(self.data[pos:colon]) start = colon + 1 end = start + n if end > len(self.data): raise BencodeError("byte string exceeds input") return self.data[start:end], end def _list(self, pos: int) -> tuple[list[Any], int]: out: list[Any] = [] pos += 1 while pos < len(self.data) and self.data[pos] != ord("e"): value, pos = self._value(pos) out.append(value) if pos >= len(self.data): raise BencodeError("unterminated list") return out, pos + 1 def _dict(self, pos: int, *, top: bool = False) -> tuple[dict[bytes, Any], int]: out: dict[bytes, Any] = {} pos += 1 while pos < len(self.data) and self.data[pos] != ord("e"): key, pos = self._bytes(pos) value_start = pos value, pos = self._value(pos) if top and key == b"info": self.info_span = (value_start, pos) out[key] = value if pos >= len(self.data): raise BencodeError("unterminated dict") return out, pos + 1 @dataclass class Metadata: info_hash: bytes piece_length: int total_size: int num_pieces: int piece_hashes: list[bytes] name: str def piece_len(self, index: int) -> int: if index + 1 == self.num_pieces: return self.total_size - index * self.piece_length return self.piece_length @dataclass class TorrentFile: path: str raw: bytes metainfo: dict[bytes, Any] info: dict[bytes, Any] info_raw: bytes metadata: Metadata trackers: list[str] files: list[tuple[str, int]] def _text(value: Any, default: str = "") -> str: if isinstance(value, bytes): return value.decode("utf-8", "replace") return default def _file_tree_size(node: Any) -> int: if not isinstance(node, dict): return 0 total = 0 file_marker = node.get(b"") if isinstance(file_marker, dict): total += int(file_marker.get(b"length", 0)) for key, child in node.items(): if key != b"": total += _file_tree_size(child) return total def _total_size(info: dict[bytes, Any]) -> int: if b"length" in info: return int(info[b"length"]) if b"files" in info: return sum(int(f.get(b"length", 0)) for f in info[b"files"]) if b"file tree" in info: return _file_tree_size(info[b"file tree"]) return 0 def _trackers(meta: dict[bytes, Any]) -> list[str]: urls: list[str] = [] announce = meta.get(b"announce") if isinstance(announce, bytes): urls.append(_text(announce)) tiers = meta.get(b"announce-list") if isinstance(tiers, list): for tier in tiers: if not isinstance(tier, list): continue for item in tier: if isinstance(item, bytes): urls.append(_text(item)) seen: set[str] = set() out: list[str] = [] for url in urls: if url and url not in seen: seen.add(url) out.append(url) return out def _path_text(parts: list[Any]) -> str: decoded = [] for part in parts: if not isinstance(part, bytes): raise BencodeError("file path component is not bytes") decoded.append(part.decode("utf-8", "replace")) return os.path.join(*decoded) if decoded else "" def _files(info: dict[bytes, Any]) -> list[tuple[str, int]]: name = _text(info.get(b"name"), "") if b"length" in info: return [(name, int(info[b"length"]))] files = info.get(b"files") if isinstance(files, list): out = [] for entry in files: if not isinstance(entry, dict): continue path = entry.get(b"path") if not isinstance(path, list): continue out.append((os.path.join(name, _path_text(path)), int(entry.get(b"length", 0)))) return out return [] def load_torrent(path: str) -> TorrentFile: raw = open(path, "rb").read() dec = BDecoder(raw) meta = dec.parse() if not isinstance(meta, dict) or dec.info_span is None: raise BencodeError("metainfo does not contain a top-level info dict") info = meta.get(b"info") if not isinstance(info, dict): raise BencodeError("metainfo info value is not a dict") info_raw = raw[dec.info_span[0]:dec.info_span[1]] pieces = info.get(b"pieces") if not isinstance(pieces, bytes) or len(pieces) % 20 != 0: raise BencodeError("only v1/hybrid torrents with a valid pieces string are supported") piece_length = int(info.get(b"piece length", 0)) if piece_length <= 0: raise BencodeError("missing or invalid piece length") piece_hashes = [pieces[i:i + 20] for i in range(0, len(pieces), 20)] total_size = _total_size(info) if total_size <= 0: raise BencodeError("missing torrent payload size") metadata = Metadata( info_hash=hashlib.sha1(info_raw).digest(), piece_length=piece_length, total_size=total_size, num_pieces=len(piece_hashes), piece_hashes=piece_hashes, name=_text(info.get(b"name"), os.path.basename(path)), ) return TorrentFile(path, raw, meta, info, info_raw, metadata, _trackers(meta), _files(info)) def load_metadata(path: str) -> Metadata: return load_torrent(path).metadata