Mpall [better] «PLUS — 2027»

def _print_summary(self, total_tasks: int, total_duration: float): """Print execution summary.""" succeeded = sum(1 for r in self.results if r.success) failed = total_tasks - succeeded self.logger.info("=" * 50) self.logger.info("SUMMARY") self.logger.info(f"Total tasks: total_tasks") self.logger.info(f"Succeeded: succeeded") self.logger.info(f"Failed: failed") self.logger.info(f"Total duration: total_duration:.2fs") if failed > 0: self.logger.info("\nFailed tasks:") for r in self.results: if not r.success: self.logger.info(f" Task r.task_id: r.stderr[:200]")

@staticmethod def run(cmd: List[str], timeout: int, env: Optional[Dict] = None) -> Tuple[int, str, str]: """Run command, return (exit_code, stdout, stderr).""" try: result = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, env=env if env else os.environ.copy(), text=True ) return result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired: return -1, "", f"Command timed out after timeout seconds" except Exception as e: return -2, "", str(e) def worker(task_id: int, args_template: List[str], replacements: Dict[str, str], timeout: int, retries: int, env: Dict) -> TaskResult: """ Worker function executed in a separate process. Builds command by replacing placeholders and runs it. """ start_time = time.time()

def _save_results_json(self): """Save detailed results to JSON file.""" data = "timestamp": datetime.now().isoformat(), "command": self.args.command, "total_tasks": len(self.results), "successful": sum(1 for r in self.results if r.success), "failed": sum(1 for r in self.results if not r.success), "results": [ "task_id": r.task_id, "args": dict(r.args), "success": r.success, "stdout": r.stdout, "stderr": r.stderr, "exit_code": r.exit_code, "duration": r.duration, "retries": r.retries for r in self.results ] with open(self.args.output_json, 'w') as f: json.dump(data, f, indent=2) self.logger.info(f"Results saved to self.args.output_json") env: Optional[Dict] = None) -&gt

parser.add_argument( "--retries", type=int, default=0, help="Number of retries on failure (default: 0)" )

parser.add_argument( "-o", "--output-json", help="Save detailed results to JSON file" ) str]: """Run command

# Validate arguments if not args.replace and not args.replace_file: parser.error("Either --replace or --replace-file must be provided")

parser.add_argument( "-r", "--replace", action="append", default=[], help="Replacements as key1=val1,key2=val2 (can specify multiple times)" ) stderr).""" try: result = subprocess.run( cmd

# All retries failed return TaskResult( task_id=task_id, args=tuple(replacements.items()), success=False, stdout="", stderr=stderr.strip() if 'stderr' in locals() else "Unknown error", exit_code=exit_code if 'exit_code' in locals() else -1, duration=time.time() - start_time, retries=retries ) class Mpall: """Main application class."""