log_grouping_large.py

# log_grouping_large.py
import re
import csv
import argparse
from collections import Counter, defaultdict

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.metrics.pairwise import cosine_similarity


# -----------------------------
# 1. Select only enriched app lines
# -----------------------------
def is_application_log(line: str) -> bool:
    """
    Keeps only the enriched app/runtime lines, e.g.
    [2026-04-10 08:14:21.123] ERROR AuthService Login failed ...
    Excludes Apache access lines.
    """
    return bool(re.match(
        r"^\[\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d{3})?\]\s+(ERROR|WARN|CRITICAL)\s+\S+",
        line.strip()
    ))


def load_logs(path: str, limit: int = None):
    logs = []
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            if is_application_log(line):
                logs.append(line)
                if limit and len(logs) >= limit:
                    break
    return logs


# -----------------------------
# 2. Preprocess
# -----------------------------
def preprocess_log(log: str) -> str:
    text = log.lower()

    # Remove timestamp
    text = re.sub(r"\[\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d{3})?\]", " ", text)

    # Remove level
    text = re.sub(r"\b(error|warn|critical|info|debug)\b", " ", text)

    # Remove IPs
    text = re.sub(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", " ", text)

    # Remove trace/request/session/user IDs and generic numbers
    text = re.sub(r"\b(trace_id|request_id|session_id|user_id|id|uid)=[a-z0-9\-_]+\b", " ", text)
    text = re.sub(r"\b[0-9]+\b", " ", text)

    # Normalize durations and paths a bit
    text = re.sub(r"\b\d+ms\b", " timeoutms ", text)
    text = re.sub(r"/api/[a-z0-9_\-\/]+", " api_path ", text)

    # Remove cause= value noise but keep the word cause
    text = re.sub(r"cause=[^\s]+", " cause ", text)
    text = re.sub(r"downstream=[^\s]+", " downstream ", text)

    # Remove punctuation
    text = re.sub(r"[^a-z\s]", " ", text)

    # Collapse whitespace
    return re.sub(r"\s+", " ", text).strip()


# -----------------------------
# 3. Light parsing / normalization
# -----------------------------
def parse_log(clean_log: str) -> str:
    text = clean_log

    replacements = {
        r"\bauthservice\b": "authservice",
        r"\buserservice\b": "userservice",
        r"\bpaymentservice\b": "paymentservice",
        r"\binventoryservice\b": "inventoryservice",
        r"\bdatabasepool\b": "databasepool",
        r"\bdb\b": "database",
        r"\blogin failed\b": "login failed",
        r"\binvalid credentials\b": "invalid password",
        r"\bpermission denied\b": "forbidden",
        r"\baccess denied\b": "forbidden",
        r"\bconnection lost\b": "database connection failure",
        r"\bconnection timeout\b": "database timeout",
        r"\bupstream timeout\b": "upstream service timeout",
        r"\bservice unavailable\b": "service unavailable",
        r"\bnull reference\b": "null reference",
        r"\bundefined\b": "null reference",
    }

    for pattern, replacement in replacements.items():
        text = re.sub(pattern, replacement, text)

    return re.sub(r"\s+", " ", text).strip()


# -----------------------------
# 4. Optional rough label assignment for evaluation
# -----------------------------
def assign_label(parsed_log: str) -> str:
    t = parsed_log

    if any(k in t for k in ["login failed", "invalid password", "authentication", "authservice"]):
        return "authentication"
    if any(k in t for k in ["forbidden", "unauthorized", "securityaudit", "permission"]):
        return "authorization"
    if any(k in t for k in ["not found", "missing resource", "resource unavailable"]):
        return "resource_not_found"
    if any(k in t for k in ["database timeout", "database connection failure", "query failed", "databasepool"]):
        return "database"
    if any(k in t for k in ["upstream service timeout", "service unavailable", "apigateway", "downstream"]):
        return "api_network"
    if any(k in t for k in ["null reference", "type error", "exception", "internal server"]):
        return "application_exception"

    return "other"


# -----------------------------
# 5. Threshold grouping with connected components
# -----------------------------
def cluster_logs(similarity_matrix, threshold=0.75):
    n = len(similarity_matrix)
    visited = [False] * n
    groups = []

    adjacency = [[] for _ in range(n)]
    for i in range(n):
        for j in range(i + 1, n):
            if similarity_matrix[i][j] >= threshold:
                adjacency[i].append(j)
                adjacency[j].append(i)

    for i in range(n):
        if visited[i]:
            continue

        stack = [i]
        visited[i] = True
        component = []

        while stack:
            node = stack.pop()
            component.append(node)
            for neighbor in adjacency[node]:
                if not visited[neighbor]:
                    visited[neighbor] = True
                    stack.append(neighbor)

        groups.append(sorted(component))

    return groups


# -----------------------------
# 6. Evaluation
# -----------------------------
def evaluate_groups(groups, labels):
    correctly_grouped = 0

    for g in groups:
        group_labels = [labels[i] for i in g]
        majority_label, majority_count = Counter(group_labels).most_common(1)[0]
        correctly_grouped += majority_count

    total_logs = len(labels)
    accuracy = correctly_grouped / total_logs if total_logs else 0
    effort_reduction = total_logs - len(groups)

    return accuracy, effort_reduction


# -----------------------------
# 7. Save outputs
# -----------------------------
def save_grouped_results(output_csv, raw_logs, clean_logs, parsed_logs, labels, groups):
    group_map = {}
    for gid, members in enumerate(groups, start=1):
        for idx in members:
            group_map[idx] = gid

    with open(output_csv, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "log_id", "group_id", "raw_message",
            "preprocessed_message", "parsed_message", "label"
        ])

        for i in range(len(raw_logs)):
            writer.writerow([
                f"L{i+1:05d}",
                group_map[i],
                raw_logs[i],
                clean_logs[i],
                parsed_logs[i],
                labels[i]
            ])


def print_group_samples(raw_logs, parsed_logs, labels, groups, max_groups=10, max_items=5):
    print("\n=== GROUP SAMPLES ===")
    for gid, members in enumerate(groups[:max_groups], start=1):
        group_labels = [labels[i] for i in members]
        majority = Counter(group_labels).most_common(1)[0][0]
        print(f"\nGroup {gid} | size={len(members)} | majority_label={majority}")
        for idx in members[:max_items]:
            print(f"  - {raw_logs[idx]}")
            print(f"    Parsed: {parsed_logs[idx]}")


# -----------------------------
# 8. Threshold sweep
# -----------------------------
def threshold_sweep(parsed_logs, labels, thresholds):
    vectorizer = CountVectorizer()
    vectors = vectorizer.fit_transform(parsed_logs)
    similarity_matrix = cosine_similarity(vectors)

    print("\n=== THRESHOLD SWEEP ===")
    print("Threshold | Groups | Accuracy | Effort Reduction")
    print("-" * 50)

    results = []
    for t in thresholds:
        groups = cluster_logs(similarity_matrix, threshold=t)
        accuracy, effort = evaluate_groups(groups, labels)
        print(f"{t:8.2f} | {len(groups):6d} | {accuracy:8.4f} | {effort:16d}")
        results.append((t, len(groups), accuracy, effort))

    return results


# -----------------------------
# 9. Main
# -----------------------------
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True, help="Path to enriched log file")
    parser.add_argument("--threshold", type=float, default=0.75, help="Cosine similarity threshold")
    parser.add_argument("--limit", type=int, default=5000, help="Max number of app logs to process")
    parser.add_argument("--output", default="grouped_logs.csv", help="CSV output file")
    parser.add_argument("--sweep", action="store_true", help="Run threshold sweep as well")
    args = parser.parse_args()

    raw_logs = load_logs(args.input, limit=args.limit)

    if not raw_logs:
        print("No application log lines found. Check the log format.")
        return

    clean_logs = [preprocess_log(log) for log in raw_logs]
    parsed_logs = [parse_log(log) for log in clean_logs]
    labels = [assign_label(log) for log in parsed_logs]

    vectorizer = CountVectorizer()
    vectors = vectorizer.fit_transform(parsed_logs)
    similarity_matrix = cosine_similarity(vectors)

    groups = cluster_logs(similarity_matrix, threshold=args.threshold)
    accuracy, effort_reduction = evaluate_groups(groups, labels)

    print("=== RESULTS ===")
    print(f"Total application logs processed : {len(raw_logs)}")
    print(f"Threshold                        : {args.threshold}")
    print(f"Number of groups                 : {len(groups)}")
    print(f"Grouping accuracy                : {accuracy:.4f}")
    print(f"Inspection effort reduction      : {effort_reduction}")

    label_counts = Counter(labels)
    print("\n=== LABEL DISTRIBUTION ===")
    for label, count in label_counts.items():
        print(f"{label:24s}: {count}")

    save_grouped_results(args.output, raw_logs, clean_logs, parsed_logs, labels, groups)
    print(f"\nSaved grouped output to: {args.output}")

    print_group_samples(raw_logs, parsed_logs, labels, groups, max_groups=10, max_items=3)

    if args.sweep:
        threshold_sweep(parsed_logs, labels, thresholds=[0.60, 0.65, 0.70, 0.75, 0.80])


if __name__ == "__main__":
    main()