# database.py — MySQL Connection & CRUD Operations for Scraper and Forecast import pandas as pd import sqlalchemy import bcrypt from sqlalchemy import create_engine, text from app.config import get_settings settings = get_settings() # Global Singleton DB Engine engine = create_engine( settings.DATABASE_URL, pool_pre_ping=True, pool_recycle=3600 ) def get_db_engine(): return engine def get_engine(): """Compatibility alias for forecast operations.""" return engine # ── READ OPERATIONS ─────────────────────────────────────────────────────────── def get_all_komoditas(): """Ambil semua komoditas aktif.""" return pd.read_sql( "SELECT id, nama, slug, unit, volatile, volatilitas_skor FROM komoditas WHERE is_active=1", engine ) def get_harga_harian(komoditas_id: int, days: int = 730): """Ambil data harga harian untuk satu komoditas (default 2 tahun terakhir).""" query = f""" SELECT hh.tanggal AS date, p.nama AS pasar_nama, hh.harga FROM harga_harian hh JOIN pasar p ON p.id = hh.pasar_id WHERE hh.komoditas_id = {komoditas_id} AND hh.tanggal >= DATE_SUB(CURDATE(), INTERVAL {days} DAY) ORDER BY hh.tanggal ASC """ df_raw = pd.read_sql(query, engine) if df_raw.empty: return pd.DataFrame() # Pivot → wide format df = df_raw.pivot_table( index='date', columns='pasar_nama', values='harga' ).reset_index() df.columns.name = None df['date'] = pd.to_datetime(df['date']) df = df.sort_values('date').reset_index(drop=True) return df def get_pasar_list(komoditas_id: int): """Ambil daftar pasar yang punya data untuk komoditas ini.""" return pd.read_sql(f""" SELECT DISTINCT p.id, p.nama FROM harga_harian hh JOIN pasar p ON p.id = hh.pasar_id WHERE hh.komoditas_id = {komoditas_id} """, engine) # ── WRITE OPERATIONS ────────────────────────────────────────────────────────── def save_model_ml(data: dict): """Upsert ke tabel model_ml.""" with engine.begin() as conn: conn.execute(text(""" INSERT INTO model_ml ( komoditas_id, nama_model, versi, deskripsi, file_path, mape, rmse, mae, r2_score, confidence_level, stabilitas, status_validasi, catatan_validasi, tanggal_training, tanggal_evaluasi, is_active, created_at, updated_at ) VALUES ( :komoditas_id, :nama_model, :versi, :deskripsi, :file_path, :mape, :rmse, :mae, :r2_score, :confidence_level, :stabilitas, :status_validasi, :catatan_validasi, :tanggal_training, :tanggal_evaluasi, 1, NOW(), NOW() ) ON DUPLICATE KEY UPDATE nama_model = VALUES(nama_model), versi = VALUES(versi), deskripsi = VALUES(deskripsi), file_path = VALUES(file_path), mape = VALUES(mape), rmse = VALUES(rmse), mae = VALUES(mae), r2_score = VALUES(r2_score), confidence_level = VALUES(confidence_level), stabilitas = VALUES(stabilitas), status_validasi = VALUES(status_validasi), catatan_validasi = VALUES(catatan_validasi), tanggal_evaluasi = VALUES(tanggal_evaluasi), updated_at = NOW() """), data) with engine.connect() as conn: return conn.execute( text("SELECT id FROM model_ml WHERE komoditas_id=:kid ORDER BY updated_at DESC LIMIT 1"), {'kid': data['komoditas_id']} ).scalar() def save_hasil_prediksi(predictions: list, komoditas_id: int, pasar_id, model_id: int, meta: dict): """Hapus prediksi lama & insert 7 baris baru.""" with engine.begin() as conn: conn.execute(text(""" DELETE FROM hasil_prediksi WHERE komoditas_id = :kid AND pasar_id = :pid AND tanggal_target >= CURDATE() """), {'kid': komoditas_id, 'pid': pasar_id}) for p in predictions: conn.execute(text(""" INSERT INTO hasil_prediksi ( komoditas_id, pasar_id, model_id, tanggal_prediksi, tanggal_target, harga_prediksi, confidence_level, model_name, mape, rmse, created_at ) VALUES ( :kid, :pid, :model_id, CURDATE(), :tanggal_target, :harga_prediksi, :confidence_level, :model_name, :mape, :rmse, NOW() ) """), { 'kid' : komoditas_id, 'pid' : pasar_id, 'model_id' : model_id, 'tanggal_target' : p['tanggal'], 'harga_prediksi' : p['harga_prediksi'], 'confidence_level': meta['confidence_level'], 'model_name' : meta['nama_model'], 'mape' : meta['mape'], 'rmse' : meta['rmse'], }) def save_ringkasan_prediksi(data: dict, komoditas_id: int, model_id: int): """Upsert ringkasan prediksi.""" with engine.begin() as conn: conn.execute(text(""" INSERT INTO ringkasan_prediksi ( komoditas_id, model_id, harga_min, harga_max, tren, confidence_level, status_analisis, deskripsi_status, tanggal_mulai, tanggal_akhir, created_at ) VALUES ( :komoditas_id, :model_id, :harga_min, :harga_max, :tren, :confidence_level, :status_analisis, :deskripsi_status, :tanggal_mulai, :tanggal_akhir, NOW() ) ON DUPLICATE KEY UPDATE harga_min = VALUES(harga_min), harga_max = VALUES(harga_max), tren = VALUES(tren), confidence_level = VALUES(confidence_level), status_analisis = VALUES(status_analisis), deskripsi_status = VALUES(deskripsi_status), tanggal_mulai = VALUES(tanggal_mulai), tanggal_akhir = VALUES(tanggal_akhir) """), {**data, 'komoditas_id': komoditas_id, 'model_id': model_id}) def save_insight_prediksi(insights: list, komoditas_id: int, model_id: int): """Hapus insight lama & insert baru.""" with engine.begin() as conn: conn.execute(text( "DELETE FROM insight_prediksi WHERE komoditas_id = :kid" ), {'kid': komoditas_id}) for ins in insights: conn.execute(text(""" INSERT INTO insight_prediksi ( komoditas_id, model_id, konten, tipe, ikon, urutan, is_active, created_at, updated_at ) VALUES ( :kid, :model_id, :konten, :tipe, :ikon, :urutan, 1, NOW(), NOW() ) """), { 'kid' : komoditas_id, 'model_id': model_id, 'konten' : ins['konten'], 'tipe' : ins['tipe'], 'ikon' : ins['ikon'], 'urutan' : ins['urutan'], }) def update_komoditas_volatilitas(komoditas_id: int, volatile: int, skor: float): with engine.begin() as conn: conn.execute(text(""" UPDATE komoditas SET volatile=:v, volatilitas_skor=:s, updated_at=NOW() WHERE id=:kid """), {'v': volatile, 's': skor, 'kid': komoditas_id}) def save_all_model_results(all_results: dict, best_meta: dict, komoditas_id: int, pasar_id: int): """Save ALL model results to model_ml (not just best). Best gets is_active=1.""" best_name = best_meta.get('nama_model', '') with engine.begin() as conn: # Deactivate old models for this komoditas conn.execute(text( "UPDATE model_ml SET is_active=0 WHERE komoditas_id=:kid" ), {'kid': komoditas_id}) # Insert each model result for model_name, metrics in all_results.items(): is_best = 1 if model_name == best_name or f"{model_name}_Tuned" == best_name else 0 # Hitung stabilitas sederhana untuk kolom ENUM agar tidak error mape_val = metrics.get('mape', 0) stabilitas_label = ( 'optimal' if mape_val < 2 else 'baik' if mape_val < 5 else 'cukup' if mape_val < 10 else 'perlu_retrain' ) conn.execute(text(""" INSERT INTO model_ml ( komoditas_id, nama_model, versi, deskripsi, file_path, mape, rmse, mae, r2_score, confidence_level, stabilitas, status_validasi, catatan_validasi, tanggal_training, tanggal_evaluasi, is_active, created_at, updated_at ) VALUES ( :komoditas_id, :nama_model, '1.0', :deskripsi, '', :mape, :rmse, :mae, :r2_score, 0, :stabilitas, 'terverifikasi', '', CURDATE(), CURDATE(), :is_active, NOW(), NOW() ) """), { 'komoditas_id': komoditas_id, 'nama_model': model_name, 'deskripsi': f"Model {model_name} untuk komoditas ID {komoditas_id}", 'mape': round(mape_val, 4), 'rmse': round(metrics.get('rmse', 0), 4), 'mae': round(metrics.get('mae', 0), 4), 'r2_score': round(metrics.get('r2', 0), 4), 'stabilitas': stabilitas_label, 'is_active': is_best, }) # ── READ OPERATIONS FOR FRONTEND API ────────────────────────────────────────── def get_prediksi_data(komoditas_id: int): """Ambil semua data yang dibutuhkan frontend untuk satu komoditas.""" komoditas = pd.read_sql(f""" SELECT k.*, kk.nama as kategori_nama FROM komoditas k JOIN kategori_komoditas kk ON kk.id = k.kategori_id WHERE k.id = {komoditas_id} """, engine).to_dict('records') komoditas = komoditas[0] if komoditas else None harga_terkini = pd.read_sql(f""" SELECT hh.harga, hh.tanggal, p.nama as pasar_nama, p.id as pasar_id FROM harga_harian hh JOIN pasar p ON p.id = hh.pasar_id WHERE hh.komoditas_id = {komoditas_id} ORDER BY hh.tanggal DESC LIMIT 1 """, engine).to_dict('records') harga_terkini = harga_terkini[0] if harga_terkini else None # Best (active) model model = pd.read_sql(f""" SELECT * FROM model_ml WHERE komoditas_id={komoditas_id} AND is_active=1 ORDER BY updated_at DESC LIMIT 1 """, engine).to_dict('records') model = model[0] if model else None # ALL models for comparison (all runs) all_models = pd.read_sql(f""" SELECT nama_model, mape, rmse, mae, r2_score, is_active, tanggal_training, created_at FROM model_ml WHERE komoditas_id={komoditas_id} ORDER BY created_at DESC, rmse ASC """, engine).to_dict('records') ringkasan = pd.read_sql(f""" SELECT * FROM ringkasan_prediksi WHERE komoditas_id={komoditas_id} ORDER BY created_at DESC LIMIT 1 """, engine).to_dict('records') ringkasan = ringkasan[0] if ringkasan else None hasil = pd.read_sql(f""" SELECT tanggal_target, harga_prediksi, confidence_level FROM hasil_prediksi WHERE komoditas_id={komoditas_id} AND tanggal_target >= CURDATE() ORDER BY tanggal_target ASC """, engine).to_dict('records') insights = pd.read_sql(f""" SELECT * FROM insight_prediksi WHERE komoditas_id={komoditas_id} AND is_active=1 ORDER BY urutan ASC """, engine).to_dict('records') return { 'komoditas' : komoditas, 'harga_terkini' : harga_terkini, 'model' : model, 'all_models' : all_models, 'ringkasan' : ringkasan, 'prediksi_7hari': hasil, 'insights' : insights, } def verify_user_mysql(email: str, password_input: str) -> dict: """ Verify user credential using MySQL 'users' table. Checks email, bcrypt password hash (Laravel format), and validates super_admin/admin role. Auto-discovers actual column names from INFORMATION_SCHEMA to handle varying schemas. Returns user dict on success, raises Exception otherwise. """ with engine.connect() as conn: # Step 1: Discover actual column names in the 'users' table try: col_query = text(""" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'users' """) columns = [row[0].lower() for row in conn.execute(col_query).fetchall()] except Exception as e: raise Exception(f"Gagal membaca skema tabel users: {str(e)}") if not columns: raise Exception("Tabel 'users' tidak ditemukan di database") # Step 2: Map logical fields to actual column names # Name column: could be 'name', 'nama', 'full_name', 'username' name_col = None for candidate in ['name', 'nama', 'full_name', 'username', 'nama_lengkap']: if candidate in columns: name_col = candidate break if not name_col: name_col = 'email' # fallback to email as display name # Password column: could be 'password', 'kata_sandi', 'sandi', 'passwd' password_col = None for candidate in ['password', 'password_hash', 'kata_sandi', 'sandi', 'passwd', 'pass']: if candidate in columns: password_col = candidate break if not password_col: raise Exception(f"Kolom password tidak ditemukan di tabel users. Kolom yang ada: {', '.join(columns)}") # Role column: could be 'role', 'roles', 'user_role', 'level', 'tipe' role_col = None for candidate in ['role', 'roles', 'user_role', 'level', 'tipe', 'type']: if candidate in columns: role_col = candidate break if not role_col: raise Exception(f"Kolom role tidak ditemukan di tabel users. Kolom yang ada: {', '.join(columns)}") # Step 3: Build and execute the query try: sql = f"SELECT id, `{name_col}` as name, email, `{password_col}` as password, `{role_col}` as role FROM users WHERE email = :email LIMIT 1" res = conn.execute(text(sql), {"email": email}).fetchone() except Exception as e: raise Exception(f"Database error querying users table: {str(e)}") if not res: raise Exception("Email tidak terdaftar") user_data = dict(res._mapping) hashed_password = user_data.get("password") if not hashed_password: raise Exception("Password hash tidak ditemukan di database") # Step 4: Verify password using bcrypt try: # Laravel uses $2y$ prefix, python bcrypt expects $2b$ or $2a$ compat_hash = hashed_password if hashed_password.startswith("$2y$"): compat_hash = "$2b$" + hashed_password[4:] is_valid = bcrypt.checkpw( password_input.encode('utf-8'), compat_hash.encode('utf-8') ) except Exception as hash_err: raise Exception(f"Gagal memverifikasi password hash: {str(hash_err)}") if not is_valid: raise Exception("Password salah") # Step 5: Check role - must be admin or super_admin role = str(user_data.get("role", "")).lower().strip() if role not in ["admin", "super_admin", "superadmin", "super-admin"]: raise Exception(f"Akses ditolak: Role '{role}' tidak memiliki izin administrator") return { "id": user_data.get("id"), "name": user_data.get("name"), "email": user_data.get("email"), "role": user_data.get("role") } def get_active_models_summary() -> dict: """ Get summary of all active models and commodity model list for AI Center page. """ with engine.connect() as conn: try: query = text(""" SELECT k.id as komoditas_id, k.nama as komoditas_nama, k.unit, k.volatile, m.nama_model, m.mape, m.rmse, m.mae, m.r2_score, m.status_validasi, m.tanggal_training, m.is_active FROM komoditas k LEFT JOIN model_ml m ON m.komoditas_id = k.id AND m.is_active = 1 WHERE k.is_active = 1 ORDER BY k.nama ASC """) rows = conn.execute(query).fetchall() except Exception as e: raise Exception(f"Database error querying active models: {str(e)}") results = [] for r in rows: d = dict(r._mapping) # handle date serialization nicely if d.get("tanggal_training"): d["tanggal_training"] = str(d["tanggal_training"]) results.append(d) # Compute averages active_mapes = [r['mape'] for r in results if r['mape'] is not None] avg_mape = round(sum(active_mapes) / len(active_mapes), 2) if active_mapes else 0.0 active_count = sum(1 for r in results if r['is_active'] == 1) latest_date = None for r in results: if r['tanggal_training']: dt_str = str(r['tanggal_training']) if not latest_date or dt_str > latest_date: latest_date = dt_str return { "total_active_models": active_count, "average_mape": avg_mape, "latest_inference": latest_date, "commodity_models": results }