"""Reporter for logging and summary generation.""" import json from datetime import datetime from pathlib import Path from pentestgpt.benchmark.models import BenchmarkResult, BenchmarkSummary class Reporter: """Handles logging and summary generation.""" def __init__(self, output_dir: Path): """ Initialize reporter. Args: output_dir: Base output directory """ # Create run-specific directory with timestamp timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") self.run_dir = output_dir / f"benchmark_run_{timestamp}" self.run_dir.mkdir(parents=True, exist_ok=True) # Create subdirectories self.benchmarks_dir = self.run_dir / "benchmarks" self.benchmarks_dir.mkdir(exist_ok=True) # Define log files self.detailed_log = self.run_dir / "detailed.log" self.summary_txt = self.run_dir / "summary.txt" self.summary_json = self.run_dir / "summary.json" print(f"\nLogs directory: {self.run_dir}\n") def get_benchmark_log_path(self, benchmark_id: str) -> Path: """ Get path for benchmark-specific log file. Args: benchmark_id: Benchmark identifier Returns: Path to log file """ return self.benchmarks_dir / f"{benchmark_id}.log" def log_start(self, benchmark_id: str, index: int, total: int): """ Log benchmark start. Args: benchmark_id: Benchmark identifier index: Current index (1-based) total: Total number of benchmarks """ timestamp = datetime.now().isoformat() message = f"[{timestamp}] START {benchmark_id}" # Write to detailed log with open(self.detailed_log, 'a') as f: f.write(message + "\n") # Print to console print(f"\n[{index}/{total}] {benchmark_id}") def log_result(self, result: BenchmarkResult): """ Log benchmark completion. Args: result: Benchmark result """ timestamp = datetime.now().isoformat() # Format status emoji based on objective completion state if result.status == "COMPLETED": status_emoji = "✓" elif result.status == "TIMEOUT": status_emoji = "⏱" else: # ERROR status_emoji = "✗" # Format duration minutes = int(result.duration_seconds // 60) seconds = int(result.duration_seconds % 60) duration_str = f"{minutes}m {seconds}s" # Log to detailed log message = ( f"[{timestamp}] {result.status} {result.benchmark_id} " f"({duration_str}, ${result.cost_usd:.4f}, {len(result.found_flags)} flags)" ) with open(self.detailed_log, 'a') as f: f.write(message + "\n") # Print to console console_msg = f" {status_emoji} {result.status} ({duration_str}, ${result.cost_usd:.2f})" # Show flags found if any if result.found_flags: console_msg += f"\n Flags found: {len(result.found_flags)}" if result.error_message: console_msg += f"\n Error: {result.error_message}" print(console_msg) def generate_summary(self, results: list[BenchmarkResult], start_time: datetime, end_time: datetime): """ Generate summary files. Args: results: List of benchmark results start_time: Run start time end_time: Run end time """ # Calculate statistics by objective completion state total = len(results) completed = sum(1 for r in results if r.status == "COMPLETED") timeout = sum(1 for r in results if r.status == "TIMEOUT") error = sum(1 for r in results if r.status == "ERROR") # Count how many found flags (useful for user to see) with_flags = sum(1 for r in results if len(r.found_flags) > 0) total_duration = sum(r.duration_seconds for r in results) avg_duration = total_duration / total if total > 0 else 0 total_cost = sum(r.cost_usd for r in results) avg_cost = total_cost / total if total > 0 else 0 # Completion rate (completed vs timed out/errored) completion_rate = (completed / total * 100) if total > 0 else 0 # Create summary object summary = BenchmarkSummary( total_benchmarks=total, completed=completed, timeout=timeout, error=error, with_flags=with_flags, total_duration_seconds=total_duration, average_duration_seconds=avg_duration, total_cost_usd=total_cost, average_cost_usd=avg_cost, completion_rate=completion_rate, results=results, start_time=start_time, end_time=end_time ) # Write JSON summary with open(self.summary_json, 'w') as f: json.dump(summary.to_dict(), f, indent=2) # Write text summary self._write_text_summary(summary) # Print to console self._print_console_summary(summary) def _write_text_summary(self, summary: BenchmarkSummary): """Write human-readable text summary.""" lines = [] lines.append("=" * 60) lines.append("PentestGPT Benchmark Results") lines.append("=" * 60) lines.append(f"Run Date: {summary.start_time.strftime('%Y-%m-%d %H:%M:%S')}") duration_h = summary.total_duration_seconds / 3600 duration_m = (summary.total_duration_seconds % 3600) / 60 lines.append(f"Duration: {int(duration_h)}h {int(duration_m)}m") lines.append("") lines.append(f"Total Benchmarks: {summary.total_benchmarks}") lines.append(f"Completed: {summary.completed} ({summary.completion_rate:.1f}%)") lines.append(f"With Flags: {summary.with_flags}") lines.append(f"Timeout: {summary.timeout}") lines.append(f"Errors: {summary.error}") lines.append("") lines.append("Cost Analysis:") lines.append(f" Total Cost: ${summary.total_cost_usd:.2f}") lines.append(f" Average Cost: ${summary.average_cost_usd:.2f}") lines.append("") lines.append("Time Analysis:") avg_m = int(summary.average_duration_seconds // 60) avg_s = int(summary.average_duration_seconds % 60) lines.append(f" Average Duration: {avg_m}m {avg_s}s") lines.append("") # Group results by status completed_results = [r for r in summary.results if r.status == "COMPLETED"] timeout_results = [r for r in summary.results if r.status == "TIMEOUT"] error_results = [r for r in summary.results if r.status == "ERROR"] if completed_results: lines.append("Completed Benchmarks:") for r in completed_results: m = int(r.duration_seconds // 60) s = int(r.duration_seconds % 60) flags_info = f"{len(r.found_flags)} flag(s)" if r.found_flags else "no flags" lines.append(f" ✓ {r.benchmark_id} ({m}m {s}s, ${r.cost_usd:.2f}, {flags_info})") lines.append("") if timeout_results: lines.append("Timeout Details:") for r in timeout_results: m = int(r.duration_seconds // 60) flags_info = f" - {len(r.found_flags)} flag(s) found" if r.found_flags else "" lines.append(f" ⏱ {r.benchmark_id}: Timeout after {m}m{flags_info}") lines.append("") if error_results: lines.append("Error Details:") for r in error_results: lines.append(f" ✗ {r.benchmark_id}: {r.error_message or 'Unknown error'}") lines.append("") lines.append("=" * 60) lines.append(f"Detailed logs: {self.run_dir}") lines.append("=" * 60) with open(self.summary_txt, 'w') as f: f.write("\n".join(lines)) def _print_console_summary(self, summary: BenchmarkSummary): """Print summary to console.""" print("\n" + "=" * 60) print("BENCHMARK RUN COMPLETE") print("=" * 60) print(f"Total: {summary.total_benchmarks}") print(f"Completed: {summary.completed} ({summary.completion_rate:.1f}%)") print(f"With Flags: {summary.with_flags}") print(f"Timeout: {summary.timeout}") print(f"Errors: {summary.error}") print(f"Total Cost: ${summary.total_cost_usd:.2f}") duration_h = summary.total_duration_seconds / 3600 print(f"Total Time: {duration_h:.2f}h") print(f"\nDetailed results: {self.run_dir}") print("=" * 60)