import argparse
from collections import Counter
from datetime import datetime
import json
from typing import Any, Dict, List, Tuple
def parse_arguments() -> argparse.Namespace:
"""명령행 인자를 파싱합니다."""
parser = argparse.ArgumentParser(description="웹 서버 접속 로그 분석 스크립트")
parser.add_argument(
"file_path",
nargs="?",
default="access.log",
help="분석할 로그 파일 경로 (기본값: access.log)",
)
return parser.parse_args()
def process_logs(
file_path: str,
) -> Tuple[Counter, Counter, int, int, int]:
"""로그 파일을 읽고 통계 처리를 위한 데이터를 집계합니다."""
hourly_counts: Counter = Counter()
ip_counts: Counter = Counter()
total_valid_requests = 0
error_requests = 0
skipped_lines = 0
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
parts = line.strip().split()
# 열 개수가 6개가 아닌 경우 제외
if len(parts) != 6:
skipped_lines += 1
continue
time_str, ip, _, _, status_str, resp_time_str = parts
# 상태코드와 응답시간이 정수가 아닌 경우 제외
try:
status_code = int(status_str)
int(resp_time_str)
except ValueError:
skipped_lines += 1
continue
# 시각(ISO 8601) 파싱 실패 시 제외
try:
dt = datetime.fromisoformat(time_str)
except ValueError:
skipped_lines += 1
continue
# 정상 데이터 집계
hour_key = f"{dt.hour:02d}"
hourly_counts[hour_key] += 1
ip_counts[ip] += 1
total_valid_requests += 1
if status_code >= 500:
error_requests += 1
return hourly_counts, ip_counts, total_valid_requests, error_requests, skipped_lines
def generate_report(
hourly_counts: Counter,
ip_counts: Counter,
total_valid_requests: int,
error_requests: int,
skipped_lines: int,
) -> Dict[str, Any]:
"""집계된 데이터를 요구사항에 맞는 딕셔너리 구조로 변환합니다."""
# 1. 00시부터 23시까지의 요청 수
hourly = {f"{h:02d}": hourly_counts.get(f"{h:02d}", 0) for h in range(24)}
# 2. 상위 5개 IP (요청 수 내림차순, IP 오름차순 정렬)
sorted_ips = sorted(ip_counts.items(), key=lambda item: (-item[1], item[0]))
top_ips: List[List[Any]] = [[ip, count] for ip, count in sorted_ips[:5]]
# 3. 상태코드 500 이상 비율(%)
if total_valid_requests > 0:
error_rate = round((error_requests / total_valid_requests) * 100, 1)
else:
error_rate = 0.0
return {
"hourly": hourly,
"top_ips": top_ips,
"error_rate": error_rate,
"skipped": skipped_lines,
}
def main() -> None:
"""메인 실행 함수입니다."""
args = parse_arguments()
hourly_counts, ip_counts, total_valid, error_count, skipped = process_logs(
args.file_path
)
result = generate_report(
hourly_counts, ip_counts, total_valid, error_count, skipped
)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
기본 설정으로 실행하려면 터미널에 python 스크립트이름.py를 입력하시면 됩니다. 특정 로그 파일을 분석하려면 python 스크립트이름.py 파일경로와 같이 파일 경로를 인자로 전달하여 실행해 주시기 바랍니다.
