import os
import smtplib
import httpx
import asyncio
import psycopg2
import psycopg2.extras
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
# ── DB 설정 ───────────────────────────────────────────────
DB_CONFIG = {
"host": os.getenv("DB_HOST", "postgres-service"),
"port": int(os.getenv("DB_PORT", "5432")),
"database": os.getenv("DB_NAME", "portaldb"),
"user": os.getenv("DB_USER", "portaluser"),
"password": os.getenv("DB_PASSWORD", "portalpass"),
}
# ── 환경변수 fallback ─────────────────────────────────────
DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "")
GMAIL_USER = os.getenv("GMAIL_USER", "")
GMAIL_APP_PASSWORD = os.getenv("GMAIL_APP_PASSWORD", "")
ALERT_EMAIL_TO = os.getenv("ALERT_EMAIL_TO", "")
# ── DB에서 활성 채널 목록 조회 ────────────────────────────
def get_active_channels():
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute("SELECT * FROM notify_channels WHERE enabled=TRUE ORDER BY id")
rows = cur.fetchall()
conn.close()
if rows:
return [dict(r) for r in rows]
except Exception as e:
print(f"[NOTIFIER] DB channel load failed: {e}")
# fallback: 환경변수로 단일 채널 구성
if DISCORD_WEBHOOK_URL or GMAIL_USER:
return [{
"id": 0, "name": "default", "type": "both",
"discord_webhook_url": DISCORD_WEBHOOK_URL,
"gmail_user": GMAIL_USER,
"gmail_app_password": GMAIL_APP_PASSWORD,
"alert_email_to": ALERT_EMAIL_TO,
"enabled": True,
}]
return []
# ── Discord 단건 발송 ─────────────────────────────────────
async def _send_discord(webhook_url: str, title: str, message: str, color: int):
if not webhook_url:
return
payload = {
"embeds": [{
"title": title,
"description": message,
"color": color,
"footer": {"text": "Web Portal Monitor"},
"timestamp": datetime.utcnow().isoformat()
}]
}
try:
async with httpx.AsyncClient() as client:
res = await client.post(webhook_url, json=payload, timeout=10)
if res.status_code not in (200, 204):
print(f"[NOTIFIER] Discord error: {res.status_code} {res.text}")
else:
print(f"[NOTIFIER] Discord sent: {title}")
except Exception as e:
print(f"[NOTIFIER] Discord exception: {e}")
# ── Gmail 단건 발송 ───────────────────────────────────────
def _send_email(gmail_user: str, gmail_app_password: str, alert_email_to: str,
subject: str, body: str):
if not all([gmail_user, gmail_app_password, alert_email_to]):
print("[NOTIFIER] Gmail config not set, skipping")
return
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[Web Portal] {subject}"
msg["From"] = gmail_user
msg["To"] = alert_email_to
html = f"""
⚠️ {subject}
{body}
Web Portal Monitor · {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
msg.attach(MIMEText(html, "html"))
with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
smtp.login(gmail_user, gmail_app_password)
smtp.sendmail(gmail_user, alert_email_to, msg.as_string())
print(f"[NOTIFIER] Email sent: {subject}")
except Exception as e:
print(f"[NOTIFIER] Email exception: {e}")
# ── 채널 유형별 발송 로직 ─────────────────────────────────
async def _dispatch_channel(ch: dict, title: str, message: str, color: int, mode: str):
ch_type = ch.get("type", "both")
send_discord_flag = (
mode in ("both", "discord") and
ch_type in ("both", "discord") and
ch.get("discord_webhook_url")
)
send_email_flag = (
mode in ("both", "email") and
ch_type in ("both", "email") and
ch.get("gmail_user")
)
if send_discord_flag:
await _send_discord(ch["discord_webhook_url"], title, message, color)
if send_email_flag:
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None, _send_email,
ch["gmail_user"], ch["gmail_app_password"],
ch["alert_email_to"], title, message
)
# ── 공개 알림 함수 ────────────────────────────────────────
async def notify_both(title: str, message: str, color: int = 0xe74c3c):
"""Discord + Gmail 타입 채널 모두에 발송 (Pod 이상/복구)"""
channels = get_active_channels()
if not channels:
print("[NOTIFIER] No active channels, skipping")
return
for ch in channels:
await _dispatch_channel(ch, title, message, color, mode="both")
async def notify_email_only(title: str, message: str, color: int = 0xf39c12):
"""Gmail 타입 채널에만 발송 (인증서 만료 임박)"""
channels = get_active_channels()
if not channels:
print("[NOTIFIER] No active channels, skipping")
return
for ch in channels:
await _dispatch_channel(ch, title, message, color, mode="email")
async def notify_discord_only(title: str, message: str, color: int = 0xe74c3c):
"""Discord 타입 채널에만 발송 (계정 잠금, 임시 비밀번호 발급)"""
channels = get_active_channels()
if not channels:
print("[NOTIFIER] No active channels, skipping")
return
for ch in channels:
await _dispatch_channel(ch, title, message, color, mode="discord")