MIF_E31231708/ml/migrate_sqlite_to_mysql.py

69 lines
2.1 KiB
Python

from __future__ import annotations
import sqlite3
from database import DB_PATH, init_db, save_prediction_result
def migrate():
if not DB_PATH.exists():
print(f"SQLite database tidak ditemukan: {DB_PATH}")
return
init_db()
with sqlite3.connect(DB_PATH) as sqlite_connection:
sqlite_connection.row_factory = sqlite3.Row
rows = sqlite_connection.execute(
"""
SELECT
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
FROM prediction_results
ORDER BY id ASC
"""
).fetchall()
migrated_count = 0
for row in rows:
result = {
"predicted_label": row["predicted_label"],
"label": row["predicted_label"],
"description": row["description"],
"confidence": row["confidence"],
"probability_pd": row["probability_pd"],
"probability_tpd": row["probability_tpd"],
"is_valid_audio": bool(row["is_valid_audio"]),
"error_message": row["error_message"],
"audio_quality": {
"duration": row["audio_duration"],
},
"voice_indicators": {
"volume_score": row["volume_score"],
"intonation_score": row["intonation_score"],
"pause_score": row["pause_score"],
"speech_activity_ratio": row["speech_activity_ratio"],
"silence_ratio": row["silence_ratio"],
},
}
save_prediction_result(row["student_name"], result)
migrated_count += 1
print(f"Selesai migrasi {migrated_count} data dari SQLite ke MySQL.")
if __name__ == "__main__":
migrate()