使用公有云云函数创建自己的消息提醒服务

之前一直使用Server酱的消息提醒服务，前一段时间微信发布了公告模板消息可能会在21年4月份下线，到时候Server酱的提醒服务也会受到影响，虽然Server酱已经提供了Turbo版加入了企业微信的支持，利用企业微信的消息提醒实现在微信中提醒，但是Turbo版也加入了相关的限制(如每天最多5条消息提醒)。

不论是server酱还是自己建立的消息提醒，本质都相当于对企业微信的提供的API进行了封装。其实完全可以直接使用企业微信的API，不过使用起来会复杂一些。既然只是对API进行封装，为什么不自己封装一下建立自己的消息提醒服务呢？

目前各大公有云都提供了Severless服务(云函数)，完全可以使用云函数完成对API的封装。我这里使用的是腾讯云，其他公有云的操作也类似。

步骤1：企业微信相关配置

PS：相关说明直接copy hostloc论坛@arron说明：https://shimo.im/docs/38dpjtwWtRRVQ6Wy/read

第一步，注册企业

用电脑打开企业微信官网，注册一个企业

第二步，创建应用

注册成功后，点「管理企业」进入管理界面，选择「应用管理」 → 「自建」 → 「创建应用」

应用名称填入「Server酱」，应用logo到这里下载，可见范围选择公司名。

创建完成后进入应用详情页，可以得到应用ID( agentid )，应用Secret( secret )，复制并填到上方。

第三步，获取企业ID

进入「我的企业」页面，拉到最下边，可以看到企业ID，复制并填到上方。

推送UID直接填 @all ，推送给公司全员。

第四步，推送消息到微信

进入「我的企业」 → 「微信插件」，拉到下边扫描二维码，关注以后即可收到推送的消息。

步骤2：云函数相关设置

1.新建一个云函数，语言可以选择python、go、java、nodejs等，这里我使用的是python3.6

加入代码(其实比较简单)，已经放在了GitHub，https://github.com/yjcn/msg_server/，防止有的同学没法访问我也在博客里粘一下

# -*- coding: utf8 -*-
import os
import json
import requests
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 从环境变量获取相关id
CORPID = os.environ&#91;'corpid']
AGENTID = os.environ&#91;'agentid']
CORPSECRET = os.environ&#91;'corpsecret']
# 相关id设置 
# corpid = 'xxx'  # 企业id
# agentid = 'xxx'   # 应用id
# corpsecret = 'xxx' # 企业secret

# 企业微信 api url
MSG_URL = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}'

# text_card template
text_card_dict = {
       "touser" : "@all",
       # "toparty" : "PartyID1 | PartyID2",
       # "totag" : "TagID1 | TagID2",
       "msgtype" : "textcard",
       "agentid" : AGENTID,
       "textcard" : {
                "title" : "领奖通知",
                "description" : "&lt;div class=\"gray\"&gt;2016年9月26日&lt;/div&gt; &lt;div class=\"normal\"&gt;恭喜你抽中iPhone 7一台，领奖码：xxxx&lt;/div&gt;&lt;div class=\"highlight\"&gt;请于2016年10月10日前联系行政同事领取&lt;/div&gt;",
                "url" : "URL",
                            "btntxt":"更多"
       },
       "enable_id_trans": 0,
       "enable_duplicate_check": 0,
       "duplicate_check_interval": 1800
    }

def get_access_token():
    url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&amp;corpsecret={}".format(CORPID, CORPSECRET)
    resp = requests.get(url=url)
    resp_dict = resp.json()
    access_token = None
    if resp.status_code == 200:
        access_token = resp_dict&#91;'access_token']
    return access_token
def main_handler(event, context):
    logger.info("Received event: " + json.dumps(event, indent = 2))
    logger.info("Received context: " + str(context))
    # 解析请求参数
    queryString = event&#91;'queryString']
    msg_title = queryString.get('text',None)
    msg_desp = queryString.get('desp', msg_title)
    text_card_dict&#91;'textcard']&#91;'title'] = msg_title
    text_card_dict&#91;'textcard']&#91;'description'] = msg_desp
    # logger.info(str(text_card_dict))
    # 获取access_token
    access_token = get_access_token() if msg_title is not None else None
    resp_status = 400
    resp_json = dict()
    if access_token is not None:
        resp = requests.post(url=MSG_URL.format(access_token),json=text_card_dict)
        resp_status = resp.status_code
        resp_json = resp.json()
    return {
        "isBase64Encoded": False,
        "statusCode": resp_status,
        "headers": {'Content-Type': 'application/json'},
        "body": json.dumps(resp_json)
    }

# -*- coding: utf8 -*-

import os

import json

import requests

import logging

logger = logging.getLogger()

logger.setLevel(logging.INFO)

# 从环境变量获取相关id

CORPID = os.environ['corpid']

AGENTID = os.environ['agentid']

CORPSECRET = os.environ['corpsecret']

# 相关id设置

# corpid = 'xxx' # 企业id

# agentid = 'xxx' # 应用id

# corpsecret = 'xxx' # 企业secret

# 企业微信 api url

MSG_URL = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}'

# text_card template

text_card_dict = {

"touser" : "@all",

# "toparty" : "PartyID1 | PartyID2",

# "totag" : "TagID1 | TagID2",

"msgtype" : "textcard",

"agentid" : AGENTID,

"textcard" : {

"title" : "领奖通知",

"description" : "<div class=\"gray\">2016年9月26日</div> <div class=\"normal\">恭喜你抽中iPhone 7一台，领奖码：xxxx</div><div class=\"highlight\">请于2016年10月10日前联系行政同事领取</div>",

"url" : "URL",

"btntxt":"更多"

"enable_id_trans": 0,

"enable_duplicate_check": 0,

"duplicate_check_interval": 1800

}

def get_access_token():

url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}".format(CORPID, CORPSECRET)

resp = requests.get(url=url)

resp_dict = resp.json()

access_token = None

if resp.status_code == 200:

access_token = resp_dict['access_token']

return access_token

def main_handler(event, context):

logger.info("Received event: " + json.dumps(event, indent = 2))

logger.info("Received context: " + str(context))

# 解析请求参数

queryString = event['queryString']

msg_title = queryString.get('text',None)

msg_desp = queryString.get('desp', msg_title)

text_card_dict['textcard']['title'] = msg_title

text_card_dict['textcard']['description'] = msg_desp

# logger.info(str(text_card_dict))

# 获取access_token

access_token = get_access_token() if msg_title is not None else None

resp_status = 400

resp_json = dict()

if access_token is not None:

resp = requests.post(url=MSG_URL.format(access_token),json=text_card_dict)

resp_status = resp.status_code

resp_json = resp.json()

return {

"isBase64Encoded": False,

"statusCode": resp_status,

"headers": {'Content-Type': 'application/json'},

"body": json.dumps(resp_json)

}

2.设置环境变量

在函数管理-函数配置里设置环境变量，把第一步获取到的agenid(应用id)、corpid(企业id)、corpsecret(企业secret)设置为云函数的环境变量。(当然也可以直接在云函数里直接写好。)

3.创建API触发器

建立好云函数数后，点击触发管理，创建API网关触发

创建好之后，就可以使用POST或者GET的方式访问建立的API链接(访问路径的链接)，与sever酱相同，text为消息的标题，desp为具体的消息描述，访问完之后就会微信就会收到消息提醒。

除了使用默认的访问路径，还可以在API网关-服务-自定义域名里设置自定义域名，具体设置可以查看说明文档。

PS：目前的代码只实现了基本的功能，欢迎大家继续提PR继续完善

YJ's Blog

Coding Makes Me Better

使用公有云云函数创建自己的消息提醒服务

第一步，注册企业

第二步，创建应用

第三步，获取企业ID

第四步，推送消息到微信

《使用公有云云函数创建自己的消息提醒服务》有3个想法

回复 yangjin 取消回复