abdus/playground
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(torrentify): add torrent management commands

Browse Source
This commit is contained in:
abdus committed 2026-10-08 11:27:35 +00:00
1 parent 0d97094092
commit 74e887db04
2 files changed
+508 -469
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

torrentify.py
+261 -469
View File
@@ -1,501 +1,293 @@
#!/usr/bin/env python3.9
import argparse
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["click", "httpx", "torf"]
# ///
import base64
import json
import logging
import os
import re
import select
import shutil
import sys
import time
from turtle import back
import typing
from pathlib import Path
from typing import Optional
import ffmpeg
from ffmpeg_strip import clean_video
import click
import httpx
import torf
logging.basicConfig(level=logging.INFO, format=f"%(asctime)s {logging.BASIC_FORMAT}")
ANNOUNCE_URLS = ("http://tracker.empornium.sx:2710/tegqucis10uanp672qh6nhn393xlkncs/announce",)
AUTHOR = "zzzp"
def force_import(module: str):
import importlib
import subprocess
import sys
try:
return importlib.import_module(module)
except ModuleNotFoundError:
subprocess.run([sys.executable, "-m", "pip", "install", module])
importlib.invalidate_caches()
return importlib.import_module(module)
try:
import torf
except ImportError:
torf = force_import("torf")
try:
import httpx
except ImportError:
httpx = force_import("httpx")
try:
import inquirer
except ImportError:
inquirer = force_import("inquirer")
TORRENT_TRACKER_URL = os.getenv(
"TORRENT_TRACKER_URL", "http://tracker.empornium.sx:2710/tegqucis10uanp672qh6nhn393xlkncs/announce"
)
TORRENT_CREATED_BY = os.getenv("TORRENT_CREATED_BY", "zzzp")
TORRENT_DIR = Path(os.getenv("TORRENT_DIR", "/mnt/box/files/_torrents/_new2/"))
VIDEO_EXTENSIONS = {".mp4", ".mpg", ".mkv"}
def format_size(size: int) -> str:
if size < 1000:
return f"{size:,} bytes"
scaled_size = float(size)
for unit in ("KB", "MB", "GB", "TB", "PB", "EB"):
scaled_size /= 1000
if scaled_size < 1000 or unit == "EB":
return f"{scaled_size:.2f} {unit} ({size:,} bytes)"
raise ValueError(f"Cannot format size: {size}")
def make_torrent(source_dir: Path, tracker_url: str = TORRENT_TRACKER_URL) -> Path:
torrent_path = source_dir / f"{source_dir.name}.torrent"
if torrent_path.is_file():
logging.info("Torrent file is already created")
return torrent_path
t = torf.Torrent(
path=str(source_dir.resolve()),
name=source_dir.resolve().name,
trackers=[tracker_url],
private=True,
created_by=TORRENT_CREATED_BY,
exclude_globs=["*.txt", "post.txt", "*.torrent", "*.gif"],
def create_torrent(
paths: tuple[Path, ...],
output: Path | None = None,
announce_urls: tuple[str, ...] = ANNOUNCE_URLS,
author: str = AUTHOR,
private: bool = True,
name: str | None = None,
overwrite: bool = False,
) -> Path:
sources = list(dict.fromkeys(path.expanduser().absolute() for path in paths))
if not sources:
raise ValueError("Provide at least one file or directory")
for source in sources:
if not source.is_file() and not source.is_dir():
raise ValueError(f"Not a file or directory: {source}")
if len({source.parent for source in sources}) > 1:
raise ValueError("All inputs must be in the same directory")
torrent = torf.Torrent(
trackers=announce_urls,
created_by=author,
private=private,
creation_date=int(time.time()),
)
t.generate()
t.write(torrent_path, overwrite=True)
if len(sources) == 1:
torrent.path = sources[0]
else:
torrent.filepaths = sources
if name is not None:
torrent.name = name
if torrent.size <= 0:
raise ValueError("The selected paths contain no nonempty files")
torrent_path = output or Path.cwd() / f"{torrent.name}.torrent"
if torrent_path.exists() and not overwrite:
raise FileExistsError(f"Output already exists: {torrent_path}")
if torrent_path.absolute() in {Path(path).absolute() for path in torrent.filepaths}:
raise ValueError("Output must not overwrite a source file")
with click.progressbar(length=torrent.pieces, label="Hashing", show_percent=True, file=sys.stderr) as progress:
def report_progress(
torrent: torf.Torrent,
filepath: str,
pieces_done: int,
pieces_total: int,
) -> None:
progress.update(n_steps=pieces_done - progress.pos)
if not torrent.generate(callback=report_progress, interval=0.2):
raise ValueError("Torrent hashing did not complete")
torrent.write(filepath=torrent_path, overwrite=overwrite)
return torrent_path
def human_size(size: int) -> str:
suffix = "B"
for unit in ["", "K", "M", "G", "T", "P", "E", "Z"]:
if abs(size) < 1024.0:
return "%3.2f%s%s" % (size, unit, suffix)
size /= 1024.0
return "%.2f%s%s" % (size, "Yi", suffix)
def modify_torrent(
torrent_path: Path,
output: Path | None = None,
announce_urls: tuple[str, ...] = (),
author: str | None = None,
private: bool | None = None,
name: str | None = None,
) -> Path:
torrent = torf.Torrent.read(filepath=torrent_path)
if announce_urls:
torrent.trackers = announce_urls
if author is not None:
torrent.created_by = author
if private is not None:
torrent.private = private
if name is not None:
torrent.name = name
destination = output or torrent_path
torrent.write(filepath=destination, overwrite=destination.absolute() == torrent_path.absolute())
return destination
def find_video(root_dir: Path) -> Optional[Path]:
for ext in VIDEO_EXTENSIONS:
for f in root_dir.rglob(f"*{ext}"):
return f
return None
def pick_name(video_path: Path) -> str:
try:
if parse_filename(video_path.stem):
return video_path.stem
except ValueError:
pass
candidates = set()
parent = video_path.parent
while parent.exists():
try:
if parse_filename(parent.name):
candidates.add(parent.name)
except ValueError:
pass
if parent.parent == parent:
break
parent = parent.parent
if len(candidates) == 1:
return candidates.pop()
elif len(candidates) > 1:
questions = [
inquirer.List(
"best_name",
message="Pick the best filename",
choices=[*candidates, "<custom>"],
),
]
answers = inquirer.prompt(questions) or {}
best_name = answers.get("best_name")
if best_name != "<custom>":
return best_name
while True:
try:
answers = inquirer.prompt([inquirer.Editor("best_name", message="New name", default=video_path.stem)])
if not answers:
raise Exception("Cancelled")
best_name = answers["best_name"].splitlines(keepends=False)[0].strip()
parse_filename(best_name)
return best_name
except ValueError:
continue
def move_video(video_path: Path, new_name: typing.Optional[str], link: bool = False) -> Path:
"""
Move video to the torrent directory and returns the video path
"""
if new_name:
best_name = Path(new_name).name
else:
best_name = pick_name(video_path)
assert not best_name.startswith('.')
logging.info("using name: %s", best_name)
best_name = best_name.removesuffix('.nometadata')
if Path(best_name).suffix in VIDEO_EXTENSIONS:
best_name = Path(best_name).stem
renamed_path = video_path.with_stem(best_name)
if link:
video_path = video_path.link_to(renamed_path)
else:
video_path = video_path.rename(renamed_path)
video_path = renamed_path
new_loc = TORRENT_DIR / video_path.stem
new_loc.mkdir(parents=True, exist_ok=True)
target_path = new_loc / video_path.name
video_path.rename(target_path)
for f in video_path.parent.glob("*.jpg"):
if f.stem.startswith(video_path.stem):
f.rename(new_loc / f.name)
break
return target_path
def upload_image(image_path: Path) -> str:
session = httpx.Client()
res = session.get("https://jerking.empornium.ph/?agree-consent", follow_redirects=True)
try:
auth_token = re.search(r'name="auth_token" value="([^"]+)"', res.text).group(1)
except AttributeError:
auth_token = re.search(r'auth_token = "([^"]+)"', res.text).group(1)
with image_path.open("rb") as f:
res = session.post(
"https://jerking.empornium.ph/json",
headers={"accept": "application/json"},
data={
"thumb_width": "160",
"thumb_height": "160",
"thumb_crop": "false",
"medium_width": "500",
"medium_crop": "false",
"type": "file",
"action": "upload",
"timestamp": str(round(time.time() * 1000)),
"auth_token": auth_token,
"nsfw": "0",
},
files={
"source": ("thumbs.jpg", f),
},
)
data = res.json()
image_url = data["image"]["url"]
thumbnail_url = data["image"]["display_url"]
return image_url
ParsedFilename = typing.TypedDict(
'ParsedFilename', {'actors': list[str], 'studio': str, 'date': str, 'title': str, 'tags': list[str]}
)
def parse_filename(filename: str) -> ParsedFilename:
parsed = {
"tags": [],
}
filename = re.sub(r"\s*\[\d+[^]]+]", "", filename)
if m := re.search(r"(?P<actors>.+)\s+-+\s+@(?P<studio>\S+)\s+-+\s+(?P<title>.+)\s+-+\s+(?P<date>[\d-]+)", filename):
parsed.update(m.groupdict())
elif m := re.search(r"(?P<actors>.+)\s+-+\s+@(?P<studio>\S+)\s+-+\s+(?P<date>[\d-]+)", filename):
parsed.update(m.groupdict())
elif m := re.search(r"(?P<actors>.+)\s+-+\s+(?P<title>.+)\s+-+\s+(?P<date>[\d-]+)", filename):
parsed.update(m.groupdict())
elif m := re.search(r"(?P<actors>.+)\s+-+\s+(?P<title>\D.+)", filename):
parsed.update(m.groupdict())
else:
raise ValueError("Unknown filename format")
if "actors" in parsed:
names = re.split(r"\s*,\s*", parsed["actors"])
with_aliases = []
for it in names:
with_aliases.extend(re.split(r"\s+aka\s+", it))
parsed["actors"] = names
parsed["tags"].extend([it.lower().replace(" ", ".") for it in with_aliases])
if "studio" in parsed:
parsed["tags"].append(parsed["studio"].lower() + ".com")
if "date" in parsed:
year, ym = parsed["date"][:4], parsed["date"][:7].replace("-", ".")
parsed["tags"].append(year)
parsed["tags"].append(ym)
return parsed
def generate_title(parsed: ParsedFilename) -> str:
parts = []
if actors := parsed.get('actors'):
parts.append(', '.join(actors))
if studio := parsed.get('studio'):
parts.append(f'@{studio}')
if title := parsed.get('title'):
parts.append(title)
if date := parsed.get('date'):
parts.append(date)
return ' -- '.join(parts)
def generate_post_bbcode(video_path: Path, thumbnails_path: Path) -> str:
try:
image_url = upload_image(thumbnails_path)
image_bbcode = f"[img]{image_url}[/img]"
except:
logging.exception("Failed to upload image")
image_bbcode = ""
parsed = parse_filename(video_path.stem)
probe = ffmpeg.ffprobe(video_path)
if 1900 <= probe.width <= 2200:
hd = "1080p"
elif 1200 <= probe.width <= 1400:
hd = "720p"
elif 3000 <= probe.width:
hd = "4K"
else:
hd = None
if hd:
parsed["tags"].append(hd.lower())
if probe.codec == "hevc":
parsed["tags"].extend(["x265", "x265.reencode", "hevc.x265"])
table = {
"Duration": probe.duration_human,
"Format": video_path.suffix.strip("."),
"Filesize": human_size(video_path.stat().st_size),
"Resolution": f"{probe.width}x{probe.height}",
"Codec": probe.codec,
"Bit rate": f"{probe.bitrate // 1000} kbit/s",
"FPS": probe.fps,
}
release_date = parsed.get("date", "")
if release_date:
table = {
"Release Date": release_date,
**table,
}
lines = [
"[table=nball]",
*(f"[tr][th=20]{k}[/th][td]{v}[/td][/tr]" for k, v in table.items()),
"[/table]",
]
table_bbcode = "\n".join(lines)
m_performers = re.search(r"(.+?) -+\s", video_path.stem)
performers = m_performers.group(1) if m_performers else ""
title = generate_title(parsed)
metadata = {
**parsed,
"is_hevc": probe.codec == "hevc",
"hd": hd,
}
metadata_json = json.dumps(metadata)
return f"""
{metadata_json}
---
[b]{title}[/b]
[cast]
{performers}
[details]
[info]
{table_bbcode}
[screens]
{image_bbcode}
""".strip()
def add_torrent(torrent_path: Path):
import time
from functools import partial
make_id = partial(time.time_ns)
# log in
logging.debug("Connecting to Deluge")
session = httpx.Client(base_url="https://t.zzzp.win/", timeout=30)
res = session.post("/json", json={"method": "auth.login", "params": ["xAsametk50"], "id": make_id()})
res.raise_for_status()
# find first available host
res = session.post("/json", json={"method": "web.get_hosts", "params": [], "id": make_id()})
res.raise_for_status()
host_id = res.json()["result"][0][0]
# connect to a host
res = session.post("/json", json={"method": "web.connect", "params": [host_id], "id": make_id()})
res.raise_for_status()
# upload torrent
logging.debug("Uploading torrent file")
with torrent_path.open("rb") as f:
res = session.post("/upload", files={"file": f})
res.raise_for_status()
remote_path: str = res.json()["files"][0]
# add torrent
logging.debug("Adding torrent file")
res = session.post(
"/json",
json={
"method": "web.add_torrents",
"params": [
[
{
"path": remote_path,
"options": {
"file_priorities": [1],
"add_paused": True,
"sequential_download": False,
"pre_allocate_storage": False,
"download_location": "/dl/_new2",
"move_completed": False,
"move_completed_path": "/root/Downloads",
"prioritize_first_last_pieces": True,
"seed_mode": True,
"super_seeding": False,
},
}
]
],
"id": make_id(),
def enqueue_torrent(
transmission_url: str,
credentials: str | tuple[str, str] | None,
torrent_path: Path,
download_path: str | Path,
) -> str:
if isinstance(credentials, str):
username, separator, password = credentials.partition(":")
if not separator:
raise ValueError("Credentials must be username:password")
credentials = (username, password)
auth = httpx.BasicAuth(*credentials) if credentials is not None else None
rpc_url = transmission_url.rstrip("/")
if not httpx.URL(url=rpc_url).path.endswith("/rpc"):
rpc_url = f"{rpc_url}/transmission/rpc"
payload = {
"method": "torrent-add",
"arguments": {
"download-dir": str(download_path),
"metainfo": base64.b64encode(torrent_path.read_bytes()).decode("ascii"),
"paused": False,
},
)
res.raise_for_status()
torrent_id = res.json()["result"][0][1]
# force recheck
# logging.info("Triggering a forced recheck")
# time.sleep(5)
# try:
# _ = session.post(
# "/json", json={"method": "core.force_recheck", "params": [[torrent_id]], "id": make_id()}, timeout=0.1
# )
# except httpx.HTTPError:
# pass
logging.debug("Torrent file has been added successfully.")
}
with httpx.Client(auth=auth, timeout=20) as client:
for _ in range(3):
response = client.post(url=rpc_url, json=payload)
if response.status_code == 409:
client.headers["X-Transmission-Session-Id"] = response.headers["X-Transmission-Session-Id"]
continue
response.raise_for_status()
result = response.json()
if result["result"] not in ("success", "duplicate-torrent"):
raise ValueError(f"Transmission rejected torrent: {result['result']}")
added = result["arguments"].get("torrent-added") or result["arguments"].get("torrent-duplicate")
if not added:
raise ValueError("Transmission returned no torrent details")
return added["hashString"]
raise ValueError("Transmission session negotiation failed")
def parse_args(argv: list[str]):
arger = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
arger.add_argument("dir_or_video", type=Path, help="Directory or video file to add")
arger.add_argument("--name", "-n", help="Torrent name")
arger.add_argument("--no-metadata", action='store_true', dest='remove_metadata', help="Remove all metadata")
arger.add_argument("--post-only", action="store_true", help="Generate only the post bbcode")
arger.add_argument("--link", action="store_true", help="Link video file instead of moving")
arger.add_argument("--thumb-path", dest='thumbnail_path', type=Path, help="Path to thumbnail sheet")
if not argv:
arger.print_help()
sys.exit(0)
return arger.parse_args(argv)
@click.group()
def main() -> None:
"""Create, modify, inspect, and enqueue torrent files."""
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
def validate_name(base_name: str):
if len(base_name) > (128 - len(".mp4")):
raise ValueError(f"Filename is too long: {base_name} ({len(base_name)} chars)")
if "?" in base_name:
raise ValueError(f"Filename contains invalid characters: {base_name}")
@main.command(name="create")
@click.argument("paths", nargs=-1, required=True, type=click.Path(exists=True, path_type=Path))
@click.option("--output", type=click.Path(dir_okay=False, path_type=Path), help="Destination; defaults to NAME.torrent in the current directory.")
@click.option("--announce-url", "announce_urls", multiple=True, help="Tracker URL; repeat for multiple trackers. Defaults to the hardcoded URLs.")
@click.option("--author", default=AUTHOR, show_default=True)
@click.option("--private/--no-private", default=True, show_default=True)
@click.option("--name", help="Torrent name; defaults to the file or common directory name.")
@click.option("--overwrite", is_flag=True, help="Replace an existing output torrent.")
def create(
paths: tuple[Path, ...],
output: Path | None,
announce_urls: tuple[str, ...],
author: str,
private: bool,
name: str | None,
overwrite: bool,
) -> None:
"""Create one torrent from FILES or DIRECTORIES, retaining their relative paths."""
try:
destination = create_torrent(
paths=paths,
output=output,
announce_urls=announce_urls or ANNOUNCE_URLS,
author=author,
private=private,
name=name,
overwrite=overwrite,
)
except (OSError, ValueError, torf.TorfError) as error:
raise click.ClickException(message=str(error)) from error
logging.info(f"Saved torrent at {destination}")
def main():
args = parse_args(sys.argv[1:])
dir_or_video: Path = args.dir_or_video.expanduser().resolve()
if dir_or_video.is_dir():
dir_path = dir_or_video
if not dir_path.is_relative_to(TORRENT_DIR):
target_path = TORRENT_DIR / dir_path.name
logging.info(f"Linking to {target_path}")
shutil.copytree(src=dir_path, dst=target_path, copy_function=os.link, symlinks=False)
dir_path = target_path
torrent_path = make_torrent(dir_path)
logging.info(f"Saved torrent at {torrent_path}")
@main.command(name="modify")
@click.argument("torrent_path", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--output", type=click.Path(dir_okay=False, path_type=Path), help="Destination; defaults to editing the input torrent in place.")
@click.option("--announce-url", "announce_urls", multiple=True, help="Replace trackers; repeat for multiple URLs.")
@click.option("--author", help="Replace the creator.")
@click.option("--private/--no-private", default=None, help="Set or clear the private flag.")
@click.option("--name", help="Replace the torrent name.")
def modify(
torrent_path: Path,
output: Path | None,
announce_urls: tuple[str, ...],
author: str | None,
private: bool | None,
name: str | None,
) -> None:
"""Change metadata without rehashing the content; omitted fields stay unchanged."""
try:
destination = modify_torrent(
torrent_path=torrent_path,
output=output,
announce_urls=announce_urls,
author=author,
private=private,
name=name,
)
except (OSError, ValueError, torf.TorfError) as error:
raise click.ClickException(message=str(error)) from error
logging.info(f"Saved torrent at {destination}")
logging.info("Adding torrent file to Deluge")
add_torrent(torrent_path)
@main.command(name="enqueue")
@click.argument("torrent_path", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--transmission-url", "--host", required=True, envvar="TRANSMISSION_HOST", help="Transmission host or RPC URL; also reads TRANSMISSION_HOST.")
@click.option("--credentials", "--auth", envvar="TRANSMISSION_AUTH", help="Username:password; also reads TRANSMISSION_AUTH.")
@click.option("--download-path", required=True, type=click.Path(path_type=Path), help="Download directory on the remote Transmission host.")
def enqueue(
torrent_path: Path,
transmission_url: str,
credentials: str | None,
download_path: Path,
) -> None:
"""Add a torrent to Transmission at the given remote download directory."""
try:
torrent_hash = enqueue_torrent(
transmission_url=transmission_url,
credentials=credentials,
torrent_path=torrent_path,
download_path=download_path,
)
except (OSError, ValueError, httpx.HTTPError) as error:
raise click.ClickException(message=str(error)) from error
logging.info(f"Enqueued torrent {torrent_hash} at {download_path}")
@main.command(name="inspect")
@click.argument("torrent_path", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--json", "json_output", is_flag=True, help="Output JSON instead of plain text.")
def inspect(torrent_path: Path, json_output: bool) -> None:
"""Show torrent properties and content paths with readable sizes."""
try:
torrent = torf.Torrent.read(filepath=torrent_path)
except (OSError, ValueError, torf.TorfError) as error:
raise click.ClickException(message=str(error)) from error
properties = {
"name": torrent.name,
"author": torrent.created_by,
"private": torrent.private,
"infohash": torrent.infohash,
"size": torrent.size,
"piece_size": torrent.piece_size,
"pieces": torrent.pieces,
"creation_date": torrent.creation_date.isoformat() if torrent.creation_date else None,
"comment": torrent.comment,
"source": torrent.source,
"announce_urls": [list(tier) for tier in torrent.trackers],
"webseeds": list(torrent.webseeds),
"httpseeds": list(torrent.httpseeds),
}
files = [{"path": str(file), "size": file.size} for file in torrent.files]
if json_output:
click.echo(message=json.dumps({"properties": properties, "files": files}, indent=2, ensure_ascii=False))
return
if not dir_or_video.is_file():
logging.error("Invalid file or directory")
return
video_path = dir_or_video
if video_path.suffix not in VIDEO_EXTENSIONS:
logging.error('not a video')
return
if args.remove_metadata:
logging.info('Removing video metadata')
save_path = video_path.with_stem(f'{video_path.stem}.nometadata')
video_path = clean_video(video_path=video_path, save_path=save_path)
video_path = move_video(video_path, new_name=args.name, link=args.link)
dir_path = video_path.parent
if p := args.thumbnail_path:
thumbnail_path = video_path.with_suffix('.jpg')
p.link_to(thumbnail_path)
else:
logging.info("Creating thumbnail tile")
old_thumb_path = video_path.with_suffix(".thumbnail.jpg")
thumbnail_path = video_path.with_suffix(".jpg")
if old_thumb_path.is_file():
old_thumb_path.rename(thumbnail_path)
image_jpeg = ffmpeg.make_thumbnail_tile(video_path)
thumbnail_path.write_bytes(image_jpeg)
logging.info(f"Saved thumbnails at {thumbnail_path}")
post_bbcode = generate_post_bbcode(video_path, thumbnail_path)
video_path.with_name("post.txt").write_text(post_bbcode)
print(post_bbcode)
if args.post_only:
return
total_files = len(list(dir_path.rglob("*")))
logging.info(f"Creating torrent file from {dir_path}. Total files: {total_files}")
torrent_path = make_torrent(dir_path)
logging.info(f"Saved torrent at {torrent_path}")
logging.info("Adding torrent file to Deluge")
add_torrent(torrent_path)
for key, value in properties.items():
label = key.replace("_", " ").capitalize()
if isinstance(value, list):
click.echo(message=f"{label}:" if value else f"{label}: (none)")
for index, item in enumerate(value, start=1):
if isinstance(item, list):
click.echo(message=f" Tier {index}: {', '.join(item)}")
else:
click.echo(message=f" {item}")
elif value is None:
click.echo(message=f"{label}: (unset)")
elif isinstance(value, bool):
click.echo(message=f"{label}: {'yes' if value else 'no'}")
elif key in ("size", "piece_size"):
click.echo(message=f"{label}: {format_size(size=value)}")
else:
click.echo(message=f"{label}: {value}")
click.echo(message=f"\nFiles ({len(files)}):")
for file in files:
click.echo(message=f" {file['path']}: {format_size(size=file['size'])}")
if __name__ == "__main__":
torrentify_test.py
+247
View File
@@ -0,0 +1,247 @@
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["click", "httpx", "torf", "pytest"]
# ///
import base64
import hashlib
import json
from pathlib import Path
import sys
from unittest.mock import patch
from click.testing import CliRunner
import httpx
import pytest
import torf
import torrentify as script
@pytest.mark.parametrize("kind", ["file", "directory", "mixed"])
def test_create_selected_content(tmp_path: Path, kind: str) -> None:
source = tmp_path / "first.txt"
source.write_bytes(b"first")
directory = tmp_path / "folder"
directory.mkdir()
(directory / "nested.txt").write_bytes(b"nested")
(tmp_path / "unselected.txt").write_bytes(b"unselected")
paths = {"file": (source,), "directory": (directory,), "mixed": (source, directory)}[kind]
output = tmp_path / "result.torrent"
result = CliRunner().invoke(cli=script.main, args=["create", *map(str, paths), "--output", str(output)])
assert result.exit_code == 0, result.output
torrent = torf.Torrent.read(filepath=output)
assert torrent.private is True
assert torrent.created_by == script.AUTHOR
assert list(torrent.trackers.flat) == list(script.ANNOUNCE_URLS)
expected = {"file": b"first", "directory": b"nested", "mixed": b"firstnested"}[kind]
assert torrent.metainfo["info"]["pieces"] == hashlib.sha1(expected).digest()
if kind == "mixed":
assert [item["path"] for item in torrent.metainfo["info"]["files"]] == [
["first.txt"], ["folder", "nested.txt"],
]
if kind == "directory":
assert torrent.name == "folder"
assert torrent.metainfo["info"]["files"][0]["path"] == ["nested.txt"]
def test_reject_inputs_in_different_directories(tmp_path: Path) -> None:
source = tmp_path / "first.txt"
source.write_bytes(b"first")
directory = tmp_path / "folder"
directory.mkdir()
nested = directory / "nested.txt"
nested.write_bytes(b"nested")
result = CliRunner().invoke(cli=script.main, args=["create", str(source), str(nested)])
assert result.exit_code == 1
assert "All inputs must be in the same directory" in result.output
assert not list(tmp_path.glob("*.torrent"))
def test_create_rejects_empty_and_existing_output(tmp_path: Path) -> None:
source = tmp_path / "file.txt"
source.touch()
with pytest.raises(ValueError, match="no nonempty files"):
script.create_torrent(paths=(source,), output=tmp_path / "result.torrent")
source.write_bytes(b"content")
output = script.create_torrent(paths=(source,), output=tmp_path / "result.torrent")
original = output.read_bytes()
with pytest.raises(FileExistsError):
script.create_torrent(paths=(source,), output=output)
assert output.read_bytes() == original
with pytest.raises(ValueError, match="overwrite a source"):
script.create_torrent(paths=(source,), output=source, overwrite=True)
assert source.read_bytes() == b"content"
def test_modify_metadata_without_content(tmp_path: Path) -> None:
source = tmp_path / "content.txt"
source.write_bytes(b"content")
output = script.create_torrent(paths=(source,), output=tmp_path / "original.torrent")
before = torf.Torrent.read(filepath=output)
source.unlink()
result = CliRunner().invoke(cli=script.main, args=[
"modify", str(output), "--announce-url", "https://tracker.example/announce",
"--announce-url", "udp://tracker.example:80/announce", "--author", "new author",
"--name", "new name", "--no-private",
])
assert result.exit_code == 0, result.output
after = torf.Torrent.read(filepath=output)
assert after.name == "new name"
assert after.created_by == "new author"
assert after.private is False
assert list(after.trackers.flat) == ["https://tracker.example/announce", "udp://tracker.example:80/announce"]
assert after.metainfo["info"]["pieces"] == before.metainfo["info"]["pieces"]
def test_modify_preserves_omitted_fields_and_original(tmp_path: Path) -> None:
source = tmp_path / "content.txt"
source.write_bytes(b"content")
original = script.create_torrent(paths=(source,), output=tmp_path / "original.torrent")
original_bytes = original.read_bytes()
output = tmp_path / "modified.torrent"
script.modify_torrent(torrent_path=original, output=output, author="replacement")
before = torf.Torrent.read(filepath=original)
after = torf.Torrent.read(filepath=output)
assert original.read_bytes() == original_bytes
assert after.infohash == before.infohash
assert after.private == before.private
assert after.trackers == before.trackers
assert after.name == before.name
with pytest.raises(torf.WriteError):
script.modify_torrent(torrent_path=original, output=output)
@pytest.mark.parametrize("added_key", ["torrent-added", "torrent-duplicate"])
@pytest.mark.parametrize("url", ["http://transmission:9091", "http://transmission:9091/transmission/rpc"])
def test_enqueue_negotiates_session_and_sets_remote_directory(tmp_path: Path, added_key: str, url: str) -> None:
torrent_path = tmp_path / "content.torrent"
torrent_path.write_bytes(b"torrent data")
requests = []
def respond(request: httpx.Request) -> httpx.Response:
requests.append(request)
if len(requests) == 1:
return httpx.Response(status_code=409, headers={"X-Transmission-Session-Id": "session-token"})
return httpx.Response(status_code=200, json={
"result": "success", "arguments": {added_key: {"hashString": "torrent-hash"}},
})
client = httpx.Client(transport=httpx.MockTransport(handler=respond), auth=httpx.BasicAuth("user", "pass:word"))
with patch.object(script.httpx, "Client", return_value=client) as factory:
result = script.enqueue_torrent(
transmission_url=url, credentials="user:pass:word",
torrent_path=torrent_path, download_path="/remote/downloads",
)
assert result == "torrent-hash"
assert factory.call_args.kwargs["auth"]._auth_header == client.auth._auth_header
assert len(requests) == 2
assert requests[1].headers["X-Transmission-Session-Id"] == "session-token"
assert str(requests[1].url) == "http://transmission:9091/transmission/rpc"
payload = json.loads(requests[1].content)
assert payload["method"] == "torrent-add"
assert payload["arguments"]["download-dir"] == "/remote/downloads"
assert payload["arguments"]["paused"] is False
assert base64.b64decode(payload["arguments"]["metainfo"]) == torrent_path.read_bytes()
@pytest.mark.parametrize("status, result", [(401, "success"), (200, "invalid torrent"), (409, "success")])
def test_enqueue_reports_failures(tmp_path: Path, status: int, result: str) -> None:
torrent_path = tmp_path / "content.torrent"
torrent_path.write_bytes(b"torrent data")
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(status_code=status, headers={"X-Transmission-Session-Id": "token"}, json={"result": result})
client = httpx.Client(transport=httpx.MockTransport(handler=respond))
with patch.object(script.httpx, "Client", return_value=client):
with pytest.raises((httpx.HTTPStatusError, ValueError)):
script.enqueue_torrent(
transmission_url="http://transmission:9091", credentials=None,
torrent_path=torrent_path, download_path="/remote/downloads",
)
@pytest.mark.parametrize("use_environment", [False, True])
def test_enqueue_command(tmp_path: Path, use_environment: bool) -> None:
torrent_path = tmp_path / "content.torrent"
torrent_path.write_bytes(b"torrent data")
arguments = ["enqueue", str(torrent_path), "--download-path", "/remote/downloads"]
environment = {"TRANSMISSION_HOST": "http://transmission:9091", "TRANSMISSION_AUTH": "user:password"}
if not use_environment:
arguments.extend(["--transmission-url", environment["TRANSMISSION_HOST"], "--credentials", environment["TRANSMISSION_AUTH"]])
with patch.object(script, "enqueue_torrent", return_value="torrent-hash") as enqueue:
result = CliRunner().invoke(cli=script.main, args=arguments, env=environment if use_environment else {})
assert result.exit_code == 0, result.output
enqueue.assert_called_once_with(
transmission_url="http://transmission:9091", credentials="user:password",
torrent_path=torrent_path, download_path=Path("/remote/downloads"),
)
def test_enqueue_command_reports_error(tmp_path: Path) -> None:
torrent_path = tmp_path / "content.torrent"
torrent_path.write_bytes(b"torrent data")
with patch.object(script, "enqueue_torrent", side_effect=ValueError("Transmission rejected torrent")):
result = CliRunner().invoke(cli=script.main, args=[
"enqueue", str(torrent_path), "--host", "http://transmission:9091",
"--download-path", "/remote/downloads",
])
assert result.exit_code == 1
assert "Error: Transmission rejected torrent" in result.output
@pytest.mark.parametrize("directory", [False, True])
@pytest.mark.parametrize("json_output", [False, True])
def test_inspect_reports_properties_and_files_without_content(
tmp_path: Path, directory: bool, json_output: bool,
) -> None:
folder = tmp_path / "content"
folder.mkdir()
source = folder / "first.txt"
source.write_bytes(b"first")
if directory:
(folder / "second.txt").write_bytes(b"second")
output = script.create_torrent(paths=(folder if directory else source,), output=tmp_path / "content.torrent")
original = output.read_bytes()
source.unlink()
arguments = ["inspect", str(output)]
if json_output:
arguments.append("--json")
result = CliRunner().invoke(cli=script.main, args=arguments)
assert result.exit_code == 0, result.output
assert result.stderr == ""
if json_output:
report = json.loads(result.stdout)
assert report["properties"]["author"] == script.AUTHOR
assert report["properties"]["private"] is True
assert report["properties"]["announce_urls"] == [list(script.ANNOUNCE_URLS)]
assert report["properties"]["size"] == (11 if directory else 5)
assert report["files"] == (
[{"path": "content/first.txt", "size": 5}, {"path": "content/second.txt", "size": 6}]
if directory else [{"path": "first.txt", "size": 5}]
)
else:
assert f"Author: {script.AUTHOR}\n" in result.stdout
assert "Private: yes\n" in result.stdout
assert f"Size: {11 if directory else 5} bytes\n" in result.stdout
assert f"Tier 1: {script.ANNOUNCE_URLS[0]}\n" in result.stdout
assert f"Files ({2 if directory else 1}):\n" in result.stdout
assert f" {'content/' if directory else ''}first.txt: 5 bytes\n" in result.stdout
if directory:
assert " content/second.txt: 6 bytes\n" in result.stdout
assert output.read_bytes() == original
def test_inspect_rejects_invalid_torrent(tmp_path: Path) -> None:
torrent_path = tmp_path / "invalid.torrent"
torrent_path.write_bytes(b"invalid")
result = CliRunner().invoke(cli=script.main, args=["inspect", str(torrent_path)])
assert result.exit_code == 1
assert "Error:" in result.output
if __name__ == "__main__":
sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))