Django实现高效关键词过滤与消息管理实战

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_length50, uniqueTrue) word_type models.CharField(max_length10, choicesWORD_TYPES) replacement models.CharField(max_length50, blankTrue) is_active models.BooleanField(defaultTrue) class Meta: verbose_name 敏感词 verbose_name_plural 敏感词库 class Message(models.Model): sender models.ForeignKey(User, on_deletemodels.CASCADE, related_namesent_messages) recipient models.ForeignKey(User, on_deletemodels.CASCADE, related_namereceived_messages) content models.TextField() is_filtered models.BooleanField(defaultFalse) created_at models.DateTimeField(auto_now_addTrue) 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_activeTrue): 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_index1] word: replacement replacement if replacement else * * len(word) char_list[start_index:end_index1] list(replacement) elif word_type fuzzy: replacement replacement if replacement else * * len(word) char_list[start_index:end_index1] 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(idrecipient_id) message Message.objects.create( senderrequest.user, recipientrecipient, contentfiltered_content, is_filteredis_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_activeTrue) activate_selected.short_description 启用所选敏感词 def deactivate_selected(self, request, queryset): queryset.update(is_activeFalse) 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, timeout3600) # 缓存1小时 def _load_keywords(self): self.automaton Automaton() from .models import SensitiveWord for word in SensitiveWord.objects.filter(is_activeTrue): 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, timeout3600)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], senderSensitiveWord) 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_length10, choicesLANGUAGE_CHOICES, defaultzh) 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_activeTrue, languagelang) # ...其余过滤逻辑5.2 正则表达式支持增强模糊匹配能力import re class KeywordFilter: def _compile_regex(self, word): return re.compile(r(?Pprefix\w{0,3})%s(?Psuffix\w{0,3}) % word, re.IGNORECASE) def filter_text(self, text): # ...AC自动机匹配逻辑 # 额外处理正则匹配 regex_words SensitiveWord.objects.filter( is_activeTrue, word_typeregex ) 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(usernameuser1) cls.user2 User.objects.create(usernameuser2) SensitiveWord.objects.create( word敏感词, word_typeexact, replacement*** ) SensitiveWord.objects.create( word广告, word_typefuzzy ) 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(f100次过滤请求耗时: {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_typetext/csv) response[Content-Disposition] attachment; filenamesensitive_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