董宏凯andClaude Opus 4.8
25e4e3dbc9
- 后端 /api/sensitive-words：全局共享词表，开放读、admin 或 lqq 整体替换写 - 前端 desensitize 种子化：SEED_DESENSITIZE_MAP + 运行时 replaceDesensitizeMap - 启动加载后台配置回灌脱敏映射；管理页支持增删改启停 + 一键导入种子词 - 入口置于左下角设置下拉（全量/简洁两种侧边栏均有），仅用户名 lqq 可见 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""Sensitive-word API (脱敏词 / 敏感词管理).
|
|
|
|
The frontend ships a seed desensitize map in code (``src/lib/desensitize.ts``);
|
|
it is the default on a fresh deployment and the source admins can import from.
|
|
This router persists a **global / shared** word map that the frontend loads at
|
|
startup and uses to build its desensitize regex. Each entry maps an original
|
|
``word`` to its ``replacement``; ``enabled`` parks a word without deleting it.
|
|
|
|
Reads are open to any authenticated user (every page desensitizes); the whole
|
|
map is replaced atomically and only by an admin.
|
|
|
|
Routes (prefix ``/api/sensitive-words``):
|
|
GET "" list all words
|
|
PUT "" replace the entire word set (admin)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_current_user, get_optional_user_from_request
|
|
from deerflow.persistence.sensitive_words import SensitiveWordStore
|
|
|
|
router = APIRouter(prefix="/api/sensitive-words", tags=["sensitive-words"])
|
|
|
|
# 敏感词管理对应的指定账号：用户名（邮箱 @ 前缀）为该值的用户可写，admin 也可写。
|
|
# 与前端 `getAccountDisplayName() === "lqq"` 的可见性门保持一致。
|
|
_MANAGER_USERNAME = "lqq"
|
|
|
|
|
|
def _username_of(user) -> str:
|
|
email = getattr(user, "email", None) or ""
|
|
return email.split("@")[0].strip().lower()
|
|
|
|
|
|
# ── schemas ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
class SensitiveWord(BaseModel):
|
|
word: str = Field(..., min_length=1, max_length=191)
|
|
replacement: str = Field(default="", max_length=191)
|
|
enabled: bool = True
|
|
sort_order: int = 0
|
|
|
|
|
|
class SensitiveWordResponse(SensitiveWord):
|
|
updated_by: str | None = None
|
|
updated_at: datetime | str | None = None
|
|
|
|
|
|
class SensitiveWordListResponse(BaseModel):
|
|
words: list[SensitiveWordResponse]
|
|
|
|
|
|
class SensitiveWordReplaceRequest(BaseModel):
|
|
words: list[SensitiveWord] = Field(default_factory=list)
|
|
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _get_store(request: Request) -> SensitiveWordStore:
|
|
store = getattr(request.app.state, "sensitive_word_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Sensitive word store not available")
|
|
return store
|
|
|
|
|
|
async def _require_manager(request: Request) -> None:
|
|
"""Allow the designated manager username (``lqq``) or an admin.
|
|
|
|
Lets the call through when auth is disabled (no user resolved).
|
|
"""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return
|
|
if getattr(user, "system_role", None) == "admin":
|
|
return
|
|
if _username_of(user) == _MANAGER_USERNAME:
|
|
return
|
|
raise HTTPException(status_code=403, detail="敏感词管理仅限指定账号")
|
|
|
|
|
|
# ── routes ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("", response_model=SensitiveWordListResponse)
|
|
async def list_sensitive_words(request: Request) -> SensitiveWordListResponse:
|
|
store = _get_store(request)
|
|
rows = await store.list_words()
|
|
return SensitiveWordListResponse(words=[SensitiveWordResponse(**r) for r in rows])
|
|
|
|
|
|
@router.put("", response_model=SensitiveWordListResponse)
|
|
async def replace_sensitive_words(request: Request, body: SensitiveWordReplaceRequest) -> SensitiveWordListResponse:
|
|
await _require_manager(request)
|
|
store = _get_store(request)
|
|
user_id = await get_current_user(request)
|
|
rows = await store.replace_all(
|
|
[w.model_dump() for w in body.words],
|
|
updated_by=user_id,
|
|
)
|
|
return SensitiveWordListResponse(words=[SensitiveWordResponse(**r) for r in rows])
|