使用云函数SCF+COS免费运营微信公众号

2024-03-23 19:38
文章标签 函数 使用 微信 运营 免费 公众 cos scf

本文主要是介绍使用云函数SCF+COS免费运营微信公众号，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

是的，你没听错，这一次我来带大家直接上手运营微信公众号。

震惊，Awesome，哼，我才不信捏，所谓无图无真相 ~

效果展示

更多的体验，可以关注我的微信公众号：乂乂又又 （仅供测试，不要乱搞哈~）

嗯，这次我信了，快点教一下我吧，嘤嘤嘤~

操作步骤

在上一篇《使用SCF+COS快速开发全栈应用》教程中，我们用腾讯云无服务器云函数 SCF 和对象存储 COS 实现了一个后端云函数，这个云函数可以根据我们的请求返回对应的结果。

现在我们将尝试在这个云函数的基础上解析微信 XML 消息，实现公众号消息的自动回复，关键词回复，文字菜单等功能。

01

添加相关依赖

为了快速完成开发，这里我们选择 python 第三方开源库 wechatpy 来接入微信公众平台。

wechatpy 支持以下功能：

  1. 普通公众平台被动响应和主动调用 API

  2. 企业微信 API

  3. 微信支付 API

  4. 第三方平台代公众号调用接口 API

  5. 小程序云开发 API

可见功能是十分完整的，不仅支持普通公众平台主被动调用，企业微信和微信支付，甚至还支持第三方平台代公众号调用接口，拿来运营微信公众号是十分绰绰有余的~

由于腾讯云函数的运行环境中缺少第三方库，需要我们自己手动上传添加依赖，这里我们需要添加的第三方依赖有：wechatpy、optionaldict、xmltodict 以及 timeout_decorator

其中 wechatpy 需要依赖 optionaldict、xmltodict，timeout_decorator 是用来限制函数运行时长的，具体的依赖文件可以自行 pip 安装后 copy 到云函数项目根目录，如上图。

02

接入微信公众号

这里需要记下自己的 AppID、Token 和 EncodingAESKey，消息加密方式建议选为安全模式。这个页面先不要关，一会儿上线发布好云函数还需要过来再次修改配置。

03

编写云函数解析并回复微信公众号消息

这一步可以直接参考 wechatpy 的官方文档

地址：http://docs.wechatpy.org/zh_CN/master/quickstart.html#id2

Life is short, show me the code.

这里我就直接上代码了（原始业务代码已略去，可以按照自己的需求开发）

import json
import timeout_decorator
from wechatpy.replies import ArticlesReply
from wechatpy.utils import check_signature
from wechatpy.crypto import WeChatCrypto
from wechatpy import parse_message, create_reply
from wechatpy.exceptions import InvalidSignatureException, InvalidAppIdException# 是否开启本地debug模式
debug = False# 腾讯云对象存储依赖
if debug:from qcloud_cos import CosConfigfrom qcloud_cos import CosS3Clientfrom qcloud_cos import CosServiceErrorfrom qcloud_cos import CosClientError
else:from qcloud_cos_v5 import CosConfigfrom qcloud_cos_v5 import CosS3Clientfrom qcloud_cos_v5 import CosServiceErrorfrom qcloud_cos_v5 import CosClientError# 配置存储桶
appid = '66666666666'
secret_id = u'xxxxxxxxxxxxxxx'
secret_key = u'xxxxxxxxxxxxxxx'
region = u'ap-chongqing'
bucket = 'name'+'-'+appid# 对象存储实例
config = CosConfig(Secret_id=secret_id, Secret_key=secret_key, Region=region)
client = CosS3Client(config)# cos 文件读写
def cosRead(key):try:response = client.get_object(Bucket=bucket, Key=key)txtBytes = response['Body'].get_raw_stream()return txtBytes.read().decode()except CosServiceError as e:return ""def cosWrite(key, txt):try:response = client.put_object(Bucket=bucket,Body=txt.encode(encoding="utf-8"),Key=key,)return Trueexcept CosServiceError as e:return Falsedef getReplys():replyMap = {}replyTxt = cosRead('Replys.txt')  # 读取数据if len(replyTxt) > 0:replyMap = json.loads(replyTxt)return replyMapdef addReplys(reply):replyMap = getReplys()if len(replyMap) > 0:replyMap[reply]='我是黑名单'return cosWrite('Replys.txt', json.dumps(replyMap, ensure_ascii=False)) if len(replyMap) > 0 else Falsedef delReplys(reply):replyMap = getReplys()if len(replyMap) > 0:replyMap.pop(reply)return cosWrite('Replys.txt', json.dumps(replyMap, ensure_ascii=False)) if len(replyMap) > 0 else False# 微信公众号对接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'crypto = WeChatCrypto(WECHAT_TOKEN, encoding_aes_key, wecaht_id)# api网关响应集成
def apiReply(reply, txt=False, content_type='application/json', code=200):return {"isBase64Encoded": False,"statusCode": code,"headers": {'Content-Type': content_type},"body": json.dumps(reply, ensure_ascii=False) if not txt else str(reply)}def replyMessage(msg):txt = msg.contentip = msg.sourceprint('请求信息--->'+ip+'%'+txt)  # 用来在腾讯云控制台打印请求日志replysTxtMap = getReplys() # 获取回复关键词if '@' in txt:keys = txt.split('@')if keys[0] == '电影': #do somethingreturnif keys[0] == '音乐': #do somethingreturnif keys[0] == '下架': #do somethingreturnif keys[0] == '上架': #do somethingreturnif keys[0] == '回复': #do somethingreturnif keys[0] == '删除': #do somethingreturnelif txt in replysTxtMap.keys(): # 如果消息在回复关键词内则自动回复return create_reply(replysTxtMap[txt], msg)return create_reply("喵呜 ฅ'ω'ฅ", msg)def wechat(httpMethod, requestParameters, body=''):if httpMethod == 'GET':signature = requestParameters['signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']echo_str = requestParameters['echostr']try:check_signature(WECHAT_TOKEN, signature, timestamp, nonce)except InvalidSignatureException:echo_str = 'error'return apiReply(echo_str, txt=True, content_type="text/plain")elif httpMethod == 'POST':msg_signature = requestParameters['msg_signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']try:decrypted_xml = crypto.decrypt_message(body,msg_signature,timestamp,nonce)except (InvalidAppIdException, InvalidSignatureException):returnmsg = parse_message(decrypted_xml)if msg.type == 'text':reply = replyMessage(msg)elif msg.type == 'image':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发图片干啥~', msg)elif msg.type == 'voice':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发语音干啥~', msg)else:reply = create_reply('哈◔ ‸◔？\n搞不明白你给我发了啥~', msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")else:msg = parse_message(body)reply = create_reply("喵呜 ฅ'ω'ฅ", msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")@timeout_decorator.timeout(4, timeout_exception=StopIteration)
def myMain(httpMethod, requestParameters, body=''):return wechat(httpMethod, requestParameters, body=body)def timeOutReply(httpMethod, requestParameters, body=''):msg_signature = requestParameters['msg_signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']try:decrypted_xml = crypto.decrypt_message(body,msg_signature,timestamp,nonce)except (InvalidAppIdException, InvalidSignatureException):returnmsg = parse_message(decrypted_xml)reply = create_reply("出了点小问题，请稍后再试", msg).render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")def main_handler(event, context):body = ''httpMethod = event["httpMethod"]requestParameters = event['queryString']if 'body' in event.keys():body = event['body']try:response = myMain(httpMethod, requestParameters, body=body)except:response = timeOutReply(httpMethod, requestParameters, body=body)return response

请求参数解析和 COS 读写部分可参考上一篇《使用 SCF+COS 快速开发全栈应用》教程

下面我来捋一下整个云函数的思路

def main_handler(event, context):body = ''httpMethod = event["httpMethod"]requestParameters = event['queryString']if 'body' in event.keys():body = event['body']try:response = myMain(httpMethod, requestParameters, body=body)except:response = timeOutReply(httpMethod, requestParameters, body=body)return response

我们先从main_handler入手。

这里我们通过 API 网关触发云函数在 event 里拿到了微信公众号请求的方法、头部和请求体，然后传给 myMain 函数做处理，需要注意的是 myMain 是通过timeout_decorator包装的限时运行函数。

@timeout_decorator.timeout(4, timeout_exception=StopIteration)
def myMain(httpMethod, requestParameters, body=''):return wechat(httpMethod, requestParameters, body=body)

当 myMain 函数运行市场超过设定的 4 秒后，就会抛出异常。

然后我们可以通过设置一个 timeOutReply 函数来处理超时后的微信公众号消息回复，可是为什么要这么做呢？

可以看到，当云函数运行超时后，微信这边就会显示「该公众号提供的服务器出现故障，请稍后再试」

这对用户体验是极不友好的，所以我们需要一个函数超时后的回复来兜底。

那么对于一次微信公众号后台消息请求多长时间算是超时呢？答案是 5 秒左右，从云函数后台的调用日志我们可以得到这个结果。

不过需要注意的是对于用户的一次消息请求，微信可能会每隔 1 秒左右重拨一次请求，直到收到服务器第一次响应。另外，超过 3 次应该就不会再重拨了，并且在 5 秒超时后即使云函数调用成功并返回了数据，用户也不会再接收到消息了~

所以我们就很有必要将自己的云函数的运行时长限制在 5 秒之内了！

当然只通过配置云函数超时时长得方式来处理是不正确的，因为这样做云函数超时后就被系统停掉了，并不会向微信返回消息。所以从一开始我就导入了 timeout_decorator 库来限制主函数的运行时长，并用一个超时后回复函数来兜底。

另外值得一提的是，在我原始的业务代码中是有一些爬虫，这些爬虫本来我是单线程顺序执行的，考虑到超时问题，我在微信云函数版这里全部改成了多线程运行来压缩时间，所以如果你也有一些比较耗时的小任务话，也可以尝试通过多线程的方式来压缩云函数的运行时长。

我们接着向下看：

def wechat(httpMethod, requestParameters, body=''):if httpMethod == 'GET':signature = requestParameters['signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']echo_str = requestParameters['echostr']try:check_signature(WECHAT_TOKEN, signature, timestamp, nonce)except InvalidSignatureException:echo_str = 'error'return apiReply(echo_str, txt=True, content_type="text/plain")elif httpMethod == 'POST':msg_signature = requestParameters['msg_signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']try:decrypted_xml = crypto.decrypt_message(body,msg_signature,timestamp,nonce)except (InvalidAppIdException, InvalidSignatureException):returnmsg = parse_message(decrypted_xml)if msg.type == 'text':reply = replyMessage(msg)elif msg.type == 'image':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发图片干啥~', msg)elif msg.type == 'voice':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发语音干啥~', msg)else:reply = create_reply('哈◔ ‸◔？\n搞不明白你给我发了啥~', msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")else:msg = parse_message(body)reply = create_reply("喵呜 ฅ'ω'ฅ", msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")

这里的 wechat 函数就是整个微信消息的解析过程，首先判断请求方法是 GET 还是 POST，GET 方法只在第一次绑定微信后台时会用到，这时我们会从微信服务器推送的请求参数中拿到 signature, timestamp, echostr 和 nonce 参数，

check_signature(WECHAT_TOKEN, signature, timestamp, nonce)

我们只需根据自己的公众号 token 和来生成签名与微信服务器传过来的 signature 对比看是否一致，若一致就说明我们的消息加解密验证是OK的，然后再将 echostr 原样返回即可接入微信公众号后台。

接入好微信公众号后，如果有用户在后台给我们发送消息，这里云函数收到的就是 POST 方法，

elif httpMethod == 'POST':msg_signature = requestParameters['msg_signature']timestamp = requestParameters['timestamp']nonce = requestParameters['nonce']try:decrypted_xml = crypto.decrypt_message(body,msg_signature,timestamp,nonce)except (InvalidAppIdException, InvalidSignatureException):returnmsg = parse_message(decrypted_xml)if msg.type == 'text':reply = replyMessage(msg)elif msg.type == 'image':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发图片干啥~', msg)elif msg.type == 'voice':reply = create_reply('哈◔ ‸◔？\n好端端的，给我发语音干啥~', msg)else:reply = create_reply('哈◔ ‸◔？\n搞不明白你给我发了啥~', msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")

然后我们根据前面在微信公众号后台拿到的 id，token 和 aes 加密 key 来初始化消息加解密实例并解密还原用户发送的消息

# 微信公众号对接
wecaht_id = 'xxxxxxxxxxxxxxx'
WECHAT_TOKEN = 'xxxxxxxxxxxxxxxxxxx'
encoding_aes_key = 'xxxxxxxxxxxxxxxxxxxxxx'crypto = WeChatCrypto(WECHAT_TOKEN, encoding_aes_key, wecaht_id)

接着判断一下消息类型，不同类型的消息可自行处理

        msg = parse_message(decrypted_xml)if msg.type == 'text':reply = replyMessage(msg)elif msg.type == 'image':reply = create_reply('哈◔ ‸◔？ 好端端的，给我发图片干啥~', msg)elif msg.type == 'voice':reply = create_reply('哈◔ ‸◔？ 好端端的，给我发语音干啥~', msg)else:reply = create_reply('哈◔ ‸◔？ 搞不明白你给我发了啥~', msg)

需要注意的是当一个用户新关注自己的公众号时，我们收到的是一个其他类型的消息，也就是上面的最后一个判断项，这里你可以自己设置新关注用户的欢迎语

        reply = create_reply('哈◔ ‸◔？\n搞不明白你给我发了啥~', msg)reply = reply.render()print('返回结果--->'+str(reply))  # 用来在腾讯云控制台打印请求日志reply = crypto.encrypt_message(reply, nonce, timestamp)return apiReply(reply, txt=True, content_type="application/xml")

之后我们通过 create_reply 来快速创建一个文本回复，并通过 render() 来生成 xml 回复消息文本。因为我之前在后台设置的是安全模式，所以还需要把 xml 重新通过 crypto.encrypt_message 方法加密，然后才能把加密后的回复消息返回给微信服务器。

上一篇文章我有提到我们不能直接返回消息，需要按照特定的格式返回数据（API 网关需要开启响应集成）

# api网关响应集成
def apiReply(reply, txt=False, content_type='application/json', code=200):return {"isBase64Encoded": False,"statusCode": code,"headers": {'Content-Type': content_type},"body": json.dumps(reply, ensure_ascii=False) if not txt else str(reply)}

04

上线发布云函数、添加 API 网关触发器、启用响应集成

参考上一篇教程 《使用 SCF+COS 快速开发全栈应用》

05

修改微信公众号后台服务器配置

终于到最后一步了，如果你已经上线发布了好自己的云函数，那么快去微信公众号后台绑定一下自己的后台服务器配置吧~

呼~ 大功告成

 点击阅读原文，领取 COS 限时1元礼包！

这篇关于使用云函数SCF+COS免费运营微信公众号的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/839335。

相关文章

Java使用Thumbnailator库实现图片处理与压缩功能

Java使用Thumbnailator库实现图片处理与压缩功能

《Java使用Thumbnailator库实现图片处理与压缩功能》Thumbnailator是高性能Java图像处理库,支持缩放、旋转、水印添加、裁剪及格式转换,提供易用API和性能优化,适合Web应... 目录1. 图片处理库Thumbnailator介绍2. 基本和指定大小图片缩放功能2.1 图片缩放的
阅读更多...
Python使用Tenacity一行代码实现自动重试详解

Python使用Tenacity一行代码实现自动重试详解

《Python使用Tenacity一行代码实现自动重试详解》tenacity是一个专为Python设计的通用重试库,它的核心理念就是用简单、清晰的方式,为任何可能失败的操作添加重试能力,下面我们就来看... 目录一切始于一个简单的 API 调用Tenacity 入门：一行代码实现优雅重试精细控制：让重试按我
阅读更多...
MySQL中EXISTS与IN用法使用与对比分析

MySQL中EXISTS与IN用法使用与对比分析

《MySQL中EXISTS与IN用法使用与对比分析》在MySQL中,EXISTS和IN都用于子查询中根据另一个查询的结果来过滤主查询的记录,本文将基于工作原理、效率和应用场景进行全面对比... 目录一、基本用法详解1. IN 运算符2. EXISTS 运算符二、EXISTS 与 IN 的选择策略三、性能对比
阅读更多...
MySQL常用字符串函数示例和场景介绍

MySQL常用字符串函数示例和场景介绍

《MySQL常用字符串函数示例和场景介绍》MySQL提供了丰富的字符串函数帮助我们高效地对字符串进行处理、转换和分析,本文我将全面且深入地介绍MySQL常用的字符串函数,并结合具体示例和场景,帮你熟练... 目录一、字符串函数概述1.1 字符串函数的作用1.2 字符串函数分类二、字符串长度与统计函数2.1
阅读更多...
使用Python构建智能BAT文件生成器的完美解决方案

使用Python构建智能BAT文件生成器的完美解决方案

《使用Python构建智能BAT文件生成器的完美解决方案》这篇文章主要为大家详细介绍了如何使用wxPython构建一个智能的BAT文件生成器,它不仅能够为Python脚本生成启动脚本,还提供了完整的文... 目录引言运行效果图项目背景与需求分析核心需求技术选型核心功能实现1. 数据库设计2. 界面布局设计3
阅读更多...
使用IDEA部署Docker应用指南分享

使用IDEA部署Docker应用指南分享

《使用IDEA部署Docker应用指南分享》本文介绍了使用IDEA部署Docker应用的四步流程：创建Dockerfile、配置IDEADocker连接、设置运行调试环境、构建运行镜像,并强调需准备本... 目录一、创建 dockerfile 配置文件二、配置 IDEA 的 Docker 连接三、配置 Do
阅读更多...
Android Paging 分页加载库使用实践

Android Paging 分页加载库使用实践

《AndroidPaging分页加载库使用实践》AndroidPaging库是Jetpack组件的一部分,它提供了一套完整的解决方案来处理大型数据集的分页加载,本文将深入探讨Paging库... 目录前言一、Paging 库概述二、Paging 3 核心组件1. PagingSource2. Pager3.
阅读更多...
python使用try函数详解

python使用try函数详解

《python使用try函数详解》Pythontry语句用于异常处理,支持捕获特定/多种异常、else/final子句确保资源释放,结合with语句自动清理,可自定义异常及嵌套结构,灵活应对错误场景... 目录try 函数的基本语法捕获特定异常捕获多个异常使用 else 子句使用 finally 子句捕获所
阅读更多...
C++11右值引用与Lambda表达式的使用

C++11右值引用与Lambda表达式的使用

《C++11右值引用与Lambda表达式的使用》C++11引入右值引用,实现移动语义提升性能,支持资源转移与完美转发；同时引入Lambda表达式,简化匿名函数定义,通过捕获列表和参数列表灵活处理变量... 目录C++11新特性右值引用和移动语义左值 / 右值常见的左值和右值移动语义移动构造函数移动复制运算符
阅读更多...
Python对接支付宝支付之使用AliPay实现的详细操作指南

Python对接支付宝支付之使用AliPay实现的详细操作指南

《Python对接支付宝支付之使用AliPay实现的详细操作指南》支付宝没有提供PythonSDK,但是强大的github就有提供python-alipay-sdk,封装里很多复杂操作,使用这个我们就... 目录一、引言二、准备工作2.1 支付宝开放平台入驻与应用创建2.2 密钥生成与配置2.3 安装ali
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2