139 lines
5.2 KiB
Python
139 lines
5.2 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 starlette.responses import Response
|
|
|
|
from app.gateway.api_read_cache import (
|
|
get_cache_version,
|
|
get_cached_json_payload,
|
|
release_cache_fill,
|
|
set_cached_json,
|
|
wait_for_cached_json_payload_or_claim_fill,
|
|
)
|
|
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:
|
|
config = getattr(request.app.state, "config", None)
|
|
cache_key = "global"
|
|
version = await get_cache_version(config, "sensitive_words")
|
|
cached = await get_cached_json_payload(config, "sensitive_words", cache_key, version=version)
|
|
fill_token = None
|
|
if cached is None:
|
|
cached, fill_token = await wait_for_cached_json_payload_or_claim_fill(
|
|
config, "sensitive_words", cache_key, version=version, wait_ms=1000
|
|
)
|
|
if cached is not None:
|
|
return Response(content=cached, media_type="application/json") # type: ignore[return-value]
|
|
try:
|
|
store = _get_store(request)
|
|
rows = await store.list_words()
|
|
result = SensitiveWordListResponse(words=[SensitiveWordResponse(**r) for r in rows])
|
|
await set_cached_json(
|
|
config,
|
|
"sensitive_words",
|
|
cache_key,
|
|
result.model_dump(mode="json"),
|
|
expected_version=version,
|
|
ttl_seconds=3600,
|
|
)
|
|
return result
|
|
finally:
|
|
if fill_token is not None:
|
|
await release_cache_fill(config, "sensitive_words", cache_key, fill_token)
|
|
|
|
|
|
@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])
|