阿里云函数计算 使用Python开发一个基于WSGI的HTTP触发器 （实战）
本文共 11217 字，大约阅读时间需要 37 分钟。
废话不多说，直接上干货
源码下载：
链接：https://pan.baidu.com/s/1OFMCtNglHHQnBlfedpdPmw
提取码：w465
文章大纲：
1. 需求背景
2. 技术选型
3. 代码开发
4. 部署项目到阿里云
1. 需求背景
大致想实现一个 端口-手机号 配置的功能，并且可以接收到指定手机发来的短信，将消息过滤后转发至钉钉群。
ps：该项目仅是服务端，另一端是Android，Android配置硬件实现端口切换【一个端口对应一张手机卡，共16个端口】，使手机使用某张卡，随后将收到的短信发送至服务端，服务端过滤消息后，将消息转发到钉钉群。这样就可以方便整个公司的员工接收短信验证码，提高效率。
原型大致如下 ：
（1）端口配置：
此处包含 增删改查 方法，相关数据存在redis，后文会介绍为什么选用redis。
（2）号码切换
切换到指定手机号，就可以收到该手机号发来的短信。
2. 技术选型
（1）Python 3.7，想挑战一下自己，并且可以学习一门新的语言，所以选择了Python
（2）WSGI，没有选用基于WSGI的框架（flask或者django），因为时间有限
（3）Redis，没有选用数据库，是因为该项目与业务关系不大，只是为了方便接收验证码
（4）阿里云函数计算 HTTP触发器
3. 代码开发
将项目部署到阿里云之前，需要让项目跑起来，并单测完成。
本地选用的是wsgiref.simple_server，来模拟一个服务。
实现simple_server有三种方式（一个函数，一个类，或者一个重载了__call__的类的实例），我选用的是最后一种方式。
ps：由于没有使用web框架，整个项目实现思想有点类似于Java Web
项目结构：
下面直接贴代码
（1）fcIndex.py
from sms_log import SMSLogfrom urllib.parse import unquotefrom redis_util import RedisUtilfrom port_phone_conf import PortPhoneConffrom wsgiref.simple_server import make_server"""路由配置"""URL_PATTERNS = ( ('/', 'list'), ('/list', 'list'), ('/insert', 'insert'), ('/update', 'update'), ('/delete', 'delete'), ('/sms', 'sms'), ('/set_current_conf', 'set_current_conf'))class AppClass: @staticmethod def _match(path): """解析应该调用哪个app""" path = path.split('/')[1] # path = path[path.rfind('/'):] for url, app in URL_PATTERNS: if path in url: return app @staticmethod def parse_params(params): """解析请求参数""" if not params: return None result = {} items = params.split('&') for item in items: result[item.split("=")[0]] = item.split('=')[1] return result def __call__(self, environ, start_response): """入口""" path = environ.get('PATH_INFO', '/') app = self._match(path) if app: if app in ['list']: """get""" params = self.parse_params(environ['QUERY_STRING']) else: """post""" request_body_size = int(environ.get("CONTENT_LENGTH", 0)) request_body_str = unquote(environ["wsgi.input"].read(request_body_size).decode('utf-8')) params = self.parse_params(request_body_str) app = globals()[app] result = app(params) start_response("200 OK", [('Content-type', 'text/plain; charset=utf-8')]) return result else: start_response("404 NOT FOUND", [('Content-type', 'text/plain')]) return [b"Page dose not exists!"]def list(params): """端口配置列表展示 """ items = PortPhoneConf().list().encode() if not items: items = [] return [items, ]def insert(params): """新增配置""" try: port_no = params['port_no'] phone_no = params['phone_no'] key_words = params['key_words'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().insert(port_no, phone_no, key_words) return [b"success"]def update(params): """修改配置""" try: port_no = params['port_no'] phone_no = params['phone_no'] key_words = params['key_words'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().update(port_no, phone_no, key_words) return [b"success"]def delete(params): """删除配置""" try: port_no = params['port_no'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().delete(port_no) return [b"success"]def sms(params): """接收短信，并转发到钉钉群""" try: phone_no = params['phone_no'] msg = params['msg'] except Exception: raise Exception('参数错误，请刷新页面后重试') current_phone_no = RedisUtil.comm_get(RedisUtil.CURRENT_PORT_KEY) if phone_no == current_phone_no: raise Exception(phone_no + "对应端口未开启") SMSLog().redirect_to_ding_talk(msg) return [b"success"]def set_current_conf(params): """设置当前开启的端口""" try: phone_no = params['phone_no'] except Exception: raise Exception('参数错误，请刷新页面后重试') RedisUtil.comm_set(RedisUtil.CURRENT_PORT_KEY, phone_no) return [b"success"]if __name__ == "__main__": handler = AppClass() httpd = make_server('', 8080, handler) print("Serving on port 8080...") httpd.serve_forever() （2）port_phone_conf.py
import jsonimport timefrom redis_util import RedisUtilclass PortPhoneConf: """端口-手机号操作类""" def insert(self, port_no, phone_no, key_words): """新增端口配置""" conf = {'port_no': port_no, 'phone_no': phone_no, 'key_words': key_words, 'gmt_create': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), 'gmt_modified': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} items = self.list() if not items: items = [] else: items = json.loads(items) items.append(conf) RedisUtil.comm_set(RedisUtil.CONF_KEY, items) def update(self, port_no, phone_no, key_words): """修改端口配置""" items = json.loads(self.list()) for item in items: if item["port_no"] == port_no: item["phone_no"] = phone_no item["key_words"] = key_words item["gmt_modified"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) RedisUtil.comm_set(RedisUtil.CONF_KEY, items) def delete(self, port_no): """删除端口配置""" global items if port_no: items = json.loads(self.list()) for item in items: if item["port_no"] == port_no: items.remove(item) break RedisUtil.comm_set(RedisUtil.CONF_KEY, items) def list(self): """获取端口配置列表""" return RedisUtil.comm_get(RedisUtil.CONF_KEY) （3）redis_util.py
import jsonimport redisclass RedisUtil: """获取redis连接，并且操作redis""" CONF_KEY = "select::port_phone_conf" CURRENT_PORT_KEY = "select::port_phone_conf_current" @staticmethod def get_conn(): """返回连接""" conn = redis.Redis(host="106.15.176.xxx", port=6379) return conn @staticmethod def comm_set(key, values): """通用set方法""" global conn try: conn = RedisUtil.get_conn() conn.set(key, json.dumps(values, ensure_ascii=False).encode("utf-8")) finally: conn.close() @staticmethod def comm_get(key): """通用get方法""" global conn try: conn = RedisUtil.get_conn() items = conn.get(key) finally: conn.close() if items: items = items.decode("utf-8") return items （4）sms_log.py
import jsonimport requestsclass SMSLog: @staticmethod def redirect_to_ding_talk(msg): """过滤消息并转发到钉钉群""" if ("阿里云" in msg) or ("钉钉" in msg): url = 'https://oapi.dingtalk.com/robot/send?access_token=xxx' data = {"msgtype": "text", "text": {"content": msg}} headers = {'Content-Type': 'application/json'} requests.request("post", url, json=data, headers=headers) （5）调试
接下来，就可以运行 fcIndex.py#__main__ 来测试功能了！
4. 部署项目到阿里云
本地单测完成，就可以部署到阿里云了，不要沾沾自喜，后面还有好多坑等着呢。
（1）创建函数和服务
进入阿里云函数计算控制台#函数_服务 #新建服务
新增函数
特别注意：函数入口一点要写对，不然会找不到入口（重要）
（2）下载自定义模块代码包
查看函数计算的文档，我们会发现，Python只提供了标准模块和一些常用模块。如果需要使用自定义的模块，则需要将它们与代码一起打包，在这个项目中，自定义模块是requests和redis。官方推荐使用Fun，个人觉得代价有点高，放弃。文档中有这样一段话：如果没有 docker 环境，且不涉及动态链接库（.so）、编译二进制程序等，只是安装语言依赖，那么可以直接使用 pip install -t . PyMySQL 的方式进行安装。这种方式不管函数计算的执行环境中是否安装了这些 python 库，都会下载下来，会增加代码包的大小。
进入项目目录 ，执行以下命令，下载代码包。
pip install -t . redis;pip install -t . requests;
再来看看我们的项目目录：
（3）修改 fcIndex.py 文件
查看文档发现，不支持最后一种方式【ps：最后一种方式只是为了单测方便】，所以我们的代码需要变动一下，改变后的代码如下：
from sms_log import SMSLogfrom urllib.parse import unquotefrom redis_util import RedisUtilfrom port_phone_conf import PortPhoneConffrom wsgiref.simple_server import make_serverdef handler(environ, start_response): """函数计算调用入口""" return AppClass(environ, start_response)"""路由配置"""URL_PATTERNS = ( ('/', 'list'), ('/list', 'list'), ('/insert', 'insert'), ('/update', 'update'), ('/delete', 'delete'), ('/sms', 'sms'), ('/set_current_conf', 'set_current_conf'))class AppClass: @staticmethod def _match(path): """解析应该调用哪个app""" path = path.split('/')[1] # path = path[path.rfind('/'):] for url, app in URL_PATTERNS: if path in url: return app @staticmethod def parse_params(params): """解析请求参数""" if not params: return None result = {} items = params.split('&') for item in items: result[item.split("=")[0]] = item.split('=')[1] return result def __init__(self, environ, start_response): self.environ = environ self.start_response = start_response def __iter__(self): path = self.environ.get('PATH_INFO', '/') app = self._match(path) if app: if app in ['list']: """get""" params = {} # params = self.parse_params(self.environ['QUERY_STRING']) else: """post""" request_body_size = int(self.environ.get("CONTENT_LENGTH", 0)) request_body_str = unquote(self.environ["wsgi.input"].read(request_body_size).decode('utf-8')) params = self.parse_params(request_body_str) app = globals()[app] result = app(params) self.start_response("200 OK", [('Content-type', 'text/plain; charset=utf-8')]) yield result else: self.start_response("404 NOT FOUND", [('Content-type', 'text/plain')]) yield b"Page dose not exists!"def list(params): """端口配置列表展示 """ items = PortPhoneConf().list().encode() if not items: items = [] return itemsdef insert(params): """新增配置""" try: port_no = params['port_no'] phone_no = params['phone_no'] key_words = params['key_words'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().insert(port_no, phone_no, key_words) return b"success"def update(params): """修改配置""" try: port_no = params['port_no'] phone_no = params['phone_no'] key_words = params['key_words'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().update(port_no, phone_no, key_words) return b"success"def delete(params): """删除配置""" try: port_no = params['port_no'] except Exception: raise Exception('参数错误，请刷新页面后重试') PortPhoneConf().delete(port_no) return b"success"def sms(params): """接收短信，并转发到钉钉群""" try: phone_no = params['phone_no'] msg = params['msg'] except Exception: raise Exception('参数错误，请刷新页面后重试') current_phone_no = RedisUtil.comm_get(RedisUtil.CURRENT_PORT_KEY) if phone_no == current_phone_no: raise Exception(phone_no + "对应端口未开启") SMSLog().redirect_to_ding_talk(msg) return b"success"def set_current_conf(params): """设置当前开启的端口""" try: phone_no = params['phone_no'] except Exception: raise Exception('参数错误，请刷新页面后重试') RedisUtil.comm_set(RedisUtil.CURRENT_PORT_KEY, phone_no) return b"success" （4）上传代码
快成功了。。。
（5）测试
保存之后，选择在线编辑，进行测试，nice
转载地址：http://riqen.baihongyu.com/
你可能感兴趣的文章
java 线程：sleep join yield | wait notify notifyAll
查看>>
Python 包、模块 概念 from 、import 关键字
查看>>
世界各国的手机号码
查看>>
通配符与正则表达式
查看>>
c++ 与 Java 之 红黑树 哈希表 辨析
查看>>
open GL 、DirectX、open CV、 open Inventor 、cocos2dx、unity3d、3dmax辨析
查看>>
理解矩阵
查看>>
彩虹七色的RGB值
查看>>
常用正则表达式实例
查看>>
web网站架构演变过程
查看>>
c、 c++、 java 基本数据类型 对比辨析
查看>>
自然语言处理(NLP)四步流程：Embed->Encode->Attend->Predict
查看>>
python机器学习包 Windows下 pip安装 scikit-learn numpy scipy
查看>>
[转发]机器学习资源大全
查看>>
《数学之美》知识点详细总结
查看>>
机器学习 数据挖掘 数据集划分 训练集 验证集 测试集
查看>>
从不同角度看机器学习的几种学习方式
查看>>
数据挖掘 NLP 之 文本挖掘 文本处理 通用流程
查看>>
NLP 主题抽取 Topic LDA代码实践 gensim包 代码
查看>>
NLP 工具包 大调查 自然语言处理工具包合集
查看>>