app/middlewares/logging_middleware.py

import contextvars
import http
import logging
import re
import sys
import time
import uuid
from copy import copy

# config 서브 모듈을 명시적으로 임포트하거나 직접 가져옵니다.
from logging.handlers import TimedRotatingFileHandler

# config 모듈을 명시적으로 불러옵니다.
from starlette.responses import JSONResponse

import app
from app.colors import ANSI

# 라우터 내부나 예외 핸들러에서도 현재 요청의 Trace ID와 User ID를 공유할 수 있도록 컨텍스트 변수 선언
request_trace_id = contextvars.ContextVar("request_trace_id", default="-")
request_user_id = contextvars.ContextVar("request_user_id", default="Anonymous")

# 2. 메인 로거 초기화 및 재설정
logger = logging.getLogger("uvicorn_file_logger")
logger.setLevel(logging.INFO)
# 💡 중요: 부모 로거(Uvicorn 표준 스트림)로 로그가 흘러 넘쳐 중복 출력되는 현상을 완전 차단합니다.
logger.propagate = False

# 💡 [핵심 해결책] 리로드 시 기존 메모리에 남아있던 핸들러들을 완전히 싹 비워버립니다.
# 이렇게 해야 새로고침을 눌러도 무조건 딱 1줄씩만 깨끗하게 출력됩니다.
logger.handlers.clear()


# -------------------------------------------------------------
# [핸들러 A] 콘솔 전용 포맷터
# -------------------------------------------------------------
class ModernColorFormatter(logging.Formatter):
  # 유비콘 순정과 똑같은 레벨별 컬러 매핑
  LEVEL_COLORS = {
    "INFO": "\x1b[32m",  # 초록
    "WARNING": "\x1b[33m",  # 노랑
    "ERROR": "\x1b[31m",  # 빨강
  }
  COLOR_RESET = "\x1b[0m"

  def __init__(self):
    # 틀 양식에는 정렬 문법을 다 빼고, 아래에서 가공할 {levelprefix} 공간만 남겨둡니다.
    super().__init__(fmt="{levelprefix} {asctime} | {current_trace} {current_user} {message}", style="{")

  def format(self, record):
    # 1. 원본 레코드 오염 방지를 위해 유비콘 소스코드 기법대로 카피본 생성
    record_copy = copy(record)

    # 유비콘 가로채기 로그들의 문자열 포맷팅 아규먼트(% 변환) 처리 방어 코드
    if record_copy.args:
      record_copy.msg = record_copy.msg % record_copy.args
      record_copy.args = ()

    # 2. [핵심 공식 🎯] 질문하신 대로 '레벨네임 + :' 문자열을 먼저 합쳐서 만듭니다!
    # 예: "INFO" -> "INFO:" / "WARNING" -> "WARNING:"
    level_with_colon = f"{record_copy.levelname}:"

    # 3. 색상 코드를 앞뒤로 입힙니다.
    color = self.LEVEL_COLORS.get(record_copy.levelname, "")
    colored_level = f"{color}{level_with_colon}{self.COLOR_RESET}"

    # 4. 색상 특수문자가 섞이면 파이썬 너비 계산(<9)이 먹통이 되므로,
    # 글자 수 길이를 재서 남은 방 크기만큼 순수 스페이스 공백 문자열을 수동 생성합니다.
    # "INFO:"(5글자) -> 9-5 = 4칸 공백 / "WARNING:"(8글자) -> 9-8 = 1칸 공백
    spaces_needed = 9 - len(level_with_colon)
    pure_spaces = " " * spaces_needed

    current_trace = request_trace_id.get()
    current_user = request_user_id.get()

    # 5. 최종 유비콘 규격 프리픽스 패키지를 가상 변수 {levelprefix} 자리에 강제 탑재
    record_copy.__dict__["levelprefix"] = f"{colored_level}{pure_spaces}"
    record_copy.__dict__["current_trace"] = f"{ANSI.BOLD}{ANSI.MAGENTA}[{current_trace}]{ANSI.RESET}"
    record_copy.__dict__["current_user"] = f"[{current_user}]"

    # {ANSI.BOLD}{ANSI.MAGENTA}[{current_trace}]{ANSI.RESET} [{current_user}]

    # 6. 부모 포맷터 호출 (스택 트레이스가 발생해도 본문 밑에 줄바꿈으로 완벽 자동 결합됨)
    return super().format(record_copy)


console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(ModernColorFormatter())
logger.addHandler(console_handler)


# -------------------------------------------------------------
# [핸들러 B] 파일 저장 전용 포맷터
# -------------------------------------------------------------
# 2. 자식 클래스: 부모의 색상 칠하기 기능 결과를 받아 ANSI 코드만 청소하는 포맷터
class StripAnsiFormatter(ModernColorFormatter):  # 💡 상속 대상을 부모 클래스로 지정!
  # ANSI 코드를 잡아내는 정규식 패턴
  ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")

  def format(self, record: logging.LogRecord) -> str:
    # 🚀 핵심: 부모(ModernColorFormatter)가 색상을 다 입혀서 만든 글자를 가져옵니다.
    colored_message = super().format(record)

    # 가져온 색상 글자에서 ANSI 코드만 공백('')으로 치환하여 깨끗하게 리턴합니다!
    return self.ANSI_ESCAPE.sub("", colored_message)


file_handler = TimedRotatingFileHandler(
  filename=app.LOG_FILE_PATH,
  when="midnight",
  interval=1,
  backupCount=30,
  encoding="utf-8",
)

file_handler.setFormatter(StripAnsiFormatter())
logger.addHandler(file_handler)


class AdvancedLoggingMiddleware:
  def __init__(self, app):
    self.app = app

  async def __call__(self, scope, receive, send):
    # HTTP 요청이 아닌 세션(웹소켓 등)은 통과
    if scope["type"] != "http":
      await self.app(scope, receive, send)
      return

    start_time = time.perf_counter()

    # ----------------------------------------------------
    # 고도화 1: 고유 트레이스 ID(Trace ID) 발급 (앞 8자리만 슬라이싱)
    # ----------------------------------------------------
    # 1. 요청이 들어오면 무작위 고유 ID 생성
    trace_id = f"tx_{uuid.uuid4().hex[:8]}"

    # 2. 사물함에 📥 넣기 (이 순간부터 하위 모든 로직에서 자동 연동 시작)
    token = request_trace_id.set(trace_id)

    # ----------------------------------------------------
    # 고도화 2: 프록시(Nginx, ALB) 뒤에 숨은 클라이언트 실제 IP 추적
    # ----------------------------------------------------
    headers = dict(scope.get("headers", []))
    # HTTP 헤더 키는 바이트 타입이므로 디코딩 처리
    x_forwarded_for = headers.get(b"x-forwarded-for", b"").decode("utf-8")
    if x_forwarded_for:
      # 프록시 체인 중 첫 번째 IP가 실제 클라이언트 IP입니다.
      client_ip = x_forwarded_for.split(",")[0].strip()
    else:
      # 프록시가 없는 개발 환경인 경우 기본 주소 사용
      client_ip = scope.get("client", ["127.0.0.1"])[0]

    """
    # 2. 브라우저가 보낸 헤더에서 JWT 토큰을 꺼내서 검증합니다.
    token = request.headers.get("Authorization")
    if token and token.startswith("Bearer "):
        try:
            # 토큰을 해독해서 실제 회원 ID를 추출 (예시)
            payload = decode_jwt_token(token.split(" ")[1])
            user_id = payload.get("user_id") # "user_12345"

            # 🌟 [지금이 타이밍!] 회원 ID를 사물함에 집어넣습니다.
            user_token = request_user_id.set(user_id)
        except Exception:
            # 토큰이 만료되었거나 가짜라면 비회원 처리
            user_token = request_user_id.set("Anonymous")
    else:
        # 토큰이 아예 없는 비로그인 유저인 경우
        user_token = request_user_id.set("Anonymous")
    """

    # 기본 ASGI scope 데이터 추출
    method = scope.get("method", "UNKNOWN")
    path = scope.get("path", "")
    query_string = scope.get("query_string", b"").decode("utf-8")
    full_path = f"{path}?{query_string}" if query_string else path
    user_agent = headers.get(b"user-agent", b"UNKNOWN").decode("utf-8")

    # 1. request. 없이 인자로 받은 scope에서 직접 꺼냅니다.
    # scope.get('http_version')은 '1.1' 또는 '2.0' 같은 문자열을 반환합니다.
    http_version = f"HTTP/{scope.get('http_version', '1.1')}"

    # 1. 낚아챈 상태 코드를 담아둘 내 전용 주머니(리스트)를 준비합니다.
    #    (초기값은 기본 200으로 방을 만들어 둡니다)
    captured_status = [200]

    # 2. 통로를 지나가는 데이터를 감시할 도청기(래퍼 함수)를 정의합니다.
    async def send_wrapper(message):
      # 만약 응답이 시작되는 신호("http.response.start")가 통로를 지나가면!
      if message["type"] == "http.response.start":
        # 🎯 [핵심] 지나가는 데이터 주머니에서 'status' 숫자(200, 404 등)를
        # 쏙 빼내어 내 주머니(captured_status)의 0번째 방에 덮어씁니다!
        captured_status[0] = message["status"]

      # 도청한 뒤 신호가 끊기지 않도록 원래 목적지로 마저 던져줍니다.
      await send(message)

    try:
      # 실제 비즈니스 로직 실행
      # (라우터 실행 중에 토큰을 검증하여 request_user_id.set("회원ID")를 수행하면 동적 반영됩니다)
      # await self.app(scope, receive, send)
      # 3. [도청기 이식] 원래의 send 함수 자리에 내가 만든 send_wrapper를 끼워넣어 실행!
      await self.app(scope, receive, send_wrapper)

      # 4. [완벽 취득 ⭕]
      # 이제 내 주머니를 열어보면 가로챈 진짜 상태 코드(200, 404 등)가 들어있습니다!
      final_status_code = captured_status[0]

      # 정상 종료 시 응답 코드 가로채기는 ASGI 특성상 scope 상태나 별도 핸들러에서 인지하므로,
      # 미들웨어 정상 마감 로그 처리 (기본 200 가정 혹은 라우터 성공 로그 활용)
      duration_ms = (time.perf_counter() - start_time) * 1000
      self._write_log(
        method, full_path, http_version, final_status_code, duration_ms, client_ip, user_agent, logging.INFO
      )

    except Exception as exc:
      # ----------------------------------------------------
      # [포획 성공] 라우터 실행 중 문법 오류/오타가 나면 무조건 이리로 떨어집니다!
      # ----------------------------------------------------
      duration_ms = (time.perf_counter() - start_time) * 1000

      # 로그 기록 및 스택 트레이스 전송
      self._write_log(method, full_path, http_version, 500, duration_ms, client_ip, user_agent, logging.ERROR, exc=exc)

      # 강제 플러시 (VS Code Undo 버퍼 유실 방지 및 파일 즉시 기록)
      for handler in logger.handlers:
        if "FileHandler" in handler.__class__.__name__:
          handler.flush()

      # 안전하게 브레이크를 밟고 예쁜 500 JSON 응답 리턴
      response = JSONResponse(status_code=500, content={"success": False, "message": "Internal Server Error"})
      await response(scope, receive, send)

    finally:
      # 4. 사물함 비우기
      # 3. 💥 [핵심] 청소는 역순으로 깔끔하게 둘 다 실행!
      # 에러가 나든 정상 종료되든 사물함을 완벽히 비웁니다.
      # request_user_id.reset(user_token)
      request_trace_id.reset(token)

  def _write_log(self, method, path, http_version, status_code, duration, client_ip, user_agent, level, exc=None):

    try:
      status_phrase = http.HTTPStatus(status_code).phrase
    except ValueError:
      status_phrase = "Unknown"

    # 1. 상태 코드에 따라 ANSI 컬러와 로그 레벨을 동시에 결정
    if status_code >= 500:
      # 🔴 서버 에러: 굵은 빨간 배경 + 흰색 글자 / ERROR 레벨
      color = f"{ANSI.BOLD}{ANSI.BG_RED}{ANSI.WHITE}"
      # log_function = logger.error
    elif status_code >= 400:
      # 🟡 클라이언트 에러(404 등): 노란 배경 + 검은 글자 / WARNING 레벨
      color = f"{ANSI.BG_YELLOW}{ANSI.BLACK}"
      # log_function = logger.warning
    else:
      # 🟢 정상 요청 (200, 300번대): 연두 배경 + 검은 글자 / INFO 레벨
      color = f"{ANSI.BOLD}{ANSI.BG_BRIGHT_GREEN}{ANSI.BLACK}"
      # log_function = logger.info

    # 질문자님이 올려주신 환상적인 포맷 원형을 뼈대로 삼아 조립합니다.
    # 포맷터가 유비콘 초기화 공작을 당하더라도 이 원형을 복구해 냅니다.
    msg_body = f'{client_ip} - "{ANSI.BOLD}{ANSI.BLUE}{method}{ANSI.RESET} {path} {http_version}" {color}{status_code} {status_phrase}{ANSI.RESET} {ANSI.BOLD}{ANSI.YELLOW}({duration:.2f}ms){ANSI.RESET} | User-Agent: {user_agent}'

    if level == logging.ERROR:
      logger.error(msg_body, exc_info=exc)
    else:
      logger.info(msg_body)