서론: 알림 자동화의 필요성

자동화 시스템의 완성도는 알림 기능에서 결정됩니다. 아무리 훌륭한 자동화 스크립트를 만들어도, 그 결과를 적시에 확인할 수 없다면 의미가 반감됩니다. 서버 장애, 배치 작업 완료, 일일 리포트 전송 등 다양한 상황에서 적절한 알림을 받는 것은 업무 효율성을 크게 높여줍니다.

이번 6편에서는 Python을 활용하여 다양한 채널로 알림을 보내는 방법을 배웁니다. 전통적인 이메일부터 슬랙, 텔레그램, 디스코드까지, 실무에서 가장 많이 사용하는 알림 방법들을 모두 다룹니다.

1. smtplib로 이메일 보내기

Python의 내장 라이브러리인 smtplib를 사용하면 별도의 설치 없이 이메일을 보낼 수 있습니다.

1.1 기본 이메일 발송

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_simple_email(sender, password, recipient, subject, body):
    """간단한 텍스트 이메일 발송"""
    # 메시지 객체 생성
    msg = MIMEMultipart()
    msg['From'] = sender
    msg['To'] = recipient
    msg['Subject'] = subject

    # 본문 추가
    msg.attach(MIMEText(body, 'plain', 'utf-8'))

    try:
        # Gmail SMTP 서버 연결
        server = smtplib.SMTP('smtp.gmail.com', 587)
        server.starttls()  # TLS 암호화
        server.login(sender, password)

        # 이메일 발송
        server.send_message(msg)
        server.quit()
        print("이메일 발송 성공!")
        return True
    except Exception as e:
        print(f"이메일 발송 실패: {e}")
        return False

# 사용 예시
send_simple_email(
    sender='your_email@gmail.com',
    password='앱_비밀번호',  # 2단계 인증 후 앱 비밀번호 사용
    recipient='recipient@example.com',
    subject='[자동화] 일일 리포트',
    body='오늘의 작업이 정상적으로 완료되었습니다.'
)

2. Gmail/네이버 SMTP 설정

2.1 Gmail SMTP 설정

Gmail을 사용하려면 먼저 앱 비밀번호를 생성해야 합니다.

1. Google 계정 설정 접속
2. 보안 탭에서 2단계 인증 활성화
3. 앱 비밀번호 생성 (기타 앱 선택)
4. 생성된 16자리 비밀번호를 코드에서 사용

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

class GmailSender:
    """Gmail SMTP를 사용한 이메일 발송 클래스"""

    def __init__(self, email, app_password):
        self.email = email
        self.password = app_password
        self.smtp_server = 'smtp.gmail.com'
        self.smtp_port = 587

    def send(self, to, subject, body, html=False):
        """이메일 발송"""
        msg = MIMEMultipart('alternative')
        msg['From'] = self.email
        msg['To'] = to if isinstance(to, str) else ', '.join(to)
        msg['Subject'] = subject

        # 텍스트 또는 HTML 본문
        content_type = 'html' if html else 'plain'
        msg.attach(MIMEText(body, content_type, 'utf-8'))

        with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
            server.starttls()
            server.login(self.email, self.password)

            # 여러 수신자에게 발송
            recipients = [to] if isinstance(to, str) else to
            server.send_message(msg)

        return True

# 사용 예시
gmail = GmailSender('your_email@gmail.com', '앱_비밀번호')
gmail.send(
    to='recipient@example.com',
    subject='테스트 메일',
    body='Python으로 보낸 이메일입니다.'
)

2.2 네이버 SMTP 설정

class NaverSender:
    """네이버 SMTP를 사용한 이메일 발송 클래스"""

    def __init__(self, email, password):
        self.email = email
        self.password = password
        self.smtp_server = 'smtp.naver.com'
        self.smtp_port = 587

    def send(self, to, subject, body, html=False):
        """이메일 발송"""
        msg = MIMEMultipart('alternative')
        msg['From'] = self.email
        msg['To'] = to if isinstance(to, str) else ', '.join(to)
        msg['Subject'] = subject

        content_type = 'html' if html else 'plain'
        msg.attach(MIMEText(body, content_type, 'utf-8'))

        with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
            server.starttls()
            server.login(self.email, self.password)
            server.send_message(msg)

        return True

# 네이버 메일 설정:
# 1. 네이버 메일 > 환경설정 > POP3/IMAP 설정
# 2. IMAP/SMTP 사용 체크
# 3. 실제 네이버 계정 비밀번호 사용 (앱 비밀번호 아님)

naver = NaverSender('your_id@naver.com', '네이버_비밀번호')
naver.send(
    to='recipient@example.com',
    subject='네이버 메일 테스트',
    body='네이버 SMTP로 보낸 이메일입니다.'
)

3. 이메일 본문 작성 (텍스트, HTML)

3.1 HTML 형식 이메일

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib

def send_html_email(sender, password, recipient, subject, data):
    """HTML 형식의 리포트 이메일 발송"""

    # HTML 템플릿
    html_template = """
    
    
    
        
    
    
        

            

                
{title}
                
{date}
            
            

                
요약
                
총 처리 건수: {total_count}
                
성공: {success_count} / 실패: {fail_count}

                
상세 내역
                
                    {table_rows}
                

                    

                        
항목
                        
상태
                        
처리 시간
                    
            
            

                
이 메일은 자동 발송되었습니다.
                
문의: admin@example.com
            
        
    
    
    """

    # 테이블 행 생성
    table_rows = ""
    for item in data['items']:
        status_class = 'status-ok' if item['status'] == '성공' else 'status-error'
        table_rows += f"""
        
            {item['name']}
            {item['status']}
            {item['time']}
        
        """

    # HTML 완성
    html_content = html_template.format(
        title=data['title'],
        date=data['date'],
        total_count=data['total_count'],
        success_count=data['success_count'],
        fail_count=data['fail_count'],
        table_rows=table_rows
    )

    # 메시지 구성
    msg = MIMEMultipart('alternative')
    msg['From'] = sender
    msg['To'] = recipient
    msg['Subject'] = subject

    # 텍스트 버전 (HTML 미지원 클라이언트용)
    text_content = f"""
    {data['title']}
    날짜: {data['date']}

    총 처리 건수: {data['total_count']}
    성공: {data['success_count']} / 실패: {data['fail_count']}
    """

    msg.attach(MIMEText(text_content, 'plain', 'utf-8'))
    msg.attach(MIMEText(html_content, 'html', 'utf-8'))

    # 발송
    with smtplib.SMTP('smtp.gmail.com', 587) as server:
        server.starttls()
        server.login(sender, password)
        server.send_message(msg)

    return True

# 사용 예시
report_data = {
    'title': '일일 배치 작업 리포트',
    'date': '2026-01-22',
    'total_count': 5,
    'success_count': 4,
    'fail_count': 1,
    'items': [
        {'name': '데이터 수집', 'status': '성공', 'time': '2.3초'},
        {'name': '데이터 변환', 'status': '성공', 'time': '5.1초'},
        {'name': '데이터 적재', 'status': '성공', 'time': '3.7초'},
        {'name': '백업 생성', 'status': '실패', 'time': '-'},
        {'name': '알림 발송', 'status': '성공', 'time': '0.5초'}
    ]
}

send_html_email(
    sender='your_email@gmail.com',
    password='앱_비밀번호',
    recipient='recipient@example.com',
    subject='[리포트] 일일 배치 작업 결과',
    data=report_data
)

4. 첨부파일 전송

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email.mime.application import MIMEApplication
from email import encoders
import smtplib
import os

def send_email_with_attachments(sender, password, recipient, subject, body, attachments):
    """첨부파일이 포함된 이메일 발송

    Args:
        attachments: 첨부할 파일 경로 리스트
    """
    msg = MIMEMultipart()
    msg['From'] = sender
    msg['To'] = recipient
    msg['Subject'] = subject

    # 본문 추가
    msg.attach(MIMEText(body, 'plain', 'utf-8'))

    # 첨부파일 추가
    for file_path in attachments:
        if not os.path.exists(file_path):
            print(f"파일을 찾을 수 없습니다: {file_path}")
            continue

        # 파일 읽기
        with open(file_path, 'rb') as f:
            file_data = f.read()

        # MIME 타입 결정
        file_name = os.path.basename(file_path)
        part = MIMEApplication(file_data, Name=file_name)

        # 헤더 추가
        part['Content-Disposition'] = f'attachment; filename="{file_name}"'
        msg.attach(part)

    # 발송
    with smtplib.SMTP('smtp.gmail.com', 587) as server:
        server.starttls()
        server.login(sender, password)
        server.send_message(msg)

    print(f"이메일 발송 완료 (첨부파일 {len(attachments)}개)")
    return True

# 사용 예시
send_email_with_attachments(
    sender='your_email@gmail.com',
    password='앱_비밀번호',
    recipient='recipient@example.com',
    subject='월별 리포트 첨부',
    body='안녕하세요,\n\n첨부된 월별 리포트를 확인해 주세요.\n\n감사합니다.',
    attachments=[
        'reports/월별리포트_202601.xlsx',
        'reports/차트_202601.png'
    ]
)

5. 이메일 수신 및 파싱 (imaplib)

import imaplib
import email
from email.header import decode_header
from datetime import datetime

class EmailReader:
    """IMAP을 사용한 이메일 수신 및 파싱 클래스"""

    def __init__(self, email_addr, password, imap_server='imap.gmail.com'):
        self.email = email_addr
        self.password = password
        self.imap_server = imap_server
        self.mail = None

    def connect(self):
        """IMAP 서버 연결"""
        self.mail = imaplib.IMAP4_SSL(self.imap_server)
        self.mail.login(self.email, self.password)
        return self

    def disconnect(self):
        """연결 종료"""
        if self.mail:
            self.mail.logout()

    def get_folders(self):
        """폴더 목록 조회"""
        status, folders = self.mail.list()
        return [f.decode().split(' "/" ')[-1].strip('"') for f in folders]

    def search_emails(self, folder='INBOX', criteria='ALL', limit=10):
        """이메일 검색

        criteria 예시:
        - 'ALL': 모든 메일
        - 'UNSEEN': 읽지 않은 메일
        - 'FROM "sender@example.com"': 특정 발신자
        - 'SUBJECT "키워드"': 제목에 키워드 포함
        - 'SINCE "01-Jan-2026"': 특정 날짜 이후
        """
        self.mail.select(folder)
        status, messages = self.mail.search(None, criteria)

        email_ids = messages[0].split()
        # 최신순으로 정렬 후 limit 적용
        email_ids = email_ids[-limit:][::-1]

        emails = []
        for email_id in email_ids:
            email_data = self._fetch_email(email_id)
            if email_data:
                emails.append(email_data)

        return emails

    def _fetch_email(self, email_id):
        """개별 이메일 가져오기"""
        status, data = self.mail.fetch(email_id, '(RFC822)')

        if status != 'OK':
            return None

        raw_email = data[0][1]
        msg = email.message_from_bytes(raw_email)

        # 헤더 디코딩
        subject = self._decode_header(msg['Subject'])
        from_addr = self._decode_header(msg['From'])
        date_str = msg['Date']

        # 본문 추출
        body = self._get_body(msg)

        # 첨부파일 정보
        attachments = self._get_attachments(msg)

        return {
            'id': email_id.decode(),
            'subject': subject,
            'from': from_addr,
            'date': date_str,
            'body': body,
            'attachments': attachments
        }

    def _decode_header(self, header):
        """헤더 디코딩"""
        if header is None:
            return ""

        decoded_parts = decode_header(header)
        result = []
        for content, encoding in decoded_parts:
            if isinstance(content, bytes):
                content = content.decode(encoding or 'utf-8', errors='replace')
            result.append(content)
        return ''.join(result)

    def _get_body(self, msg):
        """본문 추출"""
        body = ""

        if msg.is_multipart():
            for part in msg.walk():
                content_type = part.get_content_type()
                content_disposition = str(part.get('Content-Disposition'))

                if content_type == 'text/plain' and 'attachment' not in content_disposition:
                    payload = part.get_payload(decode=True)
                    charset = part.get_content_charset() or 'utf-8'
                    body = payload.decode(charset, errors='replace')
                    break
        else:
            payload = msg.get_payload(decode=True)
            charset = msg.get_content_charset() or 'utf-8'
            body = payload.decode(charset, errors='replace')

        return body

    def _get_attachments(self, msg):
        """첨부파일 정보 추출"""
        attachments = []

        for part in msg.walk():
            if part.get_content_maintype() == 'multipart':
                continue

            filename = part.get_filename()
            if filename:
                filename = self._decode_header(filename)
                attachments.append({
                    'filename': filename,
                    'content_type': part.get_content_type(),
                    'size': len(part.get_payload(decode=True) or b'')
                })

        return attachments

    def download_attachment(self, email_id, attachment_index, save_path):
        """첨부파일 다운로드"""
        status, data = self.mail.fetch(email_id.encode(), '(RFC822)')
        msg = email.message_from_bytes(data[0][1])

        current_index = 0
        for part in msg.walk():
            if part.get_content_maintype() == 'multipart':
                continue

            filename = part.get_filename()
            if filename:
                if current_index == attachment_index:
                    payload = part.get_payload(decode=True)
                    with open(save_path, 'wb') as f:
                        f.write(payload)
                    return True
                current_index += 1

        return False

# 사용 예시
reader = EmailReader('your_email@gmail.com', '앱_비밀번호')
reader.connect()

# 읽지 않은 메일 조회
unread_emails = reader.search_emails(criteria='UNSEEN', limit=5)
for mail in unread_emails:
    print(f"제목: {mail['subject']}")
    print(f"발신자: {mail['from']}")
    print(f"날짜: {mail['date']}")
    print(f"첨부파일: {len(mail['attachments'])}개")
    print("-" * 50)

# 특정 발신자의 메일 검색
from_emails = reader.search_emails(
    criteria='FROM "important@example.com"',
    limit=10
)

reader.disconnect()

6. 슬랙 알림 보내기 (Webhook)

슬랙 웹훅은 특별한 라이브러리 없이 HTTP 요청만으로 메시지를 보낼 수 있어 매우 간편합니다.

6.1 웹훅 URL 생성

1. Slack 앱 페이지 접속 (api.slack.com/apps)
2. Create New App 클릭
3. From scratch 선택 후 앱 이름과 워크스페이스 지정
4. Incoming Webhooks 활성화
5. Add New Webhook to Workspace 클릭
6. 채널 선택 후 웹훅 URL 복사

import requests
import json
from datetime import datetime

class SlackNotifier:
    """슬랙 웹훅을 사용한 알림 클래스"""

    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send_simple(self, message):
        """간단한 텍스트 메시지 전송"""
        payload = {'text': message}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code == 200

    def send_formatted(self, title, message, color='#36a64f', fields=None):
        """포맷된 메시지 전송 (attachment 사용)"""
        attachment = {
            'fallback': title,
            'color': color,
            'title': title,
            'text': message,
            'ts': datetime.now().timestamp()
        }

        if fields:
            attachment['fields'] = [
                {'title': k, 'value': str(v), 'short': True}
                for k, v in fields.items()
            ]

        payload = {'attachments': [attachment]}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code == 200

    def send_blocks(self, blocks):
        """Block Kit을 사용한 리치 메시지 전송"""
        payload = {'blocks': blocks}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code == 200

    def send_alert(self, title, message, status='success'):
        """상태별 알림 전송"""
        colors = {
            'success': '#36a64f',  # 녹색
            'warning': '#ffcc00',  # 노란색
            'error': '#ff0000',    # 빨간색
            'info': '#3498db'      # 파란색
        }

        emojis = {
            'success': ':white_check_mark:',
            'warning': ':warning:',
            'error': ':x:',
            'info': ':information_source:'
        }

        blocks = [
            {
                'type': 'header',
                'text': {
                    'type': 'plain_text',
                    'text': f"{emojis.get(status, '')} {title}"
                }
            },
            {
                'type': 'section',
                'text': {
                    'type': 'mrkdwn',
                    'text': message
                }
            },
            {
                'type': 'context',
                'elements': [
                    {
                        'type': 'mrkdwn',
                        'text': f"발생 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
                    }
                ]
            }
        ]

        return self.send_blocks(blocks)

# 사용 예시
slack = SlackNotifier('https://hooks.slack.com/services/xxx/yyy/zzz')

# 간단한 메시지
slack.send_simple('배치 작업이 완료되었습니다.')

# 포맷된 메시지
slack.send_formatted(
    title='일일 리포트',
    message='오늘의 판매 현황입니다.',
    color='#3498db',
    fields={
        '총 주문': '150건',
        '총 매출': '15,000,000원',
        '신규 가입': '25명',
        '환불 건수': '3건'
    }
)

# 에러 알림
slack.send_alert(
    title='서버 오류 감지',
    message='*DB 연결 실패*\n```ConnectionError: Unable to connect to database```',
    status='error'
)

7. 텔레그램 봇 만들기

7.1 봇 생성 및 설정

1. 텔레그램에서 @BotFather 검색
2. /newbot 명령어 입력
3. 봇 이름과 username 설정
4. 발급받은 토큰 저장
5. 생성된 봇과 대화를 시작하고 chat_id 확인

import requests
from datetime import datetime

class TelegramBot:
    """텔레그램 봇 알림 클래스"""

    def __init__(self, token):
        self.token = token
        self.base_url = f'https://api.telegram.org/bot{token}'

    def get_updates(self):
        """최근 메시지 조회 (chat_id 확인용)"""
        response = requests.get(f'{self.base_url}/getUpdates')
        return response.json()

    def send_message(self, chat_id, text, parse_mode='HTML'):
        """메시지 전송

        parse_mode: 'HTML' 또는 'Markdown'
        """
        params = {
            'chat_id': chat_id,
            'text': text,
            'parse_mode': parse_mode
        }
        response = requests.post(f'{self.base_url}/sendMessage', data=params)
        return response.json()

    def send_photo(self, chat_id, photo_path, caption=''):
        """이미지 전송"""
        with open(photo_path, 'rb') as photo:
            params = {'chat_id': chat_id, 'caption': caption}
            files = {'photo': photo}
            response = requests.post(
                f'{self.base_url}/sendPhoto',
                data=params,
                files=files
            )
        return response.json()

    def send_document(self, chat_id, file_path, caption=''):
        """파일 전송"""
        with open(file_path, 'rb') as doc:
            params = {'chat_id': chat_id, 'caption': caption}
            files = {'document': doc}
            response = requests.post(
                f'{self.base_url}/sendDocument',
                data=params,
                files=files
            )
        return response.json()

    def send_alert(self, chat_id, title, message, status='info'):
        """포맷된 알림 전송"""
        emojis = {
            'success': '✅',
            'warning': '⚠️',
            'error': '❌',
            'info': 'ℹ️'
        }

        html_message = f"""
{emojis.get(status, '')} {title}

{message}

시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
        return self.send_message(chat_id, html_message)

    def send_report(self, chat_id, title, data):
        """리포트 형식 알림"""
        lines = [f"📊 {title}\n"]

        for key, value in data.items():
            lines.append(f"• {key}: {value}")

        lines.append(f"\n생성: {datetime.now().strftime('%Y-%m-%d %H:%M')}")

        return self.send_message(chat_id, '\n'.join(lines))

# 사용 예시
bot = TelegramBot('YOUR_BOT_TOKEN')

# chat_id 확인 (처음 한 번만)
# 봇에게 메시지를 보낸 후 실행
# updates = bot.get_updates()
# print(updates)

chat_id = 'YOUR_CHAT_ID'

# 간단한 메시지
bot.send_message(chat_id, '안녕하세요! 텔레그램 봇입니다.')

# 알림
bot.send_alert(
    chat_id,
    '배치 작업 완료',
    '일일 데이터 수집이 정상적으로 완료되었습니다.',
    status='success'
)

# 리포트
bot.send_report(
    chat_id,
    '일일 판매 현황',
    {
        '총 주문': '150건',
        '총 매출': '15,000,000원',
        '평균 객단가': '100,000원',
        '신규 고객': '25명'
    }
)

# 파일 전송
bot.send_document(chat_id, 'reports/daily_report.xlsx', '일일 리포트 파일')

8. 디스코드 웹훅

import requests
import json
from datetime import datetime

class DiscordNotifier:
    """디스코드 웹훅을 사용한 알림 클래스"""

    def __init__(self, webhook_url):
        self.webhook_url = webhook_url

    def send_simple(self, content):
        """간단한 메시지 전송"""
        payload = {'content': content}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code in [200, 204]

    def send_embed(self, title, description, color=0x3498db, fields=None, footer=None):
        """임베드 메시지 전송"""
        embed = {
            'title': title,
            'description': description,
            'color': color,
            'timestamp': datetime.utcnow().isoformat()
        }

        if fields:
            embed['fields'] = [
                {'name': k, 'value': str(v), 'inline': True}
                for k, v in fields.items()
            ]

        if footer:
            embed['footer'] = {'text': footer}

        payload = {'embeds': [embed]}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code in [200, 204]

    def send_alert(self, title, message, status='info'):
        """상태별 알림 전송"""
        colors = {
            'success': 0x2ecc71,  # 녹색
            'warning': 0xf39c12,  # 주황색
            'error': 0xe74c3c,    # 빨간색
            'info': 0x3498db      # 파란색
        }

        emojis = {
            'success': ':white_check_mark:',
            'warning': ':warning:',
            'error': ':x:',
            'info': ':information_source:'
        }

        embed = {
            'title': f"{emojis.get(status, '')} {title}",
            'description': message,
            'color': colors.get(status, 0x3498db),
            'timestamp': datetime.utcnow().isoformat(),
            'footer': {'text': '자동화 알림 시스템'}
        }

        payload = {'embeds': [embed]}
        response = requests.post(
            self.webhook_url,
            data=json.dumps(payload),
            headers={'Content-Type': 'application/json'}
        )
        return response.status_code in [200, 204]

    def send_file(self, file_path, content=''):
        """파일 전송"""
        with open(file_path, 'rb') as f:
            response = requests.post(
                self.webhook_url,
                data={'content': content},
                files={'file': f}
            )
        return response.status_code in [200, 204]

# 디스코드 웹훅 URL 생성:
# 1. 서버 설정 > 연동 > 웹후크
# 2. 새 웹후크 생성
# 3. 웹후크 URL 복사

discord = DiscordNotifier('https://discord.com/api/webhooks/xxx/yyy')

# 간단한 메시지
discord.send_simple('서버 상태 정상!')

# 임베드 메시지
discord.send_embed(
    title='일일 리포트',
    description='오늘의 시스템 현황입니다.',
    color=0x3498db,
    fields={
        'CPU 사용률': '45%',
        '메모리 사용률': '62%',
        '디스크 사용률': '78%',
        '활성 세션': '234개'
    },
    footer='모니터링 시스템'
)

# 에러 알림
discord.send_alert(
    title='데이터베이스 오류',
    message='```\nConnectionError: Unable to connect to MySQL server\n```',
    status='error'
)

9. 데스크톱 알림 (plyer)

# plyer 설치
# pip install plyer

from plyer import notification
import time

def send_desktop_notification(title, message, timeout=10, app_icon=None):
    """데스크톱 알림 전송

    Args:
        title: 알림 제목
        message: 알림 내용
        timeout: 알림 표시 시간 (초)
        app_icon: 아이콘 경로 (.ico 파일)
    """
    notification.notify(
        title=title,
        message=message,
        timeout=timeout,
        app_icon=app_icon,
        app_name='Python 자동화'
    )

# 사용 예시
send_desktop_notification(
    title='작업 완료',
    message='데이터 처리가 완료되었습니다.\n처리 건수: 1,234건',
    timeout=5
)

# 여러 단계의 작업 알림
def process_with_notifications():
    """작업 진행 상황을 데스크톱 알림으로 표시"""

    send_desktop_notification('작업 시작', '데이터 수집을 시작합니다.')
    time.sleep(2)  # 작업 시뮬레이션

    send_desktop_notification('진행 중', '데이터 변환 중... (50%)')
    time.sleep(2)

    send_desktop_notification('완료', '모든 작업이 완료되었습니다!')

# Windows 토스트 알림 (더 많은 기능)
try:
    from win10toast import ToastNotifier

    toaster = ToastNotifier()

    def send_windows_toast(title, message, duration=5, icon_path=None):
        """Windows 토스트 알림"""
        toaster.show_toast(
            title,
            message,
            icon_path=icon_path,
            duration=duration,
            threaded=True  # 비동기 실행
        )

except ImportError:
    pass  # Windows가 아니거나 win10toast 미설치

10. 실전: 모니터링 알림 시스템

지금까지 배운 내용을 종합하여 서버 모니터링 알림 시스템을 만들어 보겠습니다.

import psutil
import requests
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from datetime import datetime
import time
import json
import logging

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('monitor.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class NotificationManager:
    """통합 알림 관리 클래스"""

    def __init__(self, config):
        self.config = config
        self.last_alert_time = {}

    def send_email(self, subject, body, html=False):
        """이메일 발송"""
        try:
            cfg = self.config['email']
            msg = MIMEMultipart('alternative')
            msg['From'] = cfg['sender']
            msg['To'] = cfg['recipient']
            msg['Subject'] = subject

            content_type = 'html' if html else 'plain'
            msg.attach(MIMEText(body, content_type, 'utf-8'))

            with smtplib.SMTP(cfg['smtp_server'], cfg['smtp_port']) as server:
                server.starttls()
                server.login(cfg['sender'], cfg['password'])
                server.send_message(msg)

            logger.info(f"이메일 발송 완료: {subject}")
            return True
        except Exception as e:
            logger.error(f"이메일 발송 실패: {e}")
            return False

    def send_slack(self, message, status='info'):
        """슬랙 알림"""
        try:
            webhook_url = self.config['slack']['webhook_url']

            colors = {
                'success': '#36a64f',
                'warning': '#ffcc00',
                'error': '#ff0000',
                'info': '#3498db'
            }

            payload = {
                'attachments': [{
                    'color': colors.get(status, '#3498db'),
                    'text': message,
                    'ts': datetime.now().timestamp()
                }]
            }

            response = requests.post(
                webhook_url,
                data=json.dumps(payload),
                headers={'Content-Type': 'application/json'}
            )

            if response.status_code == 200:
                logger.info(f"슬랙 알림 전송 완료")
                return True
            return False
        except Exception as e:
            logger.error(f"슬랙 알림 실패: {e}")
            return False

    def send_telegram(self, message):
        """텔레그램 알림"""
        try:
            cfg = self.config['telegram']
            url = f"https://api.telegram.org/bot{cfg['token']}/sendMessage"
            params = {
                'chat_id': cfg['chat_id'],
                'text': message,
                'parse_mode': 'HTML'
            }
            response = requests.post(url, data=params)

            if response.status_code == 200:
                logger.info("텔레그램 알림 전송 완료")
                return True
            return False
        except Exception as e:
            logger.error(f"텔레그램 알림 실패: {e}")
            return False

    def send_all(self, title, message, status='info', html_body=None):
        """모든 채널로 알림 전송"""
        # 이메일
        if self.config.get('email', {}).get('enabled'):
            self.send_email(title, html_body or message, html=bool(html_body))

        # 슬랙
        if self.config.get('slack', {}).get('enabled'):
            self.send_slack(f"*{title}*\n{message}", status)

        # 텔레그램
        if self.config.get('telegram', {}).get('enabled'):
            self.send_telegram(f"{title}\n\n{message}")

    def can_alert(self, alert_type, cooldown_seconds=300):
        """알림 쿨다운 체크 (동일 알림 연속 방지)"""
        now = datetime.now()
        last_time = self.last_alert_time.get(alert_type)

        if last_time and (now - last_time).seconds < cooldown_seconds:
            return False

        self.last_alert_time[alert_type] = now
        return True


class SystemMonitor:
    """시스템 모니터링 클래스"""

    def __init__(self, notification_manager, thresholds):
        self.notifier = notification_manager
        self.thresholds = thresholds

    def get_system_stats(self):
        """시스템 상태 수집"""
        return {
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'network': psutil.net_io_counters(),
            'timestamp': datetime.now()
        }

    def check_thresholds(self, stats):
        """임계값 체크 및 알림"""
        alerts = []

        # CPU 체크
        if stats['cpu_percent'] > self.thresholds['cpu_critical']:
            alerts.append(('CPU 위험', f"CPU 사용률: {stats['cpu_percent']}%", 'error'))
        elif stats['cpu_percent'] > self.thresholds['cpu_warning']:
            alerts.append(('CPU 경고', f"CPU 사용률: {stats['cpu_percent']}%", 'warning'))

        # 메모리 체크
        if stats['memory_percent'] > self.thresholds['memory_critical']:
            alerts.append(('메모리 위험', f"메모리 사용률: {stats['memory_percent']}%", 'error'))
        elif stats['memory_percent'] > self.thresholds['memory_warning']:
            alerts.append(('메모리 경고', f"메모리 사용률: {stats['memory_percent']}%", 'warning'))

        # 디스크 체크
        if stats['disk_percent'] > self.thresholds['disk_critical']:
            alerts.append(('디스크 위험', f"디스크 사용률: {stats['disk_percent']}%", 'error'))
        elif stats['disk_percent'] > self.thresholds['disk_warning']:
            alerts.append(('디스크 경고', f"디스크 사용률: {stats['disk_percent']}%", 'warning'))

        return alerts

    def send_daily_report(self, stats):
        """일일 리포트 발송"""
        title = f"[일일 리포트] 시스템 현황 - {datetime.now().strftime('%Y-%m-%d')}"

        message = f"""
시스템 현황 리포트

- CPU 사용률: {stats['cpu_percent']}%
- 메모리 사용률: {stats['memory_percent']}%
- 디스크 사용률: {stats['disk_percent']}%

리포트 생성 시간: {stats['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}
"""

        html_body = f"""


    
시스템 현황 리포트
    

        

            
항목
            
사용률
            
상태
        
        

            
CPU
            
{stats['cpu_percent']}%
            
{'정상' if stats['cpu_percent'] < 70 else '주의'}
        
        

            
메모리
            
{stats['memory_percent']}%
            
{'정상' if stats['memory_percent'] < 80 else '주의'}
        
        

            
디스크
            
{stats['disk_percent']}%
            
{'정상' if stats['disk_percent'] < 85 else '주의'}
        
    
    

        리포트 생성: {stats['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}
    


"""

        self.notifier.send_all(title, message, 'info', html_body)

    def run(self, interval=60):
        """모니터링 실행"""
        logger.info("모니터링 시작")

        last_report_date = None

        while True:
            try:
                # 시스템 상태 수집
                stats = self.get_system_stats()
                logger.info(f"CPU: {stats['cpu_percent']}%, MEM: {stats['memory_percent']}%, DISK: {stats['disk_percent']}%")

                # 임계값 체크
                alerts = self.check_thresholds(stats)
                for title, message, status in alerts:
                    if self.notifier.can_alert(title, cooldown_seconds=300):
                        self.notifier.send_all(title, message, status)

                # 일일 리포트 (오전 9시)
                now = datetime.now()
                if now.hour == 9 and now.date() != last_report_date:
                    self.send_daily_report(stats)
                    last_report_date = now.date()

                time.sleep(interval)

            except KeyboardInterrupt:
                logger.info("모니터링 종료")
                break
            except Exception as e:
                logger.error(f"모니터링 오류: {e}")
                time.sleep(interval)


# 설정 예시
config = {
    'email': {
        'enabled': True,
        'sender': 'your_email@gmail.com',
        'password': '앱_비밀번호',
        'recipient': 'admin@example.com',
        'smtp_server': 'smtp.gmail.com',
        'smtp_port': 587
    },
    'slack': {
        'enabled': True,
        'webhook_url': 'https://hooks.slack.com/services/xxx/yyy/zzz'
    },
    'telegram': {
        'enabled': True,
        'token': 'YOUR_BOT_TOKEN',
        'chat_id': 'YOUR_CHAT_ID'
    }
}

thresholds = {
    'cpu_warning': 70,
    'cpu_critical': 90,
    'memory_warning': 80,
    'memory_critical': 95,
    'disk_warning': 85,
    'disk_critical': 95
}

# 실행
if __name__ == "__main__":
    notifier = NotificationManager(config)
    monitor = SystemMonitor(notifier, thresholds)
    monitor.run(interval=60)  # 60초 간격으로 모니터링

마무리

이번 편에서는 Python을 활용한 다양한 알림 자동화 방법을 배웠습니다.

• 이메일: smtplib로 텍스트/HTML 이메일 및 첨부파일 전송
• 슬랙: 웹훅을 통한 간편한 팀 알림
• 텔레그램: 봇을 통한 개인/그룹 알림
• 디스코드: 웹훅을 통한 커뮤니티 알림
• 데스크톱: plyer를 통한 로컬 알림

실무에서는 상황에 맞는 채널을 선택하거나, 여러 채널을 조합하여 사용합니다. 긴급한 장애 알림은 텔레그램이나 슬랙으로, 정기 리포트는 이메일로 보내는 것이 일반적입니다.

이번 시리즈를 통해 Python 자동화의 핵심 기술들을 배웠습니다. 파일 처리, 웹 스크래핑, 데이터베이스 연동, 엑셀 자동화, 그리고 알림 시스템까지, 이 기술들을 조합하면 다양한 업무를 자동화할 수 있습니다. 배운 내용을 실무에 적용하여 반복 작업에서 벗어나시기 바랍니다.