#!/usr/bin/env python3 """ Large-archive email extractor with crash-safe resume and persistent deduplication. Designed for large ZIP-of-ZIP datasets and Linux/AWS processing. Features: - ZIP, 7z, .gz, .tar.gz, .tgz - Nested archives with configurable recursion depth - Multiprocessing across top-level archives - SQLite-backed resume state and email deduplication - Streaming scans for ZIP/TAR/text members - Nested ZIPs are spooled to disk, never loaded wholesale into RAM - Atomic TXT/CSV/JSON export - Progress bar and logging - Per-file size limits - Linux-friendly case-insensitive extension discovery Requirements: python3 -m pip install py7zr tqdm Example: python3 email_extractor_fixed.py \ -z /data/archives \ -d charter.net \ -o /data/filtered_emails.txt \ -w 8 \ --temp-dir /data/tmp """ from __future__ import annotations import argparse import csv import fcntl import gzip import json import logging import os import re import shutil import sqlite3 import sys import tarfile import tempfile import traceback import zipfile from concurrent.futures import ProcessPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import BinaryIO, Iterable, Optional try: import py7zr PY7ZR_AVAILABLE = True except ImportError: py7zr = None PY7ZR_AVAILABLE = False try: from tqdm import tqdm TQDM_AVAILABLE = True except ImportError: tqdm = None TQDM_AVAILABLE = False # ----------------------------------------------------------------------------- # Configuration # ----------------------------------------------------------------------------- @dataclass(frozen=True) class Config: archive_folder: str output_file: str target_domain: str output_format: str = "txt" max_workers: int = 4 db_batch_size: int = 5000 resume: bool = True state_db: str = ".extraction_state.sqlite3" log_file: str = "extraction.log" nested_archives: bool = True max_nested_depth: int = 4 max_file_size_mb: int = 0 temp_dir: Optional[str] = None recursive_input: bool = False text_extensions: tuple[str, ...] = ( ".txt", ".csv", ".sql", ".log", ".dump", ".json", ".xml", ".html", ".htm", ".md", ".tsv", ".dat", ".bak", ) @dataclass(frozen=True) class WorkerConfig: target_domain: str text_extensions: tuple[str, ...] max_file_size_mb: int nested_archives: bool max_nested_depth: int temp_dir: Optional[str] db_batch_size: int live_state_db: Optional[str] = None live_output_file: Optional[str] = None @dataclass class ExtractionResult: archive_path: str result_db: Optional[str] emails_found: int files_scanned: int bytes_scanned: int nested_archives_processed: int skipped_files: int errors: list[str] = field(default_factory=list) success: bool = False # ----------------------------------------------------------------------------- # General helpers # ----------------------------------------------------------------------------- def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def setup_logging(log_file: str) -> logging.Logger: logger = logging.getLogger("EmailExtractor") logger.setLevel(logging.DEBUG) logger.handlers.clear() log_path = Path(log_file) if log_path.parent != Path("."): log_path.parent.mkdir(parents=True, exist_ok=True) fh = logging.FileHandler(log_path, encoding="utf-8") fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter( "%(asctime)s | %(levelname)-8s | %(processName)s | %(message)s" )) ch = logging.StreamHandler() ch.setLevel(logging.INFO) ch.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(fh) logger.addHandler(ch) return logger def normalize_domain(domain: str) -> str: domain = domain.strip().lower() if domain.startswith("@"): domain = domain[1:] if not domain or "." not in domain: raise ValueError(f"Invalid target domain: {domain!r}") if any(c.isspace() for c in domain): raise ValueError("Target domain must not contain whitespace") return domain def get_email_pattern(domain: str) -> re.Pattern[bytes]: """Compile an ASCII bytes regex for exact-domain email matching.""" d = re.escape(domain.encode("ascii")) local = rb"[A-Za-z0-9](?:[A-Za-z0-9._%+\-]{0,62}[A-Za-z0-9])?" return re.compile( rb"(? Optional[str]: n = name.lower() if n.endswith(".tar.gz") or n.endswith(".tgz"): return "tar" if n.endswith(".zip"): return "zip" if n.endswith(".7z"): return "7z" if n.endswith(".gz"): return "gz" return None def archive_suffix(name: str) -> str: n = name.lower() if n.endswith(".tar.gz"): return ".tar.gz" if n.endswith(".tgz"): return ".tgz" if n.endswith(".zip"): return ".zip" if n.endswith(".7z"): return ".7z" if n.endswith(".gz"): return ".gz" return ".bin" def is_text_name(name: str, text_extensions: tuple[str, ...]) -> bool: return name.lower().endswith(text_extensions) def format_bytes(value: int) -> str: size = float(value) for unit in ("B", "KiB", "MiB", "GiB", "TiB"): if size < 1024 or unit == "TiB": return f"{size:.1f} {unit}" size /= 1024 return f"{size:.1f} TiB" def remove_sqlite_files(path: str) -> None: for suffix in ("", "-wal", "-shm"): try: Path(path + suffix).unlink() except FileNotFoundError: pass def ensure_temp_dir(temp_dir: Optional[str]) -> Optional[str]: if temp_dir: p = Path(temp_dir) p.mkdir(parents=True, exist_ok=True) return str(p) return None # ----------------------------------------------------------------------------- # Worker-local email store # ----------------------------------------------------------------------------- class WorkerEmailSink: """Disk-backed per-archive deduplication plus optional live global persistence.""" def __init__( self, db_path: str, batch_size: int, live_state_db: Optional[str] = None, live_output_file: Optional[str] = None, ): self.db_path = db_path self.batch_size = max(100, batch_size) self.pending: set[str] = set() self.conn = sqlite3.connect(db_path) self.conn.execute("PRAGMA journal_mode=OFF") self.conn.execute("PRAGMA synchronous=OFF") self.conn.execute("PRAGMA temp_store=MEMORY") self.conn.execute("CREATE TABLE IF NOT EXISTS emails (email TEXT PRIMARY KEY)") self.live_state_db = live_state_db self.live_output_file = live_output_file self.live_conn: Optional[sqlite3.Connection] = None if live_state_db: self.live_conn = sqlite3.connect(live_state_db, timeout=60) self.live_conn.execute("PRAGMA busy_timeout=60000") self.live_conn.execute("PRAGMA synchronous=NORMAL") def add_matches(self, matches: Iterable[bytes]) -> None: for raw in matches: try: email = raw.decode("ascii").lower() except UnicodeDecodeError: continue self.pending.add(email) if len(self.pending) >= self.batch_size: self.flush() def _persist_live(self, emails: list[str]) -> None: if not emails or self.live_conn is None: return newly_global: list[str] = [] now = utc_now() try: self.live_conn.execute("BEGIN IMMEDIATE") for email in emails: cur = self.live_conn.execute( "INSERT OR IGNORE INTO emails(email, first_seen) VALUES (?, ?)", (email, now), ) if cur.rowcount == 1: newly_global.append(email) self.live_conn.commit() except Exception: self.live_conn.rollback() raise if newly_global and self.live_output_file: destination = Path(self.live_output_file) destination.parent.mkdir(parents=True, exist_ok=True) # Linux advisory lock prevents concurrent workers from interleaving writes. with open(destination, "a", encoding="utf-8", newline="\n") as f: fcntl.flock(f.fileno(), fcntl.LOCK_EX) try: for email in sorted(newly_global): f.write(email + "\n") f.flush() finally: fcntl.flock(f.fileno(), fcntl.LOCK_UN) def flush(self) -> None: if not self.pending: return emails = sorted(self.pending) self.conn.executemany( "INSERT OR IGNORE INTO emails(email) VALUES (?)", ((email,) for email in emails), ) self.conn.commit() self._persist_live(emails) self.pending.clear() def count(self) -> int: self.flush() return int(self.conn.execute("SELECT COUNT(*) FROM emails").fetchone()[0]) def close(self) -> None: self.flush() self.conn.close() if self.live_conn is not None: self.live_conn.close() self.live_conn = None # ----------------------------------------------------------------------------- # Streaming scanner # ----------------------------------------------------------------------------- def scan_stream( stream: BinaryIO, pattern: re.Pattern[bytes], sink: WorkerEmailSink, max_email_len: int, chunk_size: int = 4 * 1024 * 1024, ) -> int: """ Scan a binary stream without loading the whole file into RAM. A small overlap is retained between chunks so addresses crossing chunk boundaries are still detected correctly. SQLite removes any overlap dupes. """ buffer = b"" total = 0 overlap = max_email_len + 2 while True: chunk = stream.read(chunk_size) if not chunk: break total += len(chunk) buffer += chunk # Keep enough bytes to cover a full email plus one boundary character. if len(buffer) <= overlap * 3: continue safe_end = len(buffer) - overlap accepted = [] for match in pattern.finditer(buffer): if match.end() <= safe_end: accepted.append(match.group(0)) if accepted: sink.add_matches(accepted) # Persist matches after each scanned chunk so TXT/state update while running. sink.flush() # Preserve enough history to re-evaluate boundary-spanning matches. discard = max(0, safe_end - overlap) buffer = buffer[discard:] if buffer: sink.add_matches(match.group(0) for match in pattern.finditer(buffer)) sink.flush() return total def copy_stream_to_temp( src: BinaryIO, temp_dir: Optional[str], suffix: str, max_bytes: Optional[int], ) -> tuple[Optional[str], int]: """Copy a stream to a temp file, aborting if it exceeds max_bytes.""" fd, path = tempfile.mkstemp(prefix="email_nested_", suffix=suffix, dir=temp_dir) os.close(fd) total = 0 try: with open(path, "wb") as out: while True: chunk = src.read(8 * 1024 * 1024) if not chunk: break total += len(chunk) if max_bytes is not None and total > max_bytes: out.close() Path(path).unlink(missing_ok=True) return None, total out.write(chunk) return path, total except Exception: Path(path).unlink(missing_ok=True) raise # ----------------------------------------------------------------------------- # Archive processor used inside each worker process # ----------------------------------------------------------------------------- class ArchiveProcessor: def __init__(self, config: WorkerConfig, sink: WorkerEmailSink): self.config = config self.sink = sink self.pattern = get_email_pattern(config.target_domain) self.max_file_size = ( None if config.max_file_size_mb == 0 else config.max_file_size_mb * 1024 * 1024 ) self.max_email_len = 64 + 1 + len(config.target_domain) self.files_scanned = 0 self.bytes_scanned = 0 self.nested_archives_processed = 0 self.skipped_files = 0 self.errors: list[str] = [] self.had_processing_error = False def add_error(self, message: str) -> None: self.had_processing_error = True if len(self.errors) < 200: self.errors.append(message) def process(self, path: str, depth: int = 0, logical_name: Optional[str] = None) -> None: name = logical_name or os.path.basename(path) kind = archive_kind(name) if kind is None: raise ValueError(f"Unsupported archive type: {name}") if depth > self.config.max_nested_depth: self.skipped_files += 1 return if depth > 0: self.nested_archives_processed += 1 try: if kind == "zip": self._process_zip(path, depth, name) elif kind == "tar": self._process_tar(path, depth, name) elif kind == "gz": self._process_gzip(path, depth, name) elif kind == "7z": self._process_7z(path, depth, name) except Exception as exc: self.add_error(f"Archive error [{name}]: {exc}") def _handle_member_stream( self, stream: BinaryIO, member_name: str, member_size: int, depth: int, context: str, ) -> None: if self.max_file_size is not None and member_size > self.max_file_size: self.skipped_files += 1 return nested_kind = archive_kind(member_name) if nested_kind and self.config.nested_archives: if depth >= self.config.max_nested_depth: self.skipped_files += 1 return temp_path = None try: temp_path, copied = copy_stream_to_temp( stream, self.config.temp_dir, archive_suffix(member_name), self.max_file_size, ) if temp_path is None: self.skipped_files += 1 return self.process( temp_path, depth + 1, logical_name=member_name, ) except Exception as exc: self.add_error(f"Nested archive error [{context}!{member_name}]: {exc}") finally: if temp_path: Path(temp_path).unlink(missing_ok=True) return if is_text_name(member_name, self.config.text_extensions): try: scanned = scan_stream( stream, self.pattern, self.sink, self.max_email_len, ) self.files_scanned += 1 self.bytes_scanned += scanned except Exception as exc: self.add_error(f"Read error [{context}!{member_name}]: {exc}") def _process_zip(self, path: str, depth: int, logical_name: str) -> None: try: with zipfile.ZipFile(path, "r") as archive: for info in archive.infolist(): if info.is_dir(): continue try: if self.max_file_size is not None and info.file_size > self.max_file_size: self.skipped_files += 1 continue member_name = info.filename relevant = ( is_text_name(member_name, self.config.text_extensions) or (self.config.nested_archives and archive_kind(member_name) is not None) ) if not relevant: continue with archive.open(info, "r") as stream: self._handle_member_stream( stream, member_name, info.file_size, depth, logical_name, ) except Exception as exc: self.add_error( f"ZIP member error [{logical_name}!{info.filename}]: {exc}" ) except zipfile.BadZipFile as exc: raise RuntimeError(f"Bad ZIP file: {exc}") from exc def _process_tar(self, path: str, depth: int, logical_name: str) -> None: # Stream members with extractfile(); never extract paths to the filesystem. with tarfile.open(path, "r:*") as archive: for member in archive: if not member.isfile(): continue member_name = member.name relevant = ( is_text_name(member_name, self.config.text_extensions) or (self.config.nested_archives and archive_kind(member_name) is not None) ) if not relevant: continue if self.max_file_size is not None and member.size > self.max_file_size: self.skipped_files += 1 continue try: stream = archive.extractfile(member) if stream is None: self.add_error(f"Could not read TAR member [{logical_name}!{member_name}]") continue with stream: self._handle_member_stream( stream, member_name, member.size, depth, logical_name, ) except Exception as exc: self.add_error( f"TAR member error [{logical_name}!{member_name}]: {exc}" ) def _process_gzip(self, path: str, depth: int, logical_name: str) -> None: # .tar.gz is routed to _process_tar before this method. inner_name = logical_name[:-3] if logical_name.lower().endswith(".gz") else logical_name suffix = archive_suffix(inner_name) if archive_kind(inner_name) else Path(inner_name).suffix or ".bin" temp_path = None try: with gzip.open(path, "rb") as src: temp_path, uncompressed_size = copy_stream_to_temp( src, self.config.temp_dir, suffix, self.max_file_size, ) if temp_path is None: self.skipped_files += 1 return inner_archive = archive_kind(inner_name) if inner_archive and self.config.nested_archives: if depth >= self.config.max_nested_depth: self.skipped_files += 1 return self.process(temp_path, depth + 1, logical_name=inner_name) return # Extensionless .gz files are commonly compressed text dumps. if is_text_name(inner_name, self.config.text_extensions) or not Path(inner_name).suffix: with open(temp_path, "rb") as stream: scanned = scan_stream( stream, self.pattern, self.sink, self.max_email_len, ) self.files_scanned += 1 self.bytes_scanned += scanned finally: if temp_path: Path(temp_path).unlink(missing_ok=True) def _process_7z(self, path: str, depth: int, logical_name: str) -> None: if not PY7ZR_AVAILABLE: raise RuntimeError("py7zr is not installed; run: python3 -m pip install py7zr") # Current py7zr removed read()/readall(). We selectively extract eligible # members to a temp directory, scan them, then immediately delete it. with py7zr.SevenZipFile(path, mode="r") as archive: members = archive.list() selected: list[tuple[str, int, bool]] = [] total_selected_size = 0 archive_uncompressed_total = 0 for member in members: name = member.filename is_dir = bool(getattr(member, "is_directory", False)) is_file = bool(getattr(member, "is_file", not is_dir)) is_symlink = bool(getattr(member, "is_symlink", False)) if is_dir or not is_file or is_symlink: continue size = int(getattr(member, "uncompressed", 0) or 0) archive_uncompressed_total += max(0, size) if self.max_file_size is not None and size > self.max_file_size: self.skipped_files += 1 continue nested = archive_kind(name) is not None text = is_text_name(name, self.config.text_extensions) if text or (nested and self.config.nested_archives): selected.append((name, size, nested)) total_selected_size += size if not selected: return extract_dir = tempfile.mkdtemp(prefix="email_7z_", dir=self.config.temp_dir) try: targets = [name for name, _, _ in selected] # Modern py7zr can cap total decompression. For solid 7z archives, # extracting one selected file may require decompressing preceding # unselected members, so the cap must cover the archive metadata total # rather than only selected members. Per-member limits still control # what is written to the temporary filesystem. kwargs = {} try: if self.max_file_size is not None: kwargs["max_extract_size"] = max(1, archive_uncompressed_total) archive = py7zr.SevenZipFile(path, mode="r", **kwargs) except TypeError: # Compatibility fallback for older py7zr releases. archive = py7zr.SevenZipFile(path, mode="r") with archive: archive.extract(path=extract_dir, targets=targets) base = Path(extract_dir).resolve() for name, size, nested in selected: normalized = name.replace("\\", "/") candidate = (base / normalized).resolve() try: candidate.relative_to(base) except ValueError: self.add_error(f"Unsafe 7z member path skipped [{logical_name}!{name}]") continue if not candidate.is_file(): self.add_error(f"Extracted 7z member missing [{logical_name}!{name}]") continue if nested and self.config.nested_archives: if depth >= self.config.max_nested_depth: self.skipped_files += 1 continue self.process(str(candidate), depth + 1, logical_name=name) elif is_text_name(name, self.config.text_extensions): try: with open(candidate, "rb") as stream: scanned = scan_stream( stream, self.pattern, self.sink, self.max_email_len, ) self.files_scanned += 1 self.bytes_scanned += scanned except Exception as exc: self.add_error(f"7z member read error [{logical_name}!{name}]: {exc}") finally: shutil.rmtree(extract_dir, ignore_errors=True) # ----------------------------------------------------------------------------- # Worker entry point # ----------------------------------------------------------------------------- def process_archive_job(args: tuple[str, WorkerConfig]) -> ExtractionResult: archive_path, config = args fd, result_db = tempfile.mkstemp( prefix="email_worker_result_", suffix=".sqlite3", dir=config.temp_dir, ) os.close(fd) sink: Optional[WorkerEmailSink] = None processor: Optional[ArchiveProcessor] = None try: sink = WorkerEmailSink( result_db, config.db_batch_size, live_state_db=config.live_state_db, live_output_file=config.live_output_file, ) processor = ArchiveProcessor(config, sink) processor.process(archive_path, depth=0, logical_name=os.path.basename(archive_path)) emails_found = sink.count() sink.close() sink = None return ExtractionResult( archive_path=archive_path, result_db=result_db, emails_found=emails_found, files_scanned=processor.files_scanned, bytes_scanned=processor.bytes_scanned, nested_archives_processed=processor.nested_archives_processed, skipped_files=processor.skipped_files, errors=processor.errors, success=not processor.had_processing_error, ) except Exception as exc: if sink is not None: try: sink.close() except Exception: pass errors = [] if processor is None else list(processor.errors) errors.append(f"Fatal worker error: {exc}") errors.append(traceback.format_exc(limit=8)) return ExtractionResult( archive_path=archive_path, result_db=result_db if os.path.exists(result_db) else None, emails_found=0, files_scanned=0 if processor is None else processor.files_scanned, bytes_scanned=0 if processor is None else processor.bytes_scanned, nested_archives_processed=0 if processor is None else processor.nested_archives_processed, skipped_files=0 if processor is None else processor.skipped_files, errors=errors[:200], success=False, ) # ----------------------------------------------------------------------------- # Persistent state / deduplication database # ----------------------------------------------------------------------------- class StateDB: def __init__(self, path: str, reset: bool = False): self.path = path parent = Path(path).parent if parent != Path("."): parent.mkdir(parents=True, exist_ok=True) if reset: remove_sqlite_files(path) self.conn = sqlite3.connect(path, timeout=60) self.conn.execute("PRAGMA journal_mode=WAL") self.conn.execute("PRAGMA synchronous=NORMAL") self.conn.execute("PRAGMA temp_store=MEMORY") self.conn.execute("PRAGMA busy_timeout=60000") self.conn.executescript( """ CREATE TABLE IF NOT EXISTS emails ( email TEXT PRIMARY KEY, first_seen TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS archives ( path TEXT PRIMARY KEY, size INTEGER NOT NULL, mtime_ns INTEGER NOT NULL, status TEXT NOT NULL, files_scanned INTEGER NOT NULL DEFAULT 0, bytes_scanned INTEGER NOT NULL DEFAULT 0, nested_archives INTEGER NOT NULL DEFAULT 0, skipped_files INTEGER NOT NULL DEFAULT 0, emails_in_archive INTEGER NOT NULL DEFAULT 0, errors_json TEXT NOT NULL DEFAULT '[]', completed_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_archives_status ON archives(status); """ ) self.conn.commit() @staticmethod def canonical_path(path: str) -> str: return str(Path(path).resolve()) def is_successfully_processed(self, path: str) -> bool: canonical = self.canonical_path(path) st = os.stat(path) row = self.conn.execute( "SELECT status, size, mtime_ns FROM archives WHERE path = ?", (canonical,), ).fetchone() return bool( row and row[0] == "success" and int(row[1]) == int(st.st_size) and int(row[2]) == int(st.st_mtime_ns) ) def merge_result(self, result: ExtractionResult) -> int: canonical = self.canonical_path(result.archive_path) st = os.stat(result.archive_path) completed = utc_now() status = "success" if result.success else "failed" new_unique = 0 try: self.conn.execute("BEGIN IMMEDIATE") before = self.conn.total_changes if result.result_db and os.path.exists(result.result_db): self.conn.execute("ATTACH DATABASE ? AS workerdb", (result.result_db,)) try: self.conn.execute( """ INSERT OR IGNORE INTO emails(email, first_seen) SELECT email, ? FROM workerdb.emails """, (completed,), ) new_unique = self.conn.total_changes - before finally: # DETACH must happen after the transaction below is committed. pass self.conn.execute( """ INSERT INTO archives( path, size, mtime_ns, status, files_scanned, bytes_scanned, nested_archives, skipped_files, emails_in_archive, errors_json, completed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(path) DO UPDATE SET size=excluded.size, mtime_ns=excluded.mtime_ns, status=excluded.status, files_scanned=excluded.files_scanned, bytes_scanned=excluded.bytes_scanned, nested_archives=excluded.nested_archives, skipped_files=excluded.skipped_files, emails_in_archive=excluded.emails_in_archive, errors_json=excluded.errors_json, completed_at=excluded.completed_at """, ( canonical, int(st.st_size), int(st.st_mtime_ns), status, result.files_scanned, result.bytes_scanned, result.nested_archives_processed, result.skipped_files, result.emails_found, json.dumps(result.errors[-200:]), completed, ), ) self.conn.commit() if result.result_db and os.path.exists(result.result_db): try: self.conn.execute("DETACH DATABASE workerdb") except sqlite3.Error: pass return max(0, new_unique) except Exception: self.conn.rollback() try: self.conn.execute("DETACH DATABASE workerdb") except sqlite3.Error: pass raise def total_unique_emails(self) -> int: return int(self.conn.execute("SELECT COUNT(*) FROM emails").fetchone()[0]) def summary(self) -> dict[str, int]: row = self.conn.execute( """ SELECT COUNT(*), SUM(CASE WHEN status='success' THEN 1 ELSE 0 END), SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END), COALESCE(SUM(files_scanned), 0), COALESCE(SUM(bytes_scanned), 0), COALESCE(SUM(nested_archives), 0), COALESCE(SUM(skipped_files), 0) FROM archives """ ).fetchone() return { "archives_recorded": int(row[0] or 0), "archives_success": int(row[1] or 0), "archives_failed": int(row[2] or 0), "files_scanned": int(row[3] or 0), "bytes_scanned": int(row[4] or 0), "nested_archives": int(row[5] or 0), "skipped_files": int(row[6] or 0), "unique_emails": self.total_unique_emails(), } def close(self) -> None: self.conn.close() # ----------------------------------------------------------------------------- # Discovery and export # ----------------------------------------------------------------------------- def discover_archives(folder: str, recursive: bool) -> list[str]: root = Path(folder) iterator = root.rglob("*") if recursive else root.iterdir() found: list[str] = [] for path in iterator: if path.is_file() and archive_kind(path.name) is not None: found.append(str(path.resolve())) found.sort(key=lambda p: p.lower()) return found def normalized_output_path(path: str, fmt: str) -> str: p = Path(path) desired = "." + fmt if p.suffix.lower() != desired: p = p.with_suffix(desired) return str(p) def export_results(state: StateDB, output_path: str, fmt: str, domain: str) -> tuple[str, int]: output_path = normalized_output_path(output_path, fmt) destination = Path(output_path) destination.parent.mkdir(parents=True, exist_ok=True) fd, tmp_path = tempfile.mkstemp( prefix=destination.name + ".", suffix=".tmp", dir=str(destination.parent), ) os.close(fd) count = state.total_unique_emails() try: if fmt == "txt": with open(tmp_path, "w", encoding="utf-8", newline="\n") as f: for (email,) in state.conn.execute("SELECT email FROM emails ORDER BY email"): f.write(email + "\n") elif fmt == "csv": with open(tmp_path, "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(["email", "domain", "local_part", "first_seen"]) for email, first_seen in state.conn.execute( "SELECT email, first_seen FROM emails ORDER BY email" ): local_part, email_domain = email.rsplit("@", 1) writer.writerow([email, email_domain, local_part, first_seen]) elif fmt == "json": with open(tmp_path, "w", encoding="utf-8") as f: f.write('{\n "domain": ') json.dump(domain, f) f.write(',\n "count": ' + str(count)) f.write(',\n "exported_at": ') json.dump(utc_now(), f) f.write(',\n "emails": [\n') first = True for (email,) in state.conn.execute("SELECT email FROM emails ORDER BY email"): if not first: f.write(",\n") f.write(" ") json.dump(email, f) first = False f.write("\n ]\n}\n") else: raise ValueError(f"Unsupported output format: {fmt}") os.replace(tmp_path, destination) return str(destination), count except Exception: Path(tmp_path).unlink(missing_ok=True) raise # ----------------------------------------------------------------------------- # Main orchestration # ----------------------------------------------------------------------------- def run_extraction(config: Config) -> int: logger = setup_logging(config.log_file) temp_dir = ensure_temp_dir(config.temp_dir) logger.info("=" * 72) logger.info("EMAIL EXTRACTION STARTED") logger.info("Target domain: %s", config.target_domain) logger.info("Archive folder: %s", config.archive_folder) logger.info("Workers: %d", config.max_workers) logger.info("Nested archives: %s (max depth %d)", config.nested_archives, config.max_nested_depth) if config.max_file_size_mb == 0: logger.info("Max member size: unlimited") else: logger.info("Max member size: %d MB", config.max_file_size_mb) logger.info("State DB: %s", config.state_db) if temp_dir: logger.info("Temp directory: %s", temp_dir) logger.info("=" * 72) folder = Path(config.archive_folder) if not folder.exists() or not folder.is_dir(): logger.error("Archive folder not found or not a directory: %s", folder) return 2 if not PY7ZR_AVAILABLE: logger.warning("py7zr is not installed. Any .7z archive will fail until it is installed.") state = StateDB(config.state_db, reset=not config.resume) try: all_archives = discover_archives(config.archive_folder, config.recursive_input) pending = [p for p in all_archives if not state.is_successfully_processed(p)] already_done = len(all_archives) - len(pending) # For TXT output, rebuild the live file from persistent state before workers # start, then workers append each newly-global unique email as it is found. live_output_file: Optional[str] = None if config.output_format == "txt": live_output_file, live_count = export_results( state, config.output_file, "txt", config.target_domain ) logger.info("Live TXT initialized with %d existing unique email(s): %s", live_count, live_output_file) logger.info( "Found %d top-level archive(s): %d pending, %d already completed", len(all_archives), len(pending), already_done, ) worker_cfg = WorkerConfig( target_domain=config.target_domain, text_extensions=config.text_extensions, max_file_size_mb=config.max_file_size_mb, nested_archives=config.nested_archives, max_nested_depth=config.max_nested_depth, temp_dir=temp_dir, db_batch_size=config.db_batch_size, live_state_db=str(Path(config.state_db).resolve()), live_output_file=(str(Path(live_output_file).resolve()) if live_output_file else None), ) failed_this_run = 0 runtime_files = 0 runtime_bytes = 0 pbar = None if TQDM_AVAILABLE and pending: pbar = tqdm(total=len(pending), desc="Processing", unit="archive") elif pending: logger.info("Install tqdm for a progress bar: python3 -m pip install tqdm") if pending: with ProcessPoolExecutor(max_workers=max(1, config.max_workers)) as executor: futures = { executor.submit(process_archive_job, (archive_path, worker_cfg)): archive_path for archive_path in pending } for future in as_completed(futures): archive_path = futures[future] result_db_to_delete: Optional[str] = None try: result = future.result() result_db_to_delete = result.result_db new_unique = state.merge_result(result) runtime_files += result.files_scanned runtime_bytes += result.bytes_scanned if result.success: logger.debug( "Processed %s: %d archive-unique email(s), %d newly-global, " "%d files, %s scanned", os.path.basename(archive_path), result.emails_found, new_unique, result.files_scanned, format_bytes(result.bytes_scanned), ) else: failed_this_run += 1 logger.warning( "PARTIAL/FAILED: %s (%d error(s)); it will be retried on the next resume run", os.path.basename(archive_path), len(result.errors), ) for err in result.errors: logger.debug("%s", err) except Exception as exc: failed_this_run += 1 logger.exception("Failed to merge/process %s: %s", archive_path, exc) finally: if result_db_to_delete: remove_sqlite_files(result_db_to_delete) if pbar: try: pbar.set_postfix( emails=state.total_unique_emails(), files=runtime_files, ) except Exception: pass pbar.update(1) if pbar: pbar.close() output_path, output_count = export_results( state, config.output_file, config.output_format, config.target_domain, ) summary = state.summary() logger.info("=" * 72) logger.info("EXTRACTION PASS COMPLETED") logger.info("Successful top-level archives in state: %d", summary["archives_success"]) logger.info("Failed/partial top-level archives in state: %d", summary["archives_failed"]) logger.info("Internal text files scanned in state: %d", summary["files_scanned"]) logger.info("Nested archives processed in state: %d", summary["nested_archives"]) logger.info("Files skipped by limits/depth: %d", summary["skipped_files"]) logger.info("Uncompressed text bytes scanned in state: %s", format_bytes(summary["bytes_scanned"])) logger.info("Total unique emails: %d", output_count) logger.info("Output file: %s", output_path) logger.info("State database: %s", config.state_db) logger.info("=" * 72) return 1 if failed_this_run else 0 finally: state.close() # ----------------------------------------------------------------------------- # CLI # ----------------------------------------------------------------------------- def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Extract exact-domain emails from large ZIP/7z/gz/tar.gz collections", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 email_extractor_fixed.py -z /data/archives -d charter.net -w 8 python3 email_extractor_fixed.py \\ -z /data/archives \\ -d charter.net \\ -o /data/filtered_emails.txt \\ --state-db /data/extraction_state.sqlite3 \\ --temp-dir /data/tmp \\ -w 8 # Start completely fresh (deletes/recreates the state database): python3 email_extractor_fixed.py -z /data/archives -d charter.net --no-resume Requirements: python3 -m pip install py7zr tqdm """, ) parser.add_argument( "-z", "--zip-folder", "--archive-folder", dest="archive_folder", required=True, help="Directory containing top-level archives", ) parser.add_argument( "-d", "--domain", required=True, help="Exact target email domain, e.g. charter.net", ) parser.add_argument( "-o", "--output", default="filtered_emails.txt", help="Output path (default: filtered_emails.txt)", ) parser.add_argument( "-f", "--format", choices=("txt", "csv", "json"), default="txt", help="Output format (default: txt)", ) parser.add_argument( "-w", "--workers", type=int, default=4, help="Parallel top-level archive workers (default: 4)", ) parser.add_argument( "-b", "--batch-size", type=int, default=5000, help="Worker SQLite email insert batch size (default: 5000)", ) parser.add_argument( "--state-db", default=".extraction_state.sqlite3", help="SQLite resume/dedup state database", ) parser.add_argument( "--no-resume", action="store_true", help="Delete existing state DB and start fresh", ) parser.add_argument( "--no-nested", action="store_true", help="Do not inspect nested archives", ) parser.add_argument( "--max-depth", type=int, default=4, help="Maximum nested archive depth (default: 4)", ) parser.add_argument( "--max-file-size", type=int, default=0, help="Maximum uncompressed member size in MB; 0 means unlimited (default: 0)", ) parser.add_argument( "--temp-dir", default=None, help="Temporary working directory; put this on fast/local NVMe when possible", ) parser.add_argument( "--recursive-input", action="store_true", help="Recursively discover top-level archive files below the input directory", ) parser.add_argument( "--log-file", default="extraction.log", help="Detailed log file path (default: extraction.log)", ) return parser.parse_args() def main() -> int: args = parse_args() try: domain = normalize_domain(args.domain) except ValueError as exc: print(f"Error: {exc}", file=sys.stderr) return 2 if args.workers < 1: print("Error: --workers must be >= 1", file=sys.stderr) return 2 if args.batch_size < 1: print("Error: --batch-size must be >= 1", file=sys.stderr) return 2 if args.max_depth < 0: print("Error: --max-depth must be >= 0", file=sys.stderr) return 2 if args.max_file_size < 0: print("Error: --max-file-size must be >= 0 (0 means unlimited)", file=sys.stderr) return 2 config = Config( archive_folder=args.archive_folder, output_file=args.output, target_domain=domain, output_format=args.format, max_workers=args.workers, db_batch_size=args.batch_size, resume=not args.no_resume, state_db=args.state_db, log_file=args.log_file, nested_archives=not args.no_nested, max_nested_depth=args.max_depth, max_file_size_mb=args.max_file_size, temp_dir=args.temp_dir, recursive_input=args.recursive_input, ) return run_extraction(config) if __name__ == "__main__": raise SystemExit(main())