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>
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
"""알림 발송 이력 조회 + 설정 확인 엔드포인트 (ADMIN/PM 전용)."""
|
|
import os
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select, desc
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import get_current_user
|
|
from database import get_db
|
|
from models import NotificationLog, NotificationLogOut, User, UserRole
|
|
|
|
router = APIRouter(prefix="/api/notifications", tags=["notifications"])
|
|
|
|
|
|
@router.get("/log", response_model=List[NotificationLogOut])
|
|
async def list_notification_log(
|
|
sr_id: Optional[str] = Query(None),
|
|
channel: Optional[str] = Query(None), # EMAIL | MESSENGER
|
|
status: Optional[str] = Query(None), # SENT | FAILED | SKIPPED
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
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 권한이 필요합니다.")
|
|
q = select(NotificationLog).order_by(desc(NotificationLog.sent_at))
|
|
if sr_id:
|
|
q = q.where(NotificationLog.sr_id == sr_id)
|
|
if channel:
|
|
q = q.where(NotificationLog.channel == channel)
|
|
if status:
|
|
q = q.where(NotificationLog.status == status)
|
|
q = q.offset(skip).limit(limit)
|
|
r = await db.execute(q)
|
|
return r.scalars().all()
|
|
|
|
|
|
@router.get("/config")
|
|
async def get_notify_config(current_user: User = Depends(get_current_user)):
|
|
"""현재 알림 설정 확인 (비밀번호는 마스킹)."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
|
|
def _e(k, d=""): return os.environ.get(k, d).strip()
|
|
def _b(k, d=True):
|
|
v = os.environ.get(k, "").strip().lower()
|
|
if v in ("1","true","yes","on"): return True
|
|
if v in ("0","false","no","off"): return False
|
|
return d
|
|
|
|
smtp_host = _e("SMTP_HOST")
|
|
smtp_pw = _e("SMTP_PASSWORD")
|
|
return {
|
|
"email": {
|
|
"enabled": bool(smtp_host),
|
|
"smtp_host": smtp_host or "(미설정)",
|
|
"smtp_port": _e("SMTP_PORT", "25"),
|
|
"smtp_user": _e("SMTP_USER") or "(없음)",
|
|
"smtp_password": ("*" * min(len(smtp_pw), 6)) if smtp_pw else "(없음)",
|
|
"smtp_from": _e("SMTP_FROM", "noreply@guardia.local"),
|
|
"smtp_tls": _b("SMTP_TLS", False),
|
|
"smtp_starttls": _b("SMTP_STARTTLS", False),
|
|
},
|
|
"messenger": {
|
|
"enabled": _b("MESSENGER_ENABLED", True),
|
|
"webhook": _e("MESSENGER_WEBHOOK", "http://localhost:8000/api/messenger/webhook"),
|
|
"ops_room": _e("MESSENGER_OPS_ROOM", "ops"),
|
|
},
|
|
"triggers": {
|
|
"notify_on_created": _b("NOTIFY_ON_CREATED", True),
|
|
"notify_statuses": _e("NOTIFY_STATUSES", "COMPLETED,REJECTED,FAILED_ROLLBACK"),
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/test-email")
|
|
async def test_email(
|
|
to: str = Query(..., description="테스트 수신 이메일"),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""이메일 발송 테스트 (ADMIN 전용)."""
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(403, "ADMIN 권한이 필요합니다.")
|
|
from core.notify import send_email, _html_email
|
|
subject = "[GUARDiA] 이메일 발송 테스트"
|
|
html = _html_email(
|
|
"발송 테스트",
|
|
f"<p>GUARDiA ITSM 이메일 알림이 정상 동작하고 있습니다.</p>"
|
|
f"<p>수신자: <strong>{to}</strong> | 발송자: {current_user.username}</p>",
|
|
)
|
|
ok, err = await send_email([to], subject, html)
|
|
return {"success": ok, "error": err or None}
|
|
|
|
|
|
@router.post("/test-messenger")
|
|
async def test_messenger(current_user: User = Depends(get_current_user)):
|
|
"""메신저 발송 테스트 (ADMIN 전용)."""
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(403, "ADMIN 권한이 필요합니다.")
|
|
import os
|
|
from core.notify import send_messenger
|
|
room = os.environ.get("MESSENGER_OPS_ROOM", "ops")
|
|
ok, err = await send_messenger(room, {
|
|
"event": "test",
|
|
"message": f"[GUARDiA 테스트] 메신저 알림 정상 동작 확인 — 발송자: {current_user.username}",
|
|
})
|
|
return {"success": ok, "error": err or None}
|