#!/usr/bin/env python3 """ FEC Bulk Data Unzipper (High-Performance Edition) Unzips all FEC bulk data files downloaded by bulk_download_fec.py from D:/fec_data/bulk-downloads/ to D:/fec_data/unzipped/ Supports multiple extraction methods for maximum speed: - Python zipfile (default, portable) - 7-Zip (2-3x faster if installed) - Parallel processing (4-8x faster with multiple workers) Directory Structure: D:/fec_data/ ├── bulk-downloads/ # Original ZIP files (source) │ ├── candidate-master/ │ │ ├── 1980/cn80.zip │ │ └── 2024/cn24.zip │ ├── contributions-by-individuals/ │ │ └── 2024/indiv24.zip │ └── ... └── unzipped/ # Unzipped CSV files (destination) ├── candidate-master/ │ ├── 1980/ │ │ ├── cn80.txt │ │ ├── cn_header_file.csv │ │ └── ... │ └── 2024/ │ ├── cn24.txt │ └── ... ├── contributions-by-individuals/ │ └── 2024/ │ ├── indiv24.txt │ ├── indiv_header_file.csv │ └── ... └── ... Usage: # Quick start: Unzip only the latest 2 years with 8 workers (RECOMMENDED) python unzip_fec_data.py --latest 2 --workers 8 --base-dir /mnt/d/fec_data # Fast: Use 8 parallel workers (4-8x faster) python unzip_fec_data.py --workers 8 --base-dir /mnt/d/fec_data # Fastest: Use 7-Zip with 8 workers (10-15x faster if 7z installed) python unzip_fec_data.py --method 7z --workers 8 --base-dir /mnt/d/fec_data # Default (single-threaded Python) python unzip_fec_data.py --base-dir /mnt/d/fec_data # Specific category only python unzip_fec_data.py --category candidate-master --workers 4 # Specific years only python unzip_fec_data.py --years 2020,2022,2024 --workers 4 # Latest 5 years only python unzip_fec_data.py --latest 5 --workers 8 # Resume interrupted extraction python unzip_fec_data.py --resume --workers 8 # Dry run (show what would be unzipped) python unzip_fec_data.py --dry-run # Remove ZIP files after successful extraction (saves 50% disk space) python unzip_fec_data.py --remove-zips --workers 8 """ import argparse import json import zipfile import sys import subprocess import shutil from pathlib import Path from typing import Dict, List, Optional, Set, Tuple from datetime import datetime from loguru import logger from tqdm import tqdm from concurrent.futures import ProcessPoolExecutor, as_completed from functools import partial def _unzip_worker(args: Tuple[Path, Path, bool, Path, bool, bool, str]) -> Tuple[bool, Path, Path, int]: """ Worker function for parallel unzipping (must be at module level for pickling) Args: args: Tuple of (zip_path, dest_dir, dry_run, base_dir, use_7z, remove_zips, method) Returns: Tuple of (success, zip_path, dest_dir, file_count) """ zip_path, dest_dir, dry_run, base_dir, use_7z, remove_zips, method = args if dry_run: return True, zip_path, dest_dir, 0 try: # Create destination directory dest_dir.mkdir(parents=True, exist_ok=True) # Extract using chosen method if use_7z: # Use 7-Zip result = subprocess.run( ['7z', 'x', str(zip_path), f'-o{dest_dir}', '-y'], capture_output=True, text=True, check=True ) file_list = [str(f.relative_to(dest_dir)) for f in dest_dir.rglob('*') if f.is_file()] else: # Use Python zipfile with zipfile.ZipFile(zip_path, 'r') as zf: file_list = zf.namelist() zf.extractall(dest_dir) file_count = len(file_list) # Remove ZIP file if requested removed = False if remove_zips: zip_path.unlink() removed = True return True, zip_path, dest_dir, file_count except Exception as e: return False, zip_path, dest_dir, 0 class FECBulkUnzipper: """Unzip FEC bulk data files with parallel processing and 7-Zip support""" def __init__( self, base_dir: Path, resume: bool = False, remove_zips: bool = False, method: str = 'python', workers: int = 1 ): """ Initialize FEC bulk unzipper Args: base_dir: Base directory containing bulk-downloads/ (e.g., D:/fec_data/) resume: Skip already unzipped files remove_zips: Remove ZIP files after successful extraction method: Extraction method ('python', '7z', or 'auto') workers: Number of parallel workers (1 = single-threaded) """ self.base_dir = Path(base_dir) self.bulk_dir = self.base_dir / "bulk-downloads" self.unzipped_dir = self.base_dir / "unzipped" self.log_file = self.base_dir / "unzip_log.json" self.resume = resume self.remove_zips = remove_zips self.method = method self.workers = workers # Validate source directory exists if not self.bulk_dir.exists(): logger.error(f"❌ Source directory not found: {self.bulk_dir}") logger.info(f"💡 Run bulk_download_fec.py first to download FEC data") sys.exit(1) # Create destination directory self.unzipped_dir.mkdir(parents=True, exist_ok=True) # Detect extraction method self.use_7z = self._detect_extraction_method() # Load unzip log self.unzip_log = self._load_log() # Statistics self.stats = { 'total_zips': 0, 'unzipped': 0, 'skipped': 0, 'failed': 0, 'removed': 0, } def _detect_extraction_method(self) -> bool: """Detect if 7-Zip is available and choose best method""" if self.method == 'python': logger.info("📦 Using Python zipfile (portable)") return False if self.method == '7z': if shutil.which('7z'): logger.info("⚡ Using 7-Zip (2-3x faster)") return True else: logger.warning("⚠️ 7z not found, falling back to Python zipfile") logger.info("💡 Install with: sudo apt-get install p7zip-full") return False if self.method == 'auto': if shutil.which('7z'): logger.info("⚡ Using 7-Zip (auto-detected, 2-3x faster)") return True else: logger.info("📦 Using Python zipfile (7z not found)") return False logger.warning(f"⚠️ Unknown method '{self.method}', using Python zipfile") return False # Load unzip log self.unzip_log = self._load_log() # Statistics self.stats = { 'total_zips': 0, 'unzipped': 0, 'skipped': 0, 'failed': 0, 'removed': 0, } def _load_log(self) -> Dict: """Load unzip log""" if self.log_file.exists(): with open(self.log_file) as f: return json.load(f) return { 'started': datetime.now().isoformat(), 'last_updated': None, 'completed_files': {}, 'failed_files': {}, } def _save_log(self): """Save unzip log""" self.unzip_log['last_updated'] = datetime.now().isoformat() with open(self.log_file, 'w') as f: json.dump(self.unzip_log, f, indent=2) def _is_unzipped(self, zip_path: Path, dest_dir: Path) -> bool: """Check if ZIP file is already unzipped""" if not self.resume: return False # Check if in completed log zip_key = str(zip_path.relative_to(self.base_dir)) if zip_key in self.unzip_log['completed_files']: unzip_info = self.unzip_log['completed_files'][zip_key] # Verify destination directory exists and has files if dest_dir.exists() and any(dest_dir.iterdir()): # Check if all expected files exist expected_files = unzip_info.get('extracted_files', []) if expected_files: all_exist = all( (dest_dir / f).exists() for f in expected_files ) if all_exist: return True return False def _unzip_with_python(self, zip_path: Path, dest_dir: Path) -> Tuple[bool, List[str]]: """Unzip using Python's zipfile module""" with zipfile.ZipFile(zip_path, 'r') as zf: file_list = zf.namelist() zf.extractall(dest_dir) return True, file_list def _unzip_with_7z(self, zip_path: Path, dest_dir: Path) -> Tuple[bool, List[str]]: """Unzip using 7-Zip (2-3x faster)""" try: # Run 7z extract command result = subprocess.run( ['7z', 'x', str(zip_path), f'-o{dest_dir}', '-y'], capture_output=True, text=True, check=True ) # Get list of extracted files from dest_dir file_list = [ str(f.relative_to(dest_dir)) for f in dest_dir.rglob('*') if f.is_file() ] return True, file_list except subprocess.CalledProcessError as e: logger.error(f"7z extraction failed: {e.stderr}") return False, [] def _unzip_file( self, zip_path: Path, dest_dir: Path, dry_run: bool = False ) -> bool: """ Unzip a single file Args: zip_path: Path to ZIP file dest_dir: Destination directory dry_run: If True, don't actually unzip Returns: True if successful, False otherwise """ zip_key = str(zip_path.relative_to(self.base_dir)) # Check if already unzipped if self._is_unzipped(zip_path, dest_dir): self.stats['skipped'] += 1 return True if dry_run: logger.info(f"🔍 Would unzip: {zip_path} → {dest_dir}") return True try: # Create destination directory dest_dir.mkdir(parents=True, exist_ok=True) # Extract using chosen method if self.use_7z: success, file_list = self._unzip_with_7z(zip_path, dest_dir) else: success, file_list = self._unzip_with_python(zip_path, dest_dir) if not success: self.stats['failed'] += 1 return False # Log success self.unzip_log['completed_files'][zip_key] = { 'zip_path': str(zip_path), 'dest_dir': str(dest_dir), 'extracted_files': file_list[:100], # Limit log size 'file_count': len(file_list), 'unzipped_at': datetime.now().isoformat(), } self.stats['unzipped'] += 1 # Remove ZIP file if requested if self.remove_zips: zip_path.unlink() self.stats['removed'] += 1 return True except Exception as e: logger.error(f"❌ Failed to unzip {zip_path.name}: {e}") self.unzip_log['failed_files'][zip_key] = { 'error': str(e), 'failed_at': datetime.now().isoformat(), } self.stats['failed'] += 1 return False def find_zip_files( self, categories: Optional[Set[str]] = None, years: Optional[Set[str]] = None ) -> List[Path]: """ Find all ZIP files in bulk-downloads directory Args: categories: Optional set of categories to filter (e.g., {'candidate-master'}) years: Optional set of years to filter (e.g., {'2020', '2022', '2024'}) Returns: List of ZIP file paths """ zip_files = [] # Recursively find all .zip files for zip_path in self.bulk_dir.rglob("*.zip"): # Filter by category if categories: # Get category from path (e.g., bulk-downloads/candidate-master/2024/cn24.zip) relative_path = zip_path.relative_to(self.bulk_dir) category = relative_path.parts[0] if len(relative_path.parts) > 0 else None if category not in categories: continue # Filter by year if years: # Get year from path (e.g., bulk-downloads/candidate-master/2024/cn24.zip) relative_path = zip_path.relative_to(self.bulk_dir) year = relative_path.parts[1] if len(relative_path.parts) > 1 else None if year not in years: continue zip_files.append(zip_path) return sorted(zip_files) def unzip_all( self, categories: Optional[Set[str]] = None, years: Optional[Set[str]] = None, dry_run: bool = False ): """ Unzip all FEC bulk data files (with optional parallel processing) Args: categories: Optional set of categories to filter years: Optional set of years to filter dry_run: If True, don't actually unzip """ logger.info("=" * 70) logger.info("FEC BULK DATA UNZIPPER (HIGH-PERFORMANCE EDITION)") logger.info("=" * 70) logger.info(f"📂 Source: {self.bulk_dir}") logger.info(f"📁 Destination: {self.unzipped_dir}") logger.info(f"⚙️ Method: {'7-Zip' if self.use_7z else 'Python zipfile'}") logger.info(f"👷 Workers: {self.workers} {'(parallel)' if self.workers > 1 else '(single-threaded)'}") if categories: logger.info(f"📋 Categories: {', '.join(sorted(categories))}") if years: logger.info(f"📅 Years: {', '.join(sorted(years))}") if dry_run: logger.warning("🔍 DRY RUN MODE - No files will be unzipped") logger.info("") # Find all ZIP files zip_files = self.find_zip_files(categories=categories, years=years) self.stats['total_zips'] = len(zip_files) if not zip_files: logger.warning("⚠️ No ZIP files found") return logger.info(f"Found {len(zip_files)} ZIP files") logger.info("") # Prepare unzip tasks tasks = [] for zip_path in zip_files: relative_path = zip_path.relative_to(self.bulk_dir) dest_dir = self.unzipped_dir / relative_path.parent / zip_path.stem tasks.append((zip_path, dest_dir, dry_run)) # Execute unzipping (parallel or sequential) if self.workers > 1 and not dry_run: logger.info(f"🚀 Starting parallel extraction with {self.workers} workers") self._unzip_parallel(tasks) else: logger.info(f"📦 Starting sequential extraction") for zip_path, dest_dir, dry_run in tqdm(tasks, desc="Unzipping", unit="file"): self._unzip_file(zip_path, dest_dir, dry_run) # Save log periodically if not dry_run and self.stats['unzipped'] % 10 == 0: self._save_log() # Save final log if not dry_run: self._save_log() # Print summary logger.info("") logger.info("=" * 70) logger.info("SUMMARY") logger.info("=" * 70) logger.info(f"📊 Total ZIP files: {self.stats['total_zips']}") logger.info(f"✅ Unzipped: {self.stats['unzipped']}") logger.info(f"⏭️ Skipped: {self.stats['skipped']}") logger.info(f"❌ Failed: {self.stats['failed']}") if self.remove_zips: logger.info(f"🗑️ Removed: {self.stats['removed']}") logger.info("") if self.stats['failed'] > 0: logger.warning("⚠️ Some files failed to unzip. Check unzip_log.json for details.") def _unzip_parallel(self, tasks: List[Tuple[Path, Path, bool]]): """Unzip files in parallel using ProcessPoolExecutor""" # Prepare tasks with all necessary args for module-level worker worker_tasks = [ (zip_path, dest_dir, dry_run, self.base_dir, self.use_7z, self.remove_zips, self.method) for zip_path, dest_dir, dry_run in tasks ] with ProcessPoolExecutor(max_workers=self.workers) as executor: # Submit all tasks futures = {executor.submit(_unzip_worker, task): task[0] for task in worker_tasks} # Track progress with tqdm with tqdm(total=len(futures), desc="Unzipping (parallel)", unit="file") as pbar: for future in as_completed(futures): zip_path = futures[future] try: success, zip_path_result, dest_dir, file_count = future.result() if success: self.stats['unzipped'] += 1 # Log to unzip_log zip_key = str(zip_path_result.relative_to(self.base_dir)) self.unzip_log['completed_files'][zip_key] = { 'zip_path': str(zip_path_result), 'dest_dir': str(dest_dir), 'file_count': file_count, 'unzipped_at': datetime.now().isoformat(), } if self.remove_zips: self.stats['removed'] += 1 else: self.stats['failed'] += 1 logger.error(f"❌ Failed to unzip {zip_path.name}") # Log failure zip_key = str(zip_path_result.relative_to(self.base_dir)) self.unzip_log['failed_files'][zip_key] = { 'error': 'Extraction failed', 'failed_at': datetime.now().isoformat(), } except Exception as e: logger.error(f"❌ Worker exception for {zip_path.name}: {e}") self.stats['failed'] += 1 pbar.update(1) # Save log periodically if self.stats['unzipped'] % 10 == 0: self._save_log() def main(): """Main entry point""" parser = argparse.ArgumentParser( description="Unzip FEC bulk data files (High-Performance Edition)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__ ) parser.add_argument( '--base-dir', type=Path, default=Path('D:/fec_data'), help='Base directory containing bulk-downloads/ (default: D:/fec_data)' ) parser.add_argument( '--category', type=str, help='Specific category to unzip (e.g., candidate-master, contributions-by-individuals)' ) parser.add_argument( '--years', type=str, help='Comma-separated list of years to unzip (e.g., 2020,2022,2024)' ) parser.add_argument( '--latest', type=int, help='Only unzip the latest N years (e.g., --latest 2 for most recent 2 years)' ) parser.add_argument( '--workers', type=int, default=1, help='Number of parallel workers (default: 1, recommend: 4-8 for best performance)' ) parser.add_argument( '--method', type=str, default='auto', choices=['python', '7z', 'auto'], help='Extraction method: python (portable), 7z (2-3x faster), auto (use 7z if available)' ) parser.add_argument( '--resume', action='store_true', help='Skip already unzipped files' ) parser.add_argument( '--dry-run', action='store_true', help='Show what would be unzipped without actually unzipping' ) parser.add_argument( '--remove-zips', action='store_true', help='Remove ZIP files after successful extraction (saves 50%% disk space)' ) args = parser.parse_args() # Parse categories and years categories = {args.category} if args.category else None years = set(args.years.split(',')) if args.years else None # Handle --latest option (auto-determine latest N years) if args.latest: if args.years: logger.error("❌ Cannot use both --years and --latest options together") sys.exit(1) # Find all available years in the bulk-downloads directory base_dir = Path(args.base_dir) bulk_dir = base_dir / "bulk-downloads" if not bulk_dir.exists(): logger.error(f"❌ Bulk downloads directory not found: {bulk_dir}") sys.exit(1) # Scan for all year directories available_years = set() for category_dir in bulk_dir.iterdir(): if category_dir.is_dir(): for year_dir in category_dir.iterdir(): if year_dir.is_dir() and year_dir.name.isdigit(): available_years.add(year_dir.name) if not available_years: logger.error("❌ No year directories found in bulk-downloads") sys.exit(1) # Get latest N years sorted_years = sorted(available_years, reverse=True) latest_years = sorted_years[:args.latest] years = set(latest_years) logger.info(f"📅 Auto-selected latest {args.latest} years: {', '.join(sorted(latest_years, reverse=True))}") logger.info("") # Auto-detect optimal worker count if requested if args.workers == 0: args.workers = max(1, cpu_count() - 1) logger.info(f"Auto-detected {args.workers} workers (CPU count: {cpu_count()})") # Create unzipper unzipper = FECBulkUnzipper( base_dir=args.base_dir, resume=args.resume, remove_zips=args.remove_zips, method=args.method, workers=args.workers ) # Unzip all files unzipper.unzip_all( categories=categories, years=years, dry_run=args.dry_run ) if __name__ == '__main__': main()