QBT_cleanup

QBT_cleanup.py

#!/usr/bin/env python

import os
import logging
import shutil
import qbittorrentapi

QBT_USERNAME = os.environ.get("USERNAME")
QBT_PASSWORD = os.environ.get("PASSWORD")

if not QBT_USERNAME or not QBT_PASSWORD:
    raise ValueError("USERNAME and PASSWORD environment variables must be set")

log_dir = os.path.expanduser("~/.local/state/qbt-cleanup")
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, "qbt-cleanup.log")

logging.basicConfig(
    filename=log_file,
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)

logging.info("Starting qBittorrent auto-remove script")

try:
    client = qbittorrentapi.Client(
        host="http://localhost:8080",
        username=QBT_USERNAME,
        password=QBT_PASSWORD,
    )
    client.auth_log_in()
    logging.info("Logged in to qBittorrent Web UI")
except qbittorrentapi.LoginFailed as e:
    logging.error(f"Login failed: {e}")
    exit(1)

try:
    removable_states = {
        "uploading",
        "stalledUP",
        "pausedUP",
        "queuedUP",
        "checkingUP",
        "error",
    }
    delete_categories = {"prowlarr", "radarr", "tv-sonarr"}
    torrents = client.torrents_info()
    hashes_to_remove = []
    hashes_to_delete_files = []
    for torrent in torrents:
        if torrent.state in removable_states:
            hashes_to_remove.append(torrent.hash)
            delete_files = False
            if torrent.category == "seasonal":
                src = "/media" + torrent.info.content_path
                dst = os.path.join(
                    "/home/necraul/Videos/Seasonals", os.path.basename(src)
                )
                try:
                    if os.path.exists(src):
                        if os.path.isdir(src):
                            shutil.copytree(src, dst, dirs_exist_ok=True)
                        else:
                            shutil.copy2(src, dst)
                        logging.info(
                            f"Successfully merged/copied: {torrent.name}"
                        )
                    else:
                        logging.warning(
                            f"Destination already exists (likely merged by a previous torrent): {dst}"
                        )
                    delete_files = True
                    hashes_to_delete_files.append(torrent.hash)
                except Exception as e:
                    logging.error(f"Failed to copy {torrent.name}: {e}")
                    hashes_to_remove.remove(torrent.hash)
                    continue
            elif torrent.category in delete_categories:
                delete_files = True
            if delete_files:
                hashes_to_delete_files.append(torrent.hash)
            logging.info(
                "Removing torrent: %s (%s) - State: %s - Delete files: %s",
                torrent.name,
                torrent.hash,
                torrent.state,
                delete_files,
            )
    if hashes_to_delete_files:
        client.torrents_delete(
            delete_files=True, torrent_hashes=hashes_to_delete_files
        )
    remaining_to_remove = [
        h for h in hashes_to_remove if h not in hashes_to_delete_files
    ]
    if remaining_to_remove:
        client.torrents_delete(
            delete_files=False, torrent_hashes=remaining_to_remove
        )

except Exception as e:
    logging.error(f"Error while processing torrents: {e}")

logging.info("Script completed")