# app/models/spi_module.py import mysql.connector from app.core.config import A_DB_CONFIG, logger from typing import List, Dict, Optional # ======================================== # spi_log (답변 저장) # ======================================== def save_answer(session_id: str, question_id: int, selected_option: str, is_correct: Optional[bool] = None): """ SPI 답변 저장 (spi_log 테이블) Args: session_id: 세션 ID question_id: 문제 ID selected_option: 선택한 답변 is_correct: 정답 여부 (능력검사만, 성격검사는 None) """ conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor() sql = """ INSERT INTO spi_log (session_id, question_id, selected_option, is_correct) VALUES (%s, %s, %s, %s) """ cursor.execute(sql, (session_id, question_id, selected_option, is_correct)) conn.commit() return cursor.lastrowid except Exception as e: logger.error(f"SPI Answer Save Error: {e}") raise e finally: if conn: cursor.close() conn.close() # ======================================== # spi_Output (최종 결과 저장) # ======================================== def save_or_update_output(session_id: str, language_score: int, language_grade: str, nonverbal_score: int, nonverbal_grade: str, personality_score: int, personality_type: str): """ personality_type 인자는 AI 코멘트(일본어 텍스트)를 받습니다. """ conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor() sql = """ INSERT INTO spi_Output (session_id, language_score, language_grade, nonverbal_score, nonverbal_grade, personality_score, personality_type) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE language_score = VALUES(language_score), language_grade = VALUES(language_grade), nonverbal_score = VALUES(nonverbal_score), nonverbal_grade = VALUES(nonverbal_grade), personality_score = VALUES(personality_score), personality_type = VALUES(personality_type) """ cursor.execute(sql, (session_id, language_score, language_grade, nonverbal_score, nonverbal_grade, personality_score, personality_type)) conn.commit() logger.info(f"SPI Output saved for session: {session_id}") # user_profile_summary에도 동기화 (AI 코멘트 저장) sync_to_user_profile_summary( session_id, language_grade, nonverbal_grade, personality_type # AI 코멘트 (일본어 텍스트) ) except Exception as e: logger.error(f"SPI Output Save Error: {e}") raise e finally: if conn: cursor.close() conn.close() # ... (나머지 동일) ... def sync_to_user_profile_summary(session_id: str, language_grade: str, nonverbal_grade: str, personality_type: str): """SPI 완료 시 user_profile_summary 테이블에도 동기화""" conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor() sql = """ INSERT INTO user_profile_summary (session_id, spi_language, spi_nonverbal, spi_personality) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE spi_language = VALUES(spi_language), spi_nonverbal = VALUES(spi_nonverbal), spi_personality = VALUES(spi_personality), updated_at = NOW() """ cursor.execute(sql, (session_id, language_grade, nonverbal_grade, personality_type)) conn.commit() logger.info(f"SPI synced to user_profile_summary for session: {session_id}") except Exception as e: logger.error(f"SPI Summary Sync Error: {e}") # 동기화 실패해도 메인 로직은 계속 진행되도록 에러를 raise하지 않음 pass finally: if conn: cursor.close() conn.close() # ======================================== # 조회 함수들 # ======================================== def get_answer_count(session_id: str) -> int: """현재 답변한 문제 수 조회""" conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor() sql = "SELECT COUNT(*) FROM spi_log WHERE session_id = %s" cursor.execute(sql, (session_id,)) result = cursor.fetchone() return result[0] if result else 0 except Exception as e: logger.error(f"SPI Answer Count Error: {e}") return 0 finally: if conn: cursor.close() conn.close() def get_answers_by_type(session_id: str, question_type: str) -> List[Dict]: """ 타입별 답변 조회 Args: session_id: 세션 ID question_type: 'language' (1-5), 'nonverbal' (6-10), 'personality' (11-20) Returns: 답변 리스트 """ conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor(dictionary=True) # question_id 범위로 타입 구분 if question_type == "language": min_id, max_id = 1, 5 elif question_type == "nonverbal": min_id, max_id = 6, 10 elif question_type == "personality": min_id, max_id = 11, 20 else: return [] sql = """ SELECT question_id, selected_option, is_correct FROM spi_log WHERE session_id = %s AND question_id BETWEEN %s AND %s ORDER BY question_id """ cursor.execute(sql, (session_id, min_id, max_id)) results = cursor.fetchall() return results except Exception as e: logger.error(f"SPI Answers by Type Error: {e}") return [] finally: if conn: cursor.close() conn.close() def get_all_answers(session_id: str) -> List[Dict]: """전체 답변 조회""" conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor(dictionary=True) sql = """ SELECT question_id, selected_option, is_correct FROM spi_log WHERE session_id = %s ORDER BY question_id """ cursor.execute(sql, (session_id,)) results = cursor.fetchall() return results except Exception as e: logger.error(f"SPI All Answers Error: {e}") return [] finally: if conn: cursor.close() conn.close() def get_output(session_id: str) -> Optional[Dict]: """SPI 최종 결과 조회""" conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor(dictionary=True) sql = "SELECT * FROM spi_Output WHERE session_id = %s" cursor.execute(sql, (session_id,)) result = cursor.fetchone() return result except Exception as e: logger.error(f"SPI Output Get Error: {e}") return None finally: if conn: cursor.close() conn.close() # app/models/spi_module.py def clear_session_logs(session_id: str): """ 테스트 시작 시 이전 기록 초기화 (재시작용) """ conn = None try: conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor() # spi_log 테이블에서 해당 세션의 기록 모두 삭제 sql = "DELETE FROM spi_log WHERE session_id = %s" cursor.execute(sql, (session_id,)) conn.commit() logger.info(f"SPI Session Cleared: {session_id}") except Exception as e: logger.error(f"SPI Session Clear Error: {e}") # 에러가 나도 진행은 되도록 raise는 안 함 finally: if conn: cursor.close() conn.close()