""" Superadmin endpoints for system settings. """ from datetime import datetime, timedelta, timezone from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_superadmin from app.core.database import get_db from app.models.settings import Settings from app.models.user import User router = APIRouter() class AutoRegistrationResponse(BaseModel): """Auto-registration status response.""" enabled: bool last_device_at: str | None time_left: int # Seconds until auto-disable (0 if disabled) class AutoRegistrationToggleRequest(BaseModel): """Auto-registration toggle request.""" enabled: bool @router.get("/auto-registration", response_model=AutoRegistrationResponse) async def get_auto_registration_status( db: Annotated[AsyncSession, Depends(get_db)], current_user: Annotated[User, Depends(get_current_superadmin)], ): """ Get auto-registration status. Returns current state and last device registration timestamp. """ result = await db.execute( select(Settings).where(Settings.key == "auto_registration") ) setting = result.scalar_one_or_none() if not setting: # Create default setting if not exists setting = Settings( key="auto_registration", value={"enabled": False, "last_device_at": None}, ) db.add(setting) await db.commit() # Check if should auto-disable (1 hour since last device) enabled = setting.value.get("enabled", False) last_device_at_str = setting.value.get("last_device_at") time_left = 0 if enabled and last_device_at_str: last_device_at = datetime.fromisoformat(last_device_at_str) time_since = datetime.now(timezone.utc) - last_device_at time_left = max(0, int(3600 - time_since.total_seconds())) # 1 hour = 3600 seconds if time_since > timedelta(hours=1): # Auto-disable - must reassign whole dict for SQLAlchemy setting.value = {**setting.value, "enabled": False} await db.commit() enabled = False time_left = 0 return AutoRegistrationResponse( enabled=enabled, last_device_at=last_device_at_str, time_left=time_left, ) @router.post("/auto-registration", response_model=AutoRegistrationResponse) async def toggle_auto_registration( data: AutoRegistrationToggleRequest, db: Annotated[AsyncSession, Depends(get_db)], current_user: Annotated[User, Depends(get_current_superadmin)], ): """ Enable or disable auto-registration of new devices. When enabled, devices that authenticate with unknown MAC addresses will be automatically registered in the system (unassigned to any organization). Auto-registration will automatically disable after 1 hour of the last registered device. """ result = await db.execute( select(Settings).where(Settings.key == "auto_registration") ) setting = result.scalar_one_or_none() if not setting: setting = Settings( key="auto_registration", value={"enabled": data.enabled, "last_device_at": None}, ) db.add(setting) else: # Must reassign the whole dict for SQLAlchemy to detect the change new_value = {**setting.value, "enabled": data.enabled} if not data.enabled: # Reset last_device_at when disabling new_value["last_device_at"] = None setting.value = new_value await db.commit() # Calculate time_left time_left = 0 enabled = setting.value.get("enabled", False) last_device_at_str = setting.value.get("last_device_at") if enabled and last_device_at_str: last_device_at = datetime.fromisoformat(last_device_at_str) time_since = datetime.now(timezone.utc) - last_device_at time_left = max(0, int(3600 - time_since.total_seconds())) return AutoRegistrationResponse( enabled=enabled, last_device_at=last_device_at_str, time_left=time_left, ) class SettingUpdate(BaseModel): """Generic setting update request.""" value: dict class SettingResponse(BaseModel): """Generic setting response.""" key: str value: dict updated_at: str | None @router.get("/setting/{key}", response_model=SettingResponse) async def get_setting( key: str, db: Annotated[AsyncSession, Depends(get_db)], current_user: Annotated[User, Depends(get_current_superadmin)], ): """Get a specific setting by key.""" result = await db.execute(select(Settings).where(Settings.key == key)) setting = result.scalar_one_or_none() if not setting: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"setting_not_found:{key}", ) return SettingResponse( key=setting.key, value=setting.value, updated_at=setting.updated_at.isoformat() if setting.updated_at else None, ) @router.put("/setting/{key}", response_model=SettingResponse) async def update_setting( key: str, data: SettingUpdate, db: Annotated[AsyncSession, Depends(get_db)], current_user: Annotated[User, Depends(get_current_superadmin)], ): """Update a specific setting by key.""" result = await db.execute(select(Settings).where(Settings.key == key)) setting = result.scalar_one_or_none() if not setting: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"setting_not_found:{key}", ) setting.value = data.value setting.updated_at = datetime.now(timezone.utc) await db.commit() await db.refresh(setting) return SettingResponse( key=setting.key, value=setting.value, updated_at=setting.updated_at.isoformat(), )