""" 정보화전략계획(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 운영 데이터 기반 자동 생성 초안 — 담당자 검토 필요", }