Add navigation in sidebar

This commit is contained in:
handi
2025-11-30 23:03:25 +01:00
parent 5277830c50
commit ed04a75ce2
12 changed files with 643 additions and 472 deletions

View File

@@ -6,30 +6,6 @@ import bcrypt
from db import get_conn #, create_user, verify_user, get_role_for_user
# ---------------------------------------------------------------------------
# DB Initialisierung: `users`-Tabelle
# ---------------------------------------------------------------------------
# def init_auth_db():
# """
# Legt die users-Tabelle an.
# WICHTIG: password_hash wird als TEXT gespeichert, damit sowohl
# streamlit-authenticator als auch dein eigener bcrypt-Code kompatibel sind.
# """
# with closing(get_conn()) as conn, conn:
# conn.execute(
# """
# CREATE TABLE IF NOT EXISTS users (
# id INTEGER PRIMARY KEY AUTOINCREMENT,
# username TEXT UNIQUE NOT NULL,
# email TEXT,
# password_hash TEXT NOT NULL,
# role TEXT NOT NULL DEFAULT 'user'
# )
# """
# )
# ---------------------------------------------------------------------------
# Benutzer anlegen
# ---------------------------------------------------------------------------
@@ -97,6 +73,17 @@ def get_role_for_user(username: str) -> str:
).fetchone()
return row[0] if row else "user"
# ---------------------------------------------------------------------------
# Fullname für einen Benutzer holen (für Streamlit-Inhalte)
# ---------------------------------------------------------------------------
def get_fullname_for_user(username: str) -> str:
with closing(get_conn()) as conn:
row = conn.execute(
"SELECT firstname ||' '|| lastname as fullname FROM users WHERE username = ?",
(username,),
).fetchone()
return row[0] if row else "user"
# ---------------------------------------------------------------------------
# Credential-Lader für streamlit-authenticator

View File

@@ -1,212 +0,0 @@
# app/auth.py
import sqlite3
import time
import secrets
from contextlib import closing
import bcrypt
import streamlit as st
from db import get_conn
SESSION_LIFETIME_SECONDS = 8 * 60 * 60 # 8 Stunden
# ---------- User-DB ----------
def init_auth_db():
"""Legt die users-Tabelle an, falls sie noch nicht existiert."""
with closing(get_conn()) as conn, conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash BLOB NOT NULL,
role TEXT NOT NULL DEFAULT 'user'
)
"""
)
# Sessions-Tabelle
conn.execute(
"""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
username TEXT NOT NULL,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL,
FOREIGN KEY (username) REFERENCES users(username) ON DELETE CASCADE
)
"""
)
def create_user(username: str, password: str, role: str = "user") -> bool:
pw_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
try:
with closing(get_conn()) as conn, conn:
conn.execute(
"INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)",
(username, pw_hash, role),
)
return True
except sqlite3.IntegrityError:
return False
def verify_user(username: str, password: str):
with closing(get_conn()) as conn:
row = conn.execute(
"SELECT password_hash, role FROM users WHERE username = ?",
(username,),
).fetchone()
if not row:
return False, None
stored_hash, role = row
ok = bcrypt.checkpw(password.encode("utf-8"), stored_hash)
return (ok, role) if ok else (False, None)
# ---------- Session-DB ----------
def create_session(username: str) -> str:
"""Erzeugt eine neue Session-ID und speichert sie in der DB."""
sid = secrets.token_urlsafe(32) # schön zufällig, URL-tauglich
now = int(time.time())
expires_at = now + SESSION_LIFETIME_SECONDS
with closing(get_conn()) as conn, conn:
conn.execute(
"""
INSERT INTO sessions (id, username, created_at, expires_at)
VALUES (?, ?, ?, ?)
""",
(sid, username, now, expires_at),
)
return sid
def get_session(sid: str):
"""Liefert (username, role), wenn Session existiert und gültig ist, sonst (None, None)."""
now = int(time.time())
with closing(get_conn()) as conn:
row = conn.execute(
"""
SELECT s.username, u.role
FROM sessions s
JOIN users u ON u.username = s.username
WHERE s.id = ? AND s.expires_at > ?
""",
(sid, now),
).fetchone()
if not row:
return None, None
return row[0], row[1]
def delete_session(sid: str):
with closing(get_conn()) as conn, conn:
conn.execute("DELETE FROM sessions WHERE id = ?", (sid,))
# Optional: abgelaufene Sessions aufräumen
def cleanup_sessions():
now = int(time.time())
with closing(get_conn()) as conn, conn:
conn.execute("DELETE FROM sessions WHERE expires_at <= ?", (now,))
# ---------- SessionState-Helper ----------
def set_session_user(username: str, role: str, sid: str | None = None):
st.session_state["auth"] = True
st.session_state["username"] = username
st.session_state["role"] = role
if sid is not None:
st.session_state["session_id"] = sid
def clear_session_user():
for k in ["auth", "username", "role", "session_id"]:
st.session_state.pop(k, None)
def is_authenticated() -> bool:
return bool(st.session_state.get("auth"))
def current_user():
return (
st.session_state.get("username"),
st.session_state.get("role"),
)
def ensure_logged_in():
"""
Am Anfang jeder Page aufrufen.
Prüft erst st.session_state, dann ?session_id=... in der URL.
"""
if is_authenticated():
return
# Versuch: session_id aus URL
params = st.experimental_get_query_params()
sid_list = params.get("session_id")
sid = sid_list[0] if sid_list else None
if sid:
username, role = get_session(sid)
if username:
set_session_user(username, role, sid)
return
# Wenn wir hier landen: kein gültiger Login → Login-View anzeigen
login_view()
st.stop()
# ---------- UI ----------
def login_view():
st.title("Intranet Login")
with st.form("login"):
u = st.text_input("Username")
p = st.text_input("Passwort", type="password")
submitted = st.form_submit_button("Anmelden")
if submitted:
username = u.strip()
ok, role = verify_user(username, p)
if ok:
# neue Session in DB
sid = create_session(username)
# SessionState setzen
set_session_user(username, role, sid)
# URL-Parameter setzen (Arme-Leute-Cookie)
st.experimental_set_query_params(session_id=sid)
st.success("Erfolgreich angemeldet.")
st.rerun()
else:
st.error("Login fehlgeschlagen.")
def authed_header():
username, role = current_user()
if not username:
return
st.sidebar.write(f"Angemeldet als **{username}** ({role})")
if st.sidebar.button("Logout"):
# Session in DB löschen, falls vorhanden
sid = st.session_state.get("session_id")
if sid:
delete_session(sid)
clear_session_user()
# URL-Parameter entfernen
st.experimental_set_query_params()
st.rerun()

View File

@@ -1,94 +0,0 @@
import sqlite3
from contextlib import closing
import bcrypt
import streamlit as st
from db import get_conn
def create_user(username: str, password: str, role: str = "user") -> bool:
pw_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
try:
with closing(get_conn()) as conn, conn:
conn.execute(
"INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)",
(username, pw_hash, role),
)
return True
except sqlite3.IntegrityError:
return False
def verify_user(username: str, password: str):
with closing(get_conn()) as conn:
row = conn.execute(
"SELECT password_hash, role FROM users WHERE username = ?",
(username,),
).fetchone()
if not row:
return False, None
stored_hash, role = row
ok = bcrypt.checkpw(password.encode("utf-8"), stored_hash)
return (ok, role) if ok else (False, None)
# ---------- Session / Helper ----------
def set_session_user(username: str, role: str):
st.session_state["auth"] = True
st.session_state["username"] = username
st.session_state["role"] = role
def clear_session_user():
for k in ["auth", "username", "role"]:
st.session_state.pop(k, None)
def is_authenticated() -> bool:
return bool(st.session_state.get("auth"))
def current_user():
"""(username, role) oder (None, None)"""
return st.session_state.get("username"), st.session_state.get("role")
def ensure_logged_in():
"""
Am Anfang jeder Page aufrufen.
Wenn kein Login: Login-View anzeigen und Script stoppen.
"""
if not is_authenticated():
login_view()
st.stop()
# ---------- UI ----------
def login_view():
st.title("Intranet Login")
with st.form("login"):
u = st.text_input("Username")
p = st.text_input("Passwort", type="password")
submitted = st.form_submit_button("Anmelden")
if submitted:
ok, role = verify_user(u.strip(), p)
if ok:
set_session_user(u.strip(), role)
st.success("Erfolgreich angemeldet.")
st.rerun()
else:
st.error("Login fehlgeschlagen.")
def authed_header():
username, role = current_user()
if not username:
return
st.sidebar.write(f"Angemeldet als **{username}** ({role})")
if st.sidebar.button("Logout"):
st.session_state.pop("app_started_logged", None)
clear_session_user()
st.rerun()

View File

@@ -0,0 +1,34 @@
import streamlit as st
from auth import create_user
def render(username: str, role: str):
st.title("Benutzerverwaltung")
with st.expander("Neuen Nutzer anlegen"):
new_u = st.text_input("Neuer Username", key="new_u")
new_fname = st.text_input("Vorname", key="new_fname")
new_lname = st.text_input("Nachname", key="new_lname")
new_email = st.text_input("E-Mail", key="new_email")
new_p = st.text_input("Neues Passwort", type="password", key="new_p")
new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
if st.button("Anlegen"):
if new_u and new_p:
ok = create_user(
new_u.strip(),
new_p,
new_role,
new_email.strip() or None,
new_fname.strip() or None,
new_lname.strip() or None,
)
st.success("Nutzer angelegt.") if ok else st.error(
"Username bereits vorhanden oder Fehler."
)
else:
st.warning("Bitte Username und Passwort eingeben.")
st.subheader("Dein Bereich")
st.write(f"Personalisierter Content für **{username}**.")

View File

@@ -0,0 +1,24 @@
import importlib
def get_dashboard_renderer(code: str):
"""
Liefert die passende Render-Funktion zum Dashboard-Code.
- "home" -> interne Funktion home_dashboard
- alles andere -> Modul app.dashboards.<code> mit Funktion render(username, role)
"""
if code == "home":
return home_dashboard
module_name = f"app.dashboards.{code}"
try:
module = importlib.import_module(module_name)
except ModuleNotFoundError:
# Zum Debuggen:
st.error(f"Modul '{module_name}' wurde nicht gefunden.")
return None
if not hasattr(module, "render"):
st.error(f"Modul '{module_name}' hat keine Funktion render().")
return None
return module.render

View File

@@ -1,14 +1,62 @@
import importlib
import streamlit as st
import yaml
from yaml.loader import SafeLoader
import streamlit_authenticator as stauth
from auth import load_credentials_from_db, get_role_for_user, create_user, needs_password_change, update_password
from streamlit_authenticator.utilities.exceptions import LoginError
import pandas as pd
from contextlib import closing
from auth import (
load_credentials_from_db,
get_role_for_user,
create_user,
needs_password_change,
update_password,
get_fullname_for_user,
)
from version import __version__
from db import get_conn # an deine Struktur anpassen
def content_for(username: str, role: str):
# ---------------------------------------------------------------------------
# Dashboards aus der DB laden
# ---------------------------------------------------------------------------
def load_dashboards_df() -> pd.DataFrame:
"""Lädt alle aktiven Dashboards aus der DB als DataFrame."""
with closing(get_conn()) as conn:
df = pd.read_sql_query(
"""
SELECT
d.dash_id,
d.dash_text as code,
d.dash_text as title,
g.group_text as grp
from
dashboards d
left join groups g
on d.group_id = g.group_id
where
d.active = 1
order by
d.dash_id
""",
conn,
)
return df
# ---------------------------------------------------------------------------
# Home-Dashboard (als Funktion, kein eigenes .py nötig)
# ---------------------------------------------------------------------------
def home_dashboard(username: str, role: str):
st.header("Controlling-Portal")
st.info(f"Willkommen, {username}!")
if role == "admin":
st.subheader("Admin-Bereich")
st.write("Nur Admins sehen das hier.")
@@ -23,8 +71,17 @@ def content_for(username: str, role: str):
if st.button("Anlegen"):
if new_u and new_p:
ok = create_user(new_u.strip(), new_p, new_role, new_email, new_fname, new_lname)
st.success("Nutzer angelegt.") if ok else st.error("Username bereits vorhanden oder Fehler.")
ok = create_user(
new_u.strip(),
new_p,
new_role,
new_email.strip() or None,
new_fname.strip() or None,
new_lname.strip() or None,
)
st.success("Nutzer angelegt.") if ok else st.error(
"Username bereits vorhanden oder Fehler."
)
else:
st.warning("Bitte Username und Passwort eingeben.")
@@ -32,21 +89,98 @@ def content_for(username: str, role: str):
st.write(f"Personalisierter Content für **{username}**.")
# ---------------------------------------------------------------------------
# Dashboard-Resolver: code -> render-Funktion
# ---------------------------------------------------------------------------
def get_dashboard_renderer(code: str):
"""
Liefert die passende Render-Funktion zum Dashboard-Code.
Konvention:
- Home: interne Funktion home_dashboard(username, role)
- andere: Modul dashboards.<code> mit Funktion render(username, role)
"""
if code == "home":
return home_dashboard
module_name = f"dashboards.{code}"
try:
module = importlib.import_module(module_name)
except ModuleNotFoundError:
return None
# Wir erwarten eine Funktion render(username, role)
return getattr(module, "render", None)
# ---------------------------------------------------------------------------
# Sidebar: Home + Suche + DB-Dashboards (Gruppen-Expander)
# ---------------------------------------------------------------------------
def build_sidebar(authenticator, username: str, df_dashboards: pd.DataFrame):
role = get_role_for_user(username)
fullname = get_fullname_for_user(username)
with st.sidebar:
st.write(f"**{fullname}** ({role})")
authenticator.logout("Logout", "sidebar")
st.divider()
st.markdown("### Dashboards")
st.divider()
# Default-Auswahl
if "selected_dashboard_code" not in st.session_state:
st.session_state["selected_dashboard_code"] = "home"
# Home ist immer da
if st.button("🏠 Home", use_container_width=True):
st.session_state["selected_dashboard_code"] = "home"
st.markdown("---")
# Suchfeld für DB-Dashboards
query = st.text_input("Suche", placeholder="Dashboard suchen …")
filtered = df_dashboards.copy()
if query:
q = query.lower()
filtered = filtered[
filtered["title"].str.lower().str.contains(q, na=False)
| filtered["grp"].str.lower().str.contains(q, na=False)
| filtered["description"].fillna("").str.lower().str.contains(q, na=False)
]
if filtered.empty:
st.info("Keine Dashboards zur Suche gefunden.")
return
# Gruppen -> Expander, darin Buttons für Dashboards
for grp, grp_df in filtered.groupby("grp"):
with st.expander(grp, expanded=False):
for _, row in grp_df.iterrows():
code = row["code"]
title = row["title"]
if st.button(title, key=f"dash_btn_{code}", use_container_width=True):
st.session_state["selected_dashboard_code"] = code
# ---------------------------------------------------------------------------
# Login + App-Shell
# ---------------------------------------------------------------------------
def main():
st.set_page_config(
page_title=f"Co-App Start - V{__version__}",
page_icon="🔒",
layout="centered",
layout="wide",
)
# DB-Struktur sicherstellen
# init_auth_db()
# --- Config laden (Cookie, etc.) ---
# --- Authenticator vorbereiten ---
with open("config/auth.yaml", "r", encoding="utf-8") as f:
base_config = yaml.load(f, Loader=SafeLoader)
# --- Credentials dynamisch aus DB laden ---
db_creds = load_credentials_from_db()
base_config["credentials"] = db_creds
@@ -55,25 +189,20 @@ def main():
base_config["cookie"]["name"],
base_config["cookie"]["key"],
base_config["cookie"]["expiry_days"],
#base_config.get("preauthorized", {}),
)
#name, auth_status, username = authenticator.login(location="main", key="Login")
# login_result = authenticator.login(location="main", key="Login")
# if login_result is None:
# st.error("Login-Initialisierung fehlgeschlagen (keine gültigen Credentials?).")
# return
# name, auth_status, username = login_result
# --- Login-View zeichnen ---
try:
authenticator.login(location="main", key="Login")
except LoginError:
authenticator.logout("ForceLogout", "sidebar")
st.error("Sitzung ungültig. Bitte neu einloggen.")
return
auth_status = st.session_state.get("authentication_status")
name = st.session_state.get("name")
username = st.session_state.get("username")
if auth_status is False:
st.error("Login fehlgeschlagen.")
return
@@ -84,10 +213,10 @@ def main():
# ---- Ab hier eingeloggt (persistenter Cookie) ----
# 1) Passwortwechsel erzwingen?
if needs_password_change(username):
st.warning("Du musst dein Passwort ändern, bevor du die Anwendung nutzen kannst.")
# Damit das Formular nur einmal pro Run erscheint
with st.form("pw_change_form"):
pw1 = st.text_input("Neues Passwort", type="password")
pw2 = st.text_input("Neues Passwort (Wiederholung)", type="password")
@@ -106,20 +235,245 @@ def main():
update_password(username, pw1, reset_flag=True)
st.success("Passwort wurde geändert.")
# Optional: danach Seite einmal „sauber“ neu laden
st.rerun()
# Solange new_pwd=1 ist, KEINEN weiteren Content anzeigen
# solange new_pwd=1: keinen weiteren Inhalt anzeigen
return
# 2) Dashboards aus DB laden (ohne Home)
df_dashboards = load_dashboards_df()
# 3) Sidebar aufbauen (User, Logout, Navigation)
build_sidebar(authenticator, username, df_dashboards)
# 4) Ausgewähltes Dashboard rendern
role = get_role_for_user(username)
selected_code = st.session_state.get("selected_dashboard_code", "home")
renderer = get_dashboard_renderer(selected_code)
authenticator.logout("Logout", "sidebar")
st.sidebar.write(f"Angemeldet als **{name}** ({username}, Rolle: {role})")
if renderer is None:
st.error(f"Kein Dashboard-Modul für '{selected_code}' gefunden.")
return
content_for(username, role)
# Titel setzen
if selected_code == "home":
title = "Home"
else:
row = df_dashboards[df_dashboards["code"] == selected_code]
title = row["title"].iloc[0] if not row.empty else selected_code
st.title(title)
renderer(username, role)
if __name__ == "__main__":
main()
# import importlib
# import streamlit as st
# import yaml
# from yaml.loader import SafeLoader
# import streamlit_authenticator as stauth
# from streamlit_authenticator.utilities.exceptions import LoginError
# from auth import (
# load_credentials_from_db,
# get_role_for_user,
# create_user,
# needs_password_change,
# update_password,
# get_fullname_for_user,
# )
# from version import __version__
# # später: from db import get_conn, um Dashboards aus DB zu laden
# # ---------------------------------------------------------------------------
# # Home-Dashboard (als Funktion, kein eigenes .py nötig)
# # ---------------------------------------------------------------------------
# def home_dashboard(username: str, role: str):
# # st.header("Controlling-Portal")
# st.info(f"Willkommen, {username}!")
# # ---------------------------------------------------------------------------
# # Dashboard-Resolver (später: weitere Dashboards dynamisch laden)
# # ---------------------------------------------------------------------------
# def get_dashboard_renderer(code: str):
# """Liefert die passende Render-Funktion zum Dashboard-Code."""
# if code == "home":
# # internes Home-Dashboard
# return home_dashboard
# # Später: Module anhand 'code' aus dashboards-Package laden
# # Beispiel: dashboards/sales.py → code = "sales"
# try:
# module = importlib.import_module(f"dashboards.{code}")
# except ModuleNotFoundError:
# return None
# return getattr(module, "render", None)
# # ---------------------------------------------------------------------------
# # Sidebar Navigation (Home + später DB-Dashboards)
# # ---------------------------------------------------------------------------
# def build_sidebar(authenticator, username: str):
# role = get_role_for_user(username)
# fullname = get_fullname_for_user(username)
# with st.sidebar:
# st.write(f"**{fullname}** ({role})")
# authenticator.logout("Logout", "sidebar")
# st.divider()
# # Default-Auswahl
# if "selected_dashboard_code" not in st.session_state:
# st.session_state["selected_dashboard_code"] = "home"
# # Home ist immer da
# if st.button("🏠 Home", use_container_width=True):
# st.session_state["selected_dashboard_code"] = "home"
# # Platzhalter für Suche + DB-Dashboards:
# # hier später: Suchfeld + Expander aus DB-Result (DataFrame)
# # z.B.:
# # query = st.text_input("Dashboards suchen …")
# # df = load_dashboards_df()
# # ... filtern, gruppieren, Buttons setzen ...
# # ---------------------------------------------------------------------------
# # Login + App-Shell
# # ---------------------------------------------------------------------------
# def main():
# st.set_page_config(
# page_title=f"Co-App Start - V{__version__}",
# page_icon="🔒",
# layout="centered",
# )
# # --- Authenticator vorbereiten ---
# with open("config/auth.yaml", "r", encoding="utf-8") as f:
# base_config = yaml.load(f, Loader=SafeLoader)
# db_creds = load_credentials_from_db()
# base_config["credentials"] = db_creds
# authenticator = stauth.Authenticate(
# base_config["credentials"],
# base_config["cookie"]["name"],
# base_config["cookie"]["key"],
# base_config["cookie"]["expiry_days"],
# )
# # --- Login-View zeichnen ---
# try:
# authenticator.login(location="main", key="Login")
# except LoginError:
# authenticator.logout("ForceLogout", "sidebar")
# st.error("Sitzung ungültig. Bitte neu einloggen.")
# return
# auth_status = st.session_state.get("authentication_status")
# name = st.session_state.get("name")
# username = st.session_state.get("username")
# if auth_status is False:
# st.error("Login fehlgeschlagen.")
# return
# if auth_status is None:
# st.warning("Bitte Benutzername und Passwort eingeben.")
# return
# # ---- Ab hier eingeloggt (persistenter Cookie) ----
# # 1) Passwortwechsel erzwingen?
# if needs_password_change(username):
# st.warning("Du musst dein Passwort ändern, bevor du die Anwendung nutzen kannst.")
# with st.form("pw_change_form"):
# pw1 = st.text_input("Neues Passwort", type="password")
# pw2 = st.text_input("Neues Passwort (Wiederholung)", type="password")
# submitted = st.form_submit_button("Passwort ändern")
# if submitted:
# if not pw1 or not pw2:
# st.error("Bitte beide Passwortfelder ausfüllen.")
# return
# if pw1 != pw2:
# st.error("Passwörter stimmen nicht überein.")
# return
# if len(pw1) < 8:
# st.error("Passwort sollte mindestens 8 Zeichen lang sein.")
# return
# update_password(username, pw1, reset_flag=True)
# st.success("Passwort wurde geändert.")
# st.rerun()
# # solange new_pwd=1: keinen weiteren Inhalt anzeigen
# return
# # 2) Sidebar aufbauen (User, Logout, Navigation)
# build_sidebar(authenticator, username)
# # 3) Ausgewähltes Dashboard rendern
# role = get_role_for_user(username)
# selected_code = st.session_state.get("selected_dashboard_code", "home")
# renderer = get_dashboard_renderer(selected_code)
# if renderer is None:
# st.error(f"Kein Dashboard-Modul für '{selected_code}' gefunden.")
# return
# # Titel setzen (Home vs. andere Dashboards)
# if selected_code == "home":
# title = "Home"
# else:
# title = selected_code # später kannst du den Titel aus der DB holen
# st.title(title)
# renderer(username, role)
# if __name__ == "__main__":
# main()

View File

@@ -1,59 +0,0 @@
# app/main.py
import streamlit as st
from auth import (
init_auth_db,
ensure_logged_in,
authed_header,
current_user,
create_user, # falls du Admin-Funktion brauchst
)
def content_for(role: str, username: str):
st.header("Dashboard")
st.info(f"Willkommen, {username}!")
if role == "admin":
st.subheader("Admin-Bereich")
st.write("Nur Admins sehen das hier.")
with st.expander("Neuen Nutzer anlegen"):
new_u = st.text_input("Neuer Username", key="new_u")
new_p = st.text_input("Neues Passwort", type="password", key="new_p")
new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
if st.button("Anlegen"):
if new_u and new_p:
ok = create_user(new_u.strip(), new_p, new_role)
st.success("Nutzer angelegt.") if ok else st.error(
"Username bereits vorhanden."
)
else:
st.warning("Bitte Username und Passwort eingeben.")
st.subheader("Dein Bereich")
st.write(f"Personalisierter Content für **{username}**.")
# fachliche Inhalte …
def main():
st.set_page_config(
page_title="Intranet-Portal",
page_icon="🔒",
layout="centered",
)
init_auth_db()
# hier wird entweder:
# - st.session_state genutzt, oder
# - ?session_id=... aus URL geprüft, oder
# - Login-Form gezeigt (und Script gestoppt)
ensure_logged_in()
authed_header()
username, role = current_user()
content_for(role, username)
if __name__ == "__main__":
main()

View File

@@ -1,70 +1,141 @@
import streamlit as st
import yaml
from yaml.loader import SafeLoader
import streamlit_authenticator as stauth
from auth import load_credentials_from_db, get_role_for_user, create_user, needs_password_change, update_password, get_fullname_for_user
from version import __version__
from logging_config import setup_logging
import os
from auth import init_auth_db, ensure_logged_in, login_view, authed_header, current_user, create_user, clear_session_user
APP_ENV = os.environ.get("APP_ENV", "dev")
logger = setup_logging(APP_ENV)
from dashboards import get_dashboard_renderer
def content_for(role: str, username: str):
st.header("Dashboard")
st.info(f"Willkommen, {username}!")
# ---------------------------------------------------------------------------
# Login beim Starten der App
# ---------------------------------------------------------------------------
if role == "admin":
st.subheader("Admin-Bereich")
st.write("Nur Admins sehen das hier.")
with st.expander("Neuen Nutzer anlegen"):
new_u = st.text_input("Neuer Username", key="new_u")
new_p = st.text_input("Neues Passwort", type="password", key="new_p")
new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
if st.button("Anlegen"):
if new_u and new_p:
ok = create_user(new_u.strip(), new_p, new_role)
st.success("Nutzer angelegt.") if ok else st.error(
"Username bereits vorhanden."
)
else:
st.warning("Bitte Username und Passwort eingeben.")
st.subheader("Dein Bereich")
st.write(f"Personalisierter Content für **{username}**.")
# hier: weitere Seite(n), DB-Zugriffe etc.
def main():
if "app_started_logged" not in st.session_state:
logger.info(f"Starting app in {APP_ENV} mode - APP-Version {__version__}")
st.session_state["app_started_logged"] = True
# logger.info(f"User Starting app in {APP_ENV} mode - APP-Version {__version__}")
def login():
st.set_page_config(
page_title="Intranet-Portal",
page_title=f"Co-App Start - V{__version__}",
page_icon="🔒",
layout="centered",
)
# Falls kein Login: Login-View anzeigen und raus
if not st.session_state.get("auth"):
login_view()
# --- Config laden (Cookie, etc.) ---
with open("config/auth.yaml", "r", encoding="utf-8") as f:
base_config = yaml.load(f, Loader=SafeLoader)
# --- Credentials dynamisch aus DB laden ---
db_creds = load_credentials_from_db()
base_config["credentials"] = db_creds
authenticator = stauth.Authenticate(
base_config["credentials"],
base_config["cookie"]["name"],
base_config["cookie"]["key"],
base_config["cookie"]["expiry_days"],
)
authenticator.login(location="main", key="Login")
auth_status = st.session_state.get("authentication_status")
name = st.session_state.get("name")
username = st.session_state.get("username")
if auth_status is False:
st.error("Login fehlgeschlagen.")
return
# Ab hier sind wir eingeloggt
authed_header()
username, role = current_user()
content_for(role, username)
if auth_status is None:
st.warning("Bitte Benutzername und Passwort eingeben.")
return
st.write(st.session_state)
# ---- Ab hier eingeloggt (persistenter Cookie) ----
if st.sidebar.button("set state"):
st.session_state["app_started_logged"] = True
#clear_session_user()
st.write(st.session_state)
#st.rerun()
if st.sidebar.button("delete state"):
st.session_state.pop("app_started_logged", None)
#clear_session_user()
st.write(st.session_state)
#st.rerun()
if needs_password_change(username):
st.warning("Du musst dein Passwort ändern, bevor du die Anwendung nutzen kannst.")
# Damit das Formular nur einmal pro Run erscheint
with st.form("pw_change_form"):
pw1 = st.text_input("Neues Passwort", type="password")
pw2 = st.text_input("Neues Passwort (Wiederholung)", type="password")
submitted = st.form_submit_button("Passwort ändern")
if submitted:
if not pw1 or not pw2:
st.error("Bitte beide Passwortfelder ausfüllen.")
return
if pw1 != pw2:
st.error("Passwörter stimmen nicht überein.")
return
if len(pw1) < 8:
st.error("Passwort sollte mindestens 8 Zeichen lang sein.")
return
update_password(username, pw1, reset_flag=True)
st.success("Passwort wurde geändert.")
# Optional: danach Seite einmal „sauber“ neu laden
st.rerun()
# Solange new_pwd=1 ist, KEINEN weiteren Content anzeigen
return
authenticator.logout("Logout", "sidebar")
load_sidebar(username)
# def home(username: str, role: str):
# st.header("Controlling-Portal")
# st.info(f"Willkommen, {username}!")
# if role == "admin":
# st.subheader("Admin-Bereich")
# st.write("Nur Admins sehen das hier.")
# with st.expander("Neuen Nutzer anlegen"):
# new_u = st.text_input("Neuer Username", key="new_u")
# new_fname = st.text_input("Vorname", key="new_fname")
# new_lname = st.text_input("Nachname", key="new_lname")
# new_email = st.text_input("E-Mail", key="new_email")
# new_p = st.text_input("Neues Passwort", type="password", key="new_p")
# new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
# if st.button("Anlegen"):
# if new_u and new_p:
# ok = create_user(new_u.strip(), new_p, new_role, new_email, new_fname, new_lname)
# st.success("Nutzer angelegt.") if ok else st.error("Username bereits vorhanden oder Fehler.")
# else:
# st.warning("Bitte Username und Passwort eingeben.")
# st.subheader("Dein Bereich")
# st.write(f"Personalisierter Content für **{username}**.")
# ---------------------------------------------------------------------------
# Sidebar aufbauen - Daten aus app.db laden und gruppieren
# ---------------------------------------------------------------------------
def load_sidebar(username: str):
renderer = get_dashboard_renderer(selected_code)
if renderer is None:
st.error(f"Kein Dashboard-Modul für '{selected_code}' gefunden.")
return
renderer()
role = get_role_for_user(username)
fullname = get_fullname_for_user(username)
with st.sidebar:
st.write(f"**{fullname}** ({role})")
st.divider()
st.markdown("### Dashboards")
st.divider()
st.page_link("home.py", label="Home")
home(username, role)
if __name__ == "__main__":
main()
login()

Binary file not shown.

View File

@@ -0,0 +1,41 @@
begin;
create table if not exists dashboards (
dash_id integer unique not null,
dash_text text not null,
dash_description text,
group_id integer not null,
active integer not null default 1,
date_create TEXT NOT NULL DEFAULT (datetime('now')),
id integer primary key autoincrement
);
create table if not exists groups (
group_id text unique not null,
group_text text not null,
group_description text,
active integer not null default 1,
date_create TEXT NOT NULL DEFAULT (datetime('now')),
id integer primary key autoincrement
);
create table if not exists roles (
role_id integer unique not null,
role_text text not null,
role_description text,
active integer not null default 1,
date_create TEXT NOT NULL DEFAULT (datetime('now')),
id integer primary key autoincrement
);
create table if not exists permissions (
role_id integer unique not null,
dash_id integer unique not null,
active integer not null default 1,
date_create TEXT NOT NULL DEFAULT (datetime('now')),
id integer primary key autoincrement
);
INSERT INTO schema_version (version) VALUES ('20251130_191100_add_table_dashboards');
COMMIT;

View File

@@ -0,0 +1,14 @@
begin;
alter table dashboards
add column zgrp1 text;
alter table dashboards
add column zgrp2 text;
alter table dashboards
add column zgrp3 text;
INSERT INTO schema_version (version) VALUES ('20251130_200000_add_col_area');
COMMIT;

View File

@@ -0,0 +1,11 @@
begin;
alter table dashboards
add column order_no integer default 0;
alter table groups
add column order_no integer default 0;
INSERT INTO schema_version (version) VALUES ('20251130_200600_add_col_order');
COMMIT;