zioinfo-mail/itsm/routers/timetable.py

"""작업 타임테이블 CRUD + Excel 다운로드 엔드포인트."""
import io
from datetime import datetime, date
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from sqlalchemy import select, and_, or_
from sqlalchemy.ext.asyncio import AsyncSession

from core.auth import get_current_user
from database import get_db
from models import (
    Institution, Server, ShellScript,
    WorkTimetable, WorkTimetableCreate, WorkTimetableOut, WorkTimetableUpdate,
    User, UserRole,
)

router = APIRouter(prefix="/api/timetable", tags=["timetable"])

WORK_TYPE_KO = {
    "REGULAR_CHECK": "정기점검",
    "PM":            "예방정비",
    "SR":            "SR작업",
    "ADHOC":         "수시점검",
    "DEPLOY":        "배포작업",
    "EMERGENCY":     "긴급대응",
}

RESULT_STATUS_KO = {
    "PENDING":   "예정",
    "SUCCESS":   "완료",
    "FAILED":    "실패",
    "PARTIAL":   "부분완료",
    "CANCELLED": "취소",
}


async def _build_query(
    db: AsyncSession,
    current_user: User,
    inst_id: Optional[int],
    work_type: Optional[str],
    result_status: Optional[str],
    assignee: Optional[str],
    date_from: Optional[date],
    date_to: Optional[date],
    keyword: Optional[str],
):
    q = select(WorkTimetable)

    # CUSTOMER 역할: 자기 기관만
    if current_user.role == UserRole.CUSTOMER and current_user.inst_code:
        r = await db.execute(
            select(Institution).where(Institution.inst_code == current_user.inst_code)
        )
        own_inst = r.scalars().first()
        if own_inst:
            q = q.where(WorkTimetable.inst_id == own_inst.id)
        else:
            q = q.where(WorkTimetable.id == -1)
    elif inst_id:
        q = q.where(WorkTimetable.inst_id == inst_id)

    if work_type:
        q = q.where(WorkTimetable.work_type == work_type)
    if result_status:
        q = q.where(WorkTimetable.result_status == result_status)
    if assignee:
        q = q.where(WorkTimetable.assignee.contains(assignee))
    if date_from:
        q = q.where(WorkTimetable.scheduled_at >= datetime.combine(date_from, datetime.min.time()))
    if date_to:
        q = q.where(WorkTimetable.scheduled_at <= datetime.combine(date_to, datetime.max.time()))
    if keyword:
        q = q.where(
            or_(
                WorkTimetable.title.contains(keyword),
                WorkTimetable.content.contains(keyword),
            )
        )
    return q


@router.get("", response_model=List[WorkTimetableOut])
async def list_timetable(
    inst_id: Optional[int] = Query(None),
    work_type: Optional[str] = Query(None),
    result_status: Optional[str] = Query(None),
    assignee: Optional[str] = Query(None),
    date_from: Optional[date] = Query(None),
    date_to: Optional[date] = Query(None),
    keyword: Optional[str] = Query(None),
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    q = await _build_query(
        db, current_user, inst_id, work_type, result_status,
        assignee, date_from, date_to, keyword,
    )
    q = q.order_by(WorkTimetable.scheduled_at.desc()).offset(skip).limit(limit)
    result = await db.execute(q)
    return result.scalars().all()


@router.get("/stats")
async def timetable_stats(
    inst_id: Optional[int] = Query(None),
    date_from: Optional[date] = Query(None),
    date_to: Optional[date] = Query(None),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """작업 유형별·결과별 집계."""
    q = await _build_query(
        db, current_user, inst_id, None, None, None, date_from, date_to, None
    )
    result = await db.execute(q)
    rows = result.scalars().all()
    by_type = {}
    by_status = {}
    for r in rows:
        by_type[r.work_type]       = by_type.get(r.work_type, 0) + 1
        by_status[r.result_status] = by_status.get(r.result_status, 0) + 1
    return {"total": len(rows), "by_type": by_type, "by_status": by_status}


@router.get("/export/excel")
async def export_excel(
    inst_id: Optional[int] = Query(None),
    work_type: Optional[str] = Query(None),
    result_status: Optional[str] = Query(None),
    assignee: Optional[str] = Query(None),
    date_from: Optional[date] = Query(None),
    date_to: Optional[date] = Query(None),
    keyword: Optional[str] = Query(None),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Excel 파일 생성 후 다운로드 (openpyxl 사용, 외부 API 없음)."""
    if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
        raise HTTPException(403, "권한이 없습니다.")

    try:
        from openpyxl import Workbook
        from openpyxl.styles import (
            Font, PatternFill, Alignment, Border, Side,
        )
        from openpyxl.utils import get_column_letter
    except ImportError:
        raise HTTPException(500, "openpyxl 라이브러리가 설치되지 않았습니다. pip install openpyxl")

    q = await _build_query(
        db, current_user, inst_id, work_type, result_status,
        assignee, date_from, date_to, keyword,
    )
    q = q.order_by(WorkTimetable.scheduled_at)
    res = await db.execute(q)
    rows = res.scalars().all()

    # 기관·서버 이름 매핑 (ID → 이름)
    inst_map: dict[int, str] = {}
    srv_map:  dict[int, str] = {}
    inst_ids = {r.inst_id for r in rows if r.inst_id}
    srv_ids  = {r.server_id for r in rows if r.server_id}
    if inst_ids:
        ir = await db.execute(select(Institution).where(Institution.id.in_(inst_ids)))
        for inst in ir.scalars():
            inst_map[inst.id] = inst.inst_name
    if srv_ids:
        sr = await db.execute(select(Server).where(Server.id.in_(srv_ids)))
        for srv in sr.scalars():
            srv_map[srv.id] = srv.server_name

    # ── Workbook 구성 ────────────────────────────────────────────────────────────
    wb = Workbook()

    # ── 스타일 정의 ──────────────────────────────────────────────────────────────
    hdr_fill  = PatternFill("solid", fgColor="1E3A5F")
    hdr_font  = Font(name="맑은 고딕", bold=True, color="FFFFFF", size=10)
    hdr_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
    thin_side = Side(style="thin", color="CCCCCC")
    thin_bdr  = Border(left=thin_side, right=thin_side, top=thin_side, bottom=thin_side)
    even_fill = PatternFill("solid", fgColor="F2F6FA")
    data_align_center = Alignment(horizontal="center", vertical="center")
    data_align_left   = Alignment(horizontal="left",   vertical="center", wrap_text=True)

    status_fills = {
        "SUCCESS":   PatternFill("solid", fgColor="C6EFCE"),
        "FAILED":    PatternFill("solid", fgColor="FFC7CE"),
        "PARTIAL":   PatternFill("solid", fgColor="FFEB9C"),
        "PENDING":   PatternFill("solid", fgColor="F4F4F4"),
        "CANCELLED": PatternFill("solid", fgColor="E2E8F0"),
    }

    # ── 작업이력 시트 ────────────────────────────────────────────────────────────
    ws = wb.active
    ws.title = "작업이력"
    ws.freeze_panes = "A2"

    headers = [
        ("번호",        7),
        ("작업유형",    12),
        ("제목",        35),
        ("기관명",      14),
        ("대상서버",    16),
        ("처리예정",    17),
        ("시작",        17),
        ("완료",        17),
        ("소요(분)",     9),
        ("처리내용",    40),
        ("명령어/쉘",   30),
        ("처리결과",    35),
        ("결과상태",    10),
        ("담당자",      12),
        ("검토자",      12),
        ("SR번호",      20),
        ("비고",        20),
    ]
    for col_idx, (hdr, width) in enumerate(headers, 1):
        cell = ws.cell(row=1, column=col_idx, value=hdr)
        cell.font      = hdr_font
        cell.fill      = hdr_fill
        cell.alignment = hdr_align
        cell.border    = thin_bdr
        ws.column_dimensions[get_column_letter(col_idx)].width = width
    ws.row_dimensions[1].height = 30

    for i, row in enumerate(rows, start=2):
        is_even = (i % 2 == 0)
        bg_fill = even_fill if is_even else None

        def _duration_min(r):
            if r.started_at and r.completed_at:
                return int((r.completed_at - r.started_at).total_seconds() / 60)
            return ""

        def _fmt(dt):
            return dt.strftime("%Y-%m-%d %H:%M") if dt else ""

        vals = [
            i - 1,
            WORK_TYPE_KO.get(row.work_type, row.work_type),
            row.title,
            inst_map.get(row.inst_id, "") if row.inst_id else "",
            srv_map.get(row.server_id, "") if row.server_id else "",
            _fmt(row.scheduled_at),
            _fmt(row.started_at),
            _fmt(row.completed_at),
            _duration_min(row),
            row.content,
            row.command_or_shell or "",
            row.result or "",
            RESULT_STATUS_KO.get(row.result_status, row.result_status),
            row.assignee or "",
            row.reviewer or "",
            row.sr_id or "",
            row.note or "",
        ]
        center_cols = {1, 2, 4, 5, 6, 7, 8, 9, 13, 14, 15, 16}
        for col_idx, val in enumerate(vals, 1):
            cell = ws.cell(row=i, column=col_idx, value=val)
            cell.border    = thin_bdr
            cell.alignment = data_align_center if col_idx in center_cols else data_align_left
            if col_idx == 13:  # 결과상태 강조
                cell.fill = status_fills.get(row.result_status, bg_fill or PatternFill())
            elif bg_fill:
                cell.fill = bg_fill

    # ── 통계 시트 ────────────────────────────────────────────────────────────────
    ws2 = wb.create_sheet("통계요약")
    ws2.column_dimensions["A"].width = 20
    ws2.column_dimensions["B"].width = 12

    ws2.cell(1, 1, "◆ 작업 유형별 집계").font = Font(name="맑은 고딕", bold=True, size=11)
    ws2.cell(2, 1, "작업유형").fill  = hdr_fill
    ws2.cell(2, 1).font = hdr_font
    ws2.cell(2, 2, "건수").fill      = hdr_fill
    ws2.cell(2, 2).font = hdr_font
    by_type: dict[str, int] = {}
    for r in rows:
        by_type[r.work_type] = by_type.get(r.work_type, 0) + 1
    for ri, (wt, cnt) in enumerate(sorted(by_type.items()), start=3):
        ws2.cell(ri, 1, WORK_TYPE_KO.get(wt, wt))
        ws2.cell(ri, 2, cnt)

    offset = len(by_type) + 5
    ws2.cell(offset, 1, "◆ 결과 상태별 집계").font = Font(name="맑은 고딕", bold=True, size=11)
    ws2.cell(offset+1, 1, "결과상태").fill  = hdr_fill
    ws2.cell(offset+1, 1).font = hdr_font
    ws2.cell(offset+1, 2, "건수").fill      = hdr_fill
    ws2.cell(offset+1, 2).font = hdr_font
    by_status: dict[str, int] = {}
    for r in rows:
        by_status[r.result_status] = by_status.get(r.result_status, 0) + 1
    for ri, (st, cnt) in enumerate(sorted(by_status.items()), start=offset+2):
        ws2.cell(ri, 1, RESULT_STATUS_KO.get(st, st))
        ws2.cell(ri, 2, cnt)
        ws2.cell(ri, 1).fill = status_fills.get(st, PatternFill())
        ws2.cell(ri, 2).fill = status_fills.get(st, PatternFill())

    # ── 파일 이름 생성 및 반환 ────────────────────────────────────────────────────
    inst_suffix = inst_map.get(inst_id, "ALL") if inst_id else "ALL"
    today_str   = date.today().strftime("%Y%m%d")
    filename    = f"GUARDiA_작업이력_{inst_suffix}_{today_str}.xlsx"

    buf = io.BytesIO()
    wb.save(buf)
    buf.seek(0)

    from urllib.parse import quote
    encoded_name = quote(filename)
    return StreamingResponse(
        buf,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={
            "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_name}",
        },
    )


@router.get("/{timetable_id}", response_model=WorkTimetableOut)
async def get_timetable(
    timetable_id: int,
    db: AsyncSession = Depends(get_db),
    _u: User = Depends(get_current_user),
):
    r = await db.execute(select(WorkTimetable).where(WorkTimetable.id == timetable_id))
    row = r.scalars().first()
    if not row:
        raise HTTPException(404, "타임테이블 항목을 찾을 수 없습니다.")
    return row


@router.post("", response_model=WorkTimetableOut, status_code=201)
async def create_timetable(
    payload: WorkTimetableCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
        raise HTTPException(403, "권한이 없습니다.")
    row = WorkTimetable(**payload.model_dump(), created_by=current_user.username)
    db.add(row)
    # 스크립트 사용 횟수 증가
    if payload.script_id:
        sc = (await db.execute(
            select(ShellScript).where(ShellScript.id == payload.script_id)
        )).scalars().first()
        if sc:
            sc.use_count = (sc.use_count or 0) + 1
    await db.commit()
    await db.refresh(row)
    return row


@router.patch("/{timetable_id}", response_model=WorkTimetableOut)
async def update_timetable(
    timetable_id: int,
    payload: WorkTimetableUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
        raise HTTPException(403, "권한이 없습니다.")
    r = await db.execute(select(WorkTimetable).where(WorkTimetable.id == timetable_id))
    row = r.scalars().first()
    if not row:
        raise HTTPException(404, "타임테이블 항목을 찾을 수 없습니다.")
    for k, v in payload.model_dump(exclude_unset=True).items():
        setattr(row, k, v)
    row.updated_at = datetime.now()
    await db.commit()
    await db.refresh(row)
    return row


@router.delete("/{timetable_id}", status_code=204)
async def delete_timetable(
    timetable_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role not in (UserRole.ADMIN, UserRole.PM):
        raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
    r = await db.execute(select(WorkTimetable).where(WorkTimetable.id == timetable_id))
    row = r.scalars().first()
    if not row:
        raise HTTPException(404, "타임테이블 항목을 찾을 수 없습니다.")
    await db.delete(row)
    await db.commit()