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])
|