from __future__ import annotations import os import sqlite3 from pathlib import Path BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" DB_PATH = DATA_DIR / "confivoice.db" DB_DRIVER = os.getenv("CONFIVOICE_DB_DRIVER", "sqlite").lower() MYSQL_HOST = os.getenv("CONFIVOICE_MYSQL_HOST", "127.0.0.1") MYSQL_PORT = int(os.getenv("CONFIVOICE_MYSQL_PORT", "3306")) MYSQL_USER = os.getenv("CONFIVOICE_MYSQL_USER", "root") MYSQL_PASSWORD = os.getenv("CONFIVOICE_MYSQL_PASSWORD", "") MYSQL_DATABASE = os.getenv("CONFIVOICE_MYSQL_DATABASE", "confivoice") def get_database_label(): if DB_DRIVER == "mysql": return f"mysql://{MYSQL_USER}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}" return str(DB_PATH) def get_connection(): if DB_DRIVER == "mysql": return get_mysql_connection() DATA_DIR.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(DB_PATH) connection.row_factory = sqlite3.Row return connection def get_mysql_connection(database=MYSQL_DATABASE): try: import pymysql except ImportError as error: raise RuntimeError( "PyMySQL belum terinstall. Jalankan: pip install PyMySQL" ) from error return pymysql.connect( host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASSWORD, database=database, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, ) def init_mysql_db(): with get_mysql_connection(database=None) as connection: with connection.cursor() as cursor: cursor.execute( f"CREATE DATABASE IF NOT EXISTS `{MYSQL_DATABASE}` " "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" ) connection.commit() with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ CREATE TABLE IF NOT EXISTS prediction_results ( id INT AUTO_INCREMENT PRIMARY KEY, student_name VARCHAR(255) NOT NULL, predicted_label VARCHAR(20), description VARCHAR(255), confidence DOUBLE NOT NULL DEFAULT 0, probability_pd DOUBLE NOT NULL DEFAULT 0, probability_tpd DOUBLE NOT NULL DEFAULT 0, is_valid_audio TINYINT(1) NOT NULL DEFAULT 1, error_message TEXT, audio_duration DOUBLE NOT NULL DEFAULT 0, volume_score DOUBLE NOT NULL DEFAULT 0, intonation_score DOUBLE NOT NULL DEFAULT 0, pause_score DOUBLE NOT NULL DEFAULT 0, speech_activity_ratio DOUBLE NOT NULL DEFAULT 0, silence_ratio DOUBLE NOT NULL DEFAULT 0, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) connection.commit() def init_db(): if DB_DRIVER == "mysql": init_mysql_db() return with get_connection() as connection: connection.execute( """ CREATE TABLE IF NOT EXISTS prediction_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, student_name TEXT NOT NULL, predicted_label TEXT, description TEXT, confidence REAL NOT NULL DEFAULT 0, probability_pd REAL NOT NULL DEFAULT 0, probability_tpd REAL NOT NULL DEFAULT 0, is_valid_audio INTEGER NOT NULL DEFAULT 1, error_message TEXT, audio_duration REAL NOT NULL DEFAULT 0, volume_score REAL NOT NULL DEFAULT 0, intonation_score REAL NOT NULL DEFAULT 0, pause_score REAL NOT NULL DEFAULT 0, speech_activity_ratio REAL NOT NULL DEFAULT 0, silence_ratio REAL NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) connection.commit() def save_prediction_result(student_name, result): init_db() audio_quality = result.get("audio_quality") or {} indicators = result.get("voice_indicators") or {} if DB_DRIVER == "mysql": return save_prediction_result_mysql(student_name, result, audio_quality, indicators) with get_connection() as connection: cursor = connection.execute( """ INSERT INTO prediction_results ( student_name, predicted_label, description, confidence, probability_pd, probability_tpd, is_valid_audio, error_message, audio_duration, volume_score, intonation_score, pause_score, speech_activity_ratio, silence_ratio ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( student_name, result.get("predicted_label") or result.get("label"), result.get("description"), float(result.get("confidence") or 0), float(result.get("probability_pd") or 0), float(result.get("probability_tpd") or 0), 1 if result.get("is_valid_audio") is not False else 0, result.get("error_message"), float(audio_quality.get("duration") or 0), float(indicators.get("volume_score") or 0), float(indicators.get("intonation_score") or 0), float(indicators.get("pause_score") or 0), float(indicators.get("speech_activity_ratio") or 0), float(indicators.get("silence_ratio") or 0), ), ) connection.commit() return cursor.lastrowid def save_prediction_result_mysql(student_name, result, audio_quality, indicators): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ INSERT INTO prediction_results ( student_name, predicted_label, description, confidence, probability_pd, probability_tpd, is_valid_audio, error_message, audio_duration, volume_score, intonation_score, pause_score, speech_activity_ratio, silence_ratio ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( student_name, result.get("predicted_label") or result.get("label"), result.get("description"), float(result.get("confidence") or 0), float(result.get("probability_pd") or 0), float(result.get("probability_tpd") or 0), 1 if result.get("is_valid_audio") is not False else 0, result.get("error_message"), float(audio_quality.get("duration") or 0), float(indicators.get("volume_score") or 0), float(indicators.get("intonation_score") or 0), float(indicators.get("pause_score") or 0), float(indicators.get("speech_activity_ratio") or 0), float(indicators.get("silence_ratio") or 0), ), ) prediction_id = cursor.lastrowid connection.commit() return prediction_id def list_prediction_results(limit=200): init_db() if DB_DRIVER == "mysql": return list_prediction_results_mysql(limit=limit) with get_connection() as connection: rows = connection.execute( """ SELECT id, student_name, predicted_label, description, confidence, probability_pd, probability_tpd, is_valid_audio, error_message, audio_duration, volume_score, intonation_score, pause_score, speech_activity_ratio, silence_ratio, created_at FROM prediction_results ORDER BY created_at DESC, id DESC LIMIT ? """, (limit,), ).fetchall() return [dict(row) for row in rows] def list_prediction_results_mysql(limit=200): with get_connection() as connection: with connection.cursor() as cursor: cursor.execute( """ SELECT id, student_name, predicted_label, description, confidence, probability_pd, probability_tpd, is_valid_audio, error_message, audio_duration, volume_score, intonation_score, pause_score, speech_activity_ratio, silence_ratio, created_at FROM prediction_results ORDER BY created_at DESC, id DESC LIMIT %s """, (limit,), ) rows = cursor.fetchall() for row in rows: row["created_at"] = str(row["created_at"]) return rows