SQLite 세션 관리자 - 범용 세션/항목 저장 패턴

"""
SQLite 세션 관리자 - 범용 세션/항목 저장 패턴

@source voice-to-text-v2
@extracted 2026-02-15
@version 1.0.0

의존성:
    - sqlite3 (표준 라이브러리)

사용법:
    from utils.sqlite_session import SQLiteSessionManager

    # 초기화 (DB 파일 경로 지정)
    manager = SQLiteSessionManager(db_path="./data/sessions.db")

    # 세션에 항목 추가 (세션 자동 생성)
    manager.add_entry(
        session_id="abc-123",
        metadata={"language": "ko", "model": "base"},
        entry_data={"text": "안녕하세요", "confidence": 0.95},
    )

    # 세션 목록 조회
    sessions = manager.get_sessions(limit=50)

    # 세션 상세 조회 (항목 포함)
    session = manager.get_session("abc-123")

    # 검색
    results = manager.search_sessions("검색어")

    # 싱글톤 사용
    manager = get_session_manager("./data/sessions.db")
"""

import logging
import sqlite3
import uuid
from datetime import datetime
from pathlib import Path

logger = logging.getLogger(__name__)


class SQLiteSessionManager:
    """
    SQLite 기반 범용 세션 관리자.

    세션(sessions) + 항목(entries) 2-테이블 구조로
    작업 이력, 변환 기록, 사용자 활동 등을 저장.

    특징:
        - WAL 모드 (동시 읽기 성능 향상)
        - 외래키 + 인덱스 최적화
        - 트랜잭션 안전
        - 싱글톤 패턴 지원
    """

    def __init__(self, db_path: str | Path = "sessions.db"):
        """
        Args:
            db_path: SQLite 데이터베이스 파일 경로
        """
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _get_conn(self) -> sqlite3.Connection:
        """SQLite 연결 생성 (Row 팩토리 + WAL 모드)"""
        conn = sqlite3.connect(str(self.db_path))
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA foreign_keys=ON")
        return conn

    def _init_db(self):
        """데이터베이스 테이블 초기화"""
        conn = self._get_conn()
        try:
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS sessions (
                    id TEXT PRIMARY KEY,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    category TEXT DEFAULT '',
                    metadata TEXT DEFAULT '{}',
                    summary TEXT DEFAULT '',
                    notes TEXT DEFAULT ''
                );

                CREATE TABLE IF NOT EXISTS entries (
                    id TEXT PRIMARY KEY,
                    session_id TEXT NOT NULL,
                    data TEXT NOT NULL DEFAULT '{}',
                    timestamp TEXT NOT NULL,
                    FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_entries_session
                ON entries(session_id);

                CREATE INDEX IF NOT EXISTS idx_sessions_created
                ON sessions(created_at DESC);

                CREATE INDEX IF NOT EXISTS idx_sessions_updated
                ON sessions(updated_at DESC);
            """)
            conn.commit()
            logger.info(f"DB 초기화 완료: {self.db_path}")
        finally:
            conn.close()

    def add_entry(
        self,
        session_id: str,
        entry_data: dict,
        metadata: dict | None = None,
        category: str = "",
    ) -> str:
        """
        세션에 항목 추가 (세션이 없으면 자동 생성).

        Args:
            session_id: 세션 ID
            entry_data: 항목 데이터 (dict → JSON 저장)
            metadata: 세션 메타데이터 (최초 생성 시만 적용)
            category: 세션 카테고리

        Returns:
            생성된 항목 ID
        """
        import json

        conn = self._get_conn()
        now = datetime.now().isoformat()

        try:
            # 세션 존재 확인 → 없으면 생성
            existing = conn.execute(
                "SELECT id FROM sessions WHERE id = ?", (session_id,)
            ).fetchone()

            if not existing:
                meta_json = json.dumps(metadata or {}, ensure_ascii=False)
                conn.execute(
                    """INSERT INTO sessions (id, created_at, updated_at, category, metadata)
                       VALUES (?, ?, ?, ?, ?)""",
                    (session_id, now, now, category, meta_json),
                )

            # 항목 추가
            entry_id = str(uuid.uuid4())
            data_json = json.dumps(entry_data, ensure_ascii=False)

            conn.execute(
                """INSERT INTO entries (id, session_id, data, timestamp)
                   VALUES (?, ?, ?, ?)""",
                (entry_id, session_id, data_json, now),
            )

            # 세션 업데이트 시간 갱신
            conn.execute(
                "UPDATE sessions SET updated_at = ? WHERE id = ?",
                (now, session_id),
            )

            conn.commit()
            logger.info(f"항목 저장: session={session_id[:8]}...")
            return entry_id

        except Exception as e:
            logger.error(f"항목 저장 실패: {e}")
            conn.rollback()
            raise
        finally:
            conn.close()

    def get_sessions(self, limit: int = 50, offset: int = 0) -> list[dict]:
        """
        최근 세션 목록 조회 (업데이트 시간 내림차순).

        Args:
            limit: 최대 조회 수
            offset: 시작 오프셋 (페이지네이션)

        Returns:
            세션 딕셔너리 리스트
        """
        conn = self._get_conn()
        try:
            rows = conn.execute(
                """SELECT id, created_at, updated_at, category, metadata, summary, notes
                   FROM sessions
                   ORDER BY updated_at DESC
                   LIMIT ? OFFSET ?""",
                (limit, offset),
            ).fetchall()

            return [dict(row) for row in rows]
        finally:
            conn.close()

    def get_session(self, session_id: str) -> dict | None:
        """
        세션 상세 조회 (항목 포함).

        Args:
            session_id: 세션 ID

        Returns:
            세션 딕셔너리 (entries 키에 항목 리스트 포함), 없으면 None
        """
        import json

        conn = self._get_conn()
        try:
            session = conn.execute(
                "SELECT * FROM sessions WHERE id = ?", (session_id,)
            ).fetchone()

            if not session:
                return None

            entries = conn.execute(
                """SELECT * FROM entries
                   WHERE session_id = ?
                   ORDER BY timestamp ASC""",
                (session_id,),
            ).fetchall()

            result = dict(session)
            result["entries"] = []
            for entry in entries:
                entry_dict = dict(entry)
                # data 필드 JSON 파싱
                try:
                    entry_dict["data"] = json.loads(entry_dict["data"])
                except (json.JSONDecodeError, TypeError):
                    pass
                result["entries"].append(entry_dict)

            return result

        finally:
            conn.close()

    def delete_session(self, session_id: str) -> bool:
        """
        세션 및 관련 항목 삭제.

        Args:
            session_id: 삭제할 세션 ID

        Returns:
            삭제 성공 여부
        """
        conn = self._get_conn()
        try:
            conn.execute("DELETE FROM entries WHERE session_id = ?", (session_id,))
            result = conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
            conn.commit()
            deleted = result.rowcount > 0
            if deleted:
                logger.info(f"세션 삭제: {session_id[:8]}...")
            return deleted
        finally:
            conn.close()

    def search_sessions(self, query: str, limit: int = 20) -> list[dict]:
        """
        세션 검색 (summary, notes 필드 대상 LIKE 검색).

        Args:
            query: 검색어
            limit: 최대 결과 수

        Returns:
            매칭된 세션 리스트
        """
        conn = self._get_conn()
        try:
            rows = conn.execute(
                """SELECT id, created_at, updated_at, category, metadata, summary, notes
                   FROM sessions
                   WHERE summary LIKE ? OR notes LIKE ?
                   ORDER BY updated_at DESC
                   LIMIT ?""",
                (f"%{query}%", f"%{query}%", limit),
            ).fetchall()
            return [dict(row) for row in rows]
        finally:
            conn.close()

    def update_session(
        self,
        session_id: str,
        notes: str | None = None,
        summary: str | None = None,
    ) -> bool:
        """
        세션 정보 업데이트.

        Args:
            session_id: 세션 ID
            notes: 노트 (None이면 변경하지 않음)
            summary: 요약 (None이면 변경하지 않음)

        Returns:
            업데이트 성공 여부
        """
        conn = self._get_conn()
        try:
            updates = []
            params = []

            if notes is not None:
                updates.append("notes = ?")
                params.append(notes)
            if summary is not None:
                updates.append("summary = ?")
                params.append(summary)

            if not updates:
                return False

            updates.append("updated_at = ?")
            params.append(datetime.now().isoformat())
            params.append(session_id)

            conn.execute(
                f"UPDATE sessions SET {', '.join(updates)} WHERE id = ?",
                params,
            )
            conn.commit()
            return True
        finally:
            conn.close()


# --- 싱글톤 패턴 ---

_managers: dict[str, SQLiteSessionManager] = {}


def get_session_manager(db_path: str | Path = "sessions.db") -> SQLiteSessionManager:
    """
    싱글톤 세션 관리자 인스턴스 반환.

    Args:
        db_path: DB 파일 경로

    Returns:
        SQLiteSessionManager 인스턴스
    """
    global _managers
    key = str(db_path)
    if key not in _managers:
        _managers[key] = SQLiteSessionManager(db_path)
    return _managers[key]