guardia-itsm/core/sla.py

"""
GUARDiA ITSM — SLA 타이머 & 자동 에스컬레이션 (Enhancement A-2)

기능:
  1. SR 생성 시 SLA 마감 시각 계산 (기관 sla_hours × 우선순위 배수)
  2. 매 30분 스케줄러 실행 → 마감 초과 SR 탐지
  3. 에스컬레이션: assigned_to → 해당 기관 PM → ADMIN 순으로 재배정
  4. 메신저/이메일 알림 발송

우선순위별 SLA 배수:
  CRITICAL → 0.5× (가장 짧음)
  HIGH     → 0.75×
  MEDIUM   → 1.0×  (기준값)
  LOW      → 2.0×

예시: 기관 sla_hours=4, CRITICAL SR → 마감 = 생성 후 2시간
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import Optional

logger = logging.getLogger(__name__)

# 우선순위별 SLA 시간 배수
_SLA_MULTIPLIER: dict[str, float] = {
    "CRITICAL": 0.50,
    "HIGH":     0.75,
    "MEDIUM":   1.00,
    "LOW":      2.00,
}

# 기관 SLA 미설정 시 기본값 (시간)
_DEFAULT_SLA_HOURS = 8


def compute_sla_deadline(
    created_at: datetime,
    sla_hours: int,
    priority: str,
) -> datetime:
    """
    SR 생성 시각 + (sla_hours × 우선순위 배수)로 SLA 마감 시각 계산.

    Args:
        created_at: SR 생성 시각 (UTC 또는 로컬 naive datetime)
        sla_hours:  기관 SLA 기준 시간 (Institution.sla_hours)
        priority:   CRITICAL / HIGH / MEDIUM / LOW

    Returns:
        SLA 마감 datetime
    """
    multiplier = _SLA_MULTIPLIER.get(priority.upper(), 1.0)
    delta_hours = sla_hours * multiplier
    return created_at + timedelta(hours=delta_hours)


def is_sla_breached(sla_deadline: Optional[datetime]) -> bool:
    """현재 시각이 SLA 마감을 초과했는지 확인."""
    if not sla_deadline:
        return False
    now = datetime.now()
    # timezone-aware vs naive 혼용 방지
    if sla_deadline.tzinfo is not None:
        now = datetime.now(timezone.utc)
    return now > sla_deadline


def sla_remaining_minutes(sla_deadline: Optional[datetime]) -> Optional[int]:
    """
    SLA 잔여 시간(분). 음수면 초과.
    sla_deadline이 None 이면 None 반환.
    """
    if not sla_deadline:
        return None
    now = datetime.now()
    if sla_deadline.tzinfo is not None:
        now = datetime.now(timezone.utc)
    delta = sla_deadline - now
    return int(delta.total_seconds() / 60)


# ── SR 생성 시 SLA 마감 자동 계산 ─────────────────────────────────────────────

async def set_sla_on_create(sr_id: str, db) -> None:
    """
    SR 생성 직후 호출 — 기관 sla_hours 조회 후 sla_deadline 계산·저장.

    routers/tasks.py의 create_sr() 내부에서 호출:
        from core.sla import set_sla_on_create
        await set_sla_on_create(sr.sr_id, db)
    """
    from sqlalchemy import select
    from models import SRRequest, Institution

    sr = (await db.execute(
        select(SRRequest).where(SRRequest.sr_id == sr_id)
    )).scalars().first()

    if not sr or sr.sla_deadline:
        return

    # 기관 SLA 시간 조회
    sla_hours = _DEFAULT_SLA_HOURS
    if sr.inst_id:
        inst = (await db.execute(
            select(Institution).where(Institution.id == sr.inst_id)
        )).scalars().first()
        if inst and inst.sla_hours:
            sla_hours = inst.sla_hours

    sr.sla_deadline = compute_sla_deadline(sr.created_at, sla_hours, sr.priority)
    await db.commit()
    logger.debug(
        "SLA 마감 설정: sr_id=%s priority=%s deadline=%s",
        sr_id, sr.priority, sr.sla_deadline,
    )


# ── 에스컬레이션 ─────────────────────────────────────────────────────────────

async def escalate_sr(sr_id: str, db) -> Optional[str]:
    """
    SLA 초과 SR을 에스컬레이션합니다.

    에스컬레이션 체인:
      1. 기관 담당자(role=PM) → ITSM PM 사용자
      2. 없으면 → ADMIN 사용자

    Returns:
        에스컬레이션 대상 username 또는 None
    """
    from sqlalchemy import select
    from models import SRRequest, User, UserRole, Institution, InstContact

    sr = (await db.execute(
        select(SRRequest).where(SRRequest.sr_id == sr_id)
    )).scalars().first()

    if not sr or sr.escalated_at:
        return None   # 이미 에스컬레이션됨

    escalate_target: Optional[str] = None

    # 1차: 기관 PM 연락처 조회
    if sr.inst_id:
        contact = (await db.execute(
            select(InstContact).where(
                InstContact.inst_id == sr.inst_id,
                InstContact.role == "PM",
                InstContact.is_active.is_(True),
            ).order_by(InstContact.is_primary.desc())
        )).scalars().first()
        if contact and contact.email:
            escalate_target = contact.email

    # 2차: ITSM PM 사용자 조회
    if not escalate_target:
        pm_user = (await db.execute(
            select(User).where(
                User.role == UserRole.PM,
                User.is_active.is_(True),
            ).limit(1)
        )).scalars().first()
        if pm_user:
            escalate_target = pm_user.username

    # 3차: ADMIN 사용자 조회
    if not escalate_target:
        admin = (await db.execute(
            select(User).where(
                User.role == UserRole.ADMIN,
                User.is_active.is_(True),
            ).limit(1)
        )).scalars().first()
        if admin:
            escalate_target = admin.username

    if not escalate_target:
        logger.warning("SLA 에스컬레이션 대상 없음: sr_id=%s", sr_id)
        return None

    now = datetime.now()
    previous_assignee = sr.assigned_to
    sr.assigned_to   = escalate_target
    sr.escalated_at  = now
    sr.escalated_to  = escalate_target
    sr.sla_breached  = True
    sr.updated_at    = now
    await db.commit()

    logger.warning(
        "SLA 에스컬레이션: sr_id=%s %s → %s",
        sr_id, previous_assignee, escalate_target,
    )
    return escalate_target


# ── 스케줄러 작업 — 매 30분 실행 ─────────────────────────────────────────────

async def check_sla_violations() -> None:
    """
    SLA 마감이 지났으나 미완료 상태인 SR을 탐지하여 에스컬레이션.

    core/scheduler.py start_scheduler()에서 등록:
        scheduler.add_job(check_sla_violations, 'interval', minutes=30, ...)
    """
    from database import SessionLocal
    from models import SRRequest, SRStatus
    from sqlalchemy import select, and_
    from core.notify import send_messenger

    # 최종 상태가 아닌 SR 목록
    terminal = {SRStatus.COMPLETED, SRStatus.REJECTED, SRStatus.FAILED_ROLLBACK}

    try:
        async with SessionLocal() as db:
            now = datetime.now()
            # sla_deadline 초과 + 미완료 + 에스컬레이션 미발생
            candidates = (await db.execute(
                select(SRRequest).where(
                    and_(
                        SRRequest.sla_deadline.isnot(None),
                        SRRequest.sla_deadline < now,
                        SRRequest.sla_breached.is_(False),
                        SRRequest.status.notin_([s.value for s in terminal]),
                    )
                )
            )).scalars().all()

        if not candidates:
            logger.debug("SLA 위반 SR 없음")
            return

        logger.warning("SLA 위반 SR 탐지: %d건", len(candidates))

        for sr in candidates:
            async with SessionLocal() as db:
                target = await escalate_sr(sr.sr_id, db)
            if target:
                # 에스컬레이션 알림
                overdue_min = abs(sla_remaining_minutes(sr.sla_deadline) or 0)
                msg = (
                    f"🚨 SLA 위반 에스컬레이션\n"
                    f"SR: {sr.sr_id}\n"
                    f"제목: {sr.title}\n"
                    f"우선순위: {sr.priority}\n"
                    f"초과 시간: {overdue_min // 60}시간 {overdue_min % 60}분\n"
                    f"담당자 변경: {sr.escalated_to}"
                )
                try:
                    await send_messenger(
                        room="ops",
                        text=msg,
                        title=f"[SLA 위반] {sr.sr_id}",
                    )
                except Exception:
                    pass  # 알림 실패는 에스컬레이션을 막지 않음

    except Exception as exc:
        logger.error("SLA 위반 점검 오류: %s", exc, exc_info=True)