Spaces:
Sleeping
Sleeping
| import zipfile | |
| import os | |
| import logging | |
| # Configure logging | |
| class ZipExtractor: | |
| def __init__(self, zip_path, output_folder="extracted"): | |
| """ | |
| Initializes the ZipExtractor with the zip file path and output folder. | |
| Parameters: | |
| zip_path : str | |
| Path to the zip file. | |
| output_folder : str, optional | |
| Folder where extracted files will be saved (default is "extracted"). | |
| """ | |
| self.zip_path = zip_path | |
| self.output_folder = output_folder | |
| logging.info(f"Initialized ZipExtractor with zip_path: {zip_path} and output_folder: {output_folder}") | |
| def extract_files(self): | |
| """ | |
| Extracts files from the zip archive. If the zip file contains CSV files, | |
| they are extracted to the specified output folder. Handles errors if | |
| the zip file is corrupted or the path is invalid. | |
| Returns: | |
| None | |
| """ | |
| logging.basicConfig( | |
| level=logging.INFO, # Set the logging level to INFO | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.FileHandler("logging.log",'w+'), # Log to a file | |
| logging.StreamHandler() # Also log to console | |
| ] | |
| ) | |
| try: | |
| # Check if output folder exists; create if it doesn't | |
| if not os.path.exists(self.output_folder): | |
| os.makedirs(self.output_folder) | |
| logging.info(f"Created output folder: {self.output_folder}") | |
| # Open the zip file and start extraction | |
| with zipfile.ZipFile(self.zip_path, 'r') as zip_ref: | |
| csv_files = [file for file in zip_ref.namelist() if file.endswith('.csv')] | |
| if not csv_files: | |
| logging.warning("No CSV files found in the zip archive.") | |
| return | |
| for file in csv_files: | |
| # Extract only CSV files | |
| zip_ref.extract(file, self.output_folder) | |
| logging.info(f"Extracted {file} to {self.output_folder}") | |
| except zipfile.BadZipFile: | |
| logging.error("Error: The zip file is corrupted or invalid.") | |
| except FileNotFoundError: | |
| logging.error("Error: The zip file was not found at the specified path.") | |
| except Exception as e: | |
| logging.error(f"An unexpected error occurred: {e}") | |
| # Usage Example: | |
| if __name__ == "__main__": | |
| # Create an instance of ZipExtractor | |
| # extractor = ZipExtractor(zip_path='data/customer_churn_dataset-training-master.csv.zip') | |
| # # Call the extract_files method | |
| # extractor.extract_files() | |
| pass | |