o2o-negosium-original/lps/crud/ip_session.py

"""IP 세션 종료 이력 CRUD — 세션당 요청 수·종료 사유를 축적(요청 예산 상한 튜닝용)."""

from sqlalchemy import text

from common.database.db_session_manager import DB_SESSION_MNG
from common.enums import DBType, DBWRType


class IpSessionLog:
    DB = DBType.MAIN.value

    async def record(self, event: dict):
        """세션 종료 이벤트 1건 저장. 기록 실패가 검색을 막지 않도록 호출부에서 예외를 삼킨다."""
        sql = text("""
            INSERT INTO ip_session (source, proxy_port, requests, ok_count, blocked_count, elapsed_sec, end_reason)
            VALUES (:source, :proxy_port, :requests, :ok_count, :blocked_count, :elapsed_sec, :end_reason)
        """)
        params = {k: event.get(k) for k in
                  ("source", "proxy_port", "requests", "ok_count", "blocked_count", "elapsed_sec", "end_reason")}
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value)
        try:
            await s.execute(sql, params)
            await s.commit()
        except Exception:
            await s.rollback()
            raise
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)

    async def admin_stats(self, hours: int = 168) -> dict:
        """관리자 FE 용 IP 세션 통계 — 종료 사유 분포·세션당 요청 수 히스토그램·
        차단 세션 최소 요청 수(예산 튜닝 기준선)·최근 세션 목록."""
        p = {"h": hours}
        by_reason = text("""
            SELECT end_reason, count(*) AS n FROM ip_session
             WHERE created_at > now() - make_interval(hours => :h) GROUP BY end_reason
        """)
        histogram = text("""
            SELECT requests, count(*) AS n FROM ip_session
             WHERE created_at > now() - make_interval(hours => :h) GROUP BY requests ORDER BY requests
        """)
        block_min = text("""
            SELECT min(requests) FROM ip_session
             WHERE created_at > now() - make_interval(hours => :h) AND end_reason = 'block'
        """)
        recent = text("""
            SELECT source, proxy_port, requests, ok_count, blocked_count, elapsed_sec, end_reason, created_at
              FROM ip_session WHERE created_at > now() - make_interval(hours => :h)
             ORDER BY created_at DESC LIMIT 50
        """)
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
        try:
            return {
                "by_reason": {r[0]: int(r[1]) for r in (await s.execute(by_reason, p)).all()},
                "histogram": [{"requests": int(r[0]), "count": int(r[1])} for r in (await s.execute(histogram, p)).all()],
                "block_min_requests": (lambda v: int(v) if v is not None else None)((await s.execute(block_min, p)).scalar()),
                "sessions": [dict(r) for r in (await s.execute(recent, p)).mappings().all()],
            }
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)

    async def recent_stats(self, minutes: int = 60) -> dict:
        """최근 N분 세션 요약 — 종료 사유별 건수(모니터링·알림용). 예: {"budget": 12, "block": 1}"""
        sql = text("""
            SELECT end_reason, count(*) FROM ip_session
             WHERE created_at > now() - make_interval(mins => :m) GROUP BY end_reason
        """)
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
        try:
            rows = (await s.execute(sql, {"m": minutes})).all()
            return {r[0]: int(r[1]) for r in rows}
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)