dashboard-nanobot/backend/services/platform_settings_service.py

from typing import Any, Dict, List

from sqlmodel import Session, select

from core.database import engine
from core.settings import (
    DEFAULT_STT_AUDIO_FILTER,
    DEFAULT_STT_AUDIO_PREPROCESS,
    DEFAULT_STT_DEFAULT_LANGUAGE,
    DEFAULT_STT_FORCE_SIMPLIFIED,
    DEFAULT_STT_INITIAL_PROMPT,
    DEFAULT_STT_MAX_AUDIO_SECONDS,
    STT_DEVICE,
    STT_MODEL,
)
from models.platform import PlatformSetting
from schemas.platform import PlatformSettingsPayload
from services.platform_settings_core import (
    ACTIVITY_EVENT_RETENTION_SETTING_KEY,
    DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS,
    DEFAULT_ALLOWED_ATTACHMENT_EXTENSIONS,
    SETTING_KEYS,
    SYSTEM_SETTING_DEFINITIONS,
    _normalize_extension_list,
    _read_setting_value,
    _upsert_setting_row,
)
from services.platform_system_settings_service import (
    create_or_update_system_setting,
    delete_system_setting,
    get_activity_event_retention_days,
    list_system_settings,
    validate_required_system_settings,
)


def get_platform_settings(session: Session) -> PlatformSettingsPayload:
    validate_required_system_settings(session)
    rows = session.exec(select(PlatformSetting).where(PlatformSetting.key.in_(SETTING_KEYS))).all()
    data: Dict[str, Any] = {row.key: _read_setting_value(row) for row in rows}
    missing = [key for key in SETTING_KEYS if key not in data]
    if missing:
        raise RuntimeError(
            "Database seed data is not initialized. "
            f"Missing sys_setting keys: {', '.join(missing)}. "
            "Run scripts/init-full-db.sh or apply scripts/sql/init-data.sql before starting the backend."
        )
    try:
        return PlatformSettingsPayload.model_validate(
            {
                "page_size": max(1, min(100, int(data["page_size"]))),
                "chat_pull_page_size": max(10, min(500, int(data["chat_pull_page_size"]))),
                "command_auto_unlock_seconds": max(1, min(600, int(data["command_auto_unlock_seconds"]))),
                "auth_token_ttl_hours": max(1, min(720, int(data["auth_token_ttl_hours"]))),
                "auth_token_max_active": max(1, min(20, int(data["auth_token_max_active"]))),
                "upload_max_mb": int(data["upload_max_mb"]),
                "allowed_attachment_extensions": _normalize_extension_list(data["allowed_attachment_extensions"]),
                "workspace_download_extensions": _normalize_extension_list(data["workspace_download_extensions"]),
                "speech_enabled": bool(data["speech_enabled"]),
            }
        )
    except Exception as exc:
        raise RuntimeError(
            "sys_setting contains invalid platform configuration values. "
            "Fix the rows manually or reapply scripts/sql/init-data.sql."
        ) from exc


def save_platform_settings(session: Session, payload: PlatformSettingsPayload) -> PlatformSettingsPayload:
    normalized = PlatformSettingsPayload(
        page_size=max(1, min(100, int(payload.page_size))),
        chat_pull_page_size=max(10, min(500, int(payload.chat_pull_page_size))),
        command_auto_unlock_seconds=max(1, min(600, int(payload.command_auto_unlock_seconds))),
        auth_token_ttl_hours=max(1, min(720, int(payload.auth_token_ttl_hours))),
        auth_token_max_active=max(1, min(20, int(payload.auth_token_max_active))),
        upload_max_mb=payload.upload_max_mb,
        allowed_attachment_extensions=_normalize_extension_list(payload.allowed_attachment_extensions),
        workspace_download_extensions=_normalize_extension_list(payload.workspace_download_extensions),
        speech_enabled=bool(payload.speech_enabled),
    )
    payload_by_key = normalized.model_dump()
    for key in SETTING_KEYS:
        definition = SYSTEM_SETTING_DEFINITIONS[key]
        _upsert_setting_row(
            session,
            key,
            name=str(definition["name"]),
            category=str(definition["category"]),
            description=str(definition["description"]),
            value_type=str(definition["value_type"]),
            value=payload_by_key[key],
            is_public=bool(definition["is_public"]),
            sort_order=int(definition["sort_order"]),
        )
    session.commit()
    return normalized


def get_platform_settings_snapshot() -> PlatformSettingsPayload:
    with Session(engine) as session:
        return get_platform_settings(session)


def get_upload_max_mb() -> int:
    return get_platform_settings_snapshot().upload_max_mb


def get_allowed_attachment_extensions() -> List[str]:
    return get_platform_settings_snapshot().allowed_attachment_extensions


def get_workspace_download_extensions() -> List[str]:
    return get_platform_settings_snapshot().workspace_download_extensions


def get_page_size() -> int:
    return get_platform_settings_snapshot().page_size


def get_chat_pull_page_size() -> int:
    return get_platform_settings_snapshot().chat_pull_page_size


def get_auth_token_ttl_hours(session: Session) -> int:
    return get_platform_settings(session).auth_token_ttl_hours


def get_auth_token_max_active(session: Session) -> int:
    return get_platform_settings(session).auth_token_max_active


def get_speech_runtime_settings() -> Dict[str, Any]:
    settings = get_platform_settings_snapshot()
    return {
        "enabled": bool(settings.speech_enabled),
        "max_audio_seconds": int(DEFAULT_STT_MAX_AUDIO_SECONDS),
        "default_language": str(DEFAULT_STT_DEFAULT_LANGUAGE or "zh").strip().lower() or "zh",
        "force_simplified": bool(DEFAULT_STT_FORCE_SIMPLIFIED),
        "audio_preprocess": bool(DEFAULT_STT_AUDIO_PREPROCESS),
        "audio_filter": str(DEFAULT_STT_AUDIO_FILTER or "").strip(),
        "initial_prompt": str(DEFAULT_STT_INITIAL_PROMPT or "").strip(),
        "model": STT_MODEL,
        "device": STT_DEVICE,
    }