parser.add_argument( "-r", "--replace", action="append", default=[], help="Replacements as key1=val1,key2=val2 (can specify multiple times)" )
@dataclass class TaskResult: """Result of a single task execution.""" task_id: int args: Tuple success: bool stdout: str stderr: str exit_code: int duration: float retries: int timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) parser
parser.add_argument( "-e", "--env", action="append", default=[], help="Set environment variable (KEY=VALUE)" ) help="Replacements as key1=val1
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 v = part.split('='
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")
parser.add_argument( "--retries", type=int, default=0, help="Number of retries on failure (default: 0)" )
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