SQLite로 상태와 대화 기록 영구화하기

-- 1. WAL 모드 활성화 (읽기와 쓰기가 서로를 블로킹하지 않음)
PRAGMA journal_mode = WAL;
 
-- 2. 동기화 수준 최적화 (성능과 내구성의 최적 균형)
PRAGMA synchronous = NORMAL;
 
-- 3. 메모리 캐시 크기 증설 (약 64MB 할당)
PRAGMA cache_size = -64000;
 
-- 4. 외래 키 제약 조건 활성화
PRAGMA foreign_keys = ON;

aiosqlite>=0.20.0

import os
import logging
from pathlib import Path
import aiosqlite
 
logger = logging.getLogger("Database")
 
# DashCloud 영구 볼륨 디렉터리 경로 설정
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
DATA_DIR.mkdir(parents=True, exist_ok=True)
DB_PATH = DATA_DIR / "bot_database.db"
 
class DatabaseManager:
    def __init__(self, db_path: Path = DB_PATH):
        self.db_path = db_path
 
    async def get_connection(self) -> aiosqlite.Connection:
        """연결 풀 생성 및 WAL 모드 자동 주입"""
        conn = await aiosqlite.connect(self.db_path, timeout=30.0)
        conn.row_factory = aiosqlite.Row  # 딕셔너리처럼 컬럼명으로 접근 가능
        await conn.execute("PRAGMA journal_mode = WAL;")
        await conn.execute("PRAGMA synchronous = NORMAL;")
        return conn
 
    async def init_tables(self):
        """테이블 스키마 생성 및 초기화"""
        async with await self.get_connection() as conn:
            # 1. 사용자 정보 테이블
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS users (
                    user_id INTEGER PRIMARY KEY,
                    username TEXT,
                    first_name TEXT,
                    points INTEGER DEFAULT 100,
                    is_admin INTEGER DEFAULT 0,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
            """)
 
            # 2. 대화 기록 테이블 (대화형 AI 챗봇 메모리용)
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS chat_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER,
                    role TEXT CHECK(role IN ('user', 'assistant')),
                    content TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY(user_id) REFERENCES users(user_id) ON DELETE CASCADE
                );
            """)
            await conn.commit()
            logger.info("SQLite 테이블 초기화 완료 (WAL 모드 활성화)")
 
    async def upsert_user(self, user_id: int, username: str, first_name: str):
        """유저 접속 시 자동 등록 또는 방문 일시 갱신"""
        async with await self.get_connection() as conn:
            await conn.execute("""
                INSERT INTO users (user_id, username, first_name, last_seen)
                VALUES (?, ?, ?, CURRENT_TIMESTAMP)
                ON CONFLICT(user_id) DO UPDATE SET
                    username = excluded.username,
                    first_name = excluded.first_name,
                    last_seen = CURRENT_TIMESTAMP;
            """, (user_id, username, first_name))
            await conn.commit()
 
    async def get_user(self, user_id: int):
        """특정 유저 정보 조회"""
        async with await self.get_connection() as conn:
            async with conn.execute("SELECT * FROM users WHERE user_id = ?", (user_id,)) as cursor:
                return await cursor.fetchone()
 
# 싱글톤 인스턴스 생성
db = DatabaseManager()

import os
import sys
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from database import db
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
async def start_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    # 1. DB에 유저 정보 자동 기록 (영구 저장)
    await db.upsert_user(
        user_id=user.id,
        username=user.username or "",
        first_name=user.first_name
    )
    
    # 2. 유저 정보 조회
    user_data = await db.get_user(user.id)
    points = user_data["points"]
 
    await update.message.reply_text(
        f"반갑습니다 {user.first_name}님!\n"
        f"현재 보유 포인트: {points}P\n"
        f"회원 정보가 영구 디스크(/app/data)에 안전하게 보존됩니다."
    )
 
async def post_init(application: Application):
    """봇 기동 시 비동기 DB 테이블 초기화"""
    await db.init_tables()
 
def main():
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    if not token:
        logger.critical("TELEGRAM_BOT_TOKEN이 누락되었습니다.")
        sys.exit(1)
 
    app = (
        Application.builder()
        .token(token)
        .post_init(post_init)  # 테이블 자동 생성 훅 바인딩
        .build()
    )
 
    app.add_handler(CommandHandler("start", start_handler))
    app.run_polling(drop_pending_updates=True)
 
if __name__ == "__main__":
    main()

import sqlite3
import shutil
from datetime import datetime
from pathlib import Path
 
def create_online_backup():
    source_db = "/app/data/bot_database.db"
    backup_dir = Path("/app/data/backups")
    backup_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    target_file = backup_dir / f"backup_{timestamp}.db"
 
    # 잠금 없이 온라인 상태에서 안전하게 복제
    with sqlite3.connect(source_db) as src, sqlite3.connect(target_file) as dst:
        src.backup(dst)
    
    print(f"백업 완료: {target_file}")