運(yùn)維微信交互機(jī)器人

前言

今年五月份參加Oracle開(kāi)發(fā)者大會(huì),在會(huì)議上看到智能AI在運(yùn)維方面的應(yīng)用場(chǎng)景;講師現(xiàn)場(chǎng)展現(xiàn)了一款能夠結(jié)合上下文對(duì)話的智能AI,通過(guò)聊天方式完成運(yùn)維工作。

公司主營(yíng)業(yè)務(wù)：成都網(wǎng)站建設(shè)、成都網(wǎng)站設(shè)計(jì)、移動(dòng)網(wǎng)站開(kāi)發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競(jìng)爭(zhēng)能力。創(chuàng)新互聯(lián)是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開(kāi)放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對(duì)我們的高要求，感謝他們從不同領(lǐng)域給我們帶來(lái)的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶帶來(lái)驚喜。創(chuàng)新互聯(lián)推出泰來(lái)免費(fèi)做網(wǎng)站回饋大家。

會(huì)議后對(duì)該款智能AI機(jī)器人念念不忘,由于人工智能AI學(xué)習(xí)成本較高,尋思著是否能夠?qū)懸惶椎团浒?code>運(yùn)維交互機(jī)器人;

思考

初期期望該機(jī)器人能夠：

• 通過(guò)手機(jī)能夠處理簡(jiǎn)單的故障
• 不智能但至少配置能夠靈活變更

有了具體的目標(biāo), 再考慮具體實(shí)現(xiàn)方案, 主要思考幾個(gè)點(diǎn)：

應(yīng)用載體
我期望這個(gè)載體是一款常用的手機(jī)APP;現(xiàn)有環(huán)境中微信企業(yè)號(hào)適合干這個(gè)事情, 且官網(wǎng)有各種API文檔, 實(shí)施起來(lái)不是個(gè)什么巨大挑戰(zhàn).

安全性
涉及到運(yùn)維平臺(tái),控制了運(yùn)維平臺(tái)就相當(dāng)于控制了所有服務(wù)器;所以關(guān)系到運(yùn)維平臺(tái)的安全問(wèn)題不可小窺,得確保在交互過(guò)程中的安全,在交互過(guò)程中需要加密,對(duì)不信任服務(wù)器進(jìn)行策略管控.

靈活性
可以通過(guò)配置文件方式進(jìn)行配置,后續(xù)隨著功能模塊增加可以隨時(shí)進(jìn)行更改,考慮到使用配置文件方式可能太過(guò)單一,花里胡哨的功能可能無(wú)法滿足實(shí)現(xiàn),盡量考慮又能花里胡哨,又能靈活管理配置的方案.

對(duì)話上下文
一般而言,通訊都需要一個(gè)長(zhǎng)連接保證通信期間雙方可以收發(fā)數(shù)據(jù)包; 考慮到一個(gè)對(duì)話就得專門(mén)起一個(gè)線程進(jìn)行通信,這樣不但增加開(kāi)發(fā)難度,且更消耗資源, 權(quán)衡利弊后,對(duì)于上下文管理這一部分盡量選用非實(shí)時(shí)性方案去做.

架構(gòu)

列出思考的幾個(gè)關(guān)鍵點(diǎn)后,對(duì)整體的設(shè)計(jì)進(jìn)行深入思考,幾經(jīng)思考后:

• 采用微信企業(yè)號(hào)作為應(yīng)用載體

  1. 有關(guān)于企業(yè)號(hào)的開(kāi)發(fā)傳送門(mén).

• 安全加固

  1. 接口平臺(tái)只放通騰訊服務(wù)器IP訪問(wèn).
  2. 運(yùn)維平臺(tái)開(kāi)放接口平臺(tái)白名單訪問(wèn),并且采用Pythonitsdangerous生成安全令牌進(jìn)行通信交互.

• 程序設(shè)計(jì)思想

  1. 采用樹(shù)結(jié)構(gòu)設(shè)計(jì)模式,每個(gè)分叉為一個(gè)功能.
  2. 這樣就不必?fù)?dān)心無(wú)法完成花里胡哨的操作,又能夠靈活變更.

• 持久化存儲(chǔ)接收信息

  1. 對(duì)每個(gè)用戶發(fā)送的信息進(jìn)行存儲(chǔ),并作出快速響應(yīng).
  2. redis對(duì)于這個(gè)場(chǎng)景非常適用,既能夠存儲(chǔ)信息又十分高效.

架構(gòu)圖看起來(lái)大概是這樣:

實(shí)現(xiàn)

接收企業(yè)號(hào)信息API代碼片段展示

# 引用企業(yè)微信JDK
from WXcrypt.WXBizMsgCrypt import WXBizMsgCrypt

def work_weixin_api(request):
    # 獲取微信Post參數(shù)
    msg_signature = request.GET.get('msg_signature', '')
    timestamp = request.GET.get('timestamp', '')
    nonce = request.GET.get('nonce', '')
    echostr = request.GET.get('echostr', '')

    # 構(gòu)造微信信息解析方法
    wxcpt = WXBizMsgCrypt(WXTOKEN, WXENCODINGAESKEY, WXCROPID)
    if request.method == 'POST':
        eagle_branch = request.POST.get('eagle_branch', 'master')

        if eagle_branch == "master":
            request_data = request.body
            # 解析接收到的文本
            ret, msg = wxcpt.DecryptMsg(request_data, msg_signature, timestamp,
                                        nonce)
            request_xml = ET.fromstring(msg)

            # 獲取信息內(nèi)容
            content = request_xml.find("Content").text

            # 獲取信息類型
            msg_type = request_xml.find("MsgType").text

            # 獲取發(fā)送人
            from_user = request_xml.find("FromUserName").text
        else:
            content = request.POST.get('content', '')
            from_user = request.POST.get('from_user', '')

安全令牌生成

# 加密
def enc_dict(d):

    # 加密
    s = URLSafeSerializer('1234')
    st = s.dumps(d)

    # 加密后再生成基于時(shí)間戳的令牌
    t = TimestampSigner('4567')
    ts = t.sign(st)
    return ts

功能樹(shù)設(shè)計(jì)代碼片段展示

先定義一個(gè)功能樹(shù)基類

# 菜單功能的基類
class Function:
    def __init__(self, data):
        self._data = data
        self._functions = []

    # 傳入的方法的描述
    def __str__(self):
        return str(self._data())

    # 返回當(dāng)前對(duì)象類型
    def f_type(self):
        return self._data.f_type

    # 返回當(dāng)前對(duì)象
    def getData(self):
        return self._data

    # 返回所有子菜單
    def getFunctions(self):
        return self._functions

    # 新增子菜單
    def add(self, function):
        self._functions.append(function)

    # 遞歸搜索
    def go(self, num):
        for _, i in enumerate(self._functions):
            if int(num) == _ :
                return i
        return None

由于是在手機(jī)上操作, 那么交互內(nèi)容盡可能簡(jiǎn)單,所以采用全數(shù)字交互方式.
在樹(shù)結(jié)構(gòu)設(shè)計(jì)模式下,所有操作都是在遞歸搜尋,對(duì)于其他特殊的輸入,例如端口 確認(rèn)驗(yàn)證碼之類的無(wú)法實(shí)現(xiàn).

在這里需要有小小的改動(dòng)

    # 新增一個(gè)類型屬性
    def f_type(self):
        return self._data.f_type

    # 遞歸搜索
    def go(self, num):
        for _, i in enumerate(self._functions):
            f_type = i._data().f_type
            # 如果類型是默認(rèn)且存在列表中，或動(dòng)態(tài)生成類型的，直接返回
            if f_type == "default" and int(num) == _ or f_type == "dynamic":
                return i
        return None

接著,編寫(xiě)一個(gè)功能樹(shù)的類

class Menu:
    def __init__(self):
        self._head = Function(FunctionNodeBase())
        self.input_text = None

    # 鏈接
    def linkToHead(self, function):
        self._head.add(function)

    # 搜索
    def search(self, text):
        cur = self._head
        for i in text.split('-'):
            if cur.go(i) == None:
                return None
            else:
                self.input_text = i
                cur = cur.go(i)
        return cur

葉子 跟 樹(shù)的主體都有了,下面來(lái)創(chuàng)建樹(shù)頂

展示： 基礎(chǔ)功能葉 動(dòng)態(tài)功能葉 靜態(tài)功能葉

# 空的功能Node
class FunctionNodeBase:
    __metaclass__ = ABCMeta

    def __init__(self,
                 user=None,
                 f_type="default",
                 input_text=None,
                 sub_text=None):
        self.user = user
        self.sub_text = sub_text
        self.input_text = input_text
        self.f_type = f_type
        self.f_mark = []

    # 菜單通過(guò)run方法執(zhí)行與生成文本
    @abstractmethod
    def run(self):
        return self.__str__()

    # 描述
    @abstractmethod
    def __str__(self):
        return "菜單樹(shù)頂層"

# 動(dòng)態(tài)生成
class SelectDeploymentTop(FunctionNodeBase):
    # 動(dòng)態(tài)生成的菜單需要聲明f_type
    def __init__(self):
        super().__init__()
        self.f_type = "dynamic"

    def run(self):

        text                          = "請(qǐng)選擇事業(yè)部\n\n"
        deployment_list = [i for i in FunctionList.keys()]

        for _, i in enumerate(deployment_list):
            self.f_mark.append(_)
            text += "%s %s\n" % (_, i)

        return text

    # 微信顯示的文本信息
    def __str__(self):
        return "選擇事業(yè)部"

# 靜態(tài)
class MySQLFunctionTop(FunctionNodeBase):
    def __init__(self):
        super().__init__()

    def run(self):
        text = "您選擇的是%s,請(qǐng)選擇您想要操作:\n" % str(self.__str__())
        text += "%s\n" % self.sub_text
        return text

    def __str__(self):
        return "MySQL操作"

效果圖,第一層功能展示

將需要的功能逐一寫(xiě)好后需要進(jìn)行注冊(cè)

def api(tid,user):

    # 實(shí)例化
    menu             = Menu()
    top                 = Function(SelectDeploymentTop)
    function_top  = Function(FunctionTop)
    mysql_top       = Function(MySQLFunctionTop)

    # 鏈接
    top.add(function_top
    function_top.add(mysql_top)

    # 關(guān)聯(lián)菜單樹(shù)
  menu.linkToHead(top)

  # 遞歸搜索
  function = menu.search(tid)

Redis存儲(chǔ)對(duì)話代碼片段

class redis_db:
    def __init__(self):
        # 按符號(hào)隔開(kāi)
        self.mark = '-'
        self.redis_db = redis.StrictRedis(
            host = host, port=6379, db=1, decode_responses=True)

    # 默認(rèn)回話過(guò)期600秒,每次存儲(chǔ) '-'隔開(kāi)
    def add(self,key,text,Timeout=600):
        if key not in self.keys():
            self.redis_db.set(key,'',ex=Timeout)
        if self.get(key):
            self.redis_db.append(key,self.mark)
        self.redis_db.append(key,text)

同理,返回上層就刪除一格; 退出即刪除該KEY的值.

成果

下圖為：通過(guò)交互機(jī)器人連接k8s增加POD數(shù)的應(yīng)用場(chǎng)景

后記

該系統(tǒng)已經(jīng)在平臺(tái)上穩(wěn)定運(yùn)行大半年, 上線后使運(yùn)維人員能夠更高效快速解決日常中遇到的一些故障.