28. 미니 프로젝트: 로그 분석기

2025-12-31 09:15:23 INFO 서버 시작
2025-12-31 09:16:01 INFO 사용자 로그인 user=alice
2025-12-31 09:17:45 WARNING 응답 지연 endpoint=/api/data
2025-12-31 09:18:30 ERROR DB 연결 실패 host=db01
2025-12-31 09:20:55 ERROR 파일 없음 path=/tmp/x.txt
2025-12-31 10:01:15 CRITICAL 시스템 다운

# log_analyzer.py
import re
from dataclasses import dataclass
from datetime import datetime

# 로그 형식을 명명 그룹으로 표현
LOG_PATTERN = re.compile(
    r"(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+"
    r"(?P<level>DEBUG|INFO|WARNING|ERROR|CRITICAL)\s+"
    r"(?P<message>.*)"
)

@dataclass
class LogEntry:
    timestamp: datetime
    level: str
    message: str

def parse_line(line: str) -> LogEntry | None:
    m = LOG_PATTERN.match(line.strip())
    if not m:
        return None                         # 형식 안 맞으면 None
    ts = datetime.strptime(m.group("timestamp"), "%Y-%m-%d %H:%M:%S")
    return LogEntry(ts, m.group("level"), m.group("message"))

print(parse_line("2025-01-01 12:00:00 INFO 정상"))   # LogEntry(...)
print(parse_line("형식 안 맞는 줄"))                   # None

def parse_file(path: str) -> list[LogEntry]:
    entries = []
    with open(path, encoding="utf-8") as f:    # 16장 with
        for line in f:                          # 줄 단위 순회
            entry = parse_line(line)
            if entry:                           # None이 아니면 (파싱 성공)
                entries.append(entry)
    return entries

entries = parse_file("sample.log")
print(len(entries))     # 10

from collections import Counter

def level_counts(entries: list[LogEntry]) -> dict:
    return dict(Counter(e.level for e in entries))     # 제너레이터 표현식

print(level_counts(entries))
# {'INFO': 4, 'WARNING': 2, 'ERROR': 3, 'CRITICAL': 1}

def errors_only(entries: list[LogEntry]) -> list[LogEntry]:
    return [e for e in entries if e.level in ("ERROR", "CRITICAL")]

for e in errors_only(entries):
    print(f"[{e.level}] {e.message}")
# [ERROR] DB 연결 실패 host=db01
# [ERROR] 파일 없음 path=/tmp/x.txt
# [ERROR] DB 연결 실패 host=db02
# [CRITICAL] 시스템 다운

def time_span(entries: list[LogEntry]):
    if not entries:
        return None
    times = [e.timestamp for e in entries]
    return min(times), max(times)

start, end = time_span(entries)
print(f"기간: {start} ~ {end}")
print(f"지속: {end - start}")           # timedelta (23장)
# 기간: 2025-12-31 09:15:23 ~ 2025-12-31 10:01:15
# 지속: 0:45:52

from collections import defaultdict

def by_hour(entries: list[LogEntry]) -> dict:
    counts = defaultdict(int)               # 없는 키는 0으로 시작
    for e in entries:
        counts[e.timestamp.hour] += 1
    return dict(counts)

print(by_hour(entries))     # {9: 9, 10: 1}

def print_report(entries: list[LogEntry]) -> None:
    counts = level_counts(entries)
    total = sum(counts.values())
    print(f"총 {total}건의 로그")
    print("-" * 30)
    for level, cnt in sorted(counts.items()):
        pct = cnt / total * 100
        bar = "#" * int(pct / 5)
        print(f"  {level:8} {cnt:2}건 ({pct:4.1f}%) {bar}")

print_report(entries)

총 10건의 로그
------------------------------
  CRITICAL  1건 (10.0%) ##
  ERROR     3건 (30.0%) ######
  INFO      4건 (40.0%) ########
  WARNING   2건 (20.0%) ####

from collections import Counter
import re

# 확장 2: 가장 흔한 에러의 첫 단어
def top_error_keyword(entries):
    errors = [e for e in entries if e.level in ("ERROR", "CRITICAL")]
    words = Counter(e.message.split()[0] for e in errors)
    return words.most_common(1)      # [('DB', 2)]

# 확장 4: 사용자별 활동 집계
def user_activity(entries):
    counts = Counter()
    for e in entries:
        m = re.search(r"user=(\w+)", e.message)
        if m:
            counts[m.group(1)] += 1
    return dict(counts)              # {'alice': 2, 'bob': 1}