import os import subprocess # Used to run external commands like 7z import logging import sys import shutil # Used to check if 7z exists # --- Configuration --- # Configure logging # Change level=logging.DEBUG to see more detailed output, including PATH logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()], ) # --- Global Variables --- SEVEN_ZIP_EXEC = None # Initialize as None # --- Core Functions --- def find_7zip(): """Attempts to find the 7-Zip executable and allows manual input as fallback.""" global SEVEN_ZIP_EXEC # Attempt to find 7z using shutil.which logging.debug("Attempting to locate 7z using shutil.which...") found_path = shutil.which("7z") or shutil.which("7z.exe") if found_path: logging.info(f"Automatically found 7-Zip executable at: {found_path}") SEVEN_ZIP_EXEC = found_path return True else: logging.warning("--------------------------------------------------") logging.warning( "Warning: 7-Zip executable ('7z' or '7z.exe') not found automatically in system PATH." ) logging.debug( f"Current PATH environment variable seen by Python:\n{os.environ.get('PATH', 'PATH variable not found')}" ) logging.warning("--------------------------------------------------") # Ask user for manual path print("\nCould not automatically find 7z.") manual_path = input( "Please enter the FULL path to your 7z.exe file (e.g., C:\\Program Files\\7-Zip\\7z.exe) or leave blank to exit: " ).strip() if not manual_path: logging.error("No manual path provided. Exiting.") return False elif os.path.isfile(manual_path) and manual_path.lower().endswith("7z.exe"): logging.info(f"Using manually provided 7-Zip path: {manual_path}") SEVEN_ZIP_EXEC = manual_path return True else: logging.error( f"Error: The provided path '{manual_path}' is not a valid file or does not seem to be 7z.exe." ) return False def decompress_file_7z(file_path, output_dir): """ Decompresses a single file using the 7z command-line tool into a specific directory. Args: file_path (str): The full path to the compressed file. output_dir (str): The specific directory where the contents should be extracted. Returns: bool: True if decompression was successful (7z exit code 0 or 1), False otherwise. """ if not SEVEN_ZIP_EXEC: logging.error("Cannot decompress: 7-Zip executable path is not set.") return False file_name = os.path.basename(file_path) # Use the provided output_dir directly logging.info( f"Attempting to decompress: {file_name} into directory: {output_dir} using 7-Zip" ) # Ensure the output directory exists (important change: create the specific named folder) try: os.makedirs(output_dir, exist_ok=True) logging.debug(f"Ensured output directory exists: {output_dir}") except OSError as e: logging.error(f"Error creating output directory {output_dir}: {e}") return False # Construct the 7z command # 'x' - extract with full paths # '-o{output_dir}' - specify output directory (no space after -o) # '-y' - assume Yes to all queries (e.g., overwrite) command = [SEVEN_ZIP_EXEC, "x", file_path, f"-o{output_dir}", "-y"] logging.debug( f"Executing command: {' '.join(command)}" ) # Log the command for debugging try: # Run the command result = subprocess.run( command, capture_output=True, text=True, check=False, encoding="utf-8", errors="replace", ) # Check the return code from 7z # Exit code 0: Success # Exit code 1: Warning (e.g., file already exists but -y used) - Treat as success for extraction # Exit code 2: Fatal error # Exit code 7: Command line error # Exit code 8: Not enough memory # Exit code 255: User stopped process if result.returncode in [0, 1]: if result.returncode == 1: logging.warning( f"7-Zip returned with a warning (Exit Code 1) for {file_name}. Check output if needed." ) if result.stdout: logging.debug(f"7-Zip stdout:\n{result.stdout.strip()}") if result.stderr: logging.warning(f"7-Zip stderr:\n{result.stderr.strip()}") logging.info( f"Successfully decompressed {file_name} into {output_dir} (Exit Code: {result.returncode})" ) return True else: # Log 7z errors if decompression failed logging.error( f"7-Zip failed to decompress {file_name}. Exit Code: {result.returncode}" ) if result.stdout: logging.error(f"7-Zip stdout:\n{result.stdout.strip()}") if result.stderr: logging.error(f"7-Zip stderr:\n{result.stderr.strip()}") return False except FileNotFoundError: logging.error( f"Decompression Error: 7-Zip executable not found at {SEVEN_ZIP_EXEC} when trying to run. Cannot decompress {file_name}." ) return False except PermissionError: logging.error( f"Decompression Error: Permission denied when trying to run 7-Zip or access files for {file_name}." ) return False except Exception as e: # Catch any other unexpected errors during subprocess execution logging.error( f"An unexpected error occurred while running 7-Zip for {file_name}: {e}", exc_info=True, ) return False def crawl_and_decompress(root_dir): """ Crawls a directory recursively, decompresses supported archive files found within it using 7-Zip into a subfolder named after the archive, and deletes the original archive file *only* if decompression was successful. Args: root_dir (str): The root directory path to start scanning. """ if not os.path.isdir(root_dir): logging.error( f"Error: The specified path '{root_dir}' is not a valid directory." ) return # Define supported extensions (lowercase) - 7z supports many more, add as needed supported_extensions = [ ".zip", ".7z", ".rar", ".tar", ".gz", ".tgz", ".bz2", ".tbz2", ".xz", ".lzma", ".iso", ".wim", ".cab", ".arj", ".z", # Add other extensions supported by your 7z installation if needed ] logging.info(f"Starting recursive scan in directory: {root_dir}") files_processed = 0 files_deleted = 0 errors_encountered = 0 # Walk through the directory tree for subdir, _, files in os.walk(root_dir): logging.debug(f"Scanning directory: {subdir}") for filename in files: # Check if the item is actually a file (skip directories/symlinks etc) file_path = os.path.join(subdir, filename) if not os.path.isfile(file_path): logging.debug(f"Skipping non-file item: {filename}") continue lower_filename = filename.lower() # Determine if the file has a supported extension is_supported = False matched_extension = None for ext in supported_extensions: # Use a robust check to handle multi-part extensions like .tar.gz correctly if lower_filename.endswith(ext): # Check if a longer extension match was already found # (e.g., if we found .gz but filename is .tar.gz, prefer .tar.gz later) if matched_extension is None or len(ext) > len(matched_extension): matched_extension = ext is_supported = True # Don't break here, continue checking for longer matches like .tar.gz vs .gz # If it's a supported archive type, process it if is_supported: logging.info(f"Found potential archive: {filename} in {subdir}") files_processed += 1 # --- Define the output directory (NEW LOGIC) --- # Get the filename without the matched extension base_name = filename[: -len(matched_extension)] # Create the target directory path (subfolder named after the archive) output_directory = os.path.join(subdir, base_name) logging.debug(f"Target extraction directory set to: {output_directory}") # --- Decompression Step (using 7z) --- # Pass the specific output_directory calculated above success = decompress_file_7z(file_path, output_directory) # --- Deletion Step (Conditional) --- if success: try: os.remove(file_path) logging.info(f"Successfully deleted original file: {filename}") files_deleted += 1 except OSError as e: logging.error( f"Error deleting file {filename} after successful extraction: {e}" ) errors_encountered += 1 else: logging.warning( f"Skipping deletion of {filename} due to decompression failure or skip." ) # Count decompression failures as errors errors_encountered += 1 # else: # Optional: Log skipped files if debugging needed # logging.debug(f"Skipping file (unsupported extension or not a file): {filename}") # --- Summary --- logging.info("--------------------------------------------------") logging.info("Processing Summary:") logging.info(f" Total potential archives found: {files_processed}") logging.info(f" Successfully decompressed & deleted: {files_deleted}") logging.info(f" Errors or skips (decompression/deletion): {errors_encountered}") logging.info("Scan complete.") logging.info("--------------------------------------------------") # --- Main Execution --- if __name__ == "__main__": print("--------------------------------------------------") print("Recursive Decompression Script (using 7-Zip)") print("Extracts archives into subfolders named after the archive.") # ... (rest of the initial print statements remain the same) ... print("--------------------------------------------------") print("REQUIREMENT: 7-Zip must be installed.") print( " The script will try to find it in PATH or ask for the full path." ) print("--------------------------------------------------") print("WARNING: This script deletes files. Ensure you have backups") print(" or test on non-critical data first.") print("--------------------------------------------------") # Attempt to find 7-Zip before asking for directory if not find_7zip(): sys.exit(1) # Exit if 7z is not found and user didn't provide a valid path # Get directory input from user target_directory = input("Enter the full path to the directory you want to scan: ") # Basic validation of input path target_directory = target_directory.strip() if not target_directory: print("Error: No directory path provided.", file=sys.stderr) elif not os.path.isdir(target_directory): print( f"Error: The path '{target_directory}' is not a valid directory.", file=sys.stderr, ) else: # Confirm before proceeding confirm = input( f"Scan '{target_directory}', extract archives into named subfolders, and delete originals? (yes/no): " ).lower() if confirm == "yes": # Run the main process crawl_and_decompress(target_directory) else: print("Operation cancelled by user.") print("Script finished.")