阿里云函数计算使用Python开发一个基于WSGI的HTTP触发器（实战）

向右看齐 2023-07-07 05:52 5阅读 0赞

废话不多说，直接上干货

源码下载：

链接：https://pan.baidu.com/s/1OFMCtNglHHQnBlfedpdPmw   
提取码：w465

文章大纲：

1. 需求背景

2. 技术选型

3. 代码开发

4. 部署项目到阿里云

# 1. 需求背景 #

大致想实现一个 **端口-手机号** 配置的功能，并且可以接收到指定手机发来的短信，将消息过滤后转发至钉钉群。

ps：该项目仅是服务端，另一端是Android，Android配置硬件实现端口切换【一个端口对应一张手机卡，共16个端口】，使手机使用某张卡，随后将收到的短信发送至服务端，服务端过滤消息后，将消息转发到钉钉群。这样就可以方便整个公司的员工接收短信验证码，提高效率。

原型大致如下 ：

（1）端口配置：

此处包含 **增删改查** 方法，相关数据存在redis，后文会介绍为什么选用redis。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70][]

（2）号码切换

切换到指定手机号，就可以收到该手机号发来的短信。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 1][]

# 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

项目结构：

![20200221152152247.png][]

下面直接贴代码

（1）fcIndex.py

from sms_log import SMSLog
    from urllib.parse import unquote
    from redis_util import RedisUtil
    from port_phone_conf import PortPhoneConf
    from 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 json
    import time
    from redis_util import RedisUtil
    
    
    class 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 json
    import redis
    
    
    class 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 json
    import requests
    
    
    class 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）创建函数和服务

进入阿里云函数计算控制台\#函数\_服务 \#新建服务

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 2][]

新增函数

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 4][]

特别注意：函数入口一点要写对，不然会找不到入口（重要）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 5][]

（2）下载自定义模块代码包

查看函数计算的文档，我们会发现，Python只提供了标准模块和一些常用模块。如果需要使用自定义的模块，则需要将它们与代码一起打包，在这个项目中，自定义模块是requests和redis。官方推荐使用Fun，个人觉得代价有点高，放弃。文档中有这样一段话：如果没有 docker 环境，且不涉及动态链接库（.so）、编译二进制程序等，只是安装语言依赖，那么可以直接使用 `pip install -t . PyMySQL` 的方式进行安装。这种方式不管函数计算的执行环境中是否安装了这些 python 库，都会下载下来，会增加代码包的大小。

进入项目目录 ，执行以下命令，下载代码包。

pip install -t . redis;
    pip install -t . requests;

再来看看我们的项目目录：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 6][]

（3）修改 fcIndex.py 文件

查看文档发现，不支持最后一种方式【ps：最后一种方式只是为了单测方便】，所以我们的代码需要变动一下，改变后的代码如下：

from sms_log import SMSLog
    from urllib.parse import unquote
    from redis_util import RedisUtil
    from port_phone_conf import PortPhoneConf
    from wsgiref.simple_server import make_server
    
    def 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 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"

（4）上传代码

快成功了。。。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 7][]

（5）测试

保存之后，选择在线编辑，进行测试，nice

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 8][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200221150037982.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200221150131200.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[20200221152152247.png]: https://img-blog.csdnimg.cn/20200221152152247.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200221153602725.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200221153700278.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200221165631753.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200221153937671.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200221160254148.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200221160607117.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200221161148185.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTQxNjE5,size_16,color_FFFFFF,t_70