From 26af7c633b397435d400a2aaadae7513a5bb7f1b Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Wed, 9 Sep 2026 02:12:37 +0000 Subject: [PATCH] Remove rtorrent scripting from the repository. --- rtorrent_client.py | 162 ------------ scripts/rtorrent-manager.py | 473 ------------------------------------ 2 files changed, 635 deletions(-) delete mode 100644 rtorrent_client.py delete mode 100755 scripts/rtorrent-manager.py diff --git a/rtorrent_client.py b/rtorrent_client.py deleted file mode 100644 index 0380c29..0000000 --- a/rtorrent_client.py +++ /dev/null @@ -1,162 +0,0 @@ -#!/usr/bin/env python3 -"""RTorrent Client - Communicate with rtorrent via XMLRPC over SCGI socket.""" - -import socket -import xmlrpc.client -from pathlib import Path - - -class SCGITransport(xmlrpc.client.Transport): - """SCGI transport for communicating with rtorrent via Unix socket.""" - - def __init__(self, socket_path: str) -> None: - super().__init__() - self.socket_path = socket_path - - def single_request(self, _host, _handler, request_body, _verbose=False): - # Create SCGI request - headers = f"CONTENT_LENGTH\x00{len(request_body)}\x00SCGI\x001\x00" - request = f"{len(headers)}:{headers},{request_body.decode('utf-8')}" - - # Connect to socket - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.connect(self.socket_path) - sock.send(request.encode("utf-8")) - - # Read response - response = b"" - while True: - data = sock.recv(4096) - if not data: - break - response += data - sock.close() - - # Parse response - skip HTTP headers - if b"\r\n\r\n" in response: - response = response.split(b"\r\n\r\n", 1)[1] - - return self.parse_response(response) - - def parse_response(self, response_body): - p, u = xmlrpc.client.getparser() - p.feed(response_body) - p.close() - return u.close() - - -class RTorrentClient: - """Client for communicating with rtorrent via XMLRPC over SCGI socket.""" - - def __init__( - self, socket_path: str = "/home/user/rtorrent/.session/rpc.socket" - ) -> None: - self.socket_path = socket_path - transport = SCGITransport(socket_path) - self.proxy = xmlrpc.client.ServerProxy( - "http://localhost/RPC2", transport=transport - ) - - def get_loaded_hashes(self) -> set[str]: - """Get set of info hashes for all currently loaded torrents.""" - try: - downloads = self.proxy.download_list("") - return {h.upper() for h in downloads} - except (OSError, xmlrpc.client.Error) as e: - print(f"Error getting loaded torrents: {e}") - return set() - - def load_torrent(self, torrent_path: Path, download_dir: Path) -> bool: - """Load a torrent file and set its download directory. - - Uses load.start_verbose to load and immediately start/hash-check. - - Args: - torrent_path: Path to the .torrent file - download_dir: Directory where the data already exists - - Returns: - True if successful, False otherwise - - """ - try: - # load.start_verbose with d.directory.set to specify download location - # This will hash-check existing files instead of re-downloading - self.proxy.load.start_verbose( - "", str(torrent_path), f'd.directory.set="{download_dir}"' - ) - return True - except (OSError, xmlrpc.client.Error) as e: - print(f"Error loading torrent {torrent_path}: {e}") - return False - - def get_torrent_info(self, info_hash: str) -> dict | None: - """Get info about a loaded torrent.""" - try: - name = self.proxy.d.name(info_hash) - message = self.proxy.d.message(info_hash) - tied_file = self.proxy.d.tied_to_file(info_hash) - directory = self.proxy.d.directory(info_hash) - base_path = self.proxy.d.base_path(info_hash) # Actual data path - is_multi_file = self.proxy.d.is_multi_file(info_hash) - return { - "hash": info_hash, - "name": name, - "message": message, - "tied_file": tied_file, - "directory": directory, - "base_path": base_path, # Full path to data (file or folder) - "is_multi_file": is_multi_file, - } - except (OSError, xmlrpc.client.Error) as e: - print(f"Error getting torrent info for {info_hash}: {e}") - return None - - def get_unregistered_torrents(self) -> list[dict]: - """Find all torrents with 'unregistered' or 'not registered' tracker errors. - - Returns: - List of torrent info dicts for torrents with registration errors - - """ - unregistered = [] - try: - hashes = self.proxy.download_list("") - for info_hash in hashes: - try: - message = self.proxy.d.message(info_hash) - if message and ( - "unregistered" in message.lower() - or "not registered" in message.lower() - ): - info = self.get_torrent_info(info_hash) - if info: - unregistered.append(info) - except OSError, xmlrpc.client.Error: - continue - except (OSError, xmlrpc.client.Error) as e: - print(f"Error scanning for unregistered torrents: {e}") - return unregistered - - def remove_torrent(self, info_hash: str, delete_files: bool = False) -> bool: - """Remove a torrent from rtorrent. - - Args: - info_hash: The info hash of the torrent to remove - delete_files: If True, also delete downloaded files (default: False) - - Returns: - True if successful, False otherwise - - """ - try: - if delete_files: - # This would delete the data - NOT what we want - self.proxy.d.erase(info_hash) - else: - # Just remove from rtorrent, keep files - self.proxy.d.erase(info_hash) - return True - except (OSError, xmlrpc.client.Error) as e: - print(f"Error removing torrent {info_hash}: {e}") - return False diff --git a/scripts/rtorrent-manager.py b/scripts/rtorrent-manager.py deleted file mode 100755 index 758bda1..0000000 --- a/scripts/rtorrent-manager.py +++ /dev/null @@ -1,473 +0,0 @@ -#!/usr/bin/env -S uv run -# /// script -# requires-python = ">=3.14" -# dependencies = [ -# "bencodepy>=0.9.5", -# ] -# /// -"""Torrent Scanner - Scans for .torrent files and analyzes their trackers.""" - -import argparse -import hashlib -import shutil -from collections.abc import Iterator -from dataclasses import dataclass -from pathlib import Path - -import bencodepy - -from rtorrent_client import RTorrentClient - - -def _expand_path_pattern(pattern: str) -> list[Path]: - """Expand a user-provided path or glob pattern with pathlib.""" - expanded = Path(pattern).expanduser() - pattern_text = str(expanded) - has_glob = any(ch in pattern_text for ch in "*?[") - - if not has_glob: - return [expanded] if expanded.exists() else [] - - normalized = pattern_text.replace("\\", "/") - if expanded.is_absolute(): - root = Path(expanded.anchor) - remainder = normalized[len(expanded.anchor) :].lstrip("/") - return list(root.glob(remainder)) if remainder else [] - return list(Path().glob(normalized)) - - -@dataclass -class TorrentInfo: - """Information extracted from a torrent file.""" - - path: Path - name: str - trackers: list[str] - size: int | None = None - files: list[str] | None = None - info_hash: str | None = None - is_multi_file: bool = False - - def has_tracker(self, domain: str) -> bool: - """Check if any tracker URL contains the given domain.""" - return any(domain.lower() in tracker.lower() for tracker in self.trackers) - - def get_download_directory(self) -> Path: - """Get the download directory (parent of .torrents folder). - - Assumes .torrent files are in /.torrents/ - so the actual downloads are one level up. - """ - return self.path.parent.parent - - def get_expected_data_path(self) -> Path: - """Get the expected path where downloaded data should exist. - - For multi-file torrents: download_dir/torrent_name/ (directory) - For single-file torrents: download_dir/torrent_name (file) - """ - return self.get_download_directory() / self.name - - def verify_download_exists(self) -> tuple[bool, str]: - """Verify that the downloaded data exists on disk. - - Returns: - Tuple of (exists: bool, message: str) - - """ - expected_path = self.get_expected_data_path() - - if self.is_multi_file: - # Multi-file torrent: expect a directory - if not expected_path.exists(): - return False, f"Directory not found: {expected_path}" - if not expected_path.is_dir(): - return False, f"Expected directory but found file: {expected_path}" - # Optionally check if at least some files exist - existing_files = list(expected_path.rglob("*")) - file_count = sum(1 for f in existing_files if f.is_file()) - if file_count == 0: - return False, f"Directory exists but is empty: {expected_path}" - return True, f"Directory exists with {file_count} files" - # Single-file torrent: expect a file - if not expected_path.exists(): - return False, f"File not found: {expected_path}" - if expected_path.is_dir(): - return False, f"Expected file but found directory: {expected_path}" - return True, f"File exists: {expected_path}" - - -def parse_torrent(filepath: Path) -> TorrentInfo | None: - """Parse a .torrent file and extract relevant information. - - Args: - filepath: Path to the .torrent file - - Returns: - TorrentInfo object or None if parsing fails - - """ - try: - data = bencodepy.decode(Path(filepath).read_bytes()) - except (OSError, ValueError, TypeError) as e: - print(f"Error parsing {filepath}: {e}") - return None - - # Extract trackers - trackers = [] - - # Main announce URL - if b"announce" in data: - announce = data[b"announce"] - if isinstance(announce, bytes): - trackers.append(announce.decode("utf-8", errors="replace")) - - # Announce list (multiple trackers) - if b"announce-list" in data: - for tier in data[b"announce-list"]: - for tracker in tier: - if isinstance(tracker, bytes): - url = tracker.decode("utf-8", errors="replace") - if url not in trackers: - trackers.append(url) - - # Extract name - info = data.get(b"info", {}) - name = info.get(b"name", b"Unknown").decode("utf-8", errors="replace") - - # Calculate info hash - info_hash = hashlib.sha1(bencodepy.encode(info)).hexdigest().upper() - - # Extract size and files - size = None - files = None - is_multi_file = False - - if b"length" in info: - # Single file torrent - size = info[b"length"] - files = [name] - is_multi_file = False - elif b"files" in info: - # Multi-file torrent - files = [] - size = 0 - is_multi_file = True - for file_info in info[b"files"]: - file_path = "/".join( - p.decode("utf-8", errors="replace") for p in file_info.get(b"path", []) - ) - files.append(file_path) - size += file_info.get(b"length", 0) - - return TorrentInfo( - path=filepath, - name=name, - trackers=trackers, - size=size, - files=files, - info_hash=info_hash, - is_multi_file=is_multi_file, - ) - - -def scan_torrent_directories(paths: list[str]) -> Iterator[Path]: - """Scan directories for .torrent files. - - Args: - paths: List of directory paths or glob patterns to scan - - Yields: - Path objects for each .torrent file found - - """ - for pattern in paths: - for torrent_dir in _expand_path_pattern(pattern): - if torrent_dir.is_dir(): - yield from torrent_dir.glob("*.torrent") - - -def find_torrents_with_tracker( - tracker_domain: str, paths: list[str] -) -> list[TorrentInfo]: - """Find all torrents that have a specific tracker domain. - - Args: - tracker_domain: Domain to search for in tracker URLs (e.g., "hdbits.org") - paths: List of directory paths or glob patterns to scan - - Returns: - List of TorrentInfo objects for matching torrents - - """ - matching_torrents = [] - - for torrent_path in scan_torrent_directories(paths): - info = parse_torrent(torrent_path) - if info and info.has_tracker(tracker_domain): - matching_torrents.append(info) - - return matching_torrents - - -def format_size(size_bytes: int | None) -> str: - """Format bytes as human-readable size.""" - if size_bytes is None: - return "Unknown" - - for unit in ["B", "KB", "MB", "GB", "TB"]: - if size_bytes < 1024: - return f"{size_bytes:.2f} {unit}" - size_bytes /= 1024 - return f"{size_bytes:.2f} PB" - - -def main() -> None: - """Run the torrent scanner command-line workflow.""" - parser = argparse.ArgumentParser( - description="Scan and manage torrent files", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - %(prog)s /path/to/torrents*/.torrents/ - %(prog)s /mnt/disk1/torrents/.torrents/ /mnt/disk2/torrents/.torrents/ - %(prog)s /torrents*/.torrents/ --tracker hdbits.org - %(prog)s /torrents*/.torrents/ --dry - """, - ) - parser.add_argument( - "paths", - nargs="+", - help="Directories or glob patterns containing .torrent files", - ) - parser.add_argument( - "--dry", - action="store_true", - help="Dry run - show what would be done without making changes", - ) - parser.add_argument( - "--tracker", - default="hdbits.org", - help="Tracker domain to filter by (default: hdbits.org)", - ) - args = parser.parse_args() - - dry_run = args.dry - tracker_domain = args.tracker - - # Expand glob patterns - expanded_paths = [] - for pattern in args.paths: - matches = [str(path) for path in _expand_path_pattern(pattern)] - if matches: - expanded_paths.extend(matches) - else: - expanded_paths.append(pattern) - - if dry_run: - print("=" * 60) - print("DRY RUN MODE - No changes will be made") - print("=" * 60) - - print("Scanning for torrents...") - print(f"Search paths: {expanded_paths}") - print("-" * 60) - - # Parse all torrents - all_torrents: list[TorrentInfo] = [] - for torrent_path in scan_torrent_directories(expanded_paths): - info = parse_torrent(torrent_path) - if info: - all_torrents.append(info) - - # Separate by tracker - with_hdbits = [t for t in all_torrents if t.has_tracker(tracker_domain)] - without_hdbits = [t for t in all_torrents if not t.has_tracker(tracker_domain)] - - # Print stats - print(f"\n{'=' * 60}") - print("SUMMARY") - print(f"{'=' * 60}") - print(f"Total torrents scanned: {len(all_torrents)}") - print(f"With {tracker_domain}: {len(with_hdbits)}") - print(f"Without {tracker_domain}: {len(without_hdbits)}") - - # Add hdbits torrents to rtorrent - if with_hdbits: - print(f"\n{'=' * 60}") - print("VERIFYING DOWNLOADS & ADDING TO RTORRENT") - print(f"{'=' * 60}") - - # First, verify which torrents have their data - verified = [] - missing_data = [] - - for torrent in with_hdbits: - exists, message = torrent.verify_download_exists() - if exists: - verified.append(torrent) - else: - missing_data.append((torrent, message)) - - print("\nVerification results:") - print(f" Downloads found: {len(verified)}") - print(f" Downloads missing: {len(missing_data)}") - - # Report missing downloads - if missing_data: - print(f"\n{'=' * 60}") - print("TORRENTS WITH MISSING DATA (will not add)") - print(f"{'=' * 60}") - for torrent, message in missing_data: - print(f"\n Name: {torrent.name}") - print(f" Torrent: {torrent.path}") - print(f" Reason: {message}") - - # Now add verified torrents to rtorrent - if verified: - print(f"\n{'=' * 60}") - print(f"ADDING {len(verified)} VERIFIED TORRENTS TO RTORRENT") - print(f"{'=' * 60}") - - client = RTorrentClient() - loaded_hashes = client.get_loaded_hashes() - print(f"Currently loaded in rtorrent: {len(loaded_hashes)} torrents") - - added = 0 - skipped = 0 - failed = 0 - - for torrent in verified: - if torrent.info_hash and torrent.info_hash in loaded_hashes: - print(f"Skipping (already loaded): {torrent.name}") - skipped += 1 - else: - download_dir = torrent.get_download_directory() - if dry_run: - print(f"Would add: {torrent.name}") - print(f" Download dir: {download_dir}") - added += 1 - else: - print(f"Adding: {torrent.name}") - print(f" Download dir: {download_dir}") - if client.load_torrent(torrent.path, download_dir): - added += 1 - else: - failed += 1 - - if dry_run: - print(f"\nDry run: {added} would be added, {skipped} already loaded") - else: - print( - f"\nRtorrent results: {added} added, " - f"{skipped} skipped, {failed} failed" - ) - - # Clean up unregistered torrents from rtorrent - print(f"\n{'=' * 60}") - print("CHECKING FOR UNREGISTERED TORRENTS") - print(f"{'=' * 60}") - - client = RTorrentClient() - unregistered = client.get_unregistered_torrents() - - if unregistered: - print(f"Found {len(unregistered)} unregistered torrent(s):\n") - - removed_from_rtorrent = 0 - removed_torrent_files = 0 - removed_downloads = 0 - - for torrent_info in unregistered: - # Determine the download path (base_path is the actual file/folder) - download_path = ( - Path(torrent_info["base_path"]) if torrent_info["base_path"] else None - ) - - if dry_run: - status = "[DRY]" - if download_path: - print(f" {status} {download_path}") - else: - print(f" {status} {torrent_info['name']} (no data path)") - # Remove from rtorrent (keeps downloaded files) - elif client.remove_torrent(torrent_info["hash"]): - removed_from_rtorrent += 1 - - # Delete the .torrent file if it exists - tied_file = torrent_info["tied_file"] - if tied_file: - torrent_file = Path(tied_file) - if torrent_file.exists(): - try: - torrent_file.unlink() - removed_torrent_files += 1 - except OSError: - pass - - # Delete the downloaded files - if download_path and download_path.exists(): - try: - if download_path.is_dir(): - shutil.rmtree(download_path) - else: - download_path.unlink() - removed_downloads += 1 - print(f" [DEL] {download_path}") - except OSError as e: - print(f" [ERR] {download_path}: {e}") - else: - print(f" [DEL] {torrent_info['name']} (no data)") - else: - print(f" [ERR] {torrent_info['name']}: failed to remove from rtorrent") - - print() - if dry_run: - print( - f"Dry run: {len(unregistered)} would be removed " - "(rtorrent + .torrent + downloads)" - ) - else: - print( - f"Cleanup: {removed_from_rtorrent} from rtorrent, " - f"{removed_torrent_files} .torrents, {removed_downloads} downloads" - ) - else: - print("No unregistered torrents found.") - - # List torrents without hdbits.org - if without_hdbits: - print(f"\n{'=' * 60}") - print(f"TORRENTS WITHOUT {tracker_domain.upper()}") - print(f"{'=' * 60}") - for torrent in without_hdbits: - print(f"\nName: {torrent.name}") - print(f"Path: {torrent.path}") - print(f"Size: {format_size(torrent.size)}") - if torrent.trackers: - print("Trackers:") - for tracker in torrent.trackers: - print(f" - {tracker}") - else: - print("Trackers: (none)") - - # Remove the non-hdbits torrent files - print(f"\n{'=' * 60}") - if dry_run: - print(f"WOULD REMOVE {len(without_hdbits)} TORRENT FILE(S)") - print(f"{'=' * 60}") - for torrent in without_hdbits: - print(f"Would remove: {torrent.path}") - else: - print(f"REMOVING {len(without_hdbits)} TORRENT FILE(S)") - print(f"{'=' * 60}") - for torrent in without_hdbits: - try: - torrent.path.unlink() - print(f"Removed: {torrent.path}") - except OSError as e: - print(f"Failed to remove {torrent.path}: {e}") - - -if __name__ == "__main__": - main()