Remove rtorrent scripting from the repository.
This commit is contained in:
@@ -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 <download_dir>/.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()
|
||||
Reference in New Issue
Block a user