''' Author: Fatma Saafan Date: 2/25/2024 Modified: 2/29/2024 Description: - The user has to be in the directory that has this 6 folder names (tables in database): dominion, ecap, navigator, pme, waterscope, weather. - The code will check if the current directory contains any of the specified folders. - If not, it will prompt the user to navigate to the appropriate directory and exit the script. - Otherwise, it will continue with the rest of the script. - This script resets the data in a SQLite database by deleting all records from tables corresponding to the provided data folder path. - It prompts the user for confirmation before performing the reset operation. - The script ignores differences in uppercase and lowercase letters when matching table names with the provided data folder path. - It also deletes the list of scanned files for that table from the scanned_files table in the db. Input: Example-1 (for current folder): "fcltls reset-database" "fcltls reset-database --data-folder-path=." Example-2 (for non-current folder): fcltls reset-database --data-folder-path=folderName "fcltls reset-database --data-folder-path=weather" Note: "--data-folder-path": The path to the folder corresponding to the data to be reset in the SQLite database. This folder must exist and contain the tables to be reset. Example: fcltls reset-database --data-folder-path PME (In this example, PME is folder corresponding to the data to be reset in the SQLite database) Output: Upon executing the command, the script prompts the user with a confirmation message: "Are you sure you want to reset the database? This action cannot be undone. [y/N]:" - If the user responds with "y" and the reset operation is successful, the script prints the tables name that gets reset successfully. - If the user responds with "n", the script prints "Deletion canceled." and exits without making any changes. ''' import os import click import sqlite3 from loguru import logger logger.trace(f"After imports {__file__}") # Logging a trace message after imports # Define the list of folder names ALLOWED_FOLDERS = ['dominion', 'ecap', 'navigator', 'pme', 'waterscope', 'weather'] # Check if the current directory contains any of the allowed folders if not any(folder.lower() in os.listdir('.') for folder in ALLOWED_FOLDERS): print("You are not in the folder that contains the required folders: dominion, ecap, navigator, pme, waterscope, weather.") print("Please navigate to the appropriate directory.") exit() # Getting the current directory path and the database file path current_directory = os.path.dirname(os.path.abspath(__file__)) database_file_path = os.path.join(current_directory, 'scannedData.db') # Establishing a connection to the SQLite database conn = sqlite3.connect(database_file_path) cursor = conn.cursor() # Defining a command line interface function using click @click.command() @click.option("--data-folder-path", type=click.Path(exists=True), help="Folder path for tables to be deleted from database") def cli(data_folder_path): """Reset the database by deleting tables corresponding to folders with the same name in the current directory.""" logger.debug("Entering reset-database") # Logging a debug message # If a specific data folder path is provided if data_folder_path and data_folder_path != '.': folder = os.path.basename(data_folder_path) # Asking for confirmation to delete data associated with the folder if confirm_deletion(folder): delete_folder_from_database(folder) click.echo(f"'{folder}' data has been deleted from the database.") click.echo("\nScanned files table is also reset.\n") # Close SQLite connection conn.close() else: click.echo("Deletion canceled.") return # Getting matching folders in the current directory matching_folders = get_matching_folders() # If no matching folders are found if not matching_folders: click.echo("No matching folders (tables names) found in the current directory.") return # If only one matching folder is found if len(matching_folders) == 1: folder = matching_folders[0] # Asking for confirmation to delete data associated with the folder if confirm_deletion(folder): delete_folder_from_database(folder) click.echo(f"'{folder}' data has been deleted from the database.") else: click.echo("Multiple matching folders found in the current directory:") # Listing the matching folders for i, folder in enumerate(matching_folders, start=1): click.echo(f"{i}. {folder}") # Prompting user to choose a folder or 'all' choice = input("Enter the number of the folder you want to delete or 'all' to delete all of them: ").strip() if choice.lower() == "all": # Asking for confirmation to delete all folders logger.warning("Are you sure you want to delete all folders? This action cannot be undone. [y/n]") confirmation = input().strip().lower() if confirmation == 'y': for folder in matching_folders: delete_folder_from_database(folder) click.echo(f"'{folder}' data has been deleted from the database.") else: click.echo("Deletion canceled.") elif choice.isdigit(): index = int(choice) - 1 if 0 <= index < len(matching_folders): folder = matching_folders[index] # Asking for confirmation to delete data associated with the folder logger.warning("Are you sure? [y/n]") confirmation = input().strip().lower() if confirmation == 'y': delete_folder_from_database(folder) click.echo(f"'{folder}' data has been deleted from the database.") else: click.echo("Deletion canceled.") else: click.echo("Invalid choice.") else: click.echo("Invalid input.") click.echo("\nScanned files table is also reset.\n") # Close SQLite connection conn.close() def confirm_deletion(folder): """Ask user for confirmation to delete the data associated with the given folder from the database.""" confirmation = input(f"Do you want to delete the data associated with the table '{folder}' from the database? (y/n): ").strip().lower() return confirmation == "y" def get_matching_folders(): """Get the list of folders in the current directory that have the same name as tables in the database.""" # Getting table names from the database tables = [row[0] for row in cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")] # Getting folder names in the current directory folders = [folder.lower() for folder in os.listdir('.') if os.path.isdir(folder)] # Finding matching folders matching_folders = [folder for folder in folders if folder in tables] return matching_folders def delete_folder_from_database(folder): """Delete the given folder's table from the database and remove corresponding rows from scanned_files.""" # Remove rows from scanned_files where metric column names match folder name (case insensitive) cursor.execute("DELETE FROM scanned_files WHERE LOWER(metric) = LOWER(?);", (folder,)) # Delete the table cursor.execute(f"DELETE FROM {folder};") conn.commit() if __name__ == '__main__': cli()