#!/usr/bin/env python3 import os import sys import argparse import json import logging import shutil import re from pathlib import Path # Setup defaults DEFAULT_CONFIG = { "log_file": "qbit_sanitize.log", "replace_spaces": False, "replacement_char": "_", "max_filename_length": 255, "illegal_chars": "<>:\"/\\|?*", "dry_run": False } def load_config(config_path): path = Path(config_path) if not path.exists(): # Try finding it relative to the script script_dir = Path(__file__).parent path = script_dir / "qbit_sanitize_config.json" if path.exists(): with open(path, 'r') as f: user_config = json.load(f) # Merge with defaults config = DEFAULT_CONFIG.copy() config.update(user_config) return config return DEFAULT_CONFIG def setup_logging(log_file, debug=False): # If log_file is relative, put it in script dir or logs dir? # Let's put it in script dir if relative log_path = Path(log_file) if not log_path.is_absolute(): log_path = Path(__file__).parent / log_file logging.basicConfig( filename=log_path, level=logging.DEBUG if debug else logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) # Also log to stdout for manual testing console = logging.StreamHandler() console.setLevel(logging.DEBUG if debug else logging.INFO) logging.getLogger('').addHandler(console) def sanitize_name(name, config): """ Sanitizes a filename/foldername string. 1. Removes illegal characters. 2. Trims length. 3. Optionally replaces spaces. """ clean = name # 1. Replace illegal chars for char in config["illegal_chars"]: clean = clean.replace(char, config["replacement_char"]) # 2. Replace spaces if configured if config["replace_spaces"]: clean = clean.replace(" ", config["replacement_char"]) # 3. Strip leading/trailing spaces/dots (Windows issue) clean = clean.strip(" .") # 4. Truncate limit = config["max_filename_length"] if len(clean) > limit: name_part, ext = os.path.splitext(clean) # Keep extension intact if possible if len(ext) < limit: name_part = name_part[:limit - len(ext)] clean = name_part + ext else: clean = clean[:limit] return clean def process_directory(directory, config): """ Recursively sanitizes files inside a directory. """ directory = Path(directory) if not directory.exists(): logging.error(f"Directory not found: {directory}") return # Process files # We walk bottom-up so we don't rename a directory before we process its contents for root, dirs, files in os.walk(directory, topdown=False): for name in files + dirs: old_path = Path(root) / name clean_name = sanitize_name(name, config) if clean_name != name: new_path = Path(root) / clean_name # Handle collision if new_path.exists(): logging.warning(f"Skipping rename {name} -> {clean_name}: Destination exists.") continue logging.info(f"Renaming item: '{name}' -> '{clean_name}'") if not config["dry_run"]: try: old_path.rename(new_path) except OSError as e: logging.error(f"Failed to rename {old_path}: {e}") def main(): parser = argparse.ArgumentParser(description="Sanitize qBittorrent downloads.") parser.add_argument("--content-path", required=True, help="Full path to content (%F in qBit)") parser.add_argument("--torrent-name", help="Original Torrent Name (%N in qBit)") parser.add_argument("--config", help="Path to config file", default="qbit_sanitize_config.json") parser.add_argument("--dry-run", action="store_true", help="Don't actually rename files") args = parser.parse_args() config = load_config(args.config) if args.dry_run: config["dry_run"] = True setup_logging(config["log_file"]) logging.info("--- Starting Sanitization ---") logging.info(f"Input Content Path: {args.content_path}") logging.info(f"Input Torrent Name: {args.torrent_name}") content_path = Path(args.content_path) if not content_path.exists(): logging.error(f"Content path does not exist: {content_path}") sys.exit(1) # 1. Rename the Root Folder if it doesn't match the Sanitized Torrent Name # This fixes the "truncated/random file name" issue at the top level. # We assume 'content_path' is what exists on disk (potentially mangled). # We assume 'torrent_name' is the source of truth for what it SHOULD be. current_root_path = content_path if args.torrent_name: sanitized_torrent_name = sanitize_name(args.torrent_name, config) parent_dir = content_path.parent expected_path = parent_dir / sanitized_torrent_name # Check if content_path is a directory (multi-file torrent) or file if content_path.is_dir(): # If the folder name on disk doesn't match our sanitized expectation if content_path.name != sanitized_torrent_name: logging.info(f"Top-level folder mismatch detected.") logging.info(f"Current: {content_path.name}") logging.info(f"Expected: {sanitized_torrent_name}") if not expected_path.exists(): if not config["dry_run"]: try: content_path.rename(expected_path) logging.info(f"Renamed root folder to: {expected_path}") current_root_path = expected_path except OSError as e: logging.error(f"Failed to rename root folder: {e}") else: logging.warning(f"Cannot rename root folder: Destination {expected_path} already exists.") # If it exists, maybe we should move contents? # For now, we assume we continue processing inside 'content_path' # OR we switch our focus to 'expected_path' if it was already moved? # Let's verify if they are the same inode? try: if content_path.resolve() == expected_path.resolve(): logging.info("Paths resolve to same location (case insensitivity?).") except: pass # 2. Recursively sanitize contents if current_root_path.is_dir(): process_directory(current_root_path, config) else: # It's a single file clean_filename = sanitize_name(current_root_path.name, config) if clean_filename != current_root_path.name: new_path = current_root_path.parent / clean_filename logging.info(f"Renaming single file: '{current_root_path.name}' -> '{clean_filename}'") if not config["dry_run"]: try: current_root_path.rename(new_path) except OSError as e: logging.error(f"Failed to rename file: {e}") logging.info("--- Sanitization Complete ---") if __name__ == "__main__": main()