122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
"""System-wide runtime settings editable from the admin UI.
|
|
|
|
Backed by a single JSON file under the DeerFlow runtime home (default
|
|
``.deer-flow/system_settings.json``). Unlike :mod:`extensions_config` this
|
|
file is **not** part of the project source tree — it is per-deployment
|
|
state, and it stays writable at runtime.
|
|
|
|
The only feature stored here at the moment is the **prompt prefix toggle**
|
|
shown in the chat input: an administrator can configure a label and a
|
|
prefix string, and any user can flip the toggle on/off; when enabled, the
|
|
prefix is prepended to the first human message of a new thread.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from deerflow.config.runtime_paths import runtime_home
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_SETTINGS_FILE = "system_settings.json"
|
|
_lock = threading.Lock()
|
|
|
|
|
|
class PromptPrefixSettings(BaseModel):
|
|
"""Admin-configurable first-message prompt prefix."""
|
|
|
|
enabled: bool = Field(default=False, description="Whether the prefix toggle is offered to users at all")
|
|
switch_label: str = Field(default="加入立场前缀", description="Toggle label shown in the chat input")
|
|
prompt_prefix: str = Field(default="", description="Text prepended to the first user message when enabled")
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
|
|
class AppearanceSettings(BaseModel):
|
|
"""System-wide appearance defaults that any *new* user inherits.
|
|
|
|
Once a user explicitly toggles their theme in the UI, that choice is
|
|
persisted in the browser and supersedes this default — so changing
|
|
``default_theme`` here only affects users who have never touched the
|
|
toggle, and any future first-time loads.
|
|
"""
|
|
|
|
default_theme: Literal["light", "dark"] = Field(
|
|
default="light",
|
|
description="Theme applied for users who haven't picked their own (light / dark)",
|
|
)
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
|
|
class CitationDisplaySettings(BaseModel):
|
|
"""System-wide cleanup rules for citation cards and detail dialogs."""
|
|
|
|
cleanup_words: list[str] = Field(
|
|
default_factory=list,
|
|
description="Words or phrases used to remove noisy advertisement/disclaimer lines from citation display.",
|
|
)
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
|
|
class PrivateSquareSettings(BaseModel):
|
|
"""Admin-configurable settings for the agent private-square (私密广场).
|
|
|
|
The password gates read access to all agents flagged ``is_private=True``.
|
|
The hash is stored opaquely — the router layer is responsible for hashing
|
|
and verifying so the harness package keeps zero dependencies on
|
|
``app.gateway.auth``.
|
|
"""
|
|
|
|
enabled: bool = Field(default=False, description="Whether the private square is offered to users at all")
|
|
password_hash: str = Field(default="", description="Opaque hash of the admin-configured password. Empty = no password set yet.")
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
|
|
class SystemSettings(BaseModel):
|
|
"""Top-level system settings document."""
|
|
|
|
prompt_prefix: PromptPrefixSettings = Field(default_factory=PromptPrefixSettings)
|
|
private_square: PrivateSquareSettings = Field(default_factory=PrivateSquareSettings)
|
|
appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
|
|
citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
|
|
def _settings_path() -> Path:
|
|
home = runtime_home()
|
|
home.mkdir(parents=True, exist_ok=True)
|
|
return home / _SETTINGS_FILE
|
|
|
|
|
|
def load_system_settings() -> SystemSettings:
|
|
"""Read the on-disk system settings, returning defaults when missing."""
|
|
path = _settings_path()
|
|
if not path.is_file():
|
|
return SystemSettings()
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(raw, dict):
|
|
logger.warning("System settings file %s is not a JSON object; ignoring", path)
|
|
return SystemSettings()
|
|
return SystemSettings.model_validate(raw)
|
|
except Exception as exc:
|
|
logger.warning("Failed to read system settings from %s: %s", path, exc)
|
|
return SystemSettings()
|
|
|
|
|
|
def save_system_settings(settings: SystemSettings) -> SystemSettings:
|
|
"""Persist the settings document atomically, returning the saved value."""
|
|
path = _settings_path()
|
|
payload = settings.model_dump(mode="json")
|
|
with _lock:
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
tmp.replace(path)
|
|
return settings
|