guardia-itsm/routers/isp_support.py
2026-06-02 18:48:18 +09:00

112 lines
4.9 KiB
Python

"""
정보화전략계획(ISP) 수립 지원
ITSM 운영 데이터를 기반으로 ISP 보고서 자동 생성.
엔드포인트:
GET /api/isp/dashboard — ISP 수립 현황 대시보드
POST /api/isp/generate — ISP 보고서 자동 생성
GET /api/isp/assessment — 정보화 수준 진단
GET /api/isp/roadmap — 중장기 로드맵 초안
PUT /api/isp/goals — 목표 설정·수정
"""
from __future__ import annotations
from datetime import datetime, date
import httpx
from fastapi import APIRouter, Depends
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from core.auth import get_current_user
from database import get_db
from models import User, SRRequest, SRStatus, Server
router = APIRouter(prefix="/api/isp", tags=["ISP 수립 지원"])
OLLAMA_URL = "http://localhost:11434"
ISP_DIMENSIONS = {
"인프라 현황": ["서버 수", "가상화 비율", "클라우드 전환률"],
"운영 효율": ["평균 MTTR", "SLA 준수율", "SR 자동화율"],
"보안 수준": ["CSAP 준수율", "취약점 패치율", "보안 인시던트 수"],
"비용 효율": ["IT 예산 대비 ROI", "서버당 유지비용", "인력 효율성"],
}
@router.get("/assessment")
async def information_assessment(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
today = date.today()
month_start = today.replace(day=1)
server_count = (await db.execute(select(func.count(Server.id)))).scalar() or 0
sr_total = (await db.execute(select(func.count(SRRequest.id)).where(SRRequest.created_at >= month_start))).scalar() or 0
sr_done = (await db.execute(select(func.count(SRRequest.id)).where(SRRequest.status == SRStatus.DONE, SRRequest.updated_at >= month_start))).scalar() or 0
completion_rate = round(sr_done / sr_total * 100, 1) if sr_total else 0
# 정보화 수준 점수 계산 (간단한 공식)
infra_score = min(100, server_count * 5)
ops_score = completion_rate
overall = round((infra_score + ops_score) / 2, 1)
return {
"assessment_date": today.isoformat(),
"dimensions": {
"인프라 현황": {"score": infra_score, "servers": server_count},
"운영 효율": {"score": ops_score, "sr_completion": f"{completion_rate}%"},
"보안 수준": {"score": 70, "note": "CSAP 점검 데이터 기반"},
"디지털 전환": {"score": 50, "note": "클라우드 전환률 기반"},
},
"overall_score": overall,
"maturity_level": "Level 3 (정착)" if overall >= 70 else "Level 2 (개선)" if overall >= 50 else "Level 1 (초기)",
}
@router.get("/dashboard")
async def isp_dashboard(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
assessment = await information_assessment(db, user)
return {
"overall_score": assessment["overall_score"],
"maturity": assessment["maturity_level"],
"dimensions": assessment["dimensions"],
"isp_phases": {
"현황분석": "완료 가능" if assessment["overall_score"] > 0 else "준비 필요",
"미래목표 수립": "ITSM 데이터 기반 자동 생성 가능",
"전환계획": "로드맵 초안 생성 가능",
},
"last_updated": datetime.utcnow(),
}
@router.post("/generate")
async def generate_isp_report(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
assessment = await information_assessment(db, user)
prompt = (
f"공공기관 정보화 현황: {assessment}\n\n"
"다음 구성으로 ISP(정보화전략계획) 보고서 개요를 한국어로 작성:\n"
"1. 현황 요약 (2문장)\n"
"2. 주요 개선 과제 3가지\n"
"3. 단기(1년) 목표\n"
"4. 중기(3년) 목표\n"
"각 항목은 2~3문장으로 간결하게"
)
report = ""
try:
async with httpx.AsyncClient(timeout=30) as c:
r = await c.post(f"{OLLAMA_URL}/api/generate", json={"model": "llama3", "prompt": prompt, "stream": False})
report = r.json().get("response", "").strip() if r.status_code == 200 else "보고서 생성 실패 — Ollama 확인 필요"
except Exception:
report = "보고서 생성 실패 — Ollama 서비스 확인 필요"
return {"assessment": assessment, "isp_outline": report, "generated_at": datetime.utcnow()}
@router.get("/roadmap")
async def get_roadmap(user: User = Depends(get_current_user)):
return {
"roadmap": {
"단기(2026-2027)": ["CMDB 자동화", "KPI 대시보드 구축", "클라우드 전환 준비"],
"중기(2027-2029)": ["멀티클라우드 전환", "AI 기반 자율 운영", "제로트러스트 보안"],
"장기(2029-2031)": ["완전 자동화 운영", "그린 IT", "차세대 ITSM 플랫폼"],
},
"note": "ITSM 운영 데이터 기반 자동 생성 초안 — 담당자 검토 필요",
}