董宏凯andClaude Haiku 4.5
01012ff300
- 登录时从路由参数 yUserId/access_token 获取并存入 userInfo - 轻应用管理页路由参数新增 userName/gateDepartName/typeName 选项 - 添加系统默认界面模式设置（管理员可指定全量/简洁） - useLayoutMode hook 支持优先级：用户自选 > 管理员默认 > 会话级 Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
380 lines
15 KiB
Python
380 lines
15 KiB
Python
"""Admin-configurable system-wide runtime settings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.auth.password import hash_password_async
|
|
from app.gateway.deps import get_optional_user_from_request
|
|
from deerflow.config import get_app_config
|
|
from deerflow.config.system_settings import (
|
|
LeaderboardSettings,
|
|
PromptPrefixSettings,
|
|
PublishSquare,
|
|
PublishSquaresSettings,
|
|
SystemSettings,
|
|
list_publish_squares,
|
|
load_system_settings,
|
|
resolve_default_square_id,
|
|
save_system_settings,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/system-settings", tags=["system-settings"])
|
|
|
|
|
|
class PromptPrefixResponse(BaseModel):
|
|
enabled: bool = Field(..., description="Whether the toggle is exposed to users")
|
|
switch_label: str = Field(..., description="Toggle label shown in the chat input")
|
|
prompt_prefix: str = Field(..., description="Prefix text prepended to the first user message")
|
|
|
|
|
|
class PromptPrefixUpdateRequest(BaseModel):
|
|
enabled: bool | None = None
|
|
switch_label: str | None = None
|
|
prompt_prefix: str | None = None
|
|
|
|
|
|
class PrivateSquareStatusResponse(BaseModel):
|
|
"""Status of the private-square feature, readable by any logged-in user.
|
|
|
|
The password hash itself is **never** returned — the response only signals
|
|
whether the feature is enabled and whether a password has been set.
|
|
"""
|
|
|
|
enabled: bool = Field(..., description="Whether the private square is exposed to users")
|
|
password_set: bool = Field(..., description="Whether a password is currently configured")
|
|
|
|
|
|
class PrivateSquareUpdateRequest(BaseModel):
|
|
enabled: bool | None = Field(default=None, description="Toggle the private-square feature")
|
|
password: str | None = Field(default=None, description="New password in plain text. Pass an empty string to clear.")
|
|
|
|
|
|
class CitationDisplayResponse(BaseModel):
|
|
cleanup_words: list[str] = Field(default_factory=list, description="Global citation cleanup words")
|
|
|
|
|
|
class CitationDisplayUpdateRequest(BaseModel):
|
|
cleanup_words: list[str] | None = None
|
|
|
|
|
|
class LeaderboardSettingsResponse(BaseModel):
|
|
exclude_user_ids: list[str] = Field(default_factory=list)
|
|
include_scheduled: bool = True
|
|
include_admins: bool = True
|
|
include_failed: bool = True
|
|
cleanup_system_questions: bool = True
|
|
|
|
|
|
class LeaderboardSettingsUpdateRequest(BaseModel):
|
|
exclude_user_ids: list[str] | None = None
|
|
include_scheduled: bool | None = None
|
|
include_admins: bool | None = None
|
|
include_failed: bool | None = None
|
|
cleanup_system_questions: bool | None = None
|
|
|
|
|
|
def _to_response(settings: PromptPrefixSettings) -> PromptPrefixResponse:
|
|
return PromptPrefixResponse(
|
|
enabled=settings.enabled,
|
|
switch_label=settings.switch_label,
|
|
prompt_prefix=settings.prompt_prefix,
|
|
)
|
|
|
|
|
|
async def _require_admin(request: Request) -> None:
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="System settings can only be modified by an admin")
|
|
|
|
|
|
@router.get("/prompt-prefix", response_model=PromptPrefixResponse)
|
|
async def get_prompt_prefix() -> PromptPrefixResponse:
|
|
"""Return the current prompt-prefix configuration (readable by any user)."""
|
|
return _to_response(load_system_settings().prompt_prefix)
|
|
|
|
|
|
@router.put("/prompt-prefix", response_model=PromptPrefixResponse)
|
|
async def update_prompt_prefix(body: PromptPrefixUpdateRequest, request: Request) -> PromptPrefixResponse:
|
|
"""Update the prompt-prefix configuration. Admin only."""
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
fields = body.model_dump(exclude_unset=True)
|
|
updated_prefix = current.prompt_prefix.model_copy(update=fields)
|
|
saved = save_system_settings(current.model_copy(update={"prompt_prefix": updated_prefix}))
|
|
return _to_response(saved.prompt_prefix)
|
|
|
|
|
|
def _leaderboard_to_response(settings: LeaderboardSettings) -> LeaderboardSettingsResponse:
|
|
return LeaderboardSettingsResponse(
|
|
exclude_user_ids=list(dict.fromkeys([uid for uid in settings.exclude_user_ids if uid])),
|
|
include_scheduled=settings.include_scheduled,
|
|
include_admins=settings.include_admins,
|
|
include_failed=settings.include_failed,
|
|
cleanup_system_questions=settings.cleanup_system_questions,
|
|
)
|
|
|
|
|
|
@router.get("/leaderboard", response_model=LeaderboardSettingsResponse)
|
|
async def get_leaderboard_settings(request: Request) -> LeaderboardSettingsResponse:
|
|
"""Return global leaderboard aggregation settings. Admin only."""
|
|
await _require_admin(request)
|
|
return _leaderboard_to_response(load_system_settings().leaderboard)
|
|
|
|
|
|
@router.put("/leaderboard", response_model=LeaderboardSettingsResponse)
|
|
async def update_leaderboard_settings(
|
|
body: LeaderboardSettingsUpdateRequest,
|
|
request: Request,
|
|
) -> LeaderboardSettingsResponse:
|
|
"""Update global leaderboard aggregation settings. Admin only."""
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
fields = body.model_dump(exclude_unset=True)
|
|
update: dict[str, object] = {}
|
|
if "exclude_user_ids" in fields and fields["exclude_user_ids"] is not None:
|
|
update["exclude_user_ids"] = list(dict.fromkeys([str(uid).strip() for uid in fields["exclude_user_ids"] if str(uid).strip()]))
|
|
for key in ("include_scheduled", "include_admins", "include_failed", "cleanup_system_questions"):
|
|
if key in fields and fields[key] is not None:
|
|
update[key] = bool(fields[key])
|
|
leaderboard = current.leaderboard.model_copy(update=update)
|
|
saved = save_system_settings(current.model_copy(update={"leaderboard": leaderboard}))
|
|
return _leaderboard_to_response(saved.leaderboard)
|
|
|
|
|
|
@router.get("/citation-display", response_model=CitationDisplayResponse)
|
|
async def get_citation_display() -> CitationDisplayResponse:
|
|
"""Return global citation display cleanup settings."""
|
|
settings = load_system_settings().citation_display
|
|
return CitationDisplayResponse(cleanup_words=settings.cleanup_words)
|
|
|
|
|
|
@router.put("/citation-display", response_model=CitationDisplayResponse)
|
|
async def update_citation_display(
|
|
body: CitationDisplayUpdateRequest,
|
|
request: Request,
|
|
) -> CitationDisplayResponse:
|
|
"""Update global citation display cleanup settings. Admin only."""
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
words = [
|
|
str(item).strip()
|
|
for item in (body.cleanup_words or [])
|
|
if str(item).strip()
|
|
]
|
|
updated = current.citation_display.model_copy(update={"cleanup_words": words})
|
|
saved = save_system_settings(
|
|
current.model_copy(update={"citation_display": updated})
|
|
)
|
|
return CitationDisplayResponse(cleanup_words=saved.citation_display.cleanup_words)
|
|
|
|
|
|
@router.get("/private-square", response_model=PrivateSquareStatusResponse)
|
|
async def get_private_square_status() -> PrivateSquareStatusResponse:
|
|
"""Return whether the private square is enabled and a password is set."""
|
|
settings = load_system_settings().private_square
|
|
return PrivateSquareStatusResponse(
|
|
enabled=settings.enabled,
|
|
password_set=bool(settings.password_hash),
|
|
)
|
|
|
|
|
|
@router.put("/private-square", response_model=PrivateSquareStatusResponse)
|
|
async def update_private_square(body: PrivateSquareUpdateRequest, request: Request) -> PrivateSquareStatusResponse:
|
|
"""Update the private-square configuration. Admin only.
|
|
|
|
Setting ``password`` to a non-empty string rehashes and stores it.
|
|
Passing an empty string clears the password (effectively locking the
|
|
private square so no one can read it).
|
|
"""
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
updated = current.private_square
|
|
fields = body.model_dump(exclude_unset=True)
|
|
|
|
update: dict[str, object] = {}
|
|
if "enabled" in fields and fields["enabled"] is not None:
|
|
update["enabled"] = bool(fields["enabled"])
|
|
if "password" in fields and fields["password"] is not None:
|
|
plain = str(fields["password"])
|
|
if plain == "":
|
|
update["password_hash"] = ""
|
|
else:
|
|
update["password_hash"] = await hash_password_async(plain)
|
|
if update:
|
|
updated = updated.model_copy(update=update)
|
|
saved = save_system_settings(current.model_copy(update={"private_square": updated}))
|
|
else:
|
|
saved = current
|
|
return PrivateSquareStatusResponse(
|
|
enabled=saved.private_square.enabled,
|
|
password_set=bool(saved.private_square.password_hash),
|
|
)
|
|
|
|
|
|
class PublishSquareItem(BaseModel):
|
|
id: str = Field(..., description="Stable square id")
|
|
name: str = Field(..., description="Display name / tab label")
|
|
description: str = Field(default="", description="Optional description")
|
|
|
|
|
|
class PublishSquaresResponse(BaseModel):
|
|
"""The configured publish squares plus the resolved default.
|
|
|
|
Readable by any logged-in user — squares are pure category tabs with no
|
|
access control, so the gallery needs the list to render its sub-tabs and
|
|
the publish flow needs it to populate the square picker.
|
|
"""
|
|
|
|
squares: list[PublishSquareItem] = Field(default_factory=list)
|
|
default_square_id: str = Field(..., description="Square id used when a publisher picks none")
|
|
|
|
|
|
class PublishSquaresUpdateRequest(BaseModel):
|
|
squares: list[PublishSquareItem] | None = Field(default=None, description="Full ordered replacement list of squares")
|
|
default_square_id: str | None = Field(default=None, description="Which square id is the publish default")
|
|
|
|
|
|
def _publish_squares_response(settings: PublishSquaresSettings) -> PublishSquaresResponse:
|
|
squares = list_publish_squares(settings)
|
|
return PublishSquaresResponse(
|
|
squares=[PublishSquareItem(id=s.id, name=s.name, description=s.description) for s in squares],
|
|
default_square_id=resolve_default_square_id(settings),
|
|
)
|
|
|
|
|
|
@router.get("/publish-squares", response_model=PublishSquaresResponse)
|
|
async def get_publish_squares() -> PublishSquaresResponse:
|
|
"""Return the configured publish squares (readable by any logged-in user)."""
|
|
return _publish_squares_response(load_system_settings().publish_squares)
|
|
|
|
|
|
@router.put("/publish-squares", response_model=PublishSquaresResponse)
|
|
async def update_publish_squares(body: PublishSquaresUpdateRequest, request: Request) -> PublishSquaresResponse:
|
|
"""Replace the publish-squares configuration. Admin only.
|
|
|
|
``squares`` is a full ordered replacement. Each square needs a non-empty
|
|
name; a missing id is auto-generated from a slug of the name. Duplicate ids
|
|
are de-duplicated (first wins). ``default_square_id`` must reference one of
|
|
the surviving squares, else it falls back to the first.
|
|
"""
|
|
from uuid import uuid4
|
|
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
fields = body.model_dump(exclude_unset=True)
|
|
update: dict[str, object] = {}
|
|
|
|
if "squares" in fields and fields["squares"] is not None:
|
|
normalized: list[PublishSquare] = []
|
|
seen_ids: set[str] = set()
|
|
for raw in body.squares or []:
|
|
name = (raw.name or "").strip()
|
|
if not name:
|
|
continue
|
|
sid = (raw.id or "").strip() or uuid4().hex[:8]
|
|
if sid in seen_ids:
|
|
continue
|
|
seen_ids.add(sid)
|
|
normalized.append(PublishSquare(id=sid, name=name, description=(raw.description or "").strip()))
|
|
update["squares"] = normalized
|
|
|
|
if "default_square_id" in fields and fields["default_square_id"] is not None:
|
|
update["default_square_id"] = str(fields["default_square_id"]).strip()
|
|
|
|
if update:
|
|
updated = current.publish_squares.model_copy(update=update)
|
|
# Re-anchor the default to a surviving square id.
|
|
if updated.squares and updated.default_square_id not in {s.id for s in updated.squares}:
|
|
updated = updated.model_copy(update={"default_square_id": updated.squares[0].id})
|
|
saved = save_system_settings(current.model_copy(update={"publish_squares": updated}))
|
|
else:
|
|
saved = current
|
|
return _publish_squares_response(saved.publish_squares)
|
|
|
|
|
|
class AppearanceResponse(BaseModel):
|
|
"""Effective system-wide appearance defaults."""
|
|
|
|
default_theme: Literal["light", "dark", "lightblue"] = Field(..., description="System default theme — light, dark or lightblue")
|
|
default_skill_view: Literal["card", "list"] = Field(..., description="System default skill list display mode — card or list")
|
|
default_agent_view: Literal["card", "list"] = Field(..., description="System default agent list display mode — card or list")
|
|
default_layout_mode: Literal["full", "compact"] = Field(..., description="System default interface layout — full (sidebar + header) or compact (hidden)")
|
|
|
|
|
|
class AppearanceUpdateRequest(BaseModel):
|
|
default_theme: Literal["light", "dark", "lightblue"] | None = Field(default=None)
|
|
default_skill_view: Literal["card", "list"] | None = Field(default=None)
|
|
default_agent_view: Literal["card", "list"] | None = Field(default=None)
|
|
default_layout_mode: Literal["full", "compact"] | None = Field(default=None)
|
|
|
|
|
|
def _appearance_response(appearance) -> AppearanceResponse:
|
|
return AppearanceResponse(
|
|
default_theme=appearance.default_theme,
|
|
default_skill_view=appearance.default_skill_view,
|
|
default_agent_view=appearance.default_agent_view,
|
|
default_layout_mode=appearance.default_layout_mode,
|
|
)
|
|
|
|
|
|
@router.get("/appearance", response_model=AppearanceResponse)
|
|
async def get_appearance_settings() -> AppearanceResponse:
|
|
"""Return the global appearance defaults. Any logged-in user may read this."""
|
|
return _appearance_response(load_system_settings().appearance)
|
|
|
|
|
|
@router.put("/appearance", response_model=AppearanceResponse)
|
|
async def update_appearance_settings(
|
|
body: AppearanceUpdateRequest, request: Request
|
|
) -> AppearanceResponse:
|
|
"""Update the global appearance defaults. Admin only."""
|
|
await _require_admin(request)
|
|
|
|
current = load_system_settings()
|
|
fields = body.model_dump(exclude_unset=True)
|
|
update: dict[str, object] = {}
|
|
for key in ("default_theme", "default_skill_view", "default_agent_view", "default_layout_mode"):
|
|
if key in fields and fields[key] is not None:
|
|
update[key] = fields[key]
|
|
if update:
|
|
updated_appearance = current.appearance.model_copy(update=update)
|
|
saved = save_system_settings(
|
|
current.model_copy(update={"appearance": updated_appearance})
|
|
)
|
|
else:
|
|
saved = current
|
|
return _appearance_response(saved.appearance)
|
|
|
|
|
|
class LogSystemResponse(BaseModel):
|
|
"""External log-system entry shown in the admin settings menu.
|
|
|
|
Sourced from ``config.yaml`` (``log_system`` section), so it is read-only
|
|
here — there is no PUT counterpart.
|
|
"""
|
|
|
|
enabled: bool = Field(..., description="Whether the entry is shown in the settings menu")
|
|
url: str = Field(..., description="External log-system URL opened in a new tab")
|
|
label: str = Field(..., description="Menu label shown for the entry")
|
|
|
|
|
|
@router.get("/log-system", response_model=LogSystemResponse)
|
|
async def get_log_system(request: Request) -> LogSystemResponse:
|
|
"""Return the admin-configured external log-system link from config.yaml. Admin only."""
|
|
await _require_admin(request)
|
|
cfg = get_app_config().log_system
|
|
return LogSystemResponse(enabled=cfg.enabled, url=cfg.url, label=cfg.label)
|