Python을 활용한 MySQL general log 파싱

개발업무 특성상 general log를 검색할 일이 자주 발생합니다.

적은양이 아니기에 notepad를 통해서는 검색이 제한적이기에

python을 활용해 특정 단어가 포함된 쿼리만을 추출해 보았습니다.

import os
import chardet
import pandas as pd
import numpy as np
from queue import Queue
from datetime import datetime

"""
입력된 값이 날짜형 값인지 검사
"""
def f_isDateTimeCheck(_inWord):
    try:
        dt_date = datetime.strptime(_inWord, '%Y-%m-%d')
    except Exception as ex:
        # print(ex)  # 여기 오류는 날짜 형식을 검증하기위한 부분이라 오류 로그를 찍으면 안됨
        return False 
    return True

"""
입려된 단어들이 본 Line에 모두 포함되는지 여부
"""
def f_QuerToStringAndFind(_line_queue, _findWordList):
    lineStr = ""
    findCnt = 0
    # 하나의 문자열로 결합
    while _line_queue.qsize() > 0:
        lineStr = lineStr + _line_queue.get()
    for str in _findWordList:   # 검색어 목록
        if str in lineStr:
            findCnt = findCnt + 1
    
    # print(findCnt, len(_findWordList), findCnt != len(_findWordList), lineStr)
    if findCnt != len(_findWordList):
        return "" 
    return lineStr

searchWordsList = ['Execute', '34356', 'like'] # 하나의 로그에 모두 포함되어 있어야 하는 검색어 목록
_line_queue = Queue()
_line_number_queue = Queue()

file_path = r'd:\generallog\20240611_generallog.log'
# 검색한 내용을 기록할 파일 생성
new_file_name = "%s/FindGeneralLog_%s.SQL" % (os.path.dirname(file_path), datetime.now().strftime('%Y%m%d%H%M%S'))
find_word_new_file = open(new_file_name, "w")

try:
	# 하나의 컬럼으로 만들어 주기위해 열구분자(sep)를 임이로 설정하였습니다.
    df_log = pd.read_table(file_path, sep='✱%*', encoding='UTF-8', encoding_errors='ignore', header=None, engine='python')
    query_cnt = 1
    for row in df_log.itertuples():
        try:
            _line_original = str(row[1])
            _line_original = _line_original.strip()     # 원본 내용, 앞뒤공백 삭제
            if len(_line_original) == 0:
                continue     # 공백행 무시
            
            # 라인의 앞 10자리가 날짜 형식이면 로그의 시작
            if f_isDateTimeCheck(_line_original[0:10]):
                _result = f_QuerToStringAndFind(_line_queue, searchWordsList)
                if len(_result) > 3:
                    findTab = _result.find("\t")
                    dt = datetime.fromisoformat(_result[0:findTab])
                    formatted_time_str = dt.strftime("%Y-%m-%d %H:%M:%S") + f".{int(dt.microsecond):05d}"   # 밀리초까지 기록
                    find_word_new_file.write("%s %s \n" % (formatted_time_str, _result[findTab:].replace("\n", " ")) )   # 찾은 Line에서 쿼리만 추출하여 파일에 기록
                    
            _line_queue.put(_line_original + '\n')
            _line_number_queue.put(row[0])
        except Exception as ex:
            print(ex)
            continue
except Exception as ex:
    print(ex)

find_word_new_file.close()
print("검색 작업이 완료되었습니다.")