Nirodan
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
# util/db_config.py
|
|
import json
|
|
import os
|
|
import mysql.connector
|
|
from util.logger import logger
|
|
|
|
|
|
def _resolve_config_path() -> str:
|
|
"""
|
|
Prefer an explicit env override, otherwise use a docker-friendly default
|
|
(/config) and fall back to the repo-local config folder for non-docker dev.
|
|
"""
|
|
if env_path := os.environ.get("DB_CONFIG_PATH"):
|
|
return os.path.abspath(env_path)
|
|
|
|
docker_path = "/config/db_config.json"
|
|
if os.path.exists("/config"):
|
|
return docker_path
|
|
|
|
# local fallback: backend/config/db_config.json (relative to this file)
|
|
return os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "config", "db_config.json")
|
|
)
|
|
|
|
|
|
CONFIG_PATH = _resolve_config_path()
|
|
|
|
def is_configured():
|
|
return os.path.exists(CONFIG_PATH)
|
|
|
|
def load_config():
|
|
try:
|
|
with open(CONFIG_PATH, "r") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"Failed to load database config: {e}")
|
|
return None
|
|
|
|
def save_config(data):
|
|
try:
|
|
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
|
|
with open(CONFIG_PATH, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
logger.info("Database config saved.")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save database config: {e}")
|
|
|
|
def test_connection(config):
|
|
try:
|
|
conn = mysql.connector.connect(**config)
|
|
conn.close()
|
|
logger.info("Database connection successful.")
|
|
return True
|
|
except Exception as e:
|
|
logger.warning(f"Database connection failed: {e}")
|
|
return False
|
|
|
|
def initialize_admin_user(config):
|
|
try:
|
|
conn = mysql.connector.connect(**config)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(50) NOT NULL UNIQUE,
|
|
password VARCHAR(255) NOT NULL,
|
|
role VARCHAR(20) NOT NULL
|
|
)
|
|
""")
|
|
cursor.execute("SELECT COUNT(*) FROM users WHERE username = 'admin'")
|
|
if cursor.fetchone()[0] == 0:
|
|
from werkzeug.security import generate_password_hash
|
|
cursor.execute(
|
|
"INSERT INTO users (username, password, role) VALUES (%s, %s, %s)",
|
|
("admin", generate_password_hash("admin"), "admin")
|
|
)
|
|
logger.info("Admin user created (username: admin, password: admin)")
|
|
conn.commit()
|
|
cursor.close()
|
|
conn.close()
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize admin user: {e}")
|