result_processor.py

import re
import csv
import json
import io
from datetime import datetime
from typing import List, Optional, Tuple

from entropy import score_token_entropy


def extract_matches(snippet: str, pattern: str) -> list:
    """Extract all matches of the pattern in the snippet."""
    try:
        regex = re.compile(pattern)
        matches = regex.findall(snippet)
        # Flatten tuples from regex groups
        flat_matches = []
        for m in matches:
            if isinstance(m, tuple):
                # Take the first non-empty group
                for group in m:
                    if group:
                        flat_matches.append(group)
                        break
            else:
                flat_matches.append(m)
        return flat_matches
    except re.error:
        return []


def process_results(results: list, pattern: str, enable_entropy: bool = True) -> list:
    """Process search results and extract matches with optional entropy scoring."""
    processed = []
    total_matches = 0
    total_unique_matches = 0
    all_unique_tokens = set()

    for item in results:
        text_matches = item.get("text_matches", [])
        collected = []
        fragments = []
        for tm in text_matches:
            fragment = tm.get("fragment", "")
            fragments.append(fragment)
            matches = extract_matches(fragment, pattern)
            if matches:
                total_matches += len(matches)
                collected.extend(matches)

        if collected:
            # Deduplicate tokens for this file
            unique_tokens = list(set(collected))
            total_unique_matches += len(unique_tokens)
            all_unique_tokens.update(unique_tokens)

            # Score each token with entropy if enabled
            token_details = []
            for token in unique_tokens:
                detail = {"value": token}
                if enable_entropy:
                    detail["entropy"] = score_token_entropy(token)
                token_details.append(detail)

            # Try to get the most relevant date from multiple possible fields
            last_modified = None
            try:
                if item.get("repository", {}).get("pushed_at"):
                    last_modified = item["repository"]["pushed_at"]
                elif item.get("repository", {}).get("updated_at"):
                    last_modified = item["repository"]["updated_at"]
                elif item.get("repository", {}).get("created_at"):
                    last_modified = item["repository"]["created_at"]

                if last_modified:
                    last_modified = datetime.strptime(
                        last_modified,
                        "%Y-%m-%dT%H:%M:%SZ"
                    ).strftime("%Y-%m-%d %H:%M:%S UTC")
            except (ValueError, KeyError):
                pass

            result_info = {
                "repository": item.get("repository", {}).get("full_name"),
                "file_path": item.get("path"),
                "html_url": item.get("html_url"),
                "last_modified": last_modified or "N/A",
                "found_tokens": unique_tokens,
                "token_details": token_details,
                "total_matches_in_file": len(collected),
                "unique_matches_in_file": len(unique_tokens),
                "found_date": datetime.now().strftime("%Y-%m-%d"),
                "fragments": fragments,
            }
            processed.append(result_info)

    # Add match statistics to the first result if we have any results
    if processed:
        processed[0]["match_statistics"] = {
            "total_files_with_matches": len(processed),
            "total_matches_found": total_matches,
            "total_unique_matches_in_files": total_unique_matches,
            "total_unique_tokens_overall": len(all_unique_tokens),
        }

    return processed


def sanitize_filename(filename: str) -> str:
    """Replace characters that are invalid in Windows/Unix filenames."""
    invalid_chars = [':', '/', '\\', '?', '*', '"', '<', '>', '|']
    for char in invalid_chars:
        filename = filename.replace(char, '-')
    return filename


def save_results(results: list, pattern: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
    """Save results to JSON files and return file paths and error status."""
    # Use safe timestamp format (no colons — compatible with Windows)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    formatted_date = datetime.now().strftime("%Y-%m-%d")
    match_stats = results[0].get("match_statistics", {}) if results else {}

    all_tokens = []
    for result in results:
        all_tokens.extend(result["found_tokens"])
    all_tokens = list(set(all_tokens))

    tokens_file = f"tokens_results_{timestamp}.json"
    detailed_file = f"detailed_results_{timestamp}.json"

    error_occurred = False
    error_message = ""

    try:
        with open(tokens_file, "w", encoding='utf-8') as f:
            json.dump({
                "tokens": all_tokens,
                "total_unique_tokens": len(all_tokens),
                "pattern": pattern,
                "scan_date": formatted_date,
                "scan_time": datetime.now().strftime("%H:%M:%S"),
                "match_statistics": match_stats,
            }, f, indent=4)
    except (IOError, OSError) as e:
        error_occurred = True
        error_message += f"Error saving tokens file ({tokens_file}): {str(e)}\n"

    try:
        with open(detailed_file, "w", encoding='utf-8') as f:
            json.dump({
                "pattern": pattern,
                "scan_date": formatted_date,
                "scan_time": datetime.now().strftime("%H:%M:%S"),
                "match_statistics": match_stats,
                "results": results,
            }, f, indent=4)
    except (IOError, OSError) as e:
        error_occurred = True
        error_message += f"Error saving detailed results file ({detailed_file}): {str(e)}"

    if error_occurred:
        return None, None, error_message
    else:
        return tokens_file, detailed_file, None


def save_results_csv(results: list) -> Tuple[Optional[str], Optional[str]]:
    """Save results to CSV format. Returns (csv_string, error_message)."""
    if not results:
        return None, "No results to export"

    output = io.StringIO()
    writer = csv.writer(output)

    # Header row
    writer.writerow([
        "Repository",
        "File Path",
        "URL",
        "Last Modified",
        "Token",
        "Entropy",
        "Entropy Risk",
        "Found Date",
    ])

    for result in results:
        for i, token in enumerate(result.get("found_tokens", [])):
            # Get entropy info if available
            token_details = result.get("token_details", [])
            entropy_info = {}
            if i < len(token_details):
                entropy_info = token_details[i].get("entropy", {})

            writer.writerow([
                result.get("repository", "N/A"),
                result.get("file_path", "N/A"),
                result.get("html_url", "N/A"),
                result.get("last_modified", "N/A"),
                token,
                entropy_info.get("entropy", "N/A"),
                entropy_info.get("risk", "N/A"),
                result.get("found_date", "N/A"),
            ])

    csv_string = output.getvalue()
    output.close()
    return csv_string, None