import importlib import logging from contextlib import asynccontextmanager from fastapi import Depends, FastAPI, Request, WebSocket from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from app.core.config import get_settings from app.core.database import get_db from app.modules.auth.dependencies import ( RequiresLoginException, check_permission, get_current_user, get_current_user_optional, ) from app.modules.auth.router import router as auth_router logger = logging.getLogger(__name__) settings = get_settings() MODULES = [ { "slug": "rss", "icon": "📡", "name": "RSS-Feed Server", "description": "Aggregiert und verteilt Neuigkeiten der Fakultät als standardkonformen RSS 2.0 Feed.", "status": "active", "permission": ["authenticated"], }, { "slug": "kalender", "icon": "📅", "name": "Kalender", "description": "Veranstaltungen, Prüfungstermine und Fristen im iCal-Format.", "status": "planned", "permission": ["authenticated"], }, { "slug": "benachrichtigungen", "icon": "🔔", "name": "Benachrichtigungen", "description": "Push-Nachrichten und WebSocket-basierte Echtzeit-Alerts für Studierende.", "status": "planned", "permission": ["authenticated"], }, { "slug": "auslastung", "icon": "📊", "name": "Auslastung", "description": "Raum- und Ressourcenauslastung der Fakultät in Echtzeit.", "status": "planned", "permission": ["authenticated"], }, { "slug": "lehrveranstaltungen", "icon": "📚", "name": "Lehrveranstaltungen", "description": "Stundenplan-API und Kursinformationen aus dem Campus-System.", "status": "planned", "permission": ["authenticated"], }, ] NAV_ITEMS = [ {"label": "Übersicht", "url": "/", "active": True}, {"label": "RSS-Feeds", "url": "/rss", "active": False}, {"label": "Kalender", "url": "/kalender", "active": False}, {"label": "Docs", "url": "/docs", "active": False}, ] def _reset_admin() -> None: from app.core.database import Base, SessionLocal, engine from app.modules.auth.models import User # noqa: F401 from app.modules.auth.service import get_user, hash_password Base.metadata.create_all(bind=engine) with SessionLocal() as db: user = get_user(db, settings.ADMIN_USERNAME) if user is None: user = User(username=settings.ADMIN_USERNAME, full_name="Administrator") db.add(user) user.pw_hash = hash_password(settings.ADMIN_PASSWORD) user.is_admin = True user.is_active = True db.commit() @asynccontextmanager async def lifespan(app: FastAPI): _reset_admin() yield app = FastAPI( title="University Process Hub", root_path=settings.APP_PREFIX, lifespan=lifespan, ) app.mount("/static", StaticFiles(directory="app/static"), name="static") templates = Jinja2Templates(directory="app/templates") app.include_router(auth_router) for _module in MODULES: if _module["status"] == "active": try: _mod = importlib.import_module(f"app.modules.{_module['slug']}.router") app.include_router(_mod.router, prefix=f"/{_module['slug']}") except ModuleNotFoundError: logger.warning("Module router not found: app.modules.%s.router", _module["slug"]) @app.exception_handler(RequiresLoginException) async def requires_login_handler(request: Request, exc: RequiresLoginException): return RedirectResponse(url="/auth/login", status_code=307) def _db_mode() -> str: url = settings.DATABASE_URL return "SQLite (dev)" if url.startswith("sqlite") else "MariaDB (prod)" @app.get("/", response_class=HTMLResponse) async def root(request: Request, current_user=Depends(get_current_user_optional)): visible = [m for m in MODULES if check_permission(current_user, m["permission"])] return templates.TemplateResponse( request, "index.html", { "nav_items": NAV_ITEMS, "modules": visible, "db_mode": _db_mode(), "app_version": "0.1.0", "current_user": current_user, }, ) @app.websocket("/ws/hello/{name}") async def websocket_hello(websocket: WebSocket, name: str): await websocket.accept() await websocket.send_text(f"Hello, {name}!") await websocket.close()