263 lines
10 KiB
Python
263 lines
10 KiB
Python
|
import json
|
||
|
|
from typing import Any, Dict, List
|
||
|
|
|
||
|
|
from sqlmodel import Session, select
|
||
|
|
|
||
|
|
from core.database import engine
|
||
|
|
from models.platform import PlatformSetting
|
||
|
|
from schemas.platform import LoadingPageSettings, PlatformSettingsPayload, SystemSettingPayload
|
||
|
|
from services.platform_common import utcnow
|
||
|
|
|
||
|
|
from services.platform_settings_support import (
|
||
|
|
ACTIVITY_EVENT_RETENTION_SETTING_KEY,
|
||
|
|
DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS,
|
||
|
|
DEPRECATED_SETTING_KEYS,
|
||
|
|
PROTECTED_SETTING_KEYS,
|
||
|
|
SETTING_KEYS,
|
||
|
|
SYS_AUTH_TOKEN_TTL_DAYS_SETTING_KEY,
|
||
|
|
SYSTEM_SETTING_DEFINITIONS,
|
||
|
|
bootstrap_platform_setting_values,
|
||
|
|
build_speech_runtime_settings,
|
||
|
|
default_platform_settings,
|
||
|
|
normalize_activity_event_retention_days,
|
||
|
|
normalize_extension_list,
|
||
|
|
normalize_setting_key,
|
||
|
|
read_setting_value,
|
||
|
|
setting_item_from_row,
|
||
|
|
upsert_setting_row,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def ensure_default_system_settings(session: Session) -> None:
|
||
|
|
bootstrap_values = bootstrap_platform_setting_values()
|
||
|
|
legacy_row = session.get(PlatformSetting, "global")
|
||
|
|
if legacy_row is not None:
|
||
|
|
try:
|
||
|
|
legacy_data = json.loads(legacy_row.value_json or "{}")
|
||
|
|
except Exception:
|
||
|
|
legacy_data = {}
|
||
|
|
if isinstance(legacy_data, dict):
|
||
|
|
for key in SETTING_KEYS:
|
||
|
|
meta = SYSTEM_SETTING_DEFINITIONS[key]
|
||
|
|
upsert_setting_row(
|
||
|
|
session,
|
||
|
|
key,
|
||
|
|
name=str(meta["name"]),
|
||
|
|
category=str(meta["category"]),
|
||
|
|
description=str(meta["description"]),
|
||
|
|
value_type=str(meta["value_type"]),
|
||
|
|
value=legacy_data.get(key, bootstrap_values.get(key, meta["value"])),
|
||
|
|
is_public=bool(meta["is_public"]),
|
||
|
|
sort_order=int(meta["sort_order"]),
|
||
|
|
)
|
||
|
|
session.delete(legacy_row)
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
dirty = False
|
||
|
|
for key in DEPRECATED_SETTING_KEYS:
|
||
|
|
legacy_row = session.get(PlatformSetting, key)
|
||
|
|
if legacy_row is not None:
|
||
|
|
session.delete(legacy_row)
|
||
|
|
dirty = True
|
||
|
|
|
||
|
|
for key, meta in SYSTEM_SETTING_DEFINITIONS.items():
|
||
|
|
row = session.get(PlatformSetting, key)
|
||
|
|
if row is None:
|
||
|
|
upsert_setting_row(
|
||
|
|
session,
|
||
|
|
key,
|
||
|
|
name=str(meta["name"]),
|
||
|
|
category=str(meta["category"]),
|
||
|
|
description=str(meta["description"]),
|
||
|
|
value_type=str(meta["value_type"]),
|
||
|
|
value=bootstrap_values.get(key, meta["value"]),
|
||
|
|
is_public=bool(meta["is_public"]),
|
||
|
|
sort_order=int(meta["sort_order"]),
|
||
|
|
)
|
||
|
|
dirty = True
|
||
|
|
continue
|
||
|
|
changed = False
|
||
|
|
for field in ("name", "category", "description", "value_type"):
|
||
|
|
value = str(meta[field])
|
||
|
|
if not getattr(row, field):
|
||
|
|
setattr(row, field, value)
|
||
|
|
changed = True
|
||
|
|
if getattr(row, "sort_order", None) is None:
|
||
|
|
row.sort_order = int(meta["sort_order"])
|
||
|
|
changed = True
|
||
|
|
if getattr(row, "is_public", None) is None:
|
||
|
|
row.is_public = bool(meta["is_public"])
|
||
|
|
changed = True
|
||
|
|
if changed:
|
||
|
|
row.updated_at = utcnow()
|
||
|
|
session.add(row)
|
||
|
|
dirty = True
|
||
|
|
if dirty:
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def list_system_settings(session: Session, search: str = "") -> List[Dict[str, Any]]:
|
||
|
|
ensure_default_system_settings(session)
|
||
|
|
stmt = select(PlatformSetting).order_by(PlatformSetting.sort_order.asc(), PlatformSetting.key.asc())
|
||
|
|
rows = session.exec(stmt).all()
|
||
|
|
keyword = str(search or "").strip().lower()
|
||
|
|
items = [setting_item_from_row(row) for row in rows]
|
||
|
|
if not keyword:
|
||
|
|
return items
|
||
|
|
return [
|
||
|
|
item
|
||
|
|
for item in items
|
||
|
|
if keyword in str(item["key"]).lower()
|
||
|
|
or keyword in str(item["name"]).lower()
|
||
|
|
or keyword in str(item["category"]).lower()
|
||
|
|
or keyword in str(item["description"]).lower()
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def create_or_update_system_setting(session: Session, payload: SystemSettingPayload) -> Dict[str, Any]:
|
||
|
|
ensure_default_system_settings(session)
|
||
|
|
normalized_key = normalize_setting_key(payload.key)
|
||
|
|
definition = SYSTEM_SETTING_DEFINITIONS.get(normalized_key, {})
|
||
|
|
row = upsert_setting_row(
|
||
|
|
session,
|
||
|
|
payload.key,
|
||
|
|
name=payload.name or str(definition.get("name") or payload.key),
|
||
|
|
category=payload.category or str(definition.get("category") or "general"),
|
||
|
|
description=payload.description or str(definition.get("description") or ""),
|
||
|
|
value_type=payload.value_type or str(definition.get("value_type") or "json"),
|
||
|
|
value=payload.value if payload.value is not None else definition.get("value"),
|
||
|
|
is_public=payload.is_public,
|
||
|
|
sort_order=payload.sort_order or int(definition.get("sort_order") or 100),
|
||
|
|
)
|
||
|
|
if normalized_key == ACTIVITY_EVENT_RETENTION_SETTING_KEY:
|
||
|
|
from services.platform_activity_service import prune_expired_activity_events
|
||
|
|
|
||
|
|
prune_expired_activity_events(session, force=True)
|
||
|
|
session.commit()
|
||
|
|
session.refresh(row)
|
||
|
|
return setting_item_from_row(row)
|
||
|
|
|
||
|
|
|
||
|
|
def delete_system_setting(session: Session, key: str) -> None:
|
||
|
|
normalized_key = normalize_setting_key(key)
|
||
|
|
if normalized_key in PROTECTED_SETTING_KEYS:
|
||
|
|
raise ValueError("Core platform settings cannot be deleted")
|
||
|
|
row = session.get(PlatformSetting, normalized_key)
|
||
|
|
if row is None:
|
||
|
|
raise ValueError("Setting not found")
|
||
|
|
session.delete(row)
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def get_platform_settings(session: Session) -> PlatformSettingsPayload:
|
||
|
|
defaults = default_platform_settings()
|
||
|
|
ensure_default_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}
|
||
|
|
|
||
|
|
merged = defaults.model_dump()
|
||
|
|
merged["page_size"] = max(1, min(100, int(data.get("page_size") or merged["page_size"])))
|
||
|
|
merged["chat_pull_page_size"] = max(10, min(500, int(data.get("chat_pull_page_size") or merged["chat_pull_page_size"])))
|
||
|
|
merged["command_auto_unlock_seconds"] = max(
|
||
|
|
1,
|
||
|
|
min(600, int(data.get("command_auto_unlock_seconds") or merged["command_auto_unlock_seconds"])),
|
||
|
|
)
|
||
|
|
merged["upload_max_mb"] = int(data.get("upload_max_mb") or merged["upload_max_mb"])
|
||
|
|
merged["allowed_attachment_extensions"] = normalize_extension_list(
|
||
|
|
data.get("allowed_attachment_extensions", merged["allowed_attachment_extensions"]),
|
||
|
|
)
|
||
|
|
merged["workspace_download_extensions"] = normalize_extension_list(
|
||
|
|
data.get("workspace_download_extensions", merged["workspace_download_extensions"]),
|
||
|
|
)
|
||
|
|
merged["speech_enabled"] = bool(data.get("speech_enabled", merged["speech_enabled"]))
|
||
|
|
loading_page = data.get("loading_page")
|
||
|
|
if isinstance(loading_page, dict):
|
||
|
|
current = dict(merged["loading_page"])
|
||
|
|
for key in ("title", "subtitle", "description"):
|
||
|
|
value = str(loading_page.get(key) or "").strip()
|
||
|
|
if value:
|
||
|
|
current[key] = value
|
||
|
|
merged["loading_page"] = current
|
||
|
|
return PlatformSettingsPayload.model_validate(merged)
|
||
|
|
|
||
|
|
|
||
|
|
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))),
|
||
|
|
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),
|
||
|
|
loading_page=LoadingPageSettings.model_validate(payload.loading_page.model_dump()),
|
||
|
|
)
|
||
|
|
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_speech_runtime_settings() -> Dict[str, Any]:
|
||
|
|
return build_speech_runtime_settings(get_platform_settings_snapshot())
|
||
|
|
|
||
|
|
|
||
|
|
def get_activity_event_retention_days(session: Session) -> int:
|
||
|
|
row = session.get(PlatformSetting, ACTIVITY_EVENT_RETENTION_SETTING_KEY)
|
||
|
|
if row is None:
|
||
|
|
return DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS
|
||
|
|
try:
|
||
|
|
value = read_setting_value(row)
|
||
|
|
except Exception:
|
||
|
|
value = DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS
|
||
|
|
return normalize_activity_event_retention_days(value)
|
||
|
|
|
||
|
|
|
||
|
|
def get_sys_auth_token_ttl_days(session: Session) -> int:
|
||
|
|
ensure_default_system_settings(session)
|
||
|
|
row = session.get(PlatformSetting, SYS_AUTH_TOKEN_TTL_DAYS_SETTING_KEY)
|
||
|
|
if row is None:
|
||
|
|
return 7
|
||
|
|
try:
|
||
|
|
value = int(read_setting_value(row))
|
||
|
|
except Exception:
|
||
|
|
value = 7
|
||
|
|
return max(1, min(365, value))
|