在上一篇“5分鐘Serverless實踐”系列文章中，我們介紹了什麽是Serverless，以及如何構建一個無服務器的圖片鑒黃Web應用，本文將延續這個話題，以敏感詞過濾為例，介紹如何構建一個無服務器API，即無服務器的後端系統。

函數工作流

函數工作流（FunctionGraph，FGS）是一項基於事件驅動的函數托管計算服務，托管函數具備以毫秒級彈性伸縮、免運維、高可靠的方式運行。通過函數工作流，開發者無需配置和管理服務器，只需關註業務邏輯，編寫函數代碼，以無服務器的方式構建應用，便能開發出一個彈性高可用的後端系統，並按實際運行消耗的資源計費。極大地提高了開發和運維效率，減小了運作成本。

相比於傳統的架構，函數工作流構建的無服務器架構具有如下優點：
1.無需關註任何服務器，只需關註核心業務邏輯，提高開發和運維效率
2.函數運行隨業務量彈性伸縮，按需付費，執行才計費，對於負載波峰波谷非常明顯的場景可以減少大量成本
3.通過簡單的配置即可連通函數工作流和其它各雲服務，甚至雲服務和雲服務

構建無服務器的敏感詞過濾後端系統

為了進一步讓大家感受函數工作流的優勢，我們將介紹如何通過函數工作流快速構建一個無服務器的敏感詞過濾系統，本文我們主要關註後端系統，前端的表現形式很多，大家可以自行構建。如下圖，該系統會識別用戶上傳的文本內容是否包含敏感信息（如×××、政治等），並對這些詞語進行過濾。

●試想，如果我們通過傳統的模式開發此應用，需要如何開發？即使是基於現在的雲平臺，我們也仍需要購買雲服務器，關註其規格、鏡像、網絡等各指標的選型和運維，然後在開發過程中可能還需要考慮與其他雲服務的集成使用問題，使代碼中耦合大量非業務代碼，並且服務器等資源也並非是按需的，特別是對於訪問量波峰波谷非常明顯的場景，會造成大量多余的費用。

現在我們可以通過函數工作流服務來快速構建這個系統，並且完全無需關註服務器，且彈性伸縮運行、按需計費，如圖：

創建函數，在函數中調用華為雲內容檢測服務提供的文本檢測接口，實現文本的敏感詞檢測，並為該函數配置一個APIG觸發器，這樣便可以對外提供一個敏感詞過濾的API，從而構建出一個完整的敏感詞過濾的無服務器後端系統。客戶端調用API，他會自動觸發函數執行，而開發者編寫的函數只需實現接收到文本之後如何處理文本的邏輯即可，最後將結果返回給客戶端。至此，我們就構建了一個完整的無服務器敏感詞過濾後端系統。

接下來，我們將介紹如何完整地將此無服務器後端系統構建出來。

1.準備工作
進入華為雲內容檢測服務，申請開通文本內容檢測，成功申請後便可以調用內容檢測服務提供的文本檢測接口了。

2.創建函數
進入函數工作流服務頁面，創建函數，實現文本檢測的接口調用和敏感詞過濾，代碼如下：

# -*- coding:utf-8 -*-
import json
import base64
import urllib
import urllib2
import ssl
import sys

reload(sys)
sys.setdefaultencoding(‘utf-8‘)

def do_filter(msg,str_list):
    result = ‘‘
    try:
        if len(str_list) <=0:
            return msg
        for str in str_list:
            str_tmp = msg.replace(str,‘‘)
            msg = str_tmp
        result = msg
    except:
        print("_do_filter catch an exception!")
    return result

def filter(context, msg):
    result = ‘‘
    try:
        ssl._create_default_https_context = ssl._create_unverified_context

        token = context.getToken();
        headers = {‘Content-Type‘:‘application/json;charset=utf8‘,‘X-Auth-Token‘:token}

        url = "https://ais.cn-north-1.myhwclouds.com/v1.0/moderation/text"

        values = {}
        values[‘categories‘] = [‘porn‘,‘ad‘,‘politics‘,‘abuse‘,‘contraband‘]
        #msg = base64.b64encode(msg)
        item = {‘type‘:‘content‘,‘text‘:msg}
        values[‘items‘] = [item]

        data = json.dumps(values)
        print("data: %s"%data)

        request = urllib2.Request(url,data,headers)
        rsp = urllib2.urlopen(request)
        http_rsp = rsp.read()
        print("http response: %s" %http_rsp)

        json_rsp = json.loads(http_rsp)
        result = json_rsp[‘result‘]

        suggestion = result[‘suggestion‘]

        if suggestion == ‘pass‘:
            print("input msg have passed the checking!")
            result = msg
        else:
            detail = result[‘detail‘]

            if detail.has_key(‘porn‘):
                list_porn = detail[‘porn‘]
                msg = do_filter(msg,list_porn)
            if detail.has_key(‘ad‘):
                list_ad = detail[‘ad‘]
                msg = do_filter(msg,list_ad)
            if detail.has_key(‘politics‘):
                list_politics = detail[‘politics‘]
                msg = do_filter(msg,list_politics)
            if detail.has_key(‘abuse‘):
                list_abuse = detail[‘abuse‘]
                msg = do_filter(msg,list_abuse)
            if detail.has_key(‘contraband‘):
                list_contraband = detail[‘contraband‘]
                msg = do_filter(msg,list_contraband)
            result = msg
    except Exception, e:
        print e
        print("filter catch an exception!")
    return result

def handler (event, context):
    print("message filter begin!")
    result = ""
    response = {}
    http_method = event.get(‘httpMethod‘)

    if http_method == ‘OPTIONS‘:
        response = {
            ‘statusCode‘: 200,
            ‘isBase64Encoded‘: True,
            ‘headers‘: {
                "Content-Type": "application/json; charset=utf-8",
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Headers": "Content-Type,Accept",
                "Access-Control-Allow-Methods": "GET,POST,PUT,DELETE"
            },
            ‘body‘: base64.b64encode(‘{"result":‘+ ‘"‘ + result +‘"}‘),
        }
        return response
    body = event.get(‘body‘)
    body_decode = base64.b64decode(body)
    json_object = json.loads(body_decode)
    msg = json_object[‘msg‘]

    print(‘msg : %s‘%msg)

    try:
        result = filter(context, msg)
        response = {
            ‘statusCode‘: 200,
            ‘isBase64Encoded‘: True,
            ‘headers‘: {
                "Content-Type": "application/json; charset=utf-8",
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Headers": "Content-Type,Accept",
                "Access-Control-Allow-Methods": "GET,POST,PUT,DELETE"
            },
            ‘body‘: base64.b64encode(‘{"result":‘+ ‘"‘ + result +‘"}‘),
        }
    except:
        print("function catch an exception!")

    return response

函數創建完成之後，為其配置具有IAM訪問權限的委托，因為本函數代碼中獲取用戶的ak、sk需要擁有訪問IAM的權限。

3.創建APIG觸發器
為函數配置一個APIG觸發器，這樣便得到一個調用該函數的HTTP（S） API，供外部調用。

創建成功後，API的URL可以在函數詳情頁面的“觸發器”欄看到：

4.測試
使用postman等工具向上一步中創建的APIG觸發器的接口發送post請求，body體為：{“msg”: “過濾檢測的文本”}，查看返回信息。

比如發送 {"msg": "just fuck ..."}， 返回體為 {"result": "just ..."}

至此，我們就完整地構建了一個無服務器的敏感詞過濾後端系統。

5分鐘Serverless實踐 | 構建無服務器的敏感詞過濾後端系統