91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""敏感词服务：DFA 算法，词库 Redis 缓存 60s（后台维护后即时生效，对齐《02-技术文档》6.7）"""
|
||
import json
|
||
import time
|
||
|
||
import redis.asyncio as aioredis
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.common import SensitiveWord
|
||
from app.services import config_service
|
||
|
||
CACHE_KEY = "words:sensitive"
|
||
CACHE_TTL = 60
|
||
|
||
# 敏感词过滤总开关（system_configs 可配，后台系统配置页可热改；"0" 关闭）
|
||
FILTER_ENABLED_KEY = "sensitive.filter.enabled"
|
||
|
||
|
||
class DFAMatcher:
|
||
"""确定性有穷自动机敏感词匹配器"""
|
||
|
||
def __init__(self) -> None:
|
||
self.root: dict = {}
|
||
|
||
def add(self, word: str) -> None:
|
||
node = self.root
|
||
for ch in word.strip().lower():
|
||
if not ch:
|
||
continue
|
||
node = node.setdefault(ch, {})
|
||
node["\x00"] = True # 词结束标记
|
||
|
||
def find(self, text: str) -> str | None:
|
||
"""返回命中的第一个敏感词，未命中返回 None"""
|
||
text = text.lower()
|
||
n = len(text)
|
||
for i in range(n):
|
||
node = self.root
|
||
j = i
|
||
while j < n and text[j] in node:
|
||
node = node[text[j]]
|
||
j += 1
|
||
if "\x00" in node:
|
||
return text[i:j]
|
||
return None
|
||
|
||
|
||
# 进程内缓存（matcher, 构建时间戳）
|
||
_local_cache: tuple[DFAMatcher, float] | None = None
|
||
|
||
|
||
async def get_matcher(db: AsyncSession, redis: aioredis.Redis) -> DFAMatcher:
|
||
"""获取 DFA 匹配器：进程缓存 → Redis → DB，60s 重建"""
|
||
global _local_cache
|
||
now = time.time()
|
||
if _local_cache and now - _local_cache[1] < CACHE_TTL:
|
||
return _local_cache[0]
|
||
|
||
words: list[str] = []
|
||
cached = await redis.get(CACHE_KEY)
|
||
if cached:
|
||
words = json.loads(cached)
|
||
else:
|
||
rows = (await db.execute(select(SensitiveWord.word))).scalars().all()
|
||
words = [w for w in rows if w]
|
||
await redis.set(CACHE_KEY, json.dumps(words, ensure_ascii=False), ex=CACHE_TTL)
|
||
|
||
matcher = DFAMatcher()
|
||
for w in words:
|
||
matcher.add(w)
|
||
_local_cache = (matcher, now)
|
||
return matcher
|
||
|
||
|
||
async def find_sensitive(db: AsyncSession, redis: aioredis.Redis, text: str) -> str | None:
|
||
"""检查文本，返回命中的敏感词或 None；后台关闭过滤开关时直接放行"""
|
||
if not text:
|
||
return None
|
||
enabled = await config_service.get_config(db, redis, FILTER_ENABLED_KEY, "1")
|
||
if str(enabled) == "0":
|
||
return None
|
||
matcher = await get_matcher(db, redis)
|
||
return matcher.find(text)
|
||
|
||
|
||
async def invalidate_cache(redis: aioredis.Redis) -> None:
|
||
"""后台增删词库后调用：清 Redis 缓存，进程缓存等 60s 内自然过期"""
|
||
global _local_cache
|
||
_local_cache = None
|
||
await redis.delete(CACHE_KEY)
|