diff --git a/torrentify.py b/torrentify.py index 7c98081..be7ac7c 100755 --- a/torrentify.py +++ b/torrentify.py @@ -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, ""], - ), - ] - - answers = inquirer.prompt(questions) or {} - best_name = answers.get("best_name") - if best_name != "": - 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.+)\s+-+\s+@(?P\S+)\s+-+\s+(?P.+)\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__": diff --git a/torrentify_test.py b/torrentify_test.py new file mode 100644 index 0000000..429c1f5 --- /dev/null +++ b/torrentify_test.py @@ -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:]]))