zioinfo-mail/itsm/core/oncall_rotate.py

"""
GUARDiA ITSM — On-Call 자동 로테이션 & 에스컬레이션 (Enhancement A-5)

기능:
  1. 매일 00:05 스케줄러 실행 → 다음날 당직 미배정 시 자동 배정
  2. OncallRotateConfig 의 engineer_list 를 current_index 기준으로 순환
  3. 배정 시 메신저/이메일 알림 발송
  4. 인시던트/SR 에스컬레이션 체인 실행:
       당직자(engineer) → backup_engineer → escalation_to → escalation_chain → ADMIN
"""
from __future__ import annotations

import json
import logging
from datetime import date, datetime, timedelta
from typing import List, Optional

logger = logging.getLogger(__name__)

# ── 로테이션 설정 조회 헬퍼 ──────────────────────────────────────────────────

async def _get_rotate_config(db):
    """OncallRotateConfig 싱글톤 레코드 조회 (없으면 None)."""
    from sqlalchemy import select
    from models import OncallRotateConfig

    result = await db.execute(
        select(OncallRotateConfig).where(OncallRotateConfig.id == 1)
    )
    return result.scalars().first()


async def get_or_create_rotate_config(db) -> "OncallRotateConfig":
    """
    설정 레코드가 없으면 기본값으로 생성 후 반환.
    routers/oncall.py 에서 설정 조회/수정 시 사용.
    """
    from models import OncallRotateConfig

    cfg = await _get_rotate_config(db)
    if not cfg:
        cfg = OncallRotateConfig(
            id=1,
            is_active=False,
            engineer_list="[]",
            current_index=0,
            rotate_days=1,
            default_shift="ALL_DAY",
            notify_on_assign=True,
            advance_days=1,
        )
        db.add(cfg)
        await db.commit()
        await db.refresh(cfg)
    return cfg


# ── 오늘/내일 당직자 조회 ────────────────────────────────────────────────────

async def get_oncall_for_date(target_date: date, db) -> List["OncallSchedule"]:
    """특정 날짜의 OncallSchedule 목록 반환."""
    from sqlalchemy import select
    from models import OncallSchedule

    result = await db.execute(
        select(OncallSchedule)
        .where(OncallSchedule.duty_date == target_date)
        .order_by(OncallSchedule.shift)
    )
    return result.scalars().all()


async def get_current_oncall(db) -> Optional["OncallSchedule"]:
    """오늘 당직자 (ALL_DAY 우선, 없으면 첫 번째 일정) 반환."""
    schedules = await get_oncall_for_date(date.today(), db)
    if not schedules:
        return None
    # ALL_DAY 우선
    for s in schedules:
        if s.shift == "ALL_DAY":
            return s
    return schedules[0]


# ── 자동 로테이션 배정 ───────────────────────────────────────────────────────

async def auto_rotate_oncall() -> None:
    """
    스케줄러에서 매일 00:05 호출.

    로직:
      1. OncallRotateConfig.is_active == True 확인
      2. (오늘 + advance_days) 날짜에 당직 미배정 확인
      3. engineer_list[current_index] 를 당직자로 OncallSchedule 등록
      4. current_index = (current_index + 1) % len(engineer_list) 업데이트
      5. notify_on_assign == True 이면 메신저 알림 발송
    """
    from database import SessionLocal
    from models import OncallSchedule, OncallRotateConfig
    from sqlalchemy import select

    try:
        async with SessionLocal() as db:
            cfg = await _get_rotate_config(db)

        if not cfg or not cfg.is_active:
            logger.debug("On-Call 자동 로테이션 비활성화 상태 — 스킵")
            return

        engineers: List[str] = json.loads(cfg.engineer_list or "[]")
        if not engineers:
            logger.debug("On-Call 로테이션 엔지니어 목록 비어있음 — 스킵")
            return

        target_date = date.today() + timedelta(days=cfg.advance_days)

        async with SessionLocal() as db:
            # 해당 날짜 당직 존재 여부 확인
            existing = (await db.execute(
                select(OncallSchedule).where(
                    OncallSchedule.duty_date == target_date,
                    OncallSchedule.shift     == cfg.default_shift,
                )
            )).scalars().first()

            if existing:
                logger.debug(
                    "On-Call 자동 로테이션 스킵: %s 날짜 당직 이미 등록됨 (engineer=%s)",
                    target_date.isoformat(), existing.engineer,
                )
                return

            # 로테이션 배정
            idx = cfg.current_index % len(engineers)
            assigned_engineer = engineers[idx]
            next_idx = (idx + 1) % len(engineers)

            # 에스컬레이션 체인에서 backup / escalation 설정
            chain: List[str] = json.loads(cfg.escalation_chain or "[]")
            backup = chain[0] if chain else None
            escalation = chain[1] if len(chain) > 1 else None

            schedule = OncallSchedule(
                duty_date       = target_date,
                shift           = cfg.default_shift,
                engineer        = assigned_engineer,
                backup_engineer = backup,
                escalation_to   = escalation,
                note            = f"[자동 배정] 로테이션 순번 {idx}",
                created_by      = "oncall-rotator",
            )
            db.add(schedule)

            # current_index 업데이트
            cfg_obj = await db.get(OncallRotateConfig, 1)
            if cfg_obj:
                cfg_obj.current_index = next_idx
                cfg_obj.updated_at = datetime.now()

            await db.commit()
            await db.refresh(schedule)

        logger.info(
            "On-Call 자동 배정 완료: date=%s shift=%s engineer=%s (next_idx=%d)",
            target_date.isoformat(), cfg.default_shift, assigned_engineer, next_idx,
        )

        # 알림 발송
        if cfg.notify_on_assign:
            await _notify_oncall_assigned(
                target_date, cfg.default_shift, assigned_engineer, backup
            )

    except Exception as exc:
        logger.error("On-Call 자동 로테이션 오류: %s", exc, exc_info=True)


# ── 에스컬레이션 체인 실행 ────────────────────────────────────────────────────

async def escalate_oncall(incident_title: str, db) -> Optional[str]:
    """
    인시던트/장애 발생 시 당직자 → 백업 → escalation_to → ADMIN 순으로 에스컬레이션.

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

    schedule = await get_current_oncall(db)

    if schedule:
        # 1차: 당직자 본인
        primary = schedule.engineer

        # 2차: 백업 담당자
        backup = schedule.backup_engineer

        # 3차: 에스컬레이션 지정 담당자
        escalation = schedule.escalation_to
    else:
        primary = backup = escalation = None

    # 4차: 로테이션 설정의 escalation_chain
    chain_fallback: List[str] = []
    cfg = await _get_rotate_config(db)
    if cfg and cfg.escalation_chain:
        chain_fallback = json.loads(cfg.escalation_chain)

    # 5차: ADMIN 사용자
    admin_user = (await db.execute(
        select(User).where(
            User.role == UserRole.ADMIN,
            User.is_active.is_(True),
        ).limit(1)
    )).scalars().first()
    admin = admin_user.username if admin_user else None

    # 순서대로 첫 번째 유효한 대상 찾기
    candidates = [primary, backup, escalation] + chain_fallback + ([admin] if admin else [])
    target = next((c for c in candidates if c), None)

    if target:
        msg = (
            f"🚨 On-Call 에스컬레이션\n"
            f"장애/인시던트: {incident_title}\n"
            f"에스컬레이션 대상: {target}\n"
            f"일시: {datetime.now().strftime('%Y-%m-%d %H:%M')}"
        )
        try:
            from core.notify import send_messenger
            await send_messenger(room="ops", text=msg, title=f"[온콜 에스컬레이션] {incident_title}")
        except Exception as e:
            logger.warning("On-Call 에스컬레이션 알림 실패: %s", e)

    return target


# ── 내부 알림 헬퍼 ────────────────────────────────────────────────────────────

async def _notify_oncall_assigned(
    duty_date: date,
    shift: str,
    engineer: str,
    backup: Optional[str],
) -> None:
    """당직 자동 배정 알림 발송."""
    try:
        from core.notify import send_messenger
        msg = (
            f"📋 On-Call 자동 배정 알림\n"
            f"날짜: {duty_date.isoformat()}\n"
            f"시프트: {shift}\n"
            f"당직자: {engineer}\n"
            f"백업: {backup or '미지정'}"
        )
        await send_messenger(
            room="ops",
            text=msg,
            title=f"[온콜 배정] {duty_date.isoformat()} {engineer}",
        )
    except Exception as exc:
        logger.warning("On-Call 배정 알림 실패 (무시): %s", exc)