""" restic 백업 디스코드 알림 봇 - 백업 성공/실패 시 각각 다른 채널로 알림 - HTTP 웹훅으로 알림 수신 """ import os import discord from discord.ext import commands from aiohttp import web import asyncio from datetime import datetime # 환경변수 BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN") SUCCESS_CHANNEL_ID = int(os.getenv("DISCORD_SUCCESS_CHANNEL_ID", "0")) FAILURE_CHANNEL_ID = int(os.getenv("DISCORD_FAILURE_CHANNEL_ID", "0")) WEBHOOK_PORT = int(os.getenv("WEBHOOK_PORT", "5001")) # 봇 설정 intents = discord.Intents.default() bot = commands.Bot(command_prefix="!", intents=intents) def format_size(size_bytes: int) -> str: """바이트를 읽기 쉬운 형식으로 변환""" for unit in ["B", "KB", "MB", "GB", "TB"]: if size_bytes < 1024: return f"{size_bytes:.2f} {unit}" size_bytes /= 1024 return f"{size_bytes:.2f} PB" @bot.event async def on_ready(): """봇 시작 시 실행""" print(f"✅ 봇 로그인: {bot.user}") try: synced = await bot.tree.sync() print(f"✅ {len(synced)}개 명령어 동기화 완료") except Exception as e: print(f"❌ 명령어 동기화 실패: {e}") async def send_success_notification(data: dict): """백업 성공 알림 전송""" if not SUCCESS_CHANNEL_ID: print("❌ SUCCESS_CHANNEL_ID가 설정되지 않음") return channel = bot.get_channel(SUCCESS_CHANNEL_ID) if not channel: print(f"❌ 채널을 찾을 수 없음: {SUCCESS_CHANNEL_ID}") return try: # 스냅샷 ID 축약 (앞 8자리만) snapshot_id = data.get('snapshot_id', 'N/A') if len(snapshot_id) > 8: snapshot_id = snapshot_id[:8] embed = discord.Embed( title="✅ 백업 완료", color=0x2ecc71, # 초록색 timestamp=datetime.now() ) # 통계 정보 stats = f"📁 **{data.get('files', '0')}** 파일 • 💾 **{data.get('size', 'N/A')}** • ⏱️ **{data.get('duration', 'N/A')}**" embed.description = stats # 백업 폴더 목록 (1024자 제한) folders = data.get('folders', []) if folders: folder_list = " ".join([f"`{f}`" for f in folders]) # 디스코드 필드 길이 제한 적용 if len(folder_list) > 1000: folder_list = folder_list[:1000] + "..." embed.add_field(name="📂 백업 대상", value=folder_list, inline=False) # 하단 정보 embed.set_footer(text=f"🆔 {snapshot_id} • 🖥️ {os.uname().nodename}") await channel.send(embed=embed) print(f"✅ 성공 알림 전송") except Exception as e: print(f"❌ 알림 전송 실패: {e}") async def send_failure_notification(data: dict): """백업 실패 알림 전송""" if not FAILURE_CHANNEL_ID: print("❌ FAILURE_CHANNEL_ID가 설정되지 않음") return channel = bot.get_channel(FAILURE_CHANNEL_ID) if not channel: print(f"❌ 채널을 찾을 수 없음: {FAILURE_CHANNEL_ID}") return try: error_msg = data.get('error', 'Unknown error') # 에러 메시지 길이 제한 if len(error_msg) > 500: error_msg = error_msg[:500] + "..." embed = discord.Embed( title="🛑 백업 실패", color=0xe74c3c, # 빨간색 timestamp=datetime.now() ) # 에러 내용 embed.description = f"```\n{error_msg}\n```" # 하단 정보 embed.set_footer(text=f"🖥️ {os.uname().nodename}") await channel.send(embed=embed) print(f"✅ 실패 알림 전송") except Exception as e: print(f"❌ 실패 알림 전송 오류: {e}") # HTTP 웹훅 핸들러 async def success_handler(request): """백업 성공 웹훅""" try: data = await request.json() await send_success_notification(data) return web.Response(text="OK") except Exception as e: print(f"❌ 성공 웹훅 오류: {e}") return web.Response(text="Error", status=500) async def failure_handler(request): """백업 실패 웹훅""" try: data = await request.json() await send_failure_notification(data) return web.Response(text="OK") except Exception as e: print(f"❌ 실패 웹훅 오류: {e}") return web.Response(text="Error", status=500) async def send_sync_success_notification(data: dict): """동기화 성공 알림 전송""" if not SUCCESS_CHANNEL_ID: return channel = bot.get_channel(SUCCESS_CHANNEL_ID) if not channel: return try: embed = discord.Embed( title="✅ 동기화 완료", color=0x3498db, # 파란색 timestamp=datetime.now() ) stats = f"💾 **{data.get('size', 'N/A')}** • ⏱️ **{data.get('duration', 'N/A')}**" embed.description = stats embed.add_field(name="📍 대상", value=f"`{data.get('target', 'N/A')}`", inline=False) embed.set_footer(text=f"🖥️ {os.uname().nodename}") await channel.send(embed=embed) print(f"✅ 동기화 성공 알림 전송") except Exception as e: print(f"❌ 동기화 성공 알림 전송 오류: {e}") async def send_sync_failure_notification(data: dict): """동기화 실패 알림 전송""" if not FAILURE_CHANNEL_ID: return channel = bot.get_channel(FAILURE_CHANNEL_ID) if not channel: return try: error_msg = data.get('error', 'Unknown error') if len(error_msg) > 500: error_msg = error_msg[:500] + "..." embed = discord.Embed( title="🛑 동기화 실패", color=0xe74c3c, timestamp=datetime.now() ) embed.description = f"```\n{error_msg}\n```" embed.add_field(name="📍 대상", value=f"`{data.get('target', 'N/A')}`", inline=False) embed.set_footer(text=f"🖥️ {os.uname().nodename}") await channel.send(embed=embed) print(f"✅ 동기화 실패 알림 전송") except Exception as e: print(f"❌ 동기화 실패 알림 전송 오류: {e}") async def sync_success_handler(request): """동기화 성공 웹훅""" try: data = await request.json() await send_sync_success_notification(data) return web.Response(text="OK") except Exception as e: print(f"❌ 동기화 성공 웹훅 오류: {e}") return web.Response(text="Error", status=500) async def sync_failure_handler(request): """동기화 실패 웹훅""" try: data = await request.json() await send_sync_failure_notification(data) return web.Response(text="OK") except Exception as e: print(f"❌ 동기화 실패 웹훅 오류: {e}") return web.Response(text="Error", status=500) async def start_webhook_server(): """웹훅 서버 시작""" app = web.Application() app.router.add_post("/success", success_handler) app.router.add_post("/failure", failure_handler) app.router.add_post("/sync-success", sync_success_handler) app.router.add_post("/sync-failure", sync_failure_handler) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, "0.0.0.0", WEBHOOK_PORT) await site.start() print(f"✅ 웹훅 서버 시작: http://0.0.0.0:{WEBHOOK_PORT}") print(f" - 백업 성공: POST /success") print(f" - 백업 실패: POST /failure") print(f" - 동기화 성공: POST /sync-success") print(f" - 동기화 실패: POST /sync-failure") async def main(): """메인 함수""" # 웹훅 서버 시작 await start_webhook_server() # 봇 시작 await bot.start(BOT_TOKEN) if __name__ == "__main__": asyncio.run(main())