log_analyzer.py

'''
    log analyzer module, has Analyzer class.
    xInstantiated by file path of log file
'''
from pathlib import Path
import time
import util
from enum import Enum
from typing import List


class Priority(Enum):
    Verbose = 'V'
    Debug = 'D'
    Info = 'I'
    Warning = 'W'
    Error = 'E'
    Fatal = 'F'
    Silent = 'S'


class Log():

    '''
        Log class is instantiated by single logs generated by Logcat
        in `threadtime` format. It cleans the line, and assigns the
        proper values to the following properties:
        log_time: time
        log_pid: str
        log_tid: str
        log_priority: str
        log_tag: str
        log_message: str
        eg. "06-14 22:08:27.424  2987  3046 W System.err:
        at java.lang.Thread.run(Thread.java:761)"
        will be used to create:

        ```
        log_time = time.struct_time(tm_year=1900, tm_mon=6, tm_mday=14,
        tm_hour=22, tm_min=8, tm_sec=27, tm_wday=3,tm_yday=165, tm_isdst=-1)
        log_pid = "2987"
        log_tid = "3046"
        log_priority = "W"
        log_tag = "System.err"
        log_message = "at java.lang.Thread.run(Thread.java:761)"
        ```
    '''

    def __init__(self, log_line: str, mode: int=None)-> None:
        if len(log_line) < 1:
            raise ValueError("Error: Empty Line Detected from Log.")
        # "06-14 22:08:27.424  2987  3046 W System.err:    at java.lang.Thread.run(Thread.java:761) # noqa
        log_line_split = log_line.split(' ', maxsplit=8)
        current_line = log_line_split[0] + " " + log_line_split[1][:-4]
        # print(log_line[1][:-4])
        try:
            self.log_time = time.strptime(current_line, "%m-%d %H:%M:%S")

        except ValueError:
            self.log_time = None
            self.log_pid = None
            self.log_tid = None
            self.log_priority = None
            self.log_tag = None
            self.log_message = None
            return None
        if mode is 1:
            try:
                self.log_pid = log_line_split[3]
                self.log_tid = log_line_split[5]
                self.log_priority = log_line_split[6]
                self.log_tag = log_line_split[7]
                self.log_message = log_line_split[8].strip().strip(":").strip()
            except IndexError:
                self.log_time = None
                self.log_pid = None
                self.log_tid = None
                self.log_priority = None
                self.log_tag = None
                self.log_message = None
                return None
        if mode is 2:
            log_line_split = log_line.split(' ', maxsplit=6)
            self.log_pid = log_line_split[2]
            self.log_tid = log_line_split[3]
            self.log_priority = log_line_split[4]
            self.log_tag = log_line_split[5].strip('\t')
            self.log_message = log_line_split[6].strip().strip(":").strip()

    def __str__(self):
        return "{}, {}, {}, {}, {}, {}".format(self.log_time,
                                               self.log_pid,
                                               self.log_tid,
                                               self.log_priority,
                                               self.log_tag,
                                               self.log_message)

    __repr__ = __str__


class Analyzer():

    '''
        Analyzer class, instantiated by file path of log file
        when object is returned as string, it gives output in following format:

        ```
        return "Unique Warnings: {}\nUnique Errors: {}\
        \nTotal Warnings: \
        {}\nTotal Errors: {}\n".format(self.count_unique_warnings,
                                                         self.count_unique_errors,
                                                         self.count_warnings,
                                                         self.count_errors)
        ```
    '''

    def __init__(self, file_path):
        self.file_path = file_path
        if not util.check_file_directory_exists(file_path, False):
            raise ValueError("File path does not exist.")
        self.file_contents = Path(file_path).read_text().split('\n')
        self.unique_warnings = set()
        self.all_warnings = list()
        self.unique_errors = set()
        self.all_errors = list()
        self.unique_fatals = set()
        self.all_fatals = list()
        self.count_warnings = None
        self.count_errors = None
        self.count_fatals = None
        self.count_unique_warnings = None
        self.count_unique_errors = None
        self.count_unique_fatals = None
        self.logs = self.__logfile_to_logs(mode=1)
        self.__calculate_stats()
        if self.count_errors == 0 and\
            self.count_unique_errors == 0 and\
                self.count_unique_warnings == 0 and self.count_warnings == 0:
            print("something went wrong. recalculating")
            self.logs = self.__logfile_to_logs(mode=2)
            self.__calculate_stats()

    def return_unique_stats(self):
        '''
            Returns a tuple containing numbers of unique
            `(warnings, errors, fatals)`
        '''
        return (self.count_unique_warnings, self.count_unique_errors,
                self.count_unique_fatals)

    def __logfile_to_logs(self, mode: int=None) -> List[Log]:
        '''
        converts logfile entries to a `List[Log]`
        '''
        log_list = []
        for line in self.file_contents:
            if len(line) < 1:
                continue
            if mode is 1:
                log_list.append(Log(line, mode))
            else:
                log_list.append(Log(line, 2))
        return log_list

    def return_stats(self):
        '''
            Returns a tuple containing numbers of `(warnings, errors, fatals)`
        '''
        return (self.count_warnings, self.count_errors, self.count_fatals)

    def __calculate_stats(self):
        avc_counter = 0
        non_jni_counter = 0
        text_speech_counter = 0
        long_monitor_counter = 0
        ad_counter = 0
        # failed_load_counter = 0
        # time_out_ad = 0
        facebook_katana_counter = 0

        for log in self.logs:
            if log.log_message is None or log.log_pid is None:
                continue
            if log.log_message.startswith('at ') or\
                log.log_message.startswith('*') or\
                    log.log_message.endswith("more"):
                continue
            if "avc: denied" in log.log_message.lower():
                if avc_counter > 0:
                    continue
                avc_counter += 1
            if "long monitor contention" in log.log_message.lower():

                if long_monitor_counter > 0:
                    continue
                long_monitor_counter += 1
            if "remove non-jni" in log.log_message.lower():

                if non_jni_counter > 0:
                    continue
                non_jni_counter += 1
            if "texttospeech" in log.log_tag.lower():

                if text_speech_counter > 0:
                    continue
                text_speech_counter += 1
            # if "failed to load ad" in log.log_message.lower():
            #     failed_load_counter += 1
            #     if failed_load_counter > 0:
            #         continue
            # if "timed out waiting for ad response" in log.log_message.lower():# noqa
            #     time_out_ad += 1
            #     if time_out_ad > 0:
            #         continue

            if "Ads".lower() is log.log_tag.lower() or(
                    "MMSDK-PlayList".lower() in log.log_tag.lower()):
                if ad_counter > 0:
                    continue
                ad_counter += 1

            if "com.facebook.katana.provider.AttributionIdProvider".lower() \
                    in log.log_message.lower():
                if facebook_katana_counter > 0:
                    continue
                facebook_katana_counter += 1
            # print(log.log_message)

            if "attempted to finish an input event" in log.log_message.lower()\
                    or "bluetooth" in log.log_message.lower() \
                    or "dropping event" in log.log_message.lower() \
                    or "cancelling event" in log.log_message.lower() \
                    or "discarding hit" in log.log_message.lower():
                # print("dropping/cancelling event")
                continue
            # print(log.log_message)

            if log.log_priority is 'E':
                self.unique_errors.add(log.log_message)
                self.all_errors.append(log.log_message)
            if log.log_priority is 'W':
                self.unique_warnings.add(log.log_message)
                self.all_warnings.append(log.log_message)
            if log.log_priority is 'F':
                self.unique_fatals.add(log.log_message)
                self.all_fatals.append(log.log_message)
        self.count_unique_errors = len(self.unique_errors)
        self.count_unique_warnings = len(self.unique_warnings)
        self.count_unique_fatals = len(self.unique_fatals)
        self.count_errors = len(self.all_errors)
        self.count_warnings = len(self.all_warnings)
        self.count_fatals = len(self.all_fatals)
        # for line in self.file_contents:
        #     if len(line) < 1 or ':' not in line:
        #         continue
        #     line = str(line)
        #     message = line.split(':', maxsplit=1)[1].strip()
        #     if message.startswith('at '):
        #         continue
        #     if line[0] == 'W':
        #         self.unique_warnings.add(message)
        #         self.all_warnings.append(message)
        #     if line[0] == 'E':
        #         self.unique_errors.add(message)
        #         self.all_errors.append(message)
        #     self.count_unique_errors = len(self.unique_errors)
        #     self.count_unique_warnings = len(self.unique_warnings)
        #     self.count_errors = len(self.all_errors)
        #     self.count_warnings = len(self.all_warnings)

    def return_file_contents(self) -> List:
        '''
        returns the contents of the log file in List
        '''
        return self.file_contents

    # def print_file_contents_experimental(self):
    #     for line in self.file_contents:
    #         Log(line)

    def __str__(self):
        return "Total Warnings: {}\nUnique Warnings: {}\
        \nTotal Errors: {}\nUnique Errors: {}\
        \nTotal Fatals: {}\nUnique Fatals: {}".format(self.count_warnings,
                                                      self.count_unique_warnings,  # noqa
                                                      self.count_errors,
                                                      self.count_unique_errors,
                                                      self.count_fatals,
                                                      self.count_unique_fatals)