写点什么

Serverless 实战:利用触发器定制一个专属的企业微信机器人

  • 2020-06-18
  • 本文字数:9554 字

    阅读完需:约 31 分钟

Serverless实战:利用触发器定制一个专属的企业微信机器人

利用定时触发器可以快速建立一个企业微信机器人,我们可以在这个机器人中实现很多定制化的功能,例如按时提醒我们喝水吃饭、定时推送新闻天气、实现监控告警等等。

使用企业微信机器人

企业微信中可以选择添加机器人:



添加机器人之后,我们可以根据文档进行企业微信机器人的基础功能定制:


以下是用 curl 工具往群组推送文本消息的示例(注意要将 url 替换成你的机器人 webhook 地址,content 必须是 utf8 编码):


curl '企业微信机器人地址' \   -H 'Content-Type: application/json' \   -d '   {        "msgtype": "text",        "text": {            "content": "hello world"        }   }'
复制代码


通过 Python 语言实现:


url = ""data = {    "msgtype": "markdown",    "markdown": {        "content": "hello world",    }}data = json.dumps(data).encode("utf-8")req_attr = urllib.request.Request(url, data)resp_attr = urllib.request.urlopen(req_attr)return_msg = resp_attr.read().decode("utf-8")
复制代码


此时,我们可以通过 Serverless Framework 部署一个机器人的基本功能,并且设置好 API 网关触发器:


index.py文件:


import osimport jsonimport urllib.request
def main_handler(event, context): url = os.environ.get("url") data = { "msgtype": "markdown", "markdown": { "content": "hello world", } } data = json.dumps(data).encode("utf-8") req_attr = urllib.request.Request(url, data) resp_attr = urllib.request.urlopen(req_attr) return resp_attr.read().decode("utf-8")
复制代码


serverless.yaml文件:


MyRobot_Base:  component: '@serverless/tencent-scf'  inputs:    name: MyRobot_Base    runtime: Python3.6    timeout: 3    codeUri: ./base_robot    description: 机器人推送接口    region: ap-guangzhou    environment:      variables:        url: webhook地址    handler: index.main_handler    memorySize: 64    tags:      app: myrobot    events:      - apigw:          name: MyRobot          parameters:            protocols:              - http              - https            description: 机器人推送接口            environment: release            endpoints:              - path: /push                method: ANY

复制代码


部署成功之后,就可以看到系统给我们的地址:



通过浏览器打开这个地址,就可以看到企业微信机器人被触发:



至此,我们完成了一个简单的hello world功能,接下来,就对这个基础函数进行额外的改造:


import osimport jsonimport urllib.request
def main_handler(event, context): url = os.environ.get("url") data = { "msgtype": "markdown", "markdown": { "content": event['body'], } } data = json.dumps(data).encode("utf-8") req_attr = urllib.request.Request(url, data) resp_attr = urllib.request.urlopen(req_attr) return resp_attr.read().decode("utf-8")
复制代码


data中的content字段更改为event['body']之后,其他模块就可以请求该接口,实现机器人推送功能。当然,这个基础函数还是可以继续完善的,例如不仅仅是markdown格式,封装更多支持的格式:


机器人功能拓展

提醒喝水/吃饭功能

通过定时触发器访问云函数实现该功能,例如index.py代码:


import osimport jsonimport urllib.request
def main_handler(event, context): url = os.environ.get("url") data = "每天都要多喝水哦,不要忘记补充水分".encode("utf-8") req_attr = urllib.request.Request(url, data) resp_attr = urllib.request.urlopen(req_attr) return resp_attr.read().decode("utf-8")
复制代码


serverless.yaml文件:


MyRobot_Water:  component: '@serverless/tencent-scf'  inputs:    name: MyRobot_Water    runtime: Python3.6    timeout: 3    codeUri: ./water    description: 提醒喝水的机器人    region: ap-guangzhou    environment:      variables:        url: https://service-lf3ug84s-1256773370.gz.apigw.tencentcs.com/release/push    handler: index.main_handler    memorySize: 64    tags:      app: myrobot    events:      - timer:          name: timer          parameters:            cronExpression: '0 */30 9-17 * * * *'            enable: true
复制代码


这个函数就是每天上午 9 点到下午 5 点,每 30 分钟提醒我们喝一次水。

天气预报/当地新闻功能

想要实现天气预报/新闻播报的功能,我们可以通过已有的新闻接口来实现。以腾讯云的云市场为例,寻找一个新闻类的 API 接口:



根据 API 文档,可以看到请求地址是:https://service-aqvnjmiq-1257101137.gz.apigw.tencentcs.com/release/news/search


Get 方法可以携带一个参数:keyword,作为目标的关键词,代码编写:


import ssl, hmac, base64, hashlib, os, jsonfrom datetime import datetime as pydatetimefrom urllib.parse import urlencodefrom urllib.request import Request, urlopen

def main_handler(event, context): source = "market"
datetime = pydatetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT') signStr = "x-date: %s\nx-source: %s" % (datetime, source) sign = base64.b64encode(hmac.new(os.environ.get('secretKey').encode('utf-8'), signStr.encode('utf-8'), hashlib.sha1).digest()) auth = 'hmac id="%s", algorithm="hmac-sha1", headers="x-date x-source", signature="%s"' % (os.environ.get("secretId"), sign.decode('utf-8'))
headers = { 'X-Source': source, 'X-Date': datetime, 'Authorization': auth, } queryParams = {'keyword': '科技新闻'} url = 'https://service-aqvnjmiq-1257101137.gz.apigw.tencentcs.com/release/news/search' if len(queryParams.keys()) > 0: url = url + '?' + urlencode(queryParams)
content = "" for eve in json.loads(urlopen(Request(url, headers=headers)).read().decode("utf-8"))["result"]["list"][0:5]: content = content + "* [](%s) \n"%(eve['title'], eve['url'])
if content: urlopen(Request(os.environ.get('url'), content.encode("utf-8")))
复制代码


serverless.yaml文件:


MyRobot_News:  component: '@serverless/tencent-scf'  inputs:    name: MyRobot_News    runtime: Python3.6    timeout: 3    codeUri: ./news    description: 新闻推送    region: ap-guangzhou    environment:      variables:        url: https://service-lf3ug84s-1256773370.gz.apigw.tencentcs.com/release/push        secretId: 云市场密钥信息        secretKey: 云市场密钥信息    handler: index.main_handler    memorySize: 64    tags:      app: myrobot    events:      - timer:          name: timer          parameters:            cronExpression: '0 0 */8 * * * *'            enable: true
复制代码


运行效果:每天早晨 8 点为我们推送当日科技新闻:



点击新闻就可以进入到对应的新闻页面。

监控告警功能

除了上面的能力,企业微信机器人还可以被赋予监控告警的能力:


index.py文件:


import osimport urllib.request
def getStatusCode(url): return urllib.request.urlopen(url).getcode()
def main_handler(event, context): url = "http://www.anycodes.cn" if getStatusCode(url) == 200: print("您的网站%s可以访问!" % (url)) else: urllib.request.urlopen(urllib.request.Request(os.environ.get('url'), ("您的网站%s 不可以访问!" % (url)).encode("utf-8"))) return None
复制代码


serverless.yaml文件:


MyRobot_Monitor:  component: '@serverless/tencent-scf'  inputs:    name: MyRobot_Monitor    runtime: Python3.6    timeout: 3    codeUri: ./monitor    description: 网站监控    region: ap-guangzhou    environment:      variables:        url: https://service-lf3ug84s-1256773370.gz.apigw.tencentcs.com/release/push    handler: index.main_handler    memorySize: 64    tags:      app: myrobot    events:      - timer:          name: timer          parameters:            cronExpression: '0 */30 * * * * *'            enable: true
复制代码


部署完成之后,网站监控脚本就启动了,每 30 分钟检查一次网站是否可用,如果不可用,则会发送告警:


思路发散

通过 Serverless 架构,企业微信机器人可以被赋予更多更有趣的功能,那么我们身边还有那些产品可以和 Serverless 架构结合,被我们赋予有趣的功能呢?


随着网络技术的不断发展,IoT 技术也逐渐走进了千家万户,例如扫地机器人、智能窗帘等智能家居,智能音箱等娱乐设施,Serverless 和智能设备能碰撞出什么样的火花?


以智能音箱为例,我们可以利用 Serverless 架构快速开发小爱同学的专属新功能。


首先,我们需要在小爱同学的开放平台注册一个账号,并提交认证:



完成之后,我们开始对小爱同学的定制化功能进行研究。如图所示,在开发文档中,小爱同学开发者平台为我们提供了相关的能力信息,同样我们也可以查看到 request 以及 response 的详细信息:



了解了小爱同学的开发者规范之后,我们就可以开始对项目进行设计。我们的目标是:当我们对小爱同学说出“进入云+社区”关键词,就可以得到腾讯云云+社区的最新热门文章的题目和简介。整个流程如图所示:



函数代码编写:


# -*- coding: utf8 -*-import jsonimport loggingimport urllib.requestimport urllib.parse
logging.basicConfig(level=logging.NOTSET)

def main_handler(event, context): host = "https://cloud.tencent.com/" path = "developer/services/ajax/column/article?action=FetchColumnHomeArticleList" json_data = { "action": "FetchColumnHomeArticleList", "payload": { "pageNumber": 1, "pageSize": 20, "version": 1 } } data = json.dumps(json_data).encode("utf-8") request_attr = urllib.request.Request(url=host + path, data=data) response_attr = urllib.request.urlopen(request_attr).read().decode("utf-8") json_resp = json.loads(response_attr) logging.debug(json_resp) temp_str = "文章题目为%s,主要内容是%s" list_data = json_resp["data"]["list"][0:5] art_list = [temp_str % (eve["title"], eve["abstract"]) for eve in list_data] news_str = '''今日腾讯云加社区热门文章如下:%s''' % ("、".join(art_list)) logging.debug(news_str)
xiaoai_response = {"version": "1.0", "response": { "open_mic": False, "to_speak": { "type": 0, "text": news_str } }, "is_session_end": False } return xiaoai_response

复制代码


完成之后,通过 Serverless Framework 进行部署,并绑定 API 网关触发器,然后通过请求地址就可以看到测试结果:



这时,我们已经获得了目标数据,在小爱同学官网创建技能开发,填写好和保存好基本信息之后,就可以选择配置服务,在配置信息处填写 HTTPS 中的测试环境地址:



配置完成之后,就可以进入到测试页面开始测试。如下图所示,当我们输入预定的命令“打开云加社区”,系统会正确去到结果信息,并返回结果:



这样,我们就通过 Serverless 架构为小爱同学成功开发了一项新功能,如果想让其它用户也能享受此功能,可以继续发布、审核、上线。

总结

通过 Serverless 架构,我们可以快速为产品增加一些新的功能,赋予新的生机。本文仅仅是抛砖引玉,通过企业微信机器人与 Serverless 架构的结合,使用若干代码实现了提醒吃饭/喝水功能、新闻/天气功能以及业务的监控告警功能,同时发散思维到小爱同学等设备上,通过 Serverless 架构为其赋予新的能力。


2020-06-18 11:392699

评论 2 条评论

发布
用户头像
setverless让用户更关注业务函数的实现,不用考虑其他的了,哈哈
2020-06-21 00:20
回复
这是一个目标,现在是做不到的,例如各大厂商宣传的无需资源评估,但是实际上目前来看,就目前的Serverless还是要的,毕竟愿景和现实有一定差距
2020-08-14 21:32
回复
没有更多了
发现更多内容

MyCat的超强功能与应用场景入门篇

肖哥弹架构

Java mycat

在Mac 上打开FPS 监视器

柠檬与橘子

YashanDB异常故障排查流程,快速定位并解决问题

数据库砖家

YashanDB支持的编程语言:灵活使用的优势

数据库砖家

YashanDB支持的多种索引类型及应用场景

数据库砖家

YashanDB支持的开发语言与工具:提升开发效率

数据库砖家

微擎面板赋能下的智慧法律服务

微擎应用市场

得物社区活动:组件化的演进与实践

得物技术

前端 JAV A 大数据 开源

YashanDB数据压缩与存储优化详解

数据库砖家

YashanDB异地多活部署方案,保障业务连续性

数据库砖家

YashanDB支持的编程语言及开发框架介绍

数据库砖家

YashanDB支持的缓存机制如何提高性能?

数据库砖家

微擎生态下的酒店民宿预定

微擎应用市场

告警管理不止降噪:如何从零散事件中挖出关键信息

嘉为蓝鲸

告警管理 #WeOps

微擎生态下的新抽奖助手

微擎应用市场

磨锋短视频微擎开源生态驱动的短视频

微擎应用市场

AI智能体的开发流程

北京木奇移动技术有限公司

AI智能体 软件外包公司 AI智能软件

AI智能体的开发框架

北京木奇移动技术有限公司

软件外包公司 AI技术应用 AI技术公司

YashanDB支持的编程语言与框架,方便开发者使用

数据库砖家

Dropzone 4:Mac更加高效地管理文件

Rose

Gödel Rescheduler:适用于云原生系统的全局最优重调度框架

字节跳动开源

Kubernetes 云原生 GödelRescheduler 合并部署 重调度

YashanDB性能优化技巧,提升数据库响应速度的方法

数据库砖家

YashanDB支持的API接口及集成方式介绍

数据库砖家

YashanDB支持的存储格式及压缩技术解析

数据库砖家

HTML文本编辑器 BBEdit for Mac强大的智能搜索、代码折叠、FTP上传等

Rose

YashanDB数据同步工具使用指南,实现数据一致性

数据库砖家

YashanDB支持的国际标准和合规性分析

数据库砖家

淘宝天猫店铺订单列表、订单详情、订单物流 API 接口全攻略

tbapi

淘宝店铺订单接口 淘宝店铺订单交易接口 淘宝店铺订单详情接口 天猫订单接口

Safari 扩展 Noir – Dark Mode :将页面变为黑暗模式,减少眼睛疲劳

Rose

YashanDB支持的DR灾难恢复方案详解

数据库砖家

YashanDB支持的多租户管理功能介绍

数据库砖家

Serverless实战:利用触发器定制一个专属的企业微信机器人_服务革新_刘宇_InfoQ精选文章