admin-scripts/update_tracker.py
2024-11-08 08:54:56 -06:00

129 lines
4.4 KiB
Python
Executable File

# /// script
# dependencies = [
# "qbittorrent-api",
# "docopt",
# "rich",
# ]
# ///
"""update_tracker.py
Description:
This script collects infohashes of all torrents in each qBittorrent instance,
updates opentracker, and reannounces all torrents to their trackers.
Expectations:
- A JSON qBittorrent authentication file at ~/.config/qbittorrent_auth.json
- SSH pubkey access to torrent tracker server
- rsync installed on the host system running this script
Usage:
update_tracker.py (--add-tracker DOMAIN)
update_tracker.py -h
Options:
--add-tracker DOMAIN ensure the provided tracker domain is added to each torrent's tracker list
-h, --help show this help message and exit
Examples:
update_tracker.py --add-tracker hyperreal.coffee
"""
import json
import subprocess
import tempfile
from pathlib import Path
import qbittorrentapi
from docopt import docopt
from rich.console import Console
from rich.text import Text
if __name__ == "__main__":
args = docopt(__doc__) # type: ignore
tracker_domain = args["--add-tracker"]
console = Console()
with console.status("[bold green]Executing the tasks...") as status:
# JSON file containing authentication info for each qBittorrent instance
QBITTORRENT_AUTH_FILE = Path.home().joinpath(".config/qbittorrent_auth.json")
# Open authentication file and load JSON data
with open(QBITTORRENT_AUTH_FILE, "r") as qbt_auth:
auth_data = json.load(qbt_auth)
# Collect infohashes of all torrents in each qBittorrent instance
console.log(
"Collecting infohashes of all torrents in each qBittorrent instance."
)
torrent_infohashes = []
for item in auth_data["instances"]:
with qbittorrentapi.Client(
host=item["hostname"],
username=item["username"],
password=item["password"],
) as qbt_client:
try:
qbt_client.auth_log_in()
except qbittorrentapi.LoginFailed as e:
print(e)
for torrent in qbt_client.torrents_info():
torrent_infohashes.append(torrent.hash)
# Format the infohashes to have a \n at the end
console.log("Formatting infohashes to have a newline at the end.")
format_infohashes = set([f"{infohash}\n" for infohash in torrent_infohashes])
# Create a NamedTemporaryFile and write all infohashes to it, one per line
console.log("Creating temporary file to write infohashes to.")
with tempfile.NamedTemporaryFile() as ntf:
with open(ntf.name, "w") as tf:
tf.writelines(format_infohashes)
# Use `sudo cp -f` to copy the infohashes file to the torrent tracker's config
# directory, overwriting the whitelist.txt file.
console.log(
"Copying the temporary infohashes file to the torrent tracker's whitelist."
)
subprocess.run(
["sudo", "cp", "-f", ntf.name, "/etc/opentracker/whitelist.txt"]
)
# Run `sudo systemctl restart opentracker.service`
console.log("Restarting opentracker.service")
subprocess.run(["sudo", "systemctl", "restart", "opentracker.service"])
# Reannounce all torrents in each qBittorrent instance to their trackers
console.log("Reannouncing all torrents to their trackers.")
for item in auth_data["instances"]:
with qbittorrentapi.Client(
host=item["hostname"],
username=item["username"],
password=item["password"],
) as qbt_client:
for torrent in qbt_client.torrents_info():
torrent.reannounce(torrent.hash)
console.log("Done!")
# Print output and make it look sexy ;)
console = Console()
tasks = Text("\nTasks completed:\n")
tasks.stylize("bold magenta")
console.print(tasks)
console.print(":white_check_mark: update the tracker's whitelist")
if tracker_domain:
console.print(
f":white_check_mark: ensure {tracker_domain}:6969/announce is in each torrent's tracker list"
)
console.print(":white_check_mark: reannounce all torrents to their trackers")
torrents = Text(str(len(torrent_infohashes)))
torrents.stylize("bold green")
console.print(torrents + " torrents were updated")