1. 项目概述：Django消息管理器与关键词过滤实战
在Web应用开发中，用户生成内容(UGC)的安全处理是每个开发者必须面对的挑战。最近在为一个社区平台升级消息管理系统时，我实现了一套基于Django的关键词过滤机制，能够有效拦截不当内容同时保持系统性能。这个方案不仅适用于站内信系统，同样可以应用于评论、帖子等所有用户输入场景。
消息过滤器的核心需求很明确：当用户发送包含预设敏感词的消息时，系统需要自动识别并执行替换或拦截操作。但在实际开发中，我发现需要解决几个关键问题：
- 如何在不影响性能的前提下实现实时过滤
- 敏感词库的易维护性和扩展性
- 过滤规则的灵活配置（如全匹配/模糊匹配）
- 多语言内容的处理能力
2. 技术架构设计
2.1 基础模型设计
首先在models.py中建立消息模型和关键词模型：
from django.db import models from django.contrib.auth import get_user_model User = get_user_model() class SensitiveWord(models.Model): WORD_TYPES = ( ('exact', '精确匹配'), ('fuzzy', '模糊匹配') ) word = models.CharField(max_length=50, unique=True) word_type = models.CharField(max_length=10, choices=WORD_TYPES) replacement = models.CharField(max_length=50, blank=True) is_active = models.BooleanField(default=True) class Meta: verbose_name = '敏感词' verbose_name_plural = '敏感词库' class Message(models.Model): sender = models.ForeignKey(User, on_delete=models.CASCADE, related_name='sent_messages') recipient = models.ForeignKey(User, on_delete=models.CASCADE, related_name='received_messages') content = models.TextField() is_filtered = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ['-created_at'] verbose_name = '站内消息' verbose_name_plural = '消息管理'2.2 过滤引擎实现
创建核心过滤工具类，采用AC自动机算法提高匹配效率：
from ahocorasick import Automaton class KeywordFilter: def __init__(self): self.automaton = Automaton() self._load_keywords() def _load_keywords(self): from .models import SensitiveWord for word in SensitiveWord.objects.filter(is_active=True): self.automaton.add_word(word.word, (word.word_type, word.replacement)) self.automaton.make_automaton() def filter_text(self, text): filtered = False result = text matches = list(self.automaton.iter(text)) if matches: filtered = True char_list = list(text) for end_index, (word_type, replacement) in matches: start_index = end_index - len(self.automaton.get(word)) + 1 if word_type == 'exact' and text[start_index:end_index+1] == word: replacement = replacement if replacement else '*' * len(word) char_list[start_index:end_index+1] = list(replacement) elif word_type == 'fuzzy': replacement = replacement if replacement else '*' * len(word) char_list[start_index:end_index+1] = list(replacement) result = ''.join(char_list) return result, filtered3. 业务逻辑集成
3.1 消息发送处理
在views.py中实现消息发送逻辑，集成过滤功能：
from django.views.decorators.http import require_POST from django.contrib.auth.decorators import login_required from django.http import JsonResponse from .models import Message from .utils import KeywordFilter filter_engine = KeywordFilter() @login_required @require_POST def send_message(request): recipient_id = request.POST.get('recipient') content = request.POST.get('content', '').strip() if not content: return JsonResponse({'status': 'error', 'message': '消息内容不能为空'}) filtered_content, is_filtered = filter_engine.filter_text(content) try: recipient = User.objects.get(id=recipient_id) message = Message.objects.create( sender=request.user, recipient=recipient, content=filtered_content, is_filtered=is_filtered ) return JsonResponse({ 'status': 'success', 'message': '消息发送成功', 'is_filtered': is_filtered }) except User.DoesNotExist: return JsonResponse({'status': 'error', 'message': '接收用户不存在'})3.2 管理员界面优化
在admin.py中配置敏感词管理界面，方便运营人员维护：
from django.contrib import admin from .models import SensitiveWord, Message @admin.register(SensitiveWord) class SensitiveWordAdmin(admin.ModelAdmin): list_display = ('word', 'word_type', 'replacement', 'is_active') list_filter = ('word_type', 'is_active') search_fields = ('word',) actions = ['activate_selected', 'deactivate_selected'] def activate_selected(self, request, queryset): queryset.update(is_active=True) activate_selected.short_description = "启用所选敏感词" def deactivate_selected(self, request, queryset): queryset.update(is_active=False) deactivate_selected.short_description = "禁用所选敏感词" @admin.register(Message) class MessageAdmin(admin.ModelAdmin): list_display = ('sender', 'recipient', 'content_preview', 'is_filtered', 'created_at') list_filter = ('is_filtered', 'created_at') search_fields = ('sender__username', 'recipient__username', 'content') readonly_fields = ('created_at',) def content_preview(self, obj): return obj.content[:50] + '...' if len(obj.content) > 50 else obj.content content_preview.short_description = '内容预览'4. 性能优化实践
4.1 缓存敏感词库
使用Django缓存框架避免每次请求都查询数据库：
from django.core.cache import cache class KeywordFilter: CACHE_KEY = 'sensitive_words_automaton' def __init__(self): self.automaton = cache.get(self.CACHE_KEY) if not self.automaton: self._load_keywords() cache.set(self.CACHE_KEY, self.automaton, timeout=3600) # 缓存1小时 def _load_keywords(self): self.automaton = Automaton() from .models import SensitiveWord for word in SensitiveWord.objects.filter(is_active=True): self.automaton.add_word(word.word, (word.word_type, word.replacement)) self.automaton.make_automaton() def refresh_cache(self): cache.delete(self.CACHE_KEY) self._load_keywords() cache.set(self.CACHE_KEY, self.automaton, timeout=3600)4.2 信号机制处理词库更新
使用Django信号在敏感词变更时自动刷新缓存：
from django.db.models.signals import post_save, post_delete from django.dispatch import receiver from .models import SensitiveWord from .utils import filter_engine @receiver([post_save, post_delete], sender=SensitiveWord) def update_filter_cache(sender, **kwargs): filter_engine.refresh_cache()5. 高级功能扩展
5.1 多语言支持
扩展模型支持多语言敏感词：
class SensitiveWord(models.Model): LANGUAGE_CHOICES = ( ('zh', '中文'), ('en', '英文'), ('ja', '日文') ) # 原有字段... language = models.CharField(max_length=10, choices=LANGUAGE_CHOICES, default='zh') class Meta: unique_together = ('word', 'language')更新过滤引擎支持语言识别：
from langdetect import detect class KeywordFilter: def filter_text(self, text): try: lang = detect(text) except: lang = 'zh' # 只加载对应语言的敏感词 words = SensitiveWord.objects.filter(is_active=True, language=lang) # ...其余过滤逻辑5.2 正则表达式支持
增强模糊匹配能力：
import re class KeywordFilter: def _compile_regex(self, word): return re.compile(r'(?P<prefix>\w{0,3})%s(?P<suffix>\w{0,3})' % word, re.IGNORECASE) def filter_text(self, text): # ...AC自动机匹配逻辑 # 额外处理正则匹配 regex_words = SensitiveWord.objects.filter( is_active=True, word_type='regex' ) for word_obj in regex_words: pattern = self._compile_regex(word_obj.word) if pattern.search(text): filtered = True replacement = word_obj.replacement if word_obj.replacement else '[已过滤]' text = pattern.sub(replacement, text) return text, filtered6. 测试与部署
6.1 单元测试用例
from django.test import TestCase from django.contrib.auth import get_user_model from .models import SensitiveWord from .utils import KeywordFilter User = get_user_model() class KeywordFilterTest(TestCase): @classmethod def setUpTestData(cls): cls.user1 = User.objects.create(username='user1') cls.user2 = User.objects.create(username='user2') SensitiveWord.objects.create( word='敏感词', word_type='exact', replacement='***' ) SensitiveWord.objects.create( word='广告', word_type='fuzzy' ) def test_exact_match(self): filter = KeywordFilter() result, filtered = filter.filter_text('这是一个敏感词测试') self.assertTrue(filtered) self.assertEqual(result, '这是一个***测试') def test_fuzzy_match(self): filter = KeywordFilter() result, filtered = filter.filter_text('不要发广告') self.assertTrue(filtered) self.assertEqual(result, '不要发**') def test_no_match(self): filter = KeywordFilter() result, filtered = filter.filter_text('正常内容') self.assertFalse(filtered) self.assertEqual(result, '正常内容')6.2 性能测试建议
使用Django测试客户端进行压力测试：
from django.test import Client from django.urls import reverse import time def test_filter_performance(): client = Client() url = reverse('send_message') # 登录逻辑... start = time.time() for i in range(100): # 模拟100次请求 response = client.post(url, { 'recipient': 2, 'content': f'测试消息{i}包含敏感词和广告内容' }) duration = time.time() - start print(f'100次过滤请求耗时: {duration:.2f}秒')7. 生产环境注意事项
词库更新策略：建议在低峰期批量更新敏感词库，避免频繁触发缓存刷新
性能监控：添加日志记录过滤耗时
import logging logger = logging.getLogger(__name__) class KeywordFilter: def filter_text(self, text): start = time.time() # ...过滤逻辑 elapsed = (time.time() - start) * 1000 # 毫秒 logger.info(f'过滤耗时: {elapsed:.2f}ms, 原始长度: {len(text)}') return result, filtered- 应急方案：当过滤系统出现异常时，应有降级策略
try: filtered_content, is_filtered = filter_engine.filter_text(content) except Exception as e: logger.error(f'内容过滤失败: {str(e)}') filtered_content = content is_filtered = False- 敏感词导入导出：为运营人员提供CSV导入导出功能
from django.http import HttpResponse from django.contrib import admin import csv @admin.register(SensitiveWord) class SensitiveWordAdmin(admin.ModelAdmin): # ...原有配置 actions = ['export_as_csv'] def export_as_csv(self, request, queryset): response = HttpResponse(content_type='text/csv') response['Content-Disposition'] = 'attachment; filename="sensitive_words.csv"' writer = csv.writer(response) writer.writerow(['word', 'word_type', 'replacement', 'is_active']) for obj in queryset: writer.writerow([obj.word, obj.word_type, obj.replacement, obj.is_active]) return response export_as_csv.short_description = "导出为CSV"
