mingle-website/bot/discord_cleanup.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
밍글 스튜디오 Discord 문의 자동 삭제 봇
- 지정된 채널의 메시지를 7일 후 자동 삭제
- 개인정보보호법(PIPA) 준수를 위한 데이터 보관기간 관리
"""

import os
import sys
import json
import urllib.request
import ssl
import time
from datetime import datetime, timedelta, timezone

# ─── 설정 ───────────────────────────────────────
BOT_TOKEN = ''
CHANNEL_ID = ''
DELETE_AFTER_DAYS = 7
CHECK_INTERVAL_HOURS = 6  # 삭제 체크 주기 (시간)

def load_env():
    """Load .env file from parent directory"""
    global BOT_TOKEN, CHANNEL_ID, DELETE_AFTER_DAYS, CHECK_INTERVAL_HOURS

    env_paths = [
        os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '.env'),
        os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env'),
    ]

    for env_path in env_paths:
        if os.path.exists(env_path):
            with open(env_path, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        key, value = line.split('=', 1)
                        key = key.strip()
                        value = value.strip().strip('"').strip("'")
                        if key == 'DISCORD_BOT_TOKEN':
                            BOT_TOKEN = value
                        elif key == 'DISCORD_INQUIRY_CHANNEL_ID':
                            CHANNEL_ID = value
                        elif key == 'DELETE_AFTER_DAYS':
                            DELETE_AFTER_DAYS = int(value)
                        elif key == 'CHECK_INTERVAL_HOURS':
                            CHECK_INTERVAL_HOURS = int(value)
            break

def discord_api(method, endpoint, data=None):
    """Discord API 호출"""
    url = f'https://discord.com/api/v10{endpoint}'
    headers = {
        'Authorization': f'Bot {BOT_TOKEN}',
        'Content-Type': 'application/json',
    }

    body = json.dumps(data).encode('utf-8') if data else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)

    ctx = ssl.create_default_context()
    try:
        with urllib.request.urlopen(req, context=ctx, timeout=15) as resp:
            if resp.status == 204:
                return None
            return json.loads(resp.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        error_body = e.read().decode('utf-8') if e.fp else ''
        print(f"[ERROR] Discord API {method} {endpoint}: {e.code} {error_body}")

        # Rate limit 처리
        if e.code == 429:
            retry_data = json.loads(error_body) if error_body else {}
            retry_after = retry_data.get('retry_after', 5)
            print(f"[RATE LIMIT] Waiting {retry_after}s...")
            time.sleep(retry_after + 0.5)
            return discord_api(method, endpoint, data)

        raise

def get_old_messages():
    """지정 채널에서 만료된 메시지 조회"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=DELETE_AFTER_DAYS)
    old_messages = []
    last_id = None

    while True:
        params = f'?limit=100'
        if last_id:
            params += f'&before={last_id}'

        messages = discord_api('GET', f'/channels/{CHANNEL_ID}/messages{params}')

        if not messages:
            break

        for msg in messages:
            # Discord 타임스탬프 파싱
            msg_time = datetime.fromisoformat(msg['timestamp'].replace('+00:00', '+00:00'))
            if msg_time.tzinfo is None:
                msg_time = msg_time.replace(tzinfo=timezone.utc)

            if msg_time < cutoff:
                old_messages.append({
                    'id': msg['id'],
                    'timestamp': msg['timestamp'],
                    'content': msg.get('content', '')[:50]
                })

        last_id = messages[-1]['id']

        # 가장 오래된 메시지도 만료 기간 내라면 더 조회할 필요 없음
        oldest_time = datetime.fromisoformat(messages[-1]['timestamp'].replace('+00:00', '+00:00'))
        if oldest_time.tzinfo is None:
            oldest_time = oldest_time.replace(tzinfo=timezone.utc)
        if oldest_time >= cutoff:
            continue
        else:
            break

        time.sleep(0.5)  # Rate limit 방지

    return old_messages

def delete_messages(messages):
    """메시지 삭제"""
    deleted = 0
    for msg in messages:
        try:
            discord_api('DELETE', f'/channels/{CHANNEL_ID}/messages/{msg["id"]}')
            print(f"  [DELETE] {msg['id']} ({msg['timestamp'][:10]})")
            deleted += 1
            time.sleep(0.5)  # Rate limit 방지
        except Exception as e:
            print(f"  [ERROR] Failed to delete {msg['id']}: {e}")
    return deleted

def run_cleanup():
    """한 번의 정리 사이클 실행"""
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"\n[{now}] Cleanup check started (delete after {DELETE_AFTER_DAYS} days)")

    try:
        messages = get_old_messages()

        if messages:
            print(f"  Found {len(messages)} expired message(s)")
            deleted = delete_messages(messages)
            print(f"  Deleted {deleted}/{len(messages)} message(s)")
        else:
            print(f"  No expired messages found")

    except Exception as e:
        print(f"  [ERROR] Cleanup failed: {e}")

def main():
    """메인 실행"""
    load_env()

    if not BOT_TOKEN:
        print("Error: DISCORD_BOT_TOKEN not set in .env")
        sys.exit(1)
    if not CHANNEL_ID:
        print("Error: DISCORD_CHANNEL_ID not set in .env")
        sys.exit(1)

    print("Mingle Studio Discord Cleanup Bot")
    print("=" * 40)
    print(f"Channel ID: {CHANNEL_ID}")
    print(f"Delete after: {DELETE_AFTER_DAYS} days")
    print(f"Check interval: {CHECK_INTERVAL_HOURS} hours")
    print("=" * 40)

    # 봇 정보 확인
    try:
        bot_info = discord_api('GET', '/users/@me')
        print(f"Bot: {bot_info['username']}#{bot_info.get('discriminator', '0')}")
    except Exception as e:
        print(f"Error: Failed to connect to Discord: {e}")
        sys.exit(1)

    print("Bot started. Press Ctrl+C to stop.\n")

    try:
        while True:
            run_cleanup()
            print(f"  Next check in {CHECK_INTERVAL_HOURS} hours...")
            time.sleep(CHECK_INTERVAL_HOURS * 3600)
    except KeyboardInterrupt:
        print("\nBot stopped.")


if __name__ == '__main__':
    main()