evaluate_existing_results.py

# MIT License
#
# Copyright (c) 2025 Andy Ryan
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

#!/usr/bin/env python3
"""
Evaluate existing scan results without re-scanning
"""

import csv
import json
import argparse
import os

def evaluate_existing_results(results_file):
    """Evaluate accuracy from existing results file (supports JSONL, CSV, TSV, Parquet)."""
    
    # Detect file format
    file_ext = os.path.splitext(results_file)[1].lower()
    
    scan_results = []
    
    # Read based on file format
    if file_ext == '.jsonl' or file_ext == '.json':
        # Read JSONL format
        with open(results_file, 'r', encoding='utf-8') as f:
            for line in f:
                if line.strip():
                    try:
                        record = json.loads(line)
                        scan_results.append({
                            'prompt': record.get('prompt', ''),
                            'expected': str(record.get('expected', '')).lower(),
                            'outcome': str(record.get('outcome', '')).lower()
                        })
                    except json.JSONDecodeError:
                        continue
    
    elif file_ext == '.csv':
        # Read CSV format
        with open(results_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                scan_results.append({
                    'prompt': row.get('prompt', ''),
                    'expected': str(row.get('expected', '')).lower(),
                    'outcome': str(row.get('outcome', '')).lower()
                })
    
    elif file_ext == '.tsv':
        # Read TSV format
        with open(results_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f, delimiter='\t')
            for row in reader:
                scan_results.append({
                    'prompt': row.get('prompt', ''),
                    'expected': str(row.get('expected', '')).lower(),
                    'outcome': str(row.get('outcome', '')).lower()
                })
    
    elif file_ext == '.parquet':
        # Read Parquet format
        try:
            import pandas as pd
            df = pd.read_parquet(results_file)
            for _, row in df.iterrows():
                scan_results.append({
                    'prompt': str(row.get('prompt', '')),
                    'expected': str(row.get('expected', '')).lower(),
                    'outcome': str(row.get('outcome', '')).lower()
                })
        except ImportError:
            print("Error: pandas is required to read parquet files. Install with: pip install pandas pyarrow")
            return
    
    else:
        print(f"Unsupported file format: {file_ext}")
        print("Supported formats: .jsonl, .csv, .tsv, .parquet")
        return
    
    print(f"Loaded {len(scan_results)} results from {results_file}")
    
    if not scan_results:
        print("No results to evaluate.")
        return
    
    # Initialize counters
    true_positive = 0
    true_negative = 0
    false_positive = 0
    false_negative = 0
    
    for result in scan_results:
        expected = result["expected"]
        outcome = result["outcome"].lower()
        
        # Check conditions
        if expected == 'true' and outcome == 'flagged':
            true_positive += 1
        elif expected == 'false' and outcome == 'cleared':
            true_negative += 1
        elif expected == 'false' and outcome == 'flagged':
            false_positive += 1
        elif expected == 'true' and outcome == 'cleared':
            false_negative += 1
    
    # Calculate metrics
    total = true_positive + true_negative + false_positive + false_negative
    accuracy = (true_positive + true_negative) / total if total > 0 else 0
    
    # Calculate precision (true positives / (true positives + false positives))
    precision = true_positive / (true_positive + false_positive) if (true_positive + false_positive) > 0 else 0
    
    # Calculate recall (true positives / (true positives + false negatives))
    recall = true_positive / (true_positive + false_negative) if (true_positive + false_negative) > 0 else 0
    
    # Calculate F1 score (2 * precision * recall / (precision + recall))
    f1_score = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
    
    # Display results
    print(f"""
Evaluation Results:
------------------
True Positives: {true_positive}
True Negatives: {true_negative}
False Positives: {false_positive}
False Negatives: {false_negative}
Total Samples: {total}
Overall Accuracy: {accuracy:.2%}
Precision: {precision:.2%}
Recall: {recall:.2%}
F1 Score: {f1_score:.2%}
""")

def main():
    parser = argparse.ArgumentParser(description='Evaluate existing scan results')
    parser.add_argument('--input', '-i', type=str, required=True,
                      help='Input file containing scan results (supports .jsonl, .csv, .tsv, .parquet)')
    
    args = parser.parse_args()
    evaluate_existing_results(args.input)

if __name__ == "__main__":
    main()