Mpall Upd May 2026
parser.add_argument( "-r", "--replace", action="append", default=[], help="Replacements as key1=val1,key2=val2 (can specify multiple times)" )
class Logger: """Unified logging handler with file and console output.""" def (self, log_file: Optional[str] = None, verbose: bool = False): self.logger = logging.getLogger("mpall") self.logger.setLevel(logging.DEBUG if verbose else logging.INFO) parser
Save as `mpall.py`, make executable, and test: help="Replacements as key1=val1
import argparse import logging import sys import time import subprocess import signal import threading from concurrent.futures import ProcessPoolExecutor, as_completed from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime import json import os log_file: Optional[str] = None
def parse_replacements(self) -> List[Dict[str, str]]: """Parse replacement arguments into list of parameter dictionaries.""" replacements = [] if self.args.replace_file: with open(self.args.replace_file, 'r') as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue parts = line.split() if len(parts) < 2: self.logger.warning(f"Skipping invalid line: line") continue # Format: key1=val1 key2=val2 ... replacement = {} for part in parts: if '=' in part: k, v = part.split('=', 1) replacement[k] = v if replacement: replacements.append(replacement) elif self.args.replace: # Format: key1=val1,key2=val2 or multiple -r flags for rep in self.args.replace: rep_dict = {} for pair in rep.split(','): if '=' in pair: k, v = pair.split('=', 1) rep_dict[k] = v replacements.append(rep_dict) else: # Single run with no replacements replacements.append({}) return replacements
def _signal_handler(self, signum, frame): """Handle Ctrl+C gracefully.""" self.logger.warning("Interrupt received, finishing current tasks...") self.cancel = True