Mpall Direct

Save as `mpall.py`, make executable, and test:

```bash chmod +x mpall.py

def run(self) -> int: """Main execution entry point.""" # Parse replacements replacements_list = self.parse_replacements() if not replacements_list: self.logger.error("No replacements provided") return 1 total_tasks = len(replacements_list) self.logger.info(f"Starting total_tasks tasks with self.args.workers workers") # Prepare environment env = os.environ.copy() if self.args.env: for env_var in self.args.env: if '=' in env_var: k, v = env_var.split('=', 1) env[k] = v # Execute tasks in parallel start_time = time.time() with ProcessPoolExecutor(max_workers=self.args.workers) as executor: futures = {} for idx, replacements in enumerate(replacements_list): if self.cancel: break future = executor.submit( worker, idx, self.args.command, replacements, self.args.timeout, self.args.retries, env ) futures[future] = idx # Collect results as they complete for future in as_completed(futures): if self.cancel: break result = future.result() self.results.append(result) self._log_result(result) total_duration = time.time() - start_time # Summary self._print_summary(total_tasks, total_duration) # Save results if requested if self.args.output_json: self._save_results_json() if self.args.output_summary: self._save_summary_text() # Return exit code (0 if all succeeded) return 0 if all(r.success for r in self.results) else 1 Save as `mpall

# Build command with replacements cmd = [] for arg in args_template: replaced = arg for key, value in replacements.items(): replaced = replaced.replace(f"key", str(value)) cmd.append(replaced)

parser.add_argument( "-t", "--timeout", type=int, default=60, help="Timeout per command in seconds (default: 60)" ) Save as `mpall.py`

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)

def _log_result(self, result: TaskResult): """Log individual task result.""" status = "✓" if result.success else "✗" self.logger.info( f"status Task result.task_id - " f"Duration: result.duration:.2fs - " f"Retries: result.retries" ) if self.args.verbose: if result.stdout: self.logger.debug(f"Task result.task_id STDOUT:\nresult.stdout") if result.stderr: self.logger.debug(f"Task result.task_id STDERR:\nresult.stderr") v = env_var.split('='

parser.add_argument( "-o", "--output-json", help="Save detailed results to JSON file" )