244 lines
8.6 KiB
Python
244 lines
8.6 KiB
Python
"""System settings router - read/update settings, upload company logo."""
|
|
import json
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.backend.config import settings
|
|
from src.backend.database import get_db
|
|
from src.backend.api.middleware.api_key import get_current_user, require_admin_user
|
|
from src.backend.api.routers.files import ALLOWED_IMAGE_TYPES, detect_file_type
|
|
from src.backend.models.orm.setting import SystemSetting
|
|
from src.backend.models.orm.user import User
|
|
|
|
router = APIRouter(prefix="/api/settings", tags=["settings"])
|
|
|
|
# Domains for the settings keys the backend actually reads. Values outside these
|
|
# would only fail later (csv.writer TypeError on close_run, etc) — reject up front.
|
|
_VALID_CSV_DELIMITERS = {",", ";", "\t", "|"}
|
|
_VALID_CSV_DECIMAL_SEPARATORS = {",", "."}
|
|
|
|
|
|
def _validate_string_list(key: str, value: str) -> None:
|
|
"""#49: array JSON di stringhe non vuote, max 50 voci (vuoto = nessuna lista)."""
|
|
try:
|
|
items = json.loads(value)
|
|
except ValueError:
|
|
items = None
|
|
if (
|
|
not isinstance(items, list)
|
|
or len(items) > 50
|
|
or not all(isinstance(i, str) and i.strip() for i in items)
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"{key} deve essere vuoto o un array JSON di al massimo 50 stringhe non vuote",
|
|
)
|
|
|
|
|
|
def _validate_setting_value(key: str, value: str) -> None:
|
|
"""Raise 422 for a value that would break the code reading this key.
|
|
|
|
Unrecognised keys pass through unvalidated: the settings store is generic.
|
|
"""
|
|
if key == "csv_delimiter":
|
|
if value not in _VALID_CSV_DELIMITERS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="csv_delimiter deve essere un carattere singolo fra ',' ';' TAB '|'",
|
|
)
|
|
elif key == "csv_decimal_separator":
|
|
if value not in _VALID_CSV_DECIMAL_SEPARATORS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="csv_decimal_separator deve essere ',' o '.'",
|
|
)
|
|
elif key == "auto_logout_minutes":
|
|
if value != "" and (not value.isdigit() or not (1 <= int(value) <= 480)):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="auto_logout_minutes deve essere vuoto (disabilitato) o un intero fra 1 e 480",
|
|
)
|
|
elif key == "default_measurement_interval_minutes":
|
|
# #47: vuoto = nessun default
|
|
if value != "" and (not value.isdigit() or not (1 <= int(value) <= 1440)):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="default_measurement_interval_minutes deve essere vuoto o un intero fra 1 e 1440",
|
|
)
|
|
elif key in ("general_tolerance_plus", "general_tolerance_minus"):
|
|
# #48: decimale >= 0 con punto; vuoto = nessun default
|
|
if value != "":
|
|
try:
|
|
number = Decimal(value)
|
|
ok = number.is_finite() and number >= 0
|
|
except InvalidOperation:
|
|
ok = False
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"{key} deve essere vuoto o un numero decimale maggiore o uguale a 0 (punto decimale)",
|
|
)
|
|
elif key in ("list_measurement_types", "list_units"):
|
|
if value != "":
|
|
_validate_string_list(key, value)
|
|
elif key in ("company_name", "company_logo_path"):
|
|
if not value.strip() or len(value) > 200:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"{key} deve essere una stringa non vuota di massimo 200 caratteri",
|
|
)
|
|
|
|
|
|
@router.get("/")
|
|
async def get_settings(
|
|
user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Get all system settings as a dictionary.
|
|
|
|
Returns key-value pairs of all settings.
|
|
Available to all authenticated users.
|
|
"""
|
|
result = await db.execute(select(SystemSetting))
|
|
settings_list = result.scalars().all()
|
|
|
|
settings_dict = {
|
|
setting.setting_key: setting.setting_value
|
|
for setting in settings_list
|
|
}
|
|
|
|
return settings_dict
|
|
|
|
|
|
@router.put("/")
|
|
async def update_settings(
|
|
settings_data: dict[str, str],
|
|
user: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update multiple system settings.
|
|
|
|
Expects a JSON object with setting_key: setting_value pairs.
|
|
Only admins can update settings.
|
|
"""
|
|
updated_keys = []
|
|
|
|
for key, value in settings_data.items():
|
|
_validate_setting_value(key, value)
|
|
|
|
# Check if setting exists
|
|
result = await db.execute(
|
|
select(SystemSetting).where(SystemSetting.setting_key == key)
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
|
|
if setting is None:
|
|
# Create new setting (assume type 'string' for new settings)
|
|
setting = SystemSetting(
|
|
setting_key=key,
|
|
setting_value=value,
|
|
setting_type="string",
|
|
updated_by=user.id,
|
|
)
|
|
db.add(setting)
|
|
else:
|
|
# Update existing setting
|
|
setting.setting_value = value
|
|
setting.updated_by = user.id
|
|
|
|
updated_keys.append(key)
|
|
|
|
await db.flush()
|
|
|
|
return {
|
|
"message": f"Updated {len(updated_keys)} settings",
|
|
"updated_keys": updated_keys,
|
|
}
|
|
|
|
|
|
@router.post("/logo")
|
|
async def upload_company_logo(
|
|
file: UploadFile = File(...),
|
|
user: User = Depends(require_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Upload company logo.
|
|
|
|
Saves logo to uploads/logos/ and updates company_logo_path setting.
|
|
Only admins can upload the logo. The type is determined from the file's
|
|
magic bytes, not the client-declared Content-Type (#36 point 2). SVG is
|
|
refused outright: it can itself carry script content, and there is no
|
|
safe way to serve it inline for a logo shown on every page.
|
|
"""
|
|
# Read in chunks, stopping as soon as the size limit is crossed instead
|
|
# of buffering an oversized body into memory first.
|
|
max_bytes = settings.max_upload_size_mb * 1024 * 1024
|
|
buffer = bytearray()
|
|
chunk_size = 1024 * 1024
|
|
while True:
|
|
chunk = await file.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
buffer.extend(chunk)
|
|
if len(buffer) > max_bytes:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"File size exceeds maximum {settings.max_upload_size_mb}MB",
|
|
)
|
|
content = bytes(buffer)
|
|
file_size = len(content)
|
|
|
|
# Validate file type from the actual bytes (SVG never matches: refused)
|
|
detected = detect_file_type(content)
|
|
if detected is None or detected[0] not in ALLOWED_IMAGE_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="File type not allowed. Must be a recognised image (SVG is not accepted).",
|
|
)
|
|
_detected_mime, detected_ext = detected
|
|
|
|
# Create logos directory
|
|
logos_dir = settings.upload_path / "logos"
|
|
logos_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Fixed filename for the company logo, extension forced to what was
|
|
# actually detected (not what the client claimed or the filename said).
|
|
logo_filename = f"company_logo{detected_ext}"
|
|
logo_path = logos_dir / logo_filename
|
|
|
|
# Write file
|
|
logo_path.write_bytes(content)
|
|
|
|
# Update setting in database
|
|
relative_path = logo_path.relative_to(settings.upload_path)
|
|
logo_path_str = str(relative_path).replace("\\", "/")
|
|
|
|
result = await db.execute(
|
|
select(SystemSetting).where(SystemSetting.setting_key == "company_logo_path")
|
|
)
|
|
setting = result.scalar_one_or_none()
|
|
|
|
if setting is None:
|
|
setting = SystemSetting(
|
|
setting_key="company_logo_path",
|
|
setting_value=logo_path_str,
|
|
setting_type="string",
|
|
description="Path to company logo file",
|
|
updated_by=user.id,
|
|
)
|
|
db.add(setting)
|
|
else:
|
|
setting.setting_value = logo_path_str
|
|
setting.updated_by = user.id
|
|
|
|
await db.flush()
|
|
|
|
return {
|
|
"message": "Company logo uploaded successfully",
|
|
"logo_path": logo_path_str,
|
|
"file_size": file_size,
|
|
}
|