682 lines
23 KiB
Python
682 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import secrets
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
PROJECT_DIR = BASE_DIR.parent
|
|
WEB_DATA_DIR = PROJECT_DIR / "cv_web" / "data"
|
|
DB_PATH = WEB_DATA_DIR / "confivoice.db"
|
|
|
|
DB_DRIVER = os.getenv("CONFIVOICE_DB_DRIVER", "mysql").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")
|
|
PASSWORD_HASH_ITERATIONS = 120_000
|
|
|
|
|
|
def hash_password(password):
|
|
salt = secrets.token_hex(16)
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256",
|
|
password.encode("utf-8"),
|
|
salt.encode("utf-8"),
|
|
PASSWORD_HASH_ITERATIONS,
|
|
).hex()
|
|
return f"pbkdf2_sha256${PASSWORD_HASH_ITERATIONS}${salt}${digest}"
|
|
|
|
|
|
def verify_password(password, stored_hash):
|
|
try:
|
|
algorithm, iterations, salt, digest = stored_hash.split("$", 3)
|
|
if algorithm != "pbkdf2_sha256":
|
|
return False
|
|
candidate = hashlib.pbkdf2_hmac(
|
|
"sha256",
|
|
password.encode("utf-8"),
|
|
salt.encode("utf-8"),
|
|
int(iterations),
|
|
).hex()
|
|
return hmac.compare_digest(candidate, digest)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
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()
|
|
|
|
WEB_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,
|
|
user_id INT NULL,
|
|
student_name VARCHAR(255) NOT NULL,
|
|
student_gender VARCHAR(20),
|
|
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
|
|
)
|
|
"""
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
full_name VARCHAR(255) NOT NULL,
|
|
username VARCHAR(120) NOT NULL UNIQUE,
|
|
password_hash VARCHAR(255) NOT NULL,
|
|
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*) AS total
|
|
FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s
|
|
AND TABLE_NAME = 'prediction_results'
|
|
AND COLUMN_NAME = 'user_id'
|
|
""",
|
|
(MYSQL_DATABASE,),
|
|
)
|
|
if cursor.fetchone()["total"] == 0:
|
|
cursor.execute("ALTER TABLE prediction_results ADD COLUMN user_id INT NULL AFTER id")
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*) AS total
|
|
FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s
|
|
AND TABLE_NAME = 'prediction_results'
|
|
AND COLUMN_NAME = 'student_gender'
|
|
""",
|
|
(MYSQL_DATABASE,),
|
|
)
|
|
if cursor.fetchone()["total"] == 0:
|
|
cursor.execute(
|
|
"ALTER TABLE prediction_results "
|
|
"ADD COLUMN student_gender VARCHAR(20) NULL AFTER student_name"
|
|
)
|
|
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,
|
|
user_id INTEGER,
|
|
student_name TEXT NOT NULL,
|
|
student_gender TEXT,
|
|
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.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
full_name TEXT NOT NULL,
|
|
username TEXT NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
columns = {
|
|
row["name"]
|
|
for row in connection.execute("PRAGMA table_info(prediction_results)").fetchall()
|
|
}
|
|
if "user_id" not in columns:
|
|
connection.execute("ALTER TABLE prediction_results ADD COLUMN user_id INTEGER")
|
|
if "student_gender" not in columns:
|
|
connection.execute(
|
|
"ALTER TABLE prediction_results ADD COLUMN student_gender TEXT"
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
def _clean_username(username):
|
|
return username.strip().lower()
|
|
|
|
|
|
def _clean_full_name(full_name):
|
|
return " ".join(full_name.strip().split())
|
|
|
|
|
|
def _user_row_to_dict(row):
|
|
if not row:
|
|
return None
|
|
return {
|
|
"id": row["id"],
|
|
"full_name": row["full_name"],
|
|
"username": row["username"],
|
|
"created_at": str(row["created_at"]),
|
|
}
|
|
|
|
|
|
def create_user(full_name, username, password):
|
|
init_db()
|
|
clean_full_name = _clean_full_name(full_name)
|
|
clean_username = _clean_username(username)
|
|
password_hash = hash_password(password)
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO users (full_name, username, password_hash)
|
|
VALUES (%s, %s, %s)
|
|
""",
|
|
(clean_full_name, clean_username, password_hash),
|
|
)
|
|
user_id = cursor.lastrowid
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
return get_user_by_id(user_id)
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO users (full_name, username, password_hash)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(clean_full_name, clean_username, password_hash),
|
|
)
|
|
connection.commit()
|
|
return get_user_by_id(cursor.lastrowid)
|
|
|
|
|
|
def get_user_by_username(username):
|
|
init_db()
|
|
clean_username = _clean_username(username)
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, full_name, username, password_hash, created_at
|
|
FROM users
|
|
WHERE username = %s
|
|
""",
|
|
(clean_username,),
|
|
)
|
|
return cursor.fetchone()
|
|
|
|
with get_connection() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT id, full_name, username, password_hash, created_at
|
|
FROM users
|
|
WHERE username = ?
|
|
""",
|
|
(clean_username,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def get_user_by_id(user_id):
|
|
init_db()
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, full_name, username, created_at
|
|
FROM users
|
|
WHERE id = %s
|
|
""",
|
|
(user_id,),
|
|
)
|
|
return _user_row_to_dict(cursor.fetchone())
|
|
|
|
with get_connection() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT id, full_name, username, created_at
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
return _user_row_to_dict(dict(row) if row else None)
|
|
|
|
|
|
def list_users(limit=500):
|
|
init_db()
|
|
safe_limit = max(1, min(int(limit or 500), 1000))
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, full_name, username, created_at
|
|
FROM users
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT %s
|
|
""",
|
|
(safe_limit,),
|
|
)
|
|
rows = cursor.fetchall()
|
|
return [_user_row_to_dict(row) for row in rows]
|
|
|
|
with get_connection() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT id, full_name, username, created_at
|
|
FROM users
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT ?
|
|
""",
|
|
(safe_limit,),
|
|
).fetchall()
|
|
return [_user_row_to_dict(dict(row)) for row in rows]
|
|
|
|
|
|
def update_user(user_id, full_name, username, password=None):
|
|
init_db()
|
|
clean_full_name = _clean_full_name(full_name)
|
|
clean_username = _clean_username(username)
|
|
clean_password = (password or "").strip()
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
if clean_password:
|
|
cursor.execute(
|
|
"""
|
|
UPDATE users
|
|
SET full_name = %s, username = %s, password_hash = %s
|
|
WHERE id = %s
|
|
""",
|
|
(
|
|
clean_full_name,
|
|
clean_username,
|
|
hash_password(clean_password),
|
|
user_id,
|
|
),
|
|
)
|
|
else:
|
|
cursor.execute(
|
|
"""
|
|
UPDATE users
|
|
SET full_name = %s, username = %s
|
|
WHERE id = %s
|
|
""",
|
|
(clean_full_name, clean_username, user_id),
|
|
)
|
|
affected = cursor.rowcount
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
return affected > 0
|
|
|
|
with get_connection() as connection:
|
|
if clean_password:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE users
|
|
SET full_name = ?, username = ?, password_hash = ?
|
|
WHERE id = ?
|
|
""",
|
|
(clean_full_name, clean_username, hash_password(clean_password), user_id),
|
|
)
|
|
else:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE users
|
|
SET full_name = ?, username = ?
|
|
WHERE id = ?
|
|
""",
|
|
(clean_full_name, clean_username, user_id),
|
|
)
|
|
connection.commit()
|
|
return cursor.rowcount > 0
|
|
|
|
|
|
def delete_user(user_id):
|
|
init_db()
|
|
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
|
|
affected = cursor.rowcount
|
|
connection.commit()
|
|
return affected > 0
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
connection.commit()
|
|
return cursor.rowcount > 0
|
|
|
|
|
|
def authenticate_user(username, password):
|
|
row = get_user_by_username(username)
|
|
if not row or not verify_password(password, row["password_hash"]):
|
|
return None
|
|
return _user_row_to_dict(row)
|
|
|
|
|
|
def save_prediction_result(student_name, result):
|
|
init_db()
|
|
audio_quality = result.get("audio_quality") or {}
|
|
indicators = result.get("voice_indicators") or {}
|
|
|
|
values = (
|
|
result.get("user_id"),
|
|
student_name,
|
|
result.get("student_gender"),
|
|
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),
|
|
)
|
|
|
|
if DB_DRIVER == "mysql":
|
|
return save_prediction_result_mysql(values)
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO prediction_results (
|
|
user_id, student_name, student_gender, 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
values,
|
|
)
|
|
connection.commit()
|
|
return cursor.lastrowid
|
|
|
|
|
|
def save_prediction_result_mysql(values):
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO prediction_results (
|
|
user_id, student_name, student_gender, 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, %s, %s)
|
|
""",
|
|
values,
|
|
)
|
|
prediction_id = cursor.lastrowid
|
|
connection.commit()
|
|
return prediction_id
|
|
|
|
|
|
def get_prediction_result(prediction_id):
|
|
init_db()
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
id, user_id, student_name, student_gender, 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
|
|
WHERE id = %s
|
|
""",
|
|
(prediction_id,),
|
|
)
|
|
row = cursor.fetchone()
|
|
if row:
|
|
row["created_at"] = str(row["created_at"])
|
|
return row
|
|
|
|
with get_connection() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT
|
|
id, user_id, student_name, student_gender, 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
|
|
WHERE id = ?
|
|
""",
|
|
(prediction_id,),
|
|
).fetchone()
|
|
|
|
return dict(row) if row else None
|
|
|
|
|
|
def update_prediction_result(prediction_id, values):
|
|
init_db()
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
UPDATE prediction_results
|
|
SET
|
|
student_name = %s,
|
|
student_gender = %s,
|
|
predicted_label = %s,
|
|
description = %s,
|
|
confidence = %s,
|
|
probability_pd = %s,
|
|
probability_tpd = %s,
|
|
is_valid_audio = %s,
|
|
error_message = %s,
|
|
audio_duration = %s,
|
|
volume_score = %s,
|
|
intonation_score = %s,
|
|
pause_score = %s,
|
|
speech_activity_ratio = %s,
|
|
silence_ratio = %s
|
|
WHERE id = %s
|
|
""",
|
|
(*values, prediction_id),
|
|
)
|
|
affected = cursor.rowcount
|
|
connection.commit()
|
|
return affected > 0
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE prediction_results
|
|
SET
|
|
student_name = ?,
|
|
student_gender = ?,
|
|
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 = ?
|
|
WHERE id = ?
|
|
""",
|
|
(*values, prediction_id),
|
|
)
|
|
connection.commit()
|
|
return cursor.rowcount > 0
|
|
|
|
|
|
def delete_prediction_result(prediction_id):
|
|
init_db()
|
|
if DB_DRIVER == "mysql":
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"DELETE FROM prediction_results WHERE id = %s",
|
|
(prediction_id,),
|
|
)
|
|
affected = cursor.rowcount
|
|
connection.commit()
|
|
return affected > 0
|
|
|
|
with get_connection() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM prediction_results WHERE id = ?",
|
|
(prediction_id,),
|
|
)
|
|
connection.commit()
|
|
return cursor.rowcount > 0
|
|
|
|
|
|
def list_prediction_results(limit=200, user_id=None):
|
|
init_db()
|
|
if DB_DRIVER == "mysql":
|
|
return list_prediction_results_mysql(limit=limit, user_id=user_id)
|
|
|
|
with get_connection() as connection:
|
|
where_clause = "WHERE user_id = ?" if user_id is not None else ""
|
|
params = (user_id, limit) if user_id is not None else (limit,)
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT
|
|
id, user_id, student_name, student_gender, 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
|
|
{where_clause}
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT ?
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def list_prediction_results_mysql(limit=200, user_id=None):
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
where_clause = "WHERE user_id = %s" if user_id is not None else ""
|
|
params = (user_id, limit) if user_id is not None else (limit,)
|
|
cursor.execute(
|
|
f"""
|
|
SELECT
|
|
id, user_id, student_name, student_gender, 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
|
|
{where_clause}
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT %s
|
|
""",
|
|
params,
|
|
)
|
|
rows = cursor.fetchall()
|
|
|
|
for row in rows:
|
|
row["created_at"] = str(row["created_at"])
|
|
return rows
|