85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""
|
|
API 에러 로깅 유틸리티 (JSONL 기록 + 로테이션 + 클라이언트 IP 추출)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import json
|
|
import re
|
|
from typing import Dict, Any
|
|
from fastapi import Request
|
|
|
|
|
|
LOG_DIR = "logs"
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
|
|
API_ERROR_LOG_PATH = os.path.join(LOG_DIR, "api_errors.jsonl")
|
|
API_ERROR_MAX_BYTES = 10 * 1024 * 1024 # 10MB
|
|
API_ERROR_BACKUP_COUNT = 5
|
|
|
|
|
|
def _rotate_if_needed() -> None:
|
|
try:
|
|
if os.path.exists(API_ERROR_LOG_PATH) and os.path.getsize(API_ERROR_LOG_PATH) >= API_ERROR_MAX_BYTES:
|
|
ts = time.strftime("%Y%m%d-%H%M%S")
|
|
rotated_path = os.path.join(LOG_DIR, f"api_errors_{ts}.jsonl")
|
|
os.replace(API_ERROR_LOG_PATH, rotated_path)
|
|
|
|
rotated = [
|
|
os.path.join(LOG_DIR, f) for f in os.listdir(LOG_DIR)
|
|
if f.startswith("api_errors_") and f.endswith(".jsonl")
|
|
]
|
|
rotated.sort(key=lambda p: os.path.getmtime(p), reverse=True)
|
|
for old in rotated[API_ERROR_BACKUP_COUNT:]:
|
|
try:
|
|
os.remove(old)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
# 로테이션 실패는 치명적이지 않으므로 무시
|
|
pass
|
|
|
|
|
|
def append_api_error_log(record: Dict[str, Any]) -> None:
|
|
try:
|
|
_rotate_if_needed()
|
|
with open(API_ERROR_LOG_PATH, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def extract_client_ip(request: Request) -> str:
|
|
try:
|
|
xff = request.headers.get("x-forwarded-for") or request.headers.get("X-Forwarded-For")
|
|
if xff:
|
|
first_ip = xff.split(",")[0].strip()
|
|
if first_ip:
|
|
return first_ip
|
|
xri = request.headers.get("x-real-ip") or request.headers.get("X-Real-IP")
|
|
if xri:
|
|
return xri.strip()
|
|
fwd = request.headers.get("forwarded") or request.headers.get("Forwarded")
|
|
if fwd:
|
|
m = re.search(r"for=([^;,\s]+)", fwd)
|
|
if m:
|
|
return m.group(1).strip('"')
|
|
if request.client and request.client.host:
|
|
return request.client.host
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def get_content_length(request: Request) -> int:
|
|
try:
|
|
v = request.headers.get("content-length") or request.headers.get("Content-Length")
|
|
if v is None:
|
|
return 0
|
|
return int(v)
|
|
except Exception:
|
|
return 0
|
|
|
|
|