Initial import: UDP tracker scraper

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.py[cod]
+.venv/

README.md

@@ -0,0 +1,13 @@
+# Torrent Tracker Scraper
+
+Современная, компактная переработка `/_torrent/torrent.py` и `GetPeers.py`.
+Утилита отправляет стандартный UDP scrape-запрос трекеру и выводит число seeders,
+completed и leechers для указанного info hash. Она не скачивает файлы и не
+подключается к пиринговой сети.
+
+```bash
+python -m pip install -e .
+torrent-tracker-scrape tracker.example.org 6969 0123456789abcdef0123456789abcdef01234567
+```
+
+Используйте только с трекерами и материалами, доступ к которым у вас есть.

pyproject.toml

@@ -0,0 +1,15 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "torrent-tracker-scraper"
+version = "0.1.0"
+description = "Inspect peer statistics from UDP BitTorrent trackers."
+requires-python = ">=3.10"
+
+[project.scripts]
+torrent-tracker-scrape = "torrent_tracker_scraper.cli:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/torrent_tracker_scraper/__init__.py

@@ -0,0 +1 @@
+"""Клиент UDP scrape-протокола BitTorrent-трекеров."""

src/torrent_tracker_scraper/cli.py

@@ -0,0 +1,24 @@
+"""Командный интерфейс одноразовой диагностики UDP-трекера."""
+
+from __future__ import annotations
+
+import argparse
+
+from .udp import scrape
+
+
+def main() -> None:
+    """Выполнить scrape и напечатать ответ в удобном для скриптов виде."""
+
+    parser = argparse.ArgumentParser(description="Read one BitTorrent UDP tracker scrape record")
+    parser.add_argument("host")
+    parser.add_argument("port", type=int)
+    parser.add_argument("info_hash")
+    parser.add_argument("--timeout", type=float, default=8.0)
+    args = parser.parse_args()
+    result = scrape(args.host, args.port, args.info_hash, args.timeout)
+    print(f"seeders={result.seeders} completed={result.completed} leechers={result.leechers}")
+
+
+if __name__ == "__main__":
+    main()

src/torrent_tracker_scraper/udp.py

@@ -0,0 +1,63 @@
+"""Небольшая реализация BEP 15 для чтения агрегированной статистики трекера."""
+
+from __future__ import annotations
+
+import os
+import socket
+import struct
+from dataclasses import dataclass
+
+_PROTOCOL_ID = 0x41727101980
+
+
+@dataclass(frozen=True)
+class ScrapeStats:
+    """Сводные значения, возвращаемые трекером для одного info hash."""
+
+    seeders: int
+    completed: int
+    leechers: int
+
+
+def _transaction_id() -> int:
+    """Создать случайный идентификатор, связывающий запрос и ответ."""
+
+    return int.from_bytes(os.urandom(4), "big", signed=False)
+
+
+def _receive(sock: socket.socket, transaction_id: int, minimum_size: int) -> bytes:
+    """Проверить длину и ID ответа, чтобы не принять чужой UDP-пакет."""
+
+    packet, _ = sock.recvfrom(2048)
+    if len(packet) < minimum_size:
+        raise RuntimeError(f"Tracker response is too short: {len(packet)} bytes")
+    if struct.unpack_from("!I", packet, 4)[0] != transaction_id:
+        raise RuntimeError("Tracker response transaction ID does not match")
+    if struct.unpack_from("!I", packet, 0)[0] == 3:
+        raise RuntimeError(packet[8:].decode("utf-8", errors="replace"))
+    return packet
+
+
+def scrape(host: str, port: int, info_hash: str, timeout: float = 8.0) -> ScrapeStats:
+    """Запросить статистику одного 40-символьного hex info hash у UDP-трекера."""
+
+    try:
+        encoded_hash = bytes.fromhex(info_hash)
+    except ValueError as exc:
+        raise ValueError("info_hash must be hexadecimal") from exc
+    if len(encoded_hash) != 20:
+        raise ValueError("info_hash must contain exactly 40 hexadecimal characters")
+
+    address = (socket.gethostbyname(host), port)
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+        sock.settimeout(timeout)
+        connect_tx = _transaction_id()
+        sock.sendto(struct.pack("!QII", _PROTOCOL_ID, 0, connect_tx), address)
+        connect_response = _receive(sock, connect_tx, 16)
+        connection_id = struct.unpack_from("!Q", connect_response, 8)[0]
+
+        scrape_tx = _transaction_id()
+        sock.sendto(struct.pack("!QII20s", connection_id, 2, scrape_tx, encoded_hash), address)
+        response = _receive(sock, scrape_tx, 20)
+        seeders, completed, leechers = struct.unpack_from("!III", response, 8)
+        return ScrapeStats(seeders=seeders, completed=completed, leechers=leechers)