写点什么

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:392407

评论 2 条评论

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

Python 入门指南之Python 简介

海拥(haiyong.site)

Python 7月月更

超分辨率技术在实时音视频领域的研究与实践

网易云信

实时音视频

单商户 V4.4,初心未变,实力依旧!

CRMEB

养不起真猫,就用代码吸猫 -Unity 粒子实现画猫咪

芝麻粒儿

游戏 Unity 特效 7月月更

详解SQL中Groupings Sets 语句的功能和底层实现逻辑

华为云开发者联盟

数据库 sql 聚合函数

无人自助洗车机投放选址有啥要点

共享电单车厂家

自助洗车加盟 车白兔自助洗车 自助洗车机投放

Web3.0时代来了,看天翼云存储资源盘活系统如何赋能新基建(下)

天翼云开发者社区

数字化 云存储

我们为什么要学习数学建模?

图灵教育

数学 数学建模

RedHat7.4配置yum软件仓库(RHEL7.4)

Albert Edison

7月月更

践行自主可控3.0,真正开创中国人自己的开源事业

极狐GitLab

git DevOps gitlab 自主可控 极狐GitLab

HiEngine:可媲美本地的云原生内存数据库引擎

华为云开发者联盟

数据库 华为云 内容数据库引擎

自助洗车加盟门槛如何?高不高

共享电单车厂家

自助洗车加盟费 车白兔自助洗车 自助洗车加盟门槛

来看一看智能自助洗车怎么加盟?

共享电单车厂家

自助洗车加盟 加盟自助洗车 车白兔自助洗车

5. 数据访问 - EntityFramework集成

MASA技术团队

C# .net 微软 后端 Framework

企业级数据安全,天翼云是这样理解的

天翼云开发者社区

数据安全

Python|数据结构——列表和元组

AXYZdong

7月月更

华律网牵手观测云,上线系统全链路可观测平台

观测云

CODING DevSecOps 助力金融企业跑出数字加速度

CODING DevOps

研发效能 DevSecOps CODING 代码评审 持续安全交付

视觉体验全面升级,豪威集团与英特尔Evo 3.0共同加速PC产业变革

科技之家

2022-Java后端工程师面试指南-(并发-多线程)

自然

多线程 并发 7月月更

尚硅谷尚优选项目教程发布

小谷哥

EMQX 5.0 发布:单集群支持 1 亿 MQTT 连接的开源物联网消息服务器

EMQ映云科技

物联网 IoT mqtt #开源 7月月更

更严苛的英特尔Evo 3.0,正在让合作伙伴们上瘾

科技之家

Vue和小程序的关系

Geek_99967b

小程序 Vue

MySQL数据库索引教程(超详细)

Albert Edison

7月月更

【网易云信】超分辨率技术在实时音视频领域的研究与实践

网易智企

实时音视频

现在加盟自助洗车是否还来得及

共享电单车厂家

自助洗车加盟 加盟自助洗车 车白兔自助洗车

自助共享洗车能挣钱么?分析下

共享电单车厂家

自助洗车加盟 车白兔自助洗车 自助洗车费用

通过的英特尔Evo 3.0整机认证到底有多难?忆联科技告诉你

科技之家

跨境支付平台 XTransfer 的低代码实践:如何与其他中台融合是核心

XTransfer技术

技术 前端

彻底理解为什么网络 I/O 会被阻塞?

C++后台开发

网络编程 socket 非阻塞网络I/O C++后台开发 C++开发

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