pharmacy-pos-qr-system/backend/db/kims_logger.py

"""
KIMS API 로깅 모듈
- API 호출/응답 SQLite 저장
- AI 학습용 데이터 수집
"""
import sqlite3
import json
import os
from datetime import datetime
from pathlib import Path

# DB 파일 경로
DB_PATH = Path(__file__).parent / 'kims_logs.db'

def init_db():
    """DB 초기화 (테이블 생성)"""
    schema_path = Path(__file__).parent / 'kims_logs_schema.sql'
    
    conn = sqlite3.connect(str(DB_PATH))
    cursor = conn.cursor()
    
    with open(schema_path, 'r', encoding='utf-8') as f:
        schema = f.read()
        cursor.executescript(schema)
    
    conn.commit()
    conn.close()
    print(f"KIMS 로그 DB 초기화 완료: {DB_PATH}")

def log_kims_call(
    pre_serial: str = None,
    user_id: int = None,
    source: str = 'admin',
    drug_codes: list = None,
    drug_names: list = None,
    api_status: str = 'SUCCESS',
    http_status: int = 200,
    response_time_ms: int = 0,
    interactions: list = None,
    response_raw: dict = None,
    error_message: str = None
) -> int:
    """
    KIMS API 호출 로그 저장
    
    Returns:
        log_id: 생성된 로그 ID
    """
    # DB 없으면 초기화
    if not DB_PATH.exists():
        init_db()
    
    conn = sqlite3.connect(str(DB_PATH))
    cursor = conn.cursor()
    
    interactions = interactions or []
    drug_codes = drug_codes or []
    drug_names = drug_names or []
    
    # 심각한 상호작용 여부 (severity 1 또는 2)
    has_severe = any(
        str(i.get('severity', '5')) in ['1', '2'] 
        for i in interactions
    )
    
    # 메인 로그 삽입
    cursor.execute("""
        INSERT INTO kims_api_logs (
            pre_serial, user_id, source,
            request_drug_codes, request_drug_names, request_drug_count,
            api_status, http_status, response_time_ms,
            interaction_count, has_severe_interaction,
            interactions_json, response_raw, error_message
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        pre_serial,
        user_id,
        source,
        json.dumps(drug_codes, ensure_ascii=False),
        json.dumps(drug_names, ensure_ascii=False),
        len(drug_codes),
        api_status,
        http_status,
        response_time_ms,
        len(interactions),
        1 if has_severe else 0,
        json.dumps(interactions, ensure_ascii=False),
        json.dumps(response_raw, ensure_ascii=False) if response_raw else None,
        error_message
    ))
    
    log_id = cursor.lastrowid
    
    # 상호작용 상세 삽입 (정규화)
    for inter in interactions:
        cursor.execute("""
            INSERT INTO kims_interactions (
                log_id,
                drug1_code, drug1_name, drug1_generic,
                drug2_code, drug2_name, drug2_generic,
                severity_level, severity_desc,
                likelihood_level, likelihood_desc,
                observation, observation_generic,
                clinical_management, action_to_take, reference
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            log_id,
            inter.get('drug1_code'),
            inter.get('drug1_name'),
            inter.get('generic1'),
            inter.get('drug2_code'),
            inter.get('drug2_name'),
            inter.get('generic2'),
            int(inter.get('severity', 5)) if str(inter.get('severity', '')).isdigit() else None,
            inter.get('severity_text'),
            None,  # likelihood_level
            inter.get('likelihood'),
            inter.get('description'),
            None,  # observation_generic
            inter.get('management'),
            inter.get('action'),
            None   # reference
        ))
    
    conn.commit()
    conn.close()
    
    return log_id

def get_recent_logs(limit: int = 50):
    """최근 로그 조회"""
    if not DB_PATH.exists():
        return []
    
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT * FROM kims_api_logs
        ORDER BY created_at DESC
        LIMIT ?
    """, (limit,))
    
    rows = cursor.fetchall()
    conn.close()
    
    return [dict(row) for row in rows]

def get_log_detail(log_id: int):
    """로그 상세 조회 (상호작용 포함)"""
    if not DB_PATH.exists():
        return None
    
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    # 메인 로그
    cursor.execute("SELECT * FROM kims_api_logs WHERE id = ?", (log_id,))
    log = cursor.fetchone()
    
    if not log:
        conn.close()
        return None
    
    # 상호작용 상세
    cursor.execute("""
        SELECT * FROM kims_interactions 
        WHERE log_id = ?
        ORDER BY severity_level ASC
    """, (log_id,))
    interactions = cursor.fetchall()
    
    conn.close()
    
    result = dict(log)
    result['interactions_detail'] = [dict(i) for i in interactions]
    
    return result

def get_stats():
    """통계 조회"""
    if not DB_PATH.exists():
        return {}
    
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    # 전체 통계
    cursor.execute("""
        SELECT 
            COUNT(*) as total_calls,
            SUM(CASE WHEN api_status = 'SUCCESS' THEN 1 ELSE 0 END) as success_count,
            SUM(CASE WHEN interaction_count > 0 THEN 1 ELSE 0 END) as with_interaction,
            SUM(CASE WHEN has_severe_interaction = 1 THEN 1 ELSE 0 END) as with_severe,
            AVG(response_time_ms) as avg_response_ms
        FROM kims_api_logs
    """)
    stats = dict(cursor.fetchone())
    
    # 최근 7일 일별 통계
    cursor.execute("""
        SELECT * FROM kims_stats
        ORDER BY date DESC
        LIMIT 7
    """)
    daily = [dict(row) for row in cursor.fetchall()]
    
    conn.close()
    
    stats['daily'] = daily
    return stats


if __name__ == '__main__':
    # DB 초기화 테스트
    init_db()
    print("KIMS 로그 DB 초기화 완료!")