import dash
from dash import Dash, html, dcc, Input, Output, State
import dash_mantine_components as dmc
from flask import Flask, redirect, url_for, request, session
from components.header import create_header
from components.sidebar import create_sidebar
from components.breadcrumb import create_breadcrumb
from components.chatbot import create_chatbot_panel
from components.footer import create_footer
from data.caching import setup_caching
from agents.rac_generation_agent import rac_graph_agent
from sockets import socketio, init_sockets
from flask_login import LoginManager, UserMixin, login_user, logout_user, current_user, login_required
from dotenv import load_dotenv
import os
import json
import datetime
from datetime import timedelta
load_dotenv()
server = Flask(__name__)
server.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'supersecretkey') # Default for dev
server.config['SESSION_COOKIE_SECURE'] = True
server.config['SESSION_COOKIE_HTTPONLY'] = True
server.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(seconds=3600)
login_manager = LoginManager()
login_manager.init_app(server)
login_manager.login_view = 'login'
app = Dash(__name__, server=server, use_pages=True,
external_scripts=[
"https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"
])
init_sockets(server)
users_data = json.loads(os.getenv('USERS', '{}'))
class User(UserMixin):
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password = password
@staticmethod
def get(user_id):
for user_data in users_data.values():
if user_data['id'] == int(user_id):
return User(user_data['id'], user_data['username'], user_data['password'])
return None
@staticmethod
def get_by_username(username):
for user_id, user_data in users_data.items():
if user_data['username'] == username:
return User(user_data['id'], user_data['username'], user_data['password'])
return None
@login_manager.user_loader
def load_user(user_id):
user = User.get(user_id)
return user
@server.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect('/') # Changed from dash_app
error_message = None
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
user = User.get_by_username(username)
if user and user.password == password:
login_user(user)
return redirect('/') # Redirect to the protected Dash app root
else:
error_message = 'Invalid username or password'
# HTML content for the login page
html_content = f'''
Gustave IA - Create Account
Gustave HUB
{f'
{error_message}
' if error_message else ''}
Propulsé par le CIPEN de l'université Gustave Eiffel
datacipen@univ-eiffel.fr
Gustave IA au service des analyses automatiques et des datavisualisations dynamiques.
'''
return html_content
@server.route('/logout')
def logout():
logout_user()
return redirect(url_for('login')) # Redirect to login page after logout
@app.server.before_request
def before_request():
# Si la requête concerne la page de connexion ou si l'utilisateur est authentifié, autoriser
if request.endpoint == 'login' or current_user.is_authenticated or request.endpoint == 'logout':
return None
# Si la requête concerne les ressources internes de Dash (assets, suites de composants, callbacks, rechargement), autoriser
# C'est crucial pour que Dash fonctionne même sans authentification, mais uniquement pour les requêtes internes
if request.path.startswith('/assets/') or \
request.path.startswith('/_dash-component-suites/') or \
request.path.startswith('/_favicon.ico') or \
request.path.startswith('/_dash-update-component') or \
request.path.startswith('/_dash-reload-app') or \
request.path.startswith('/static/'):
return None
# Pour toutes les autres requêtes (y compris toutes les pages Dash et la racine '/'), rediriger vers la connexion
return redirect(url_for('login'))
get_data = setup_caching(app)
# You can now use get_data('path/to/your/excel.xlsx') in your page layouts
# and the data will be cached.
app.layout = dmc.MantineProvider(
id="mantine-provider",
theme={
"primaryColor": "indigo",
"colorScheme": "dark",
"components": {
"NavLink": {
"styles": {
"root": {
"&:hover": {"backgroundColor": "#181F2F"}
}
}
}
}
},
forceColorScheme="dark",
withGlobalClasses=True,
children=[
dcc.Location(id='url', refresh=True),
dcc.Store(id='client-session-id', data=None),
create_chatbot_panel(),
dmc.Drawer(
id="history-drawer",
title="Historique de consultation",
padding="md",
position="right",
opened=False, # Explicitly set to closed for initial load
children=[
dmc.Text("Contenu de l'historique ici...")
],
),
dmc.AppShell(
id="app-shell",
padding="md",
header={"height": 70},
footer={"height": 60,"zIndex": 1000,"bgcolor": "#060621"},
navbar={
"width": 300,
"breakpoint": "sm",
"collapsed": {"mobile": False, "desktop": False}, # Initialize as NOT collapsed for both mobile and desktop
},
children=[
create_header(theme_checked=True), # Pass theme_checked value
create_sidebar(),
create_footer(),
dmc.AppShellMain(
id="app-shell-main", # Added ID here
children=[
html.Div(id='breadcrumb-container'),
dash.page_container
],
style={"backgroundColor": "#060621"}
)
]
)
]
)
@server.route("/start_rac_agent", methods=['POST'])
def start_rac_agent():
rac_graph_agent.invoke({})
return {"status": "started"}
@socketio.on('connect')
def handle_connect():
print('Client connected')
@socketio.on('disconnect')
def handle_disconnect():
print('Client disconnected')
@app.callback(
Output("history-drawer", "opened"), # Cible directement la propriété 'opened' du tiroir
Input("history-button", "n_clicks"),
State("history-drawer", "opened"), # Lit l'état actuel du tiroir
prevent_initial_call=True,
)
def toggle_history_drawer(n_clicks, is_opened):
if n_clicks is None:
return is_opened # Ne rien faire si ce n'est pas un clic
return not is_opened
@app.callback(
Output('breadcrumb-container', 'children'),
[Input('url', 'pathname')]
)
def update_breadcrumb(pathname):
return create_breadcrumb(pathname)
@app.callback(
Output("mantine-provider", "theme"),
Output("app-shell", "navbar", allow_duplicate=True),
Input("theme-switch", "checked"),
Input("sidebar-burger", "opened"), # We'll keep this input for now for theme/header updates
prevent_initial_call=True,
)
def update_theme_and_navbar(theme_checked, sidebar_opened):
scheme = "dark" if theme_checked else "light"
theme = {"primaryColor": "indigo", "colorScheme": scheme}
# The navbar config will now be controlled by this callback for responsive behavior
navbar_config = {"width": 300, "breakpoint": "sm", "collapsed": {"mobile": sidebar_opened, "desktop": sidebar_opened}}
# style = {"backgroundColor": "#020817"} if theme_checked else {"backgroundColor": "#f8f9fa"}
# if sidebar_opened:
# style["paddingLeft"] = 300
# else:
# style["paddingLeft"] = 80
# return theme, navbar_config, create_header(theme_checked).children # Removed create_header.children
return theme, navbar_config
@app.callback(
Output("main-header", "children"), # New callback for updating header children
Input("theme-switch", "checked"),
prevent_initial_call=True,
)
def update_header_theme(theme_checked):
return create_header(theme_checked).children
@app.callback(
Output("chatbot-drawer", "opened"), # Cible directement la propriété 'opened' du tiroir
Input("chatbot-button", "n_clicks"),
State("chatbot-drawer", "opened"), # Lit l'état actuel du tiroir
prevent_initial_call=True,
)
def toggle_chatbot(n_clicks, is_opened):
if n_clicks is None:
return is_opened # Ne rien faire si ce n'est pas un clic
return not is_opened
@app.callback(
Output("app-shell-main", "style", allow_duplicate=True),
Input("app-shell", "navbar"), # Changed input to app-shell's navbar property
State("theme-switch", "checked"),
prevent_initial_call='initial_duplicate',
)
def update_main_content_padding(navbar_config, theme_checked):
style = {"backgroundColor": "#060621"} if theme_checked else {"backgroundColor": "#f8f9fa"}
# Extract collapsed state from navbar_config (assuming mobile collapsed state dictates padding)
is_collapsed = navbar_config.get("collapsed", {}).get("mobile", True)
if is_collapsed:
style["paddingLeft"] = 80
else:
style["paddingLeft"] = 300
return style
@app.callback(
Output('url', 'href'),
Input('logout-button', 'n_clicks'),
prevent_initial_call=True
)
def redirect_on_logout(n_clicks):
if n_clicks:
return '/logout'
return dash.no_update
if __name__ == "__main__":
socketio.run(server, debug=True, port=5001)