一、概述
目前，网络游戏和视频直播都是很火爆的产业，然而，无论在多人在线网络游戏（MMORPG)中，还是在游戏视频直播或者其他直播中(弹幕)，玩家或者用户的在线交流和互动都是它们的核心乐趣所在，但是玩家之间的交流和互动不可避免地会涉及到一些敏感词，对玩家或者用户交流中出现的敏感词进行过滤，提供一个健康和谐的网络环境是非常重要有意义的。
二、现有技术方案
现有技术中，通常采用两种方式进行敏感词的过滤：
第一种方式：客户端完成敏感词的过滤
客户端存储有敏感词库；由玩家输入的文本内容经过客户端敏感词过滤模块过滤，然后将过滤后的文本由客户端发送至服务器，再由服务器转发至接收方客户端显示，有时为了担心发送端的过滤模块被hacker绕过，需要在接收方在接收的时候，也经过下客户端敏感词过滤模块。
- 优势:
每个客户端单独执行敏感词过滤，效率高，且不占用服务器资源，服务器没有计算消耗。 - 劣势:
敏感词具有时效性，针对最新出现的敏感词，不能实时的增加到客户端的敏感词库，只能等下次客户端打新包或者patch才可以添加进去，这个时间成本有点大。
第二种方式：服务器完成敏感词的过滤
在服务器中存储敏感词库，并设置敏感词过滤模块，将服务器接收到的文本内容进行敏感词过滤，并将执行完敏感词过滤后的文本内容发送至客户端。
- 优势：
在于可实时增加敏感词，当出现突发事件后，不需要停服就可实现动态添加新的敏感词。 - 劣势：由服务器执行敏感词过滤会导致增加大量额外计算消耗，影响服务器性能。有的特殊的输入可能会导致执行某些检测规则时间过长造成服务器进程卡顿。
三、函数计算可以提供的解决方案
将敏感词检测的模块放在函数计算实现，至于是由客户端和服务端发起函数调用看具体需求，在本示例中，我们把发起的敏感词检测过程放在客户端发起，如下图所示：
优势：不会增加服务器的计算消耗，同时只需要更新下函数，就可以达到实时更新敏感词目的。
具体步骤
1 . 函数计算作为 API 网关后端服务, 具体的教程可以参考官方教程和函数计算获取临时token
2 . 函数计算实现敏感词检测的功能，本文主要讲解这个过程, 以python作为编程语言，给出具体的示例代码。
在本教程中，选用正则式作为我们的敏感词检测，也就是python的re模块，但是做敏感词检查还存在一些性能问题需要我们去解决，问题是：当游戏或者视频直播上线的时候，敏感词正则库一般有几百个，在python中通过 re.compile 编译正则表达式是很耗时的，几百个正则编译完可能需要10几秒甚至20多秒，但是如果正则式如果是编译完的，那么直接进行匹配match或者search还是很快的，十几毫秒可以搞定。
初始方案：我们可以将正则表达式的编译结果 cPickle.dumps 序列化保存到文件中。待下次程序启动时直接从文件读取内容， cPickle.loads 反序列化成正则表达式对象。但是这个有坑，根本没有耗时改善，详情见cPickle正则表达式对象
最终方案：基于cPickle正则表达式对象的处理方案，对于最耗时的结果压缩再序列化，反之，反序列的时候记得再解压一下就行，这样的好处是，中间的序列化文件大小可以大大压缩，本教程测试一般显示能压缩5倍以上，但是对整个运行时间影响不是很大，如果对时间特别敏感的，可以不考虑压缩过程。
核心处理代码文件re_pickple.py
# -*- coding:utf-8 -*-
'''
_sre.SRE_Pattern 对象的序列化，就是编译的函数 re._compile 和输入的参数 pattern ，flags 给保存起来，
反序列化的时候 _compile(pattern, flags) ,这和直接 re.compile 没有有什么区别，赤裸裸的伪序列化
这里将re模块中的compile过程分成两个部分，dump最耗时间的code，然后load出code，再快速转换成 _sre.SRE_Pattern 对象
'''
import cPickle, re, sre_compile, sre_parse, _sre
import zlib,time
# 开启压缩的开关
zip_flag = False
# 目前函数计算的ca环境没有内置builtins
# 需要先在本地pip install -t . future
if zip_flag:
from builtins import int
'''
raw_compile, build_compiled是re模块中的compile拆分成两个部分
'''
# the first half of sre_compile.compile
def raw_compile(p, flags=0):
# internal: convert pattern list to internal format
if sre_compile.isstring(p):
pattern = p
p = sre_parse.parse(p, flags)
else:
pattern = None
# 主要耗时在_code函数
code = sre_compile._code(p, flags)
if zip_flag:
code = zlib.compress(str(code)) # code格式简单，元素都是整型的list
return code
# the second half of sre_compile.compile
def build_compiled(p, flags, code):
# print code
# 重新计算p, parse函数耗时很少，如果将p dump,占用空间较大
if sre_compile.isstring(p):
pattern = p
p = sre_parse.parse(p, flags)
else:
pattern = None
# XXX: <fl> get rid of this limitation!
if p.pattern.groups > 100:
raise AssertionError(
"sorry, but this version only supports 100 named groups"
)
# map in either direction
groupindex = p.pattern.groupdict
indexgroup = [None] * p.pattern.groups
for k, i in groupindex.items():
indexgroup[i] = k
return _sre.compile(
pattern, flags | p.pattern.flags, code,
p.pattern.groups-1,
groupindex, indexgroup
)
def dump(regexes, o_file):
picklable = []
for r in regexes:
code = raw_compile(r, 0)
picklable.append((r,code))
cPickle.dump(picklable,o_file)
def load(pkl):
regexes = []
for r, code in cPickle.load(pkl):
if zip_flag:
code= decode_str(code)
regexes.append(build_compiled(r, 0, code))
return regexes
def decode_str(code_str): #解压缩
raw_code = zlib.decompress(code_str)
raw_code.strip()
raw_code = raw_code[1:-1]
str_li = raw_code.split(',')
code = [int(item.strip()) for item in str_li]
return code假设我们有一个检测昵称的正则字库，一个有150条正则规则，部分内容如下:
# -*- coding: utf-8 -*-
data = [
'(?i)公会(强势)??入驻|招募.{0,3}?玩家|强势入驻|进群有??福利|体验号|交流群|群[^a-zA-Z一-龥0-9_丷灬一丨丿乀乄丶]*?号码|活跃工会|[qｑ][^a-zA-Z一-龥0-9_丷灬一丨丿乀乄丶]*?群|公会福利|福利[暴超]多|激活码|序列号|公会号|[百白]度搜|免费发放|发放免费|工会群|发放礼包|礼包发放|各种礼包|免费礼包|礼包免费|加入.{0,3}?公会',
'([\dⅠⅡⅢⅣⅤⅥⅧⅦⅨⅩⅪⅫ⒈⒉⒊⒋⒌⒍⒎⒏⒐⒑⒒⒓⒔⒕⒖⒗⒘⒙⒚⒛㈠㈡㈢㈤㈣㈥㈦㈧㈨㈩①②③④⑤⑥⑦⑧⑨⑩⑴⑵⑷⑶⑸⑹⑺⑻⑼⑽⑾⑿⒀⒁⒂⒃⒄⒅⒆⒇１２３４５６７８９０一二三四五六七八九十0０零oOＯｏ○〇0oOｏ０Ｏāóòáō零][^A-Za-z0-9_一-龥]{0,6}?){5,}?',
......
]2.1 我们对其中间最耗时的结果进行序列化, 序列化结果保存在name_prog.pkl文件中:
# -*- coding:utf-8 -*-
import server_nick_name
import re, time, re_pickle
import cPickle as pickle
with open(u'name_prog.pkl', 'wb') as f:
nick_name_regx = []
for pattern in server_nick_name.data:
pattern = unicode(pattern,"utf8")
nick_name_regx.append( pattern )
re_pickle.dump(nick_name_regx, f)
print "done"2.2 敏感词检测demo
# -*- coding:utf-8 -*-
import time, re_pickle
start = time.time()
with open(u'name_prog.pkl', 'r') as f:
nick_name_prog = re_pickle.load(f)
end = time.time()
print "load time = ", end - start
def check_nick_name_valid(nick_name):
nick_name = unicode(nick_name,"utf8")
for prog in nick_name_prog:
if prog.search(nick_name):
return False
return True
if __name__ == '__main__':
inputlst = ["李四", "小白兔", "李四", "王五"]
for input in inputlst:
start = time.time()
ret = check_nick_name_valid(input)
end = time.time()
print 'check ret = {} ; check_time = {}'.format(ret,end - start)输出如下：
load time = 0.238202095032
check ret = True ; check_time = 0.000257015228271
check ret = True ; check_time = 8.10623168945e-06
check ret = True ; check_time = 4.05311584473e-06
check ret = True ; check_time = 0.000158071517944注：由于网络安全审查，这里举例的名字都是合法的，用户可以在本地尝试使用下敏感的名字
因此，只要我们将name_prog.pkl和我们的检测函数、re_pickple.py一起，构建我们的函数，假设我们的函数如下，一次性检查4个名字，我们测试下调用100消耗时间的情况。
# -*- coding:utf-8 -*-
import time, re_pickle
begin = time.time()
with open(u'name_prog.pkl', 'r') as f:
nick_name_prog = re_pickle.load(f)
end = time.time()
print "load time = ", end - begin
def check_nick_name_valid(nick_name):
nick_name = unicode(nick_name,"utf8")
for prog in nick_name_prog:
if prog.search(nick_name):
return False
return True
def handler(event, context):
inputlst = ["李四", "毛大", "小白兔", "张三"]
for input in inputlst:
ret = check_nick_name_valid(input)
return time.time() - begin调用100次所消耗的时间情况如下：
avg: 0.000343136787415
min: 0.000230073928833
max: 0.000503063201904
总结
采用函数计算实现自定义的敏感词检测模块方案是一个很好的选择，自己预编译的的正则序列化中间结果除了第一次load大约是100来毫秒以外，其他的检测结果基本毫秒级就能给出结果，而且函数计算能做到自动按流量来scale，即使出现大量的文字交流，也能快速及时处理，很火的视频直播再也不怕海量弹幕了。同时，进一步扩展，针对网络游戏中出现的语音信息的翻译成文字，文字敏感性检测都可以通过函数计算来实现, 语音敏感检测。
