Files
infoscreen/server/init_db.py
2025-07-15 10:45:56 +00:00

60 lines
1.8 KiB
Python

import sys
sys.path.insert(0, '/workspace')
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from models.models import Base, User, UserRole
import os
from dotenv import load_dotenv
import bcrypt
# .env laden
load_dotenv()
# Konfiguration aus .env
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOST = os.getenv("DB_HOST")
DB_NAME = os.getenv("DB_NAME")
DEFAULT_ADMIN_USERNAME = os.getenv("DEFAULT_ADMIN_USERNAME")
DEFAULT_ADMIN_PASSWORD = os.getenv("DEFAULT_ADMIN_PASSWORD")
# Erst ohne DB verbinden, um sie ggf. zu erstellen
admin_conn_str = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/"
admin_engine = create_engine(admin_conn_str, echo=True)
with admin_engine.connect() as conn:
conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {DB_NAME} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"))
# Jetzt mit Datenbank verbinden
db_conn_str = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}"
engine = create_engine(db_conn_str, echo=True)
# Tabellen anlegen
# Base.metadata.create_all(engine) # wird von alembic verwaltet
# Session erstellen
Session = sessionmaker(bind=engine)
session = Session()
# Prüfen, ob der User bereits existiert
existing_user = session.query(User).filter_by(username=DEFAULT_ADMIN_USERNAME).first()
if not existing_user:
# Passwort hashen
hashed_pw = bcrypt.hashpw(DEFAULT_ADMIN_PASSWORD.encode('utf-8'), bcrypt.gensalt())
# Neuen User anlegen
admin_user = User(
username=DEFAULT_ADMIN_USERNAME,
password_hash=hashed_pw.decode('utf-8'),
role=UserRole.admin
)
session.add(admin_user)
session.commit()
print(f"Admin-User '{DEFAULT_ADMIN_USERNAME}' wurde erfolgreich angelegt.")
else:
print(f"Admin-User '{DEFAULT_ADMIN_USERNAME}' existiert bereits.")
session.close()