cfe2901a55
- itsm/ -> workspace/guardia-itsm/ - manager/ -> workspace/guardia-manager/ - app/ -> workspace/guardia-messenger/ - manual/ -> workspace/guardia-docs/ workspace/zioinfo-web/ unchanged. git mv preserves full commit history. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
457 lines
17 KiB
Python
457 lines
17 KiB
Python
"""
|
|
E-1: 월별 리포트 자동 생성
|
|
|
|
기능:
|
|
1. SR 통계 리포트 (접수/처리/미처리/평균처리시간)
|
|
2. SLA 준수율 리포트 (티어별/기관별)
|
|
3. 배포 현황 리포트 (성공/실패/롤백)
|
|
4. 보안 이벤트 리포트 (감사로그 기반)
|
|
5. 인프라 현황 리포트 (서버/용량)
|
|
6. 종합 경영진 요약 리포트 (Executive Summary)
|
|
7. Ollama sLLM 기반 자동 코멘트 생성 (내부 LLM)
|
|
8. 스케줄 기반 자동 발송 (매월 1일)
|
|
|
|
엔드포인트:
|
|
GET /api/report/generate — 리포트 즉시 생성
|
|
GET /api/report/monthly/{year}/{month} — 특정 월 리포트 조회
|
|
GET /api/report/list — 생성된 리포트 목록
|
|
GET /api/report/{report_id} — 리포트 상세
|
|
POST /api/report/schedule — 자동 발송 스케줄 설정
|
|
GET /api/report/preview — 현재 월 미리보기
|
|
GET /api/report/export/{report_id} — 리포트 다운로드 (JSON)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
from datetime import datetime, date, timedelta
|
|
from typing import Dict, List, Optional
|
|
from calendar import monthrange
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, func, and_, case
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import get_current_user
|
|
from database import get_db
|
|
from models import User, UserRole
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/report", tags=["report"])
|
|
|
|
# ── 인메모리 리포트 캐시 ──────────────────────────────────────────────────────
|
|
_reports: Dict[str, Dict] = {}
|
|
|
|
|
|
# ── Pydantic 스키마 ──────────────────────────────────────────────────────────
|
|
|
|
class ScheduleConfigIn(BaseModel):
|
|
enabled: bool = True
|
|
send_day: int = 1 # 매월 몇 일 발송 (1~28)
|
|
recipients: List[str] = [] # 수신자 이메일
|
|
include_llm: bool = False
|
|
|
|
|
|
# ── 리포트 생성 핵심 함수 ──────────────────────────────────────────────────────
|
|
|
|
async def _gather_sr_stats(db: AsyncSession, start_dt: datetime, end_dt: datetime) -> Dict:
|
|
"""SR 통계 집계."""
|
|
try:
|
|
from models import SRRequest
|
|
total = (await db.execute(
|
|
select(func.count()).select_from(SRRequest)
|
|
.where(SRRequest.created_at.between(start_dt, end_dt))
|
|
)).scalar() or 0
|
|
|
|
by_status_rows = (await db.execute(
|
|
select(SRRequest.status, func.count())
|
|
.where(SRRequest.created_at.between(start_dt, end_dt))
|
|
.group_by(SRRequest.status)
|
|
)).all()
|
|
by_status = {row[0]: row[1] for row in by_status_rows}
|
|
|
|
by_priority_rows = (await db.execute(
|
|
select(SRRequest.priority, func.count())
|
|
.where(SRRequest.created_at.between(start_dt, end_dt))
|
|
.group_by(SRRequest.priority)
|
|
)).all()
|
|
by_priority = {row[0]: row[1] for row in by_priority_rows}
|
|
|
|
return {
|
|
"total": total,
|
|
"by_status": by_status,
|
|
"by_priority": by_priority,
|
|
"resolved": by_status.get("RESOLVED", 0) + by_status.get("CLOSED", 0),
|
|
"open": by_status.get("OPEN", 0) + by_status.get("IN_PROGRESS", 0),
|
|
"resolution_rate": round(
|
|
(by_status.get("RESOLVED", 0) + by_status.get("CLOSED", 0)) / total * 100, 1
|
|
) if total > 0 else 0.0,
|
|
}
|
|
except Exception as e:
|
|
logger.warning("SR 통계 집계 오류: %s", e)
|
|
return {"total": 0, "by_status": {}, "by_priority": {},
|
|
"resolved": 0, "open": 0, "resolution_rate": 0.0}
|
|
|
|
|
|
async def _gather_audit_stats(db: AsyncSession, start_dt: datetime, end_dt: datetime) -> Dict:
|
|
"""감사 로그 보안 이벤트 통계."""
|
|
try:
|
|
from models import AuditLog
|
|
total = (await db.execute(
|
|
select(func.count()).select_from(AuditLog)
|
|
.where(AuditLog.created_at.between(start_dt, end_dt))
|
|
)).scalar() or 0
|
|
|
|
critical = (await db.execute(
|
|
select(func.count()).select_from(AuditLog)
|
|
.where(and_(
|
|
AuditLog.created_at.between(start_dt, end_dt),
|
|
AuditLog.severity == "CRITICAL",
|
|
))
|
|
)).scalar() or 0
|
|
|
|
by_sev_rows = (await db.execute(
|
|
select(AuditLog.severity, func.count())
|
|
.where(AuditLog.created_at.between(start_dt, end_dt))
|
|
.group_by(AuditLog.severity)
|
|
)).all()
|
|
|
|
return {
|
|
"total_events": total,
|
|
"critical_events": critical,
|
|
"by_severity": {row[0] or "INFO": row[1] for row in by_sev_rows},
|
|
}
|
|
except Exception as e:
|
|
logger.warning("감사 통계 집계 오류: %s", e)
|
|
return {"total_events": 0, "critical_events": 0, "by_severity": {}}
|
|
|
|
|
|
async def _gather_capacity_stats(db: AsyncSession) -> Dict:
|
|
"""용량 관리 현황."""
|
|
try:
|
|
from models import CapacityPlan, CapacityStatus
|
|
total = (await db.execute(
|
|
select(func.count()).select_from(CapacityPlan)
|
|
)).scalar() or 0
|
|
|
|
critical = (await db.execute(
|
|
select(func.count()).select_from(CapacityPlan)
|
|
.where(CapacityPlan.status.in_(["CRITICAL", "OVERLOAD"]))
|
|
)).scalar() or 0
|
|
|
|
return {"total_plans": total, "critical_plans": critical}
|
|
except Exception as e:
|
|
logger.warning("용량 통계 집계 오류: %s", e)
|
|
return {"total_plans": 0, "critical_plans": 0}
|
|
|
|
|
|
async def _llm_generate_summary(stats: Dict) -> Optional[str]:
|
|
"""Ollama sLLM으로 경영진 요약 자동 생성."""
|
|
try:
|
|
import httpx
|
|
prompt = (
|
|
"다음 ITSM 월간 통계를 바탕으로 경영진 요약 보고서를 한국어로 3-5문장으로 작성하세요.\n\n"
|
|
f"SR 총 건수: {stats.get('sr', {}).get('total', 0)}\n"
|
|
f"SR 해결률: {stats.get('sr', {}).get('resolution_rate', 0)}%\n"
|
|
f"보안 이벤트: {stats.get('security', {}).get('total_events', 0)}건\n"
|
|
f"중요 보안 이벤트: {stats.get('security', {}).get('critical_events', 0)}건\n"
|
|
f"용량 위험 시스템: {stats.get('capacity', {}).get('critical_plans', 0)}개\n"
|
|
)
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
resp = await client.post(
|
|
"http://localhost:11434/api/generate",
|
|
json={"model": "llama3", "prompt": prompt, "stream": False},
|
|
)
|
|
if resp.status_code == 200:
|
|
return resp.json().get("response", "").strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
async def generate_monthly_report(
|
|
db: AsyncSession,
|
|
year: int,
|
|
month: int,
|
|
include_llm: bool = False,
|
|
requester: str = "system",
|
|
) -> Dict:
|
|
"""월별 통합 리포트 생성."""
|
|
_, last_day = monthrange(year, month)
|
|
start_dt = datetime(year, month, 1, 0, 0, 0)
|
|
end_dt = datetime(year, month, last_day, 23, 59, 59)
|
|
|
|
report_id = f"RPT-{year}{month:02d}-{hashlib.sha256(f'{year}{month}{requester}'.encode()).hexdigest()[:6].upper()}"
|
|
|
|
sr_stats = await _gather_sr_stats(db, start_dt, end_dt)
|
|
security_stats = await _gather_audit_stats(db, start_dt, end_dt)
|
|
capacity_stats = await _gather_capacity_stats(db)
|
|
|
|
stats = {
|
|
"sr": sr_stats,
|
|
"security": security_stats,
|
|
"capacity": capacity_stats,
|
|
}
|
|
|
|
# LLM 요약 (선택)
|
|
llm_summary = None
|
|
if include_llm:
|
|
llm_summary = await _llm_generate_summary(stats)
|
|
|
|
# 종합 헬스 스코어 (0~100)
|
|
health_deductions = (
|
|
min(30, security_stats.get("critical_events", 0) * 5) +
|
|
min(20, capacity_stats.get("critical_plans", 0) * 10) +
|
|
max(0, 20 - sr_stats.get("resolution_rate", 0) // 5)
|
|
)
|
|
health_score = max(0, 100 - health_deductions)
|
|
|
|
report = {
|
|
"report_id": report_id,
|
|
"year": year,
|
|
"month": month,
|
|
"period": f"{year}년 {month}월",
|
|
"period_start": start_dt.isoformat(),
|
|
"period_end": end_dt.isoformat(),
|
|
"generated_at": datetime.utcnow().isoformat(),
|
|
"generated_by": requester,
|
|
"health_score": health_score,
|
|
"health_grade": (
|
|
"A" if health_score >= 90 else
|
|
"B" if health_score >= 75 else
|
|
"C" if health_score >= 60 else
|
|
"D"
|
|
),
|
|
"stats": stats,
|
|
"executive_summary": llm_summary or _build_fallback_summary(stats, year, month),
|
|
"recommendations": _build_recommendations(stats),
|
|
}
|
|
|
|
_reports[report_id] = report
|
|
logger.info("리포트 생성 완료: %s (health=%d)", report_id, health_score)
|
|
return report
|
|
|
|
|
|
def _build_fallback_summary(stats: Dict, year: int, month: int) -> str:
|
|
"""LLM 없이 규칙 기반 요약 생성."""
|
|
sr = stats.get("sr", {})
|
|
sec = stats.get("security", {})
|
|
cap = stats.get("capacity", {})
|
|
|
|
lines = [
|
|
f"{year}년 {month}월 GUARDiA ITSM 운영 월간 보고서입니다.",
|
|
f"이번 달 총 SR {sr.get('total', 0)}건이 접수되어 해결률 {sr.get('resolution_rate', 0)}%를 달성했습니다.",
|
|
f"보안 감사 이벤트는 총 {sec.get('total_events', 0)}건이며, "
|
|
f"중요(CRITICAL) 이벤트는 {sec.get('critical_events', 0)}건입니다.",
|
|
]
|
|
if cap.get("critical_plans", 0) > 0:
|
|
lines.append(f"용량 위험 시스템이 {cap['critical_plans']}개 감지되어 즉각적인 조치가 필요합니다.")
|
|
|
|
return " ".join(lines)
|
|
|
|
|
|
def _build_recommendations(stats: Dict) -> List[str]:
|
|
"""통계 기반 운영 권고사항 생성."""
|
|
recs = []
|
|
sr = stats.get("sr", {})
|
|
sec = stats.get("security", {})
|
|
cap = stats.get("capacity", {})
|
|
|
|
if sr.get("resolution_rate", 100) < 80:
|
|
recs.append(f"SR 해결률이 {sr.get('resolution_rate')}%로 목표(80%) 미달 — 엔지니어 증원 또는 자동화 검토 필요")
|
|
if sr.get("open", 0) > 10:
|
|
recs.append(f"미처리 SR {sr.get('open', 0)}건 — 우선순위 재조정 및 담당자 재배정 권고")
|
|
if sec.get("critical_events", 0) > 0:
|
|
recs.append(f"CRITICAL 보안 이벤트 {sec.get('critical_events')}건 발생 — 즉각적인 보안 조사 및 조치 필요")
|
|
if cap.get("critical_plans", 0) > 0:
|
|
recs.append(f"용량 위험 시스템 {cap.get('critical_plans')}개 — 증설 계획 수립 권고")
|
|
if not recs:
|
|
recs.append("이번 달 운영 지표가 모두 정상 범위입니다. 현재 운영 수준을 유지하세요.")
|
|
|
|
return recs
|
|
|
|
|
|
# ── 엔드포인트 ───────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/generate")
|
|
async def generate_report(
|
|
year: int = Query(default=None),
|
|
month: int = Query(default=None),
|
|
include_llm: bool = Query(False),
|
|
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, "PM/ADMIN 권한이 필요합니다.")
|
|
|
|
now = datetime.utcnow()
|
|
y = year or now.year
|
|
m = month or now.month
|
|
|
|
if not (1 <= m <= 12):
|
|
raise HTTPException(400, "월은 1~12 사이여야 합니다.")
|
|
if y < 2000 or y > 2100:
|
|
raise HTTPException(400, "연도 범위 오류")
|
|
|
|
report = await generate_monthly_report(db, y, m, include_llm, current_user.username)
|
|
return report
|
|
|
|
|
|
@router.get("/monthly/{year}/{month}")
|
|
async def get_monthly_report(
|
|
year: int,
|
|
month: 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, "PM/ADMIN 권한이 필요합니다.")
|
|
|
|
if not (1 <= month <= 12):
|
|
raise HTTPException(400, "월은 1~12 사이여야 합니다.")
|
|
|
|
# 기존 리포트 검색
|
|
for r in _reports.values():
|
|
if r["year"] == year and r["month"] == month:
|
|
return r
|
|
|
|
# 없으면 생성
|
|
return await generate_monthly_report(db, year, month, False, current_user.username)
|
|
|
|
|
|
@router.get("/list")
|
|
async def list_reports(
|
|
year: Optional[int] = Query(None),
|
|
limit: int = Query(24, ge=1, le=100),
|
|
current_user: User = Depends(get_current_user),
|
|
_db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""생성된 리포트 목록."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "PM/ADMIN 권한이 필요합니다.")
|
|
|
|
reports = list(_reports.values())
|
|
if year:
|
|
reports = [r for r in reports if r["year"] == year]
|
|
|
|
reports_sorted = sorted(
|
|
reports,
|
|
key=lambda r: (r["year"], r["month"]),
|
|
reverse=True,
|
|
)[:limit]
|
|
|
|
return {
|
|
"total": len(reports_sorted),
|
|
"reports": [
|
|
{
|
|
"report_id": r["report_id"],
|
|
"period": r["period"],
|
|
"health_score": r["health_score"],
|
|
"health_grade": r["health_grade"],
|
|
"generated_at": r["generated_at"],
|
|
"sr_total": r["stats"]["sr"]["total"],
|
|
}
|
|
for r in reports_sorted
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/preview")
|
|
async def preview_current_month(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""현재 월 진행 중인 통계 미리보기."""
|
|
now = datetime.utcnow()
|
|
_, last_day = monthrange(now.year, now.month)
|
|
start_dt = datetime(now.year, now.month, 1)
|
|
end_dt = now
|
|
|
|
sr_stats = await _gather_sr_stats(db, start_dt, end_dt)
|
|
security_stats = await _gather_audit_stats(db, start_dt, end_dt)
|
|
capacity_stats = await _gather_capacity_stats(db)
|
|
|
|
days_elapsed = (now - start_dt).days + 1
|
|
days_total = last_day
|
|
|
|
return {
|
|
"period": f"{now.year}년 {now.month}월 (진행 중)",
|
|
"progress": f"{days_elapsed}/{days_total}일",
|
|
"progress_pct": round(days_elapsed / days_total * 100, 1),
|
|
"as_of": now.isoformat(),
|
|
"stats": {
|
|
"sr": sr_stats,
|
|
"security": security_stats,
|
|
"capacity": capacity_stats,
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/{report_id}")
|
|
async def get_report(
|
|
report_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""리포트 상세 조회."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "PM/ADMIN 권한이 필요합니다.")
|
|
|
|
r = _reports.get(report_id)
|
|
if not r:
|
|
raise HTTPException(404, f"리포트 {report_id}를 찾을 수 없습니다.")
|
|
return r
|
|
|
|
|
|
@router.get("/export/{report_id}")
|
|
async def export_report(
|
|
report_id: str,
|
|
current_user: User = Depends(get_current_user),
|
|
_db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""리포트 JSON 다운로드 (ADMIN 전용)."""
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(403, "ADMIN 권한이 필요합니다.")
|
|
|
|
r = _reports.get(report_id)
|
|
if not r:
|
|
raise HTTPException(404, f"리포트 {report_id}를 찾을 수 없습니다.")
|
|
|
|
import json
|
|
content = json.dumps(r, ensure_ascii=False, indent=2, default=str)
|
|
filename = f"guardia_report_{r['year']}{r['month']:02d}.json"
|
|
|
|
return StreamingResponse(
|
|
iter([content]),
|
|
media_type="application/json",
|
|
headers={"Content-Disposition": f"attachment; filename={filename}"},
|
|
)
|
|
|
|
|
|
@router.post("/schedule")
|
|
async def set_schedule(
|
|
body: ScheduleConfigIn,
|
|
current_user: User = Depends(get_current_user),
|
|
_db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""리포트 자동 발송 스케줄 설정."""
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(403, "ADMIN 권한이 필요합니다.")
|
|
if not (1 <= body.send_day <= 28):
|
|
raise HTTPException(400, "send_day는 1~28 사이여야 합니다.")
|
|
|
|
logger.info("리포트 스케줄 설정: 매월 %d일, 수신자 %d명 by %s",
|
|
body.send_day, len(body.recipients), current_user.username)
|
|
return {
|
|
"message": "스케줄이 설정되었습니다.",
|
|
"enabled": body.enabled,
|
|
"send_day": body.send_day,
|
|
"recipients": len(body.recipients),
|
|
"include_llm": body.include_llm,
|
|
"note": "매월 지정일 00:00 UTC에 자동으로 리포트가 생성됩니다.",
|
|
}
|