微信运维交互机器人

2023-11-18 19:19:04

运维微信交互机器人

前言

今年五月份参加Oracle开发者大会,在会议上看到智能AI在运维方面的应用场景;讲师现场展现了一款能够结合上下文对话的智能AI,通过聊天方式完成运维工作。

会议后对该款智能AI机器人念念不忘,由于人工智能AI学习成本较高,寻思着是否能够写一套低配版运维交互机器人;

思考

初期期望该机器人能够：

通过手机能够处理简单的故障
不智能但至少配置能够灵活变更

有了具体的目标, 再考虑具体实现方案, 主要思考几个点：

应用载体
我期望这个载体是一款常用的手机APP;现有环境中微信企业号适合干这个事情, 且官网有各种API文档, 实施起来不是个什么巨大挑战.

安全性
涉及到运维平台,控制了运维平台就相当于控制了所有服务器;所以关系到运维平台的安全问题不可小窥,得确保在交互过程中的安全,在交互过程中需要加密,对不信任服务器进行策略管控.

灵活性
可以通过配置文件方式进行配置,后续随着功能模块增加可以随时进行更改,考虑到使用配置文件方式可能太过单一,花里胡哨的功能可能无法满足实现,尽量考虑又能花里胡哨,又能灵活管理配置的方案.

对话上下文
一般而言,通讯都需要一个长连接保证通信期间双方可以收发数据包; 考虑到一个对话就得专门起一个线程进行通信,这样不但增加开发难度,且更消耗资源, 权衡利弊后,对于上下文管理这一部分尽量选用非实时性方案去做.

架构

列出思考的几个关键点后,对整体的设计进行深入思考,几经思考后:

采用微信企业号作为应用载体
1. 有关于企业号的开发传送门.
安全加固
1. 接口平台只放通腾讯服务器IP访问.
2. 运维平台开放接口平台白名单访问,并且采用Pythonitsdangerous生成安全令牌进行通信交互.
程序设计思想
1. 采用树结构设计模式,每个分叉为一个功能.
2. 这样就不必担心无法完成花里胡哨的操作,又能够灵活变更.
持久化存储接收信息
1. 对每个用户发送的信息进行存储,并作出快速响应.
2. Redis对于这个场景非常适用,既能够存储信息又十分高效.

架构图看起来大概是这样:

实现

接收企业号信息API代码片段展示

# 引用企业微信JDK
from WXcrypt.WXBizMsgCrypt import WXBizMsgCrypt

def work_weixin_api(request):
    # 获取微信Post参数
    msg_signature = request.GET.get(‘msg_signature‘, ‘‘)
    timestamp = request.GET.get(‘timestamp‘, ‘‘)
    nonce = request.GET.get(‘nonce‘, ‘‘)
    echostr = request.GET.get(‘echostr‘, ‘‘)

    # 构造微信信息解析方法
    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)

            # 获取信息内容
            content = request_xml.find("Content").text

            # 获取信息类型
            msg_type = request_xml.find("MsgType").text

            # 获取发送人
            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)

    # 加密后再生成基于时间戳的令牌
    t = TimestampSigner(‘4567‘)
    ts = t.sign(st)
    return ts

功能树设计代码片段展示

先定义一个功能树基类

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

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

    # 返回当前对象类型
    def f_type(self):
        return self._data.f_type

    # 返回当前对象
    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

由于是在手机上操作, 那么交互内容尽可能简单,所以采用全数字交互方式.
在树结构设计模式下,所有操作都是在递归搜寻,对于其他特殊的输入,例如端口 确认验证码之类的无法实现.

在这里需要有小小的改动

    # 新增一个类型属性
    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
            # 如果类型是默认且存在列表中，或动态生成类型的，直接返回
            if f_type == "default" and int(num) == _ or f_type == "dynamic":
                return i
        return None

接着,编写一个功能树的类

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

叶子 跟 树的主体都有了,下面来创建树顶

展示： 基础功能叶 动态功能叶 静态功能叶

# 空的功能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 = []

    # 菜单通过run方法执行与生成文本
    @abstractmethod
    def run(self):
        return self.__str__()

    # 描述
    @abstractmethod
    def __str__(self):
        return "菜单树顶层"

# 动态生成
class SelectDeploymentTop(FunctionNodeBase):
    # 动态生成的菜单需要声明f_type
    def __init__(self):
        super().__init__()
        self.f_type = "dynamic"

    def run(self):

        text                          = "请选择事业部\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 "选择事业部"

# 静态
class MySQLFunctionTop(FunctionNodeBase):
    def __init__(self):
        super().__init__()

    def run(self):
        text = "您选择的是%s,请选择您想要操作:\n" % str(self.__str__())
        text += "%s\n" % self.sub_text
        return text

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

效果图,第一层功能展示

将需要的功能逐一写好后需要进行注册

def api(tid,user):

    # 实例化
    menu             = Menu()
    top                 = Function(SelectDeploymentTop)
    function_top  = Function(FunctionTop)
    mysql_top       = Function(MySQLFunctionTop)

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

    # 关联菜单树
  menu.linkToHead(top)

  # 递归搜索
  function = menu.search(tid)

Redis存储对话代码片段

class redis_db:
    def __init__(self):
        # 按符号隔开
        self.mark = ‘-‘
        self.redis_db = redis.StrictRedis(
            host = host, port=6379, db=1, decode_responses=True)

    # 默认回话过期600秒,每次存储 ‘-‘隔开
    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的值.

成果

下图为：通过交互机器人连接k8s增加POD数的应用场景

微信运维交互机器人

码农公寓

前言

思考

架构

实现

成果

相关文章