log_classification/classify.py at main · labdhiongithub7/log_classification · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
from processor_regex import classify_with_regex
from processor_bert import classify_with_bert
from processor_llm import classify_with_llm

def classify(logs):
    """Classify logs and return labels only (for backward compatibility)"""
    labels = []
    for source, log_msg in logs:
        result = classify_log_detailed(source, log_msg)
        labels.append(result["target_label"])
    return labels


def classify_log_detailed(source, log_msg):
    """
    Classify a log message with detailed information including confidence and method.

    Returns:
        dict: {
            "target_label": str,
            "confidence": float,
            "method": str  # "regex", "bert", or "llm"
        }
    """
    if source == "LegacyCRM":
        label, confidence = classify_with_llm(log_msg)
        return {
            "target_label": label,
            "confidence": confidence,
            "method": "llm"
        }
    else:
        label, confidence = classify_with_regex(log_msg)
        if label:
            return {
                "target_label": label,
                "confidence": confidence,
                "method": "regex"
            }
        else:
            label, confidence = classify_with_bert(log_msg)
            return {
                "target_label": label,
                "confidence": confidence,
                "method": "bert"
            }


def classify_log(source, log_msg):
    """Legacy function for backward compatibility - returns label only"""
    result = classify_log_detailed(source, log_msg)
    return result["target_label"]

def classify_csv(input_file):
    import pandas as pd
    df = pd.read_csv(input_file)

    # Perform classification
    df["target_label"] = classify(list(zip(df["source"], df["log_message"])))

    # Save the modified file
    output_file = "output.csv"
    df.to_csv(output_file, index=False)

    return output_file

if __name__ == '__main__':
    classify_csv("test.csv")
    # logs = [
    #     ("ModernCRM", "IP 192.168.133.114 blocked due to potential attack"),
    #     ("BillingSystem", "User 12345 logged in."),
    #     ("AnalyticsEngine", "File data_6957.csv uploaded successfully by user User265."),
    #     ("AnalyticsEngine", "Backup completed successfully."),
    #     ("ModernHR", "GET /v2/54fadb412c4e40cdbaed9335e4c35a9e/servers/detail HTTP/1.1 RCODE  200 len: 1583 time: 0.1878400"),
    #     ("ModernHR", "Admin access escalation detected for user 9429"),
    #     ("LegacyCRM", "Case escalation for ticket ID 7324 failed because the assigned support agent is no longer active."),
    #     ("LegacyCRM", "Invoice generation process aborted for order ID 8910 due to invalid tax calculation module."),
    #     ("LegacyCRM", "The 'BulkEmailSender' feature is no longer supported. Use 'EmailCampaignManager' for improved functionality."),
    #     ("LegacyCRM", " The 'ReportGenerator' module will be retired in version 4.0. Please migrate to the 'AdvancedAnalyticsSuite' by Dec 2025")
    # ]
    # labels = classify(logs)
    #
    # for log, label in zip(logs, labels):
    #     print(log[0], "->", label)