# -*- coding: utf-8 -*- """ 동물약 챗봇 로깅 모듈 - SQLite에 대화 로그 저장 - 각 단계별 소요시간, 토큰, 비용 기록 """ import os import json import sqlite3 import logging from pathlib import Path from datetime import datetime from typing import Dict, List, Optional from dataclasses import dataclass, field, asdict logger = logging.getLogger(__name__) # DB 경로 DB_PATH = Path(__file__).parent.parent / "db" / "animal_chat_logs.db" SCHEMA_PATH = Path(__file__).parent.parent / "db" / "animal_chat_logs_schema.sql" # GPT-4o-mini 가격 (USD per 1K tokens) INPUT_COST_PER_1K = 0.00015 # $0.15 / 1M = $0.00015 / 1K OUTPUT_COST_PER_1K = 0.0006 # $0.60 / 1M = $0.0006 / 1K @dataclass class ChatLogEntry: """챗봇 로그 엔트리""" session_id: str = "" # 입력 user_message: str = "" history_length: int = 0 # MSSQL mssql_drug_count: int = 0 mssql_duration_ms: int = 0 # PostgreSQL pgsql_rag_count: int = 0 pgsql_duration_ms: int = 0 # LanceDB vector_results_count: int = 0 vector_top_scores: List[float] = field(default_factory=list) vector_sources: List[str] = field(default_factory=list) vector_duration_ms: int = 0 # OpenAI openai_model: str = "" openai_prompt_tokens: int = 0 openai_completion_tokens: int = 0 openai_total_tokens: int = 0 openai_cost_usd: float = 0.0 openai_duration_ms: int = 0 # 출력 assistant_response: str = "" products_mentioned: List[str] = field(default_factory=list) # 메타 total_duration_ms: int = 0 error: str = "" def calculate_cost(self): """토큰 기반 비용 계산""" self.openai_cost_usd = ( self.openai_prompt_tokens * INPUT_COST_PER_1K / 1000 + self.openai_completion_tokens * OUTPUT_COST_PER_1K / 1000 ) def init_db(): """DB 초기화 (테이블 생성)""" try: DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(DB_PATH)) if SCHEMA_PATH.exists(): schema = SCHEMA_PATH.read_text(encoding='utf-8') conn.executescript(schema) else: # 스키마 파일 없으면 인라인 생성 conn.executescript(""" CREATE TABLE IF NOT EXISTS chat_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, user_message TEXT, history_length INTEGER, mssql_drug_count INTEGER, mssql_duration_ms INTEGER, pgsql_rag_count INTEGER, pgsql_duration_ms INTEGER, vector_results_count INTEGER, vector_top_scores TEXT, vector_sources TEXT, vector_duration_ms INTEGER, openai_model TEXT, openai_prompt_tokens INTEGER, openai_completion_tokens INTEGER, openai_total_tokens INTEGER, openai_cost_usd REAL, openai_duration_ms INTEGER, assistant_response TEXT, products_mentioned TEXT, total_duration_ms INTEGER, error TEXT, created_at TEXT DEFAULT (datetime('now', 'localtime')) ); CREATE INDEX IF NOT EXISTS idx_chat_created ON chat_logs(created_at); """) conn.commit() conn.close() logger.info(f"동물약 챗봇 로그 DB 초기화: {DB_PATH}") return True except Exception as e: logger.error(f"DB 초기화 실패: {e}") return False def log_chat(entry: ChatLogEntry) -> Optional[int]: """ 챗봇 대화 로그 저장 Returns: 저장된 로그 ID (실패시 None) """ try: # 비용 계산 entry.calculate_cost() conn = sqlite3.connect(str(DB_PATH)) cursor = conn.cursor() cursor.execute(""" INSERT INTO chat_logs ( session_id, user_message, history_length, mssql_drug_count, mssql_duration_ms, pgsql_rag_count, pgsql_duration_ms, vector_results_count, vector_top_scores, vector_sources, vector_duration_ms, openai_model, openai_prompt_tokens, openai_completion_tokens, openai_total_tokens, openai_cost_usd, openai_duration_ms, assistant_response, products_mentioned, total_duration_ms, error ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( entry.session_id, entry.user_message, entry.history_length, entry.mssql_drug_count, entry.mssql_duration_ms, entry.pgsql_rag_count, entry.pgsql_duration_ms, entry.vector_results_count, json.dumps(entry.vector_top_scores), json.dumps(entry.vector_sources), entry.vector_duration_ms, entry.openai_model, entry.openai_prompt_tokens, entry.openai_completion_tokens, entry.openai_total_tokens, entry.openai_cost_usd, entry.openai_duration_ms, entry.assistant_response, json.dumps(entry.products_mentioned), entry.total_duration_ms, entry.error )) log_id = cursor.lastrowid conn.commit() conn.close() logger.debug(f"챗봇 로그 저장: ID={log_id}, tokens={entry.openai_total_tokens}") return log_id except Exception as e: logger.error(f"로그 저장 실패: {e}") return None def get_logs( limit: int = 100, offset: int = 0, date_from: str = None, date_to: str = None, error_only: bool = False ) -> List[Dict]: """로그 조회""" try: conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row cursor = conn.cursor() query = "SELECT * FROM chat_logs WHERE 1=1" params = [] if date_from: query += " AND created_at >= ?" params.append(date_from) if date_to: query += " AND created_at <= ?" params.append(date_to + " 23:59:59") if error_only: query += " AND error IS NOT NULL AND error != ''" query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" params.extend([limit, offset]) cursor.execute(query, params) rows = cursor.fetchall() conn.close() return [dict(row) for row in rows] except Exception as e: logger.error(f"로그 조회 실패: {e}") return [] def get_stats(date_from: str = None, date_to: str = None) -> Dict: """통계 조회""" try: conn = sqlite3.connect(str(DB_PATH)) cursor = conn.cursor() query = """ SELECT COUNT(*) as total_chats, AVG(total_duration_ms) as avg_duration_ms, SUM(openai_total_tokens) as total_tokens, SUM(openai_cost_usd) as total_cost_usd, AVG(openai_total_tokens) as avg_tokens, SUM(CASE WHEN error IS NOT NULL AND error != '' THEN 1 ELSE 0 END) as error_count, AVG(vector_duration_ms) as avg_vector_ms, AVG(openai_duration_ms) as avg_openai_ms FROM chat_logs WHERE 1=1 """ params = [] if date_from: query += " AND created_at >= ?" params.append(date_from) if date_to: query += " AND created_at <= ?" params.append(date_to + " 23:59:59") cursor.execute(query, params) row = cursor.fetchone() conn.close() if row: return { 'total_chats': row[0] or 0, 'avg_duration_ms': round(row[1] or 0), 'total_tokens': row[2] or 0, 'total_cost_usd': round(row[3] or 0, 4), 'avg_tokens': round(row[4] or 0), 'error_count': row[5] or 0, 'avg_vector_ms': round(row[6] or 0), 'avg_openai_ms': round(row[7] or 0) } return {} except Exception as e: logger.error(f"통계 조회 실패: {e}") return {} # 모듈 로드 시 DB 초기화 init_db()