E41222753_NinikYuniarsih_Ju.../db/database.py

303 lines
12 KiB
Python

import hashlib
def hash_password(password):
"""Hash password dengan SHA-256."""
return hashlib.sha256(password.encode('utf-8')).hexdigest()
def register_admin(username, password):
"""
Tambah admin baru ke tabel admin. Password akan di-hash.
Akan melempar Exception jika username sudah ada.
"""
password_hash = hash_password(password)
with get_connection() as conn:
cur = conn.cursor()
cur.execute("INSERT INTO admin (username, password_hash) VALUES (%s, %s)", (username, password_hash))
conn.commit()
def verify_admin_login(username, password):
"""
Verifikasi login admin. Return True jika username & password cocok, else False.
"""
password_hash = hash_password(password)
with get_connection() as conn:
cur = conn.cursor()
cur.execute("SELECT id FROM admin WHERE username = %s AND password_hash = %s", (username, password_hash))
result = cur.fetchone()
return result is not None
def verify_admin_login_detail(username, password):
"""
Verifikasi login admin dengan status detail.
Return salah satu dari:
- 'success'
- 'username_not_found'
- 'wrong_password'
- 'wrong_username_and_password'
"""
input_password_hash = hash_password(password)
with get_connection() as conn:
cur = conn.cursor()
cur.execute("SELECT password_hash FROM admin WHERE username = %s", (username,))
result = cur.fetchone()
if result is None:
cur.execute("SELECT id FROM admin WHERE password_hash = %s", (input_password_hash,))
password_match = cur.fetchone()
if password_match is not None:
return 'username_not_found'
return 'wrong_username_and_password'
stored_hash = result[0]
if stored_hash != input_password_hash:
return 'wrong_password'
return 'success'
import psycopg2
from contextlib import contextmanager
DB_CONFIG = {
'host': 'localhost',
'database': 'db_klasifikasi',
'user': 'postgres',
'password': 'admin'
}
@contextmanager
def get_connection():
"""
Context manager untuk koneksi database PostgreSQL.
Otomatis menutup koneksi setelah digunakan.
Akan melempar Exception jika gagal koneksi.
"""
conn = None
try:
conn = psycopg2.connect(**DB_CONFIG)
yield conn
except psycopg2.Error as e:
raise Exception(f"Gagal koneksi database: {e}")
finally:
if conn:
conn.close()
def test_connection():
"""Cek koneksi ke database, return True jika sukses, False jika gagal."""
try:
with get_connection() as conn:
cur = conn.cursor()
cur.execute("SELECT 1")
return True
except Exception:
return False
# untuk simpan hasil prediksi
def simpan_hasil_prediksi(nama, nilai_dict, paket_prediksi):
with get_connection() as conn:
cur = conn.cursor()
sql = """
INSERT INTO hasil_prediksi (
nama_siswa,
nilai_informatika_sem1, nilai_informatika_sem2, nilai_informatika,
nilai_fisika_sem1, nilai_fisika_sem2, nilai_fisika,
nilai_kimia_sem1, nilai_kimia_sem2, nilai_kimia,
nilai_biologi_sem1, nilai_biologi_sem2, nilai_biologi,
nilai_big_lanjut_sem1, nilai_big_lanjut_sem2, nilai_big_lanjut,
nilai_ekonomi_sem1, nilai_ekonomi_sem2, nilai_ekonomi,
nilai_mat_lanjut_sem1, nilai_mat_lanjut_sem2, nilai_mat_lanjut,
nilai_sej_lanjut_sem1, nilai_sej_lanjut_sem2, nilai_sej_lanjut,
nilai_sosiologi_sem1, nilai_sosiologi_sem2, nilai_sosiologi,
nilai_geografi_sem1, nilai_geografi_sem2, nilai_geografi,
paket_prediksi
) VALUES (
%s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
)
"""
values = (
str(nama),
# Informatika
float(nilai_dict.get('nilai_informatika_sem1', 0)),
float(nilai_dict.get('nilai_informatika_sem2', 0)),
float(nilai_dict.get('nilai_informatika', 0)),
# Fisika
float(nilai_dict.get('nilai_fisika_sem1', 0)),
float(nilai_dict.get('nilai_fisika_sem2', 0)),
float(nilai_dict.get('nilai_fisika', 0)),
# Kimia
float(nilai_dict.get('nilai_kimia_sem1', 0)),
float(nilai_dict.get('nilai_kimia_sem2', 0)),
float(nilai_dict.get('nilai_kimia', 0)),
# Biologi
float(nilai_dict.get('nilai_biologi_sem1', 0)),
float(nilai_dict.get('nilai_biologi_sem2', 0)),
float(nilai_dict.get('nilai_biologi', 0)),
# BIG Lanjut
float(nilai_dict.get('nilai_big_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_big_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_big_lanjut', 0)),
# Ekonomi
float(nilai_dict.get('nilai_ekonomi_sem1', 0)),
float(nilai_dict.get('nilai_ekonomi_sem2', 0)),
float(nilai_dict.get('nilai_ekonomi', 0)),
# MAT Lanjut
float(nilai_dict.get('nilai_mat_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_mat_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_mat_lanjut', 0)),
# SEJ Lanjut
float(nilai_dict.get('nilai_sej_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_sej_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_sej_lanjut', 0)),
# Sosiologi
float(nilai_dict.get('nilai_sosiologi_sem1', 0)),
float(nilai_dict.get('nilai_sosiologi_sem2', 0)),
float(nilai_dict.get('nilai_sosiologi', 0)),
# Geografi
float(nilai_dict.get('nilai_geografi_sem1', 0)),
float(nilai_dict.get('nilai_geografi_sem2', 0)),
float(nilai_dict.get('nilai_geografi', 0)),
int(paket_prediksi)
)
cur.execute(sql, values)
conn.commit() #membuat data masuk ke database
def ambil_semua_hasil_prediksi():
"""
Ambil semua data hasil prediksi dari tabel hasil_prediksi.
Return: list of dict.
"""
with get_connection() as conn:
cur = conn.cursor()
sql = "SELECT * FROM hasil_prediksi ORDER BY id DESC"
cur.execute(sql)
columns = [desc[0] for desc in cur.description]
rows = cur.fetchall()
return [dict(zip(columns, row)) for row in rows]
#untuk update hasil prediksi di database berdasarkan id.
def update_hasil_prediksi(id, nama, nilai_dict, paket_prediksi):
with get_connection() as conn:
cur = conn.cursor()
sql = """
UPDATE hasil_prediksi SET
nama_siswa = %s,
nilai_informatika_sem1 = %s, nilai_informatika_sem2 = %s, nilai_informatika = %s,
nilai_fisika_sem1 = %s, nilai_fisika_sem2 = %s, nilai_fisika = %s,
nilai_kimia_sem1 = %s, nilai_kimia_sem2 = %s, nilai_kimia = %s,
nilai_biologi_sem1 = %s, nilai_biologi_sem2 = %s, nilai_biologi = %s,
nilai_big_lanjut_sem1 = %s, nilai_big_lanjut_sem2 = %s, nilai_big_lanjut = %s,
nilai_ekonomi_sem1 = %s, nilai_ekonomi_sem2 = %s, nilai_ekonomi = %s,
nilai_mat_lanjut_sem1 = %s, nilai_mat_lanjut_sem2 = %s, nilai_mat_lanjut = %s,
nilai_sej_lanjut_sem1 = %s, nilai_sej_lanjut_sem2 = %s, nilai_sej_lanjut = %s,
nilai_sosiologi_sem1 = %s, nilai_sosiologi_sem2 = %s, nilai_sosiologi = %s,
nilai_geografi_sem1 = %s, nilai_geografi_sem2 = %s, nilai_geografi = %s,
paket_prediksi = %s
WHERE id = %s
"""
values = (
str(nama),
# Informatika
float(nilai_dict.get('nilai_informatika_sem1', 0)),
float(nilai_dict.get('nilai_informatika_sem2', 0)),
float(nilai_dict.get('nilai_informatika', 0)),
# Fisika
float(nilai_dict.get('nilai_fisika_sem1', 0)),
float(nilai_dict.get('nilai_fisika_sem2', 0)),
float(nilai_dict.get('nilai_fisika', 0)),
# Kimia
float(nilai_dict.get('nilai_kimia_sem1', 0)),
float(nilai_dict.get('nilai_kimia_sem2', 0)),
float(nilai_dict.get('nilai_kimia', 0)),
# Biologi
float(nilai_dict.get('nilai_biologi_sem1', 0)),
float(nilai_dict.get('nilai_biologi_sem2', 0)),
float(nilai_dict.get('nilai_biologi', 0)),
# BIG Lanjut
float(nilai_dict.get('nilai_big_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_big_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_big_lanjut', 0)),
# Ekonomi
float(nilai_dict.get('nilai_ekonomi_sem1', 0)),
float(nilai_dict.get('nilai_ekonomi_sem2', 0)),
float(nilai_dict.get('nilai_ekonomi', 0)),
# MAT Lanjut
float(nilai_dict.get('nilai_mat_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_mat_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_mat_lanjut', 0)),
# SEJ Lanjut
float(nilai_dict.get('nilai_sej_lanjut_sem1', 0)),
float(nilai_dict.get('nilai_sej_lanjut_sem2', 0)),
float(nilai_dict.get('nilai_sej_lanjut', 0)),
# Sosiologi
float(nilai_dict.get('nilai_sosiologi_sem1', 0)),
float(nilai_dict.get('nilai_sosiologi_sem2', 0)),
float(nilai_dict.get('nilai_sosiologi', 0)),
# Geografi
float(nilai_dict.get('nilai_geografi_sem1', 0)),
float(nilai_dict.get('nilai_geografi_sem2', 0)),
float(nilai_dict.get('nilai_geografi', 0)),
int(paket_prediksi),
int(id)
)
cur.execute(sql, values)
conn.commit()
def hapus_hasil_prediksi(id):
"""
Hapus data hasil prediksi berdasarkan id.
Akan melempar Exception jika data tidak ditemukan.
"""
with get_connection() as conn:
cur = conn.cursor()
sql = "DELETE FROM hasil_prediksi WHERE id = %s"
cur.execute(sql, (int(id),))
if cur.rowcount == 0:
raise Exception(f"Data dengan ID {id} tidak ditemukan")
conn.commit()
def backup_database(backup_path):
"""
Backup seluruh data hasil_prediksi ke file CSV.
Return True jika sukses, False jika gagal.
"""
try:
data = ambil_semua_hasil_prediksi()
import pandas as pd
df = pd.DataFrame(data)
df.to_csv(backup_path, index=False)
return True
except Exception:
return False
def get_statistics():
"""
Ambil statistik ringkas dari tabel hasil_prediksi.
Return dict statistik atau None jika gagal.
"""
try:
with get_connection() as conn:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM hasil_prediksi")
total_siswa = cur.fetchone()[0]
cur.execute("""
SELECT paket_prediksi, COUNT(*) as jumlah
FROM hasil_prediksi
GROUP BY paket_prediksi
ORDER BY paket_prediksi
""")
distribusi_paket = cur.fetchall()
cur.execute("""
SELECT
AVG(nilai_informatika), AVG(nilai_fisika), AVG(nilai_kimia), AVG(nilai_biologi),
AVG(nilai_big_lanjut), AVG(nilai_ekonomi), AVG(nilai_mat_lanjut), AVG(nilai_sej_lanjut),
AVG(nilai_sosiologi), AVG(nilai_geografi)
FROM hasil_prediksi
""")
avg_nilai = cur.fetchone()
return {
'total_siswa': total_siswa,
'distribusi_paket': distribusi_paket,
'rata_rata_nilai': avg_nilai
}
except Exception:
return None