基于Serverless的验证码识别API

anycodes刚刚日常分享2706132

前言

之前和大家分享了很多的CV相关的例子,被很多小伙伴吐槽说我是调包侠,还连累了Serverless被很多人误以为也仅仅能"调包玩一玩",其实在Serverless中,开发者的自由度还是非常大的,除了调包快速实现一些东西,我们也可以通过一些代码训练一些模型,然后实现一些功能,本文将会通过简单的实验,在Serverless架构上实现一个基于卷积神经网络(CNN)算法的在线验证码识别的小工具。

验证码与识别

验证码(CAPTCHA)是“Completely Automated Public Turing test to tell Computers and Humans Apart”(全自动区分计算机和人类的图灵测试)的缩写,是一种区分用户是计算机还是人的公共全自动程序。可以防止:恶意破解密码、刷票、论坛灌水,有效防止某个黑客对某一个特定注册用户用特定程序暴力破解方式进行不断的登陆尝试,实际上用验证码是现在很多网站通行的方式,我们利用比较简易的方式实现了这个功能。这个问题可以由计算机生成并评判,但是必须只有人类才能解答。由于计算机无法解答CAPTCHA的问题,所以回答出问题的用户就可以被认为是人类。

说白了,验证码就是用来验证的码,验证是人访问的还是机器访问的码。

验证码的发展,可以说是非常迅速的,从开始的单纯数字验证码,到后来的数字+字母验证码,再到后来的数字+字母+中文的验证码以及图形图像验证码,可以说就单纯的验证码素材已经越来越多了,从验证码的形态来看,也是各不相同,输入、点击、拖拽以及短信验证码、语音验证码……

例如腾讯云后台登陆的验证码与Bilibili的登录验证码就是滑动登录:

而百度贴吧、知乎、以及Google等相关网站的验证码又各不相同,例如选择正着写的文字,选择包括指定物体的图片以及按顺序点击图片中的字符等。

验证码的识别可能会根据验证码的类型而不太一致,当然最简单的验证码可能就是最原始的文字验证码了:

即便是文字验证码,也是存在很多差异的,例如简单的数字验证码,简单的数字+字母验证码,文字验证码,验证码中包括计算,简单验证码中增加一些干扰成为复杂验证码…….

就这种比较简单的验证码的识别方法也有很多,除了目前比流行的端到端识别之外,之前比较常见的识别就是通过图像的切割,对验证码每一部分裁剪,然后再对每个裁剪单元进行相似度对比,获得最可能的结果,最后进行拼接,例如将验证码:

进行二值化等操作:

完成之后再进行切割:

切割完成在进行识别,再进行拼接,这样的做法是,针对每个字符进行识别,相对来说是比较容易容易的。但是对于某些情况,是没办法切割的,例如图片中有很多干扰线等。这个时候就可能需要深度学习,来进行端对端的识别了。

代码实现

本代码很多内容来源于Github,更多是通过搜集一些资料,发挥自己的想象,将该项目部署到Serverless架构上。

验证码生成部分

# coding:utf-8
# name:captcha_gen.py

import random
import numpy as np
from PIL import Image
from captcha.image 简单的性能测试

接下来对性能进行一波简单的测试,首先购买一个云服务器,将这个部分代码部署到云服务器上。
在云上购买服务器,保守一点买了1核2G


然后配置环境,到服务可以跑起来:


通过Post设置一下简单的Tests:

然后对接口进行测试:

非常顺利完成了接口测试:

可以通过接口测试结果进行部分可视化:

同时对数据进行统计:

可以看到,通过上图和上表,服务器测的整体响应时间都快于云函数的响应时间。而且可以看到函数存在冷启动,一按出现冷启动,其响应时间会增长20余倍。在由于上述测试,仅仅是非常简单的接口,接下来我们来测试一下稍微复杂的接口,使用了jieba分词的接口,因为jieba分词接口存在:

测试结果:

可视化结果:


通过对Jieba接口的测试,可以看到虽然服务器也会有因分词组件进行初始化而产生比较慢的响应时间,但是整体而言,速度依旧是远远低于云函数。

那么问题来了,是函数本身的性能有问题,还是增加了Flask框架+APIGW响应集成之后才有问题?

接下来,做一组新的接口测试,在函数中,直接返回内容,而不进行额外处理,看看函数+API网关性能和正常情况下的服务器性能对比


可以看出虽然最小和平均耗时的区别不是很大,但是最大耗时基本上是持平。可以看出来,框架的加载会导致函数冷启动时间长度变得异常可怕。
接下来通过Python代码,对Flask框架进行并发测试:
对函数进行3次压测,每次并发301:

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 1.2727971077
response mintime 0.573610067368

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 1.1745698452
response mintime 0.172255039215

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 1.2857568264
response mintime 0.157210826874

对服务器进行3次压测,同样是每次并发301:

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 3.41151213646
response mintime 0.255661010742

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 3.37784004211
response mintime 0.212490081787

===========task end===========
total:301,succ:301,fail:0,except:0
response maxtime: 3.39548277855
response mintime 0.439364910126

通过这一波压测,我们可以看到这样一个奇怪现象,那就是在函数和服务器预热完成之后,连续三次并发301个请求。函数的整体表现,反而比服务器的要好。这也说明了在Serverless架构下,弹性伸缩的一个非常重要的表现。传统服务器,我们如果出现了高并发现象,很容易会导致整体服务受到严重影响,例如响应时间变长,无响应,甚至是服务器直接挂掉,但是在Serverless架构下,这个弹性伸缩的能力是云厂商帮助我们做的,所以在并发量达到一定的时候,其实Serverless架构的优势变得会更加明显。

传统Web框架上云方法(以Python Web框架为例)

分析已有Component(Flask为例)

首先第一步,我们要知道其他的框架是怎么运行的,例如Flask等,我们先通过腾讯云的Flask-Component,按照他的说明部署一下:

非常简单轻松愉快的部署上线,然后在函数的控制台,我们把部署好的下载下来,研究一下:

下载解压之后,我们可以看这样一个目录结构:

蓝色框起来的,是依赖包,黄色的app.py是我们的自己写的代码,那么红色圈起来的是什么?这两个文件从哪里出来的?
api_server.py文件内容:

import app  # Replace with your actual application
import severless_wsgi

# If you need to send additional content types as text, add then directly
# to the whitelist:
#
# serverless_wsgi.TEXT_MIME_TYPES.append("application/custom+json")

def handler(event, context):
    return severless_wsgi.handle_request(app.app, event, context)

可以看到,这里面是将我们创建的app.py文件引入,并且拿到了app这个对象,并且将event和context同时传递给severless_wsgi.py中的handle_reques方法中,那么问题来了,这个方法是什么?

这个方法内容好多……看着有点眼晕,但是,我们可以直接发现这一段代码:

这一段是什么呢?这一段实际上就是将我们拿到的参数(event和context)进行转换,转换之后统一environ中,然后接下来通过werkzeug这个依赖,将这个内容变成request对象,并且与我们刚才说的app对象一起调用from_app方法。获得到反馈:

并且按照API网关的响应集成的格式,将结果返回。
此时此刻,各位看官可能有点想法了,貌似有一丢丢灵感出现了,那么我们不妨看一下Flask/Django这些框架的实现原理:

通过这个简版的原理图,和我刚才说的内容,我们可以想到,实际上正常用的时候要通过web_server,进入到下一个环节,而我们云函数更多是一个函数,本不需要启动web server,所以我们就可以直接调用wsgi_app这个方法,其中这里的environ就是我们刚才的通过对event/context等进行处理后的对象,start_response可以认为是我们的一种特殊的数据结构,例如我们的response结构形态等。所以,如果我们自己想要实现这个过程,不使用腾讯云flask-component,可以这样做:

import sys

try:
    from urllib import urlencode
except ImportError:
    from urllib.parse import urlencode

from flask import Flask

try:
    from cStringIO import StringIO
except ImportError:
    try:
        from StringIO import StringIO
    except ImportError:
        from io import StringIO

from werkzeug.wrappers import BaseRequest

__version__ = '0.0.4'


def make_environ(event):
    environ = {}
    for hdr_name, hdr_value in event['headers'].items():
        hdr_name = hdr_name.replace('-''_').upper()
        if hdr_name in ['CONTENT_TYPE''CONTENT_LENGTH']:
            environ[hdr_name] = hdr_value
            continue

        http_hdr_name = 'HTTP_%s' % hdr_name
        environ[http_hdr_name] = hdr_value

    apigateway_qs = event['queryStringParameters']
    request_qs = event['queryString']
    qs = apigateway_qs.copy()
    qs.update(request_qs)

    body = ''
    if 'body' in event:
        body = event['body']

    environ['REQUEST_METHOD'] = event['httpMethod']
    environ['PATH_INFO'] = event['path']
    environ['QUERY_STRING'] = urlencode(qs) if qs else ''
    environ['REMOTE_ADDR'] = 80
    environ['HOST'] = event['headers']['host']
    environ['SCRIPT_NAME'] = ''
    environ['SERVER_PORT'] = 80
    environ['SERVER_PROTOCOL'] = 'HTTP/1.1'
    environ['CONTENT_LENGTH'] = str(len(body))
    environ['wsgi.url_scheme'] = ''
    environ['wsgi.input'] = StringIO(body)
    environ['wsgi.version'] = (10)
    environ['wsgi.errors'] = sys.stderr
    environ['wsgi.multithread'] = False
    environ['wsgi.run_once'] = True
    environ['wsgi.multiprocess'] = False

    BaseRequest(environ)

    return environ


class LambdaResponse(object):
    def __init__(self):
        self.status = None
        self.response_headers = None

    def start_response(self, status, response_headers, exc_info=None):
        self.status = int(status[:3])
        self.response_headers = dict(response_headers)


class FlaskLambda(Flask):
    def __call__(self, event, context):
        if 'httpMethod' not in event:
            print('httpMethod not in event')
            return super(FlaskLambda, self).__call__(event, context)

        response = LambdaResponse()

        body = next(self.wsgi_app(
            make_environ(event),
            response.start_response
        ))

        return {
            'statusCode': response.status,
            'headers': response.response_headers,
            'body': body
        }

这样一个流程,就会变得更加简单,清楚。整个实现过程,可以认为是对web server部分进行了一种“截断”或者是“替换”:

这就是对Flask-Component的基本分析思路,那么按照这个思路,我们是否可以将Django框架部署上Serverless架构呢?那么Flask和Django有什么区别呢?我这里的区别特指的是在运行启动过程中。

拓展思路:实现Django-component

仔细想一下,貌似并没有区别,那么我们是不是可以直接用Flask这个转换逻辑,将flask的app替换成django的app呢?
把:

from flask import Flask
app = Flask(__name__)

替换成:

import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE''mydjango.settings')
application = get_wsgi_application()

是否就能解决问题呢?
我们不妨试一下:

建立好Django项目,直接增加index.py:

# -*- coding: utf-8 -*-

import os
import sys
import base64
from werkzeug.datastructures import Headers, MultiDict
from werkzeug.wrappers import Response
from werkzeug.urls import url_encode, url_unquote
from werkzeug.http import HTTP_STATUS_CODES
from werkzeug._compat import BytesIO, string_types, to_bytes, wsgi_encoding_dance
import mydjango.wsgi

TEXT_MIME_TYPES = [
    "application/json",
    "application/javascript",
    "application/xml",
    "application/vnd.api+json",
    "image/svg+xml",
]


def all_casings(input_string):
    if not input_string:
        yield ""
    else:
        first = input_string[:1]
        if first.lower() == first.upper():
            for sub_casing in all_casings(input_string[1:]):
                yield first + sub_casing
        else:
            for sub_casing in all_casings(input_string[1:]):
                yield first.lower() + sub_casing
                yield first.upper() + sub_casing


def split_headers(headers):
    """
    If there are multiple occurrences of headers, create case-mutated variations
    in order to pass them through APIGW. This is a hack that's currently
    needed. See: https://github.com/logandk/serverless-wsgi/issues/11
    Source: https://github.com/Miserlou/Zappa/blob/master/zappa/middleware.py
    """

    new_headers = {}

    for key in headers.keys():
        values = headers.get_all(key)
        if len(values) > 1:
            for value, casing in zip(values, all_casings(key)):
                new_headers[casing] = value
        elif len(values) == 1:
            new_headers[key] = values[0]

    return new_headers


def group_headers(headers):
    new_headers = {}

    for key in headers.keys():
        new_headers[key] = headers.get_all(key)

    return new_headers


def encode_query_string(event):
    multi = event.get(u"multiValueQueryStringParameters")
    if multi:
        return url_encode(MultiDict((i, j) for i in multi for j in multi[i]))
    else:
        return url_encode(event.get(u"queryString"or {})


def handle_request(application, event, context):

    if u"multiValueHeaders" in event:
        headers = Headers(event["multiValueHeaders"])
    else:
        headers = Headers(event["headers"])

    strip_stage_path = os.environ.get("STRIP_STAGE_PATH""").lower().strip() in [
        "yes",
        "y",
        "true",
        "t",
        "1",
    ]
    if u"apigw.tencentcs.com" in headers.get(u"Host"u""and not strip_stage_path:
        script_name = "/{}".format(event["requestContext"].get(u"stage"""))
    else:
        script_name = ""

    path_info = event["path"]
    base_path = os.environ.get("API_GATEWAY_BASE_PATH")
    if base_path:
        script_name = "/" + base_path

        if path_info.startswith(script_name):
            path_info = path_info[len(script_name) :] or "/"

    if u"body" in event:
        body = event[u"body"or ""
    else:
        body = ""

    if event.get("isBase64Encoded"False):
        body = base64.b64decode(body)
    if isinstance(body, string_types):
        body = to_bytes(body, charset="utf-8")

    environ = {
        "CONTENT_LENGTH": str(len(body)),
        "CONTENT_TYPE": headers.get(u"Content-Type"""),
        "PATH_INFO": url_unquote(path_info),
        "QUERY_STRING": encode_query_string(event),
        "REMOTE_ADDR": event["requestContext"]
        .get(u"identity", {})
        .get(u"sourceIp"""),
        "REMOTE_USER": event["requestContext"]
        .get(u"authorizer", {})
        .get(u"principalId"""),
        "REQUEST_METHOD": event["httpMethod"],
        "SCRIPT_NAME": script_name,
        "SERVER_NAME": headers.get(u"Host""lambda"),
        "SERVER_PORT": headers.get(u"X-Forwarded-Port""80"),
        "SERVER_PROTOCOL""HTTP/1.1",
        "wsgi.errors": sys.stderr,
        "wsgi.input": BytesIO(body),
        "wsgi.multiprocess"False,
        "wsgi.multithread"False,
        "wsgi.run_once"False,
        "wsgi.url_scheme": headers.get(u"X-Forwarded-Proto""http"),
        "wsgi.version": (10),
        "serverless.authorizer": event["requestContext"].get(u"authorizer"),
        "serverless.event": event,
        "serverless.context": context,
        TODO: Deprecate the following entries, as they do not comply with the WSGI
        # spec. For custom variables, the spec says:
        #
        #   Finally, the environ dictionary may also contain server-defined variables.
        #   These variables should be named using only lower-case letters, numbers, dots,
        #   and underscores, and should be prefixed with a name that is unique to the
        #   defining server or gateway.
        "API_GATEWAY_AUTHORIZER": event["requestContext"].get(u"authorizer"),
        "event": event,
        "context": context,
    }

    for key, value in environ.items():
        if isinstance(value, string_types):
            environ[key] = wsgi_encoding_dance(value)

    for key, value in headers.items():
        key = "HTTP_" + key.upper().replace("-""_")
        if key not in ("HTTP_CONTENT_TYPE""HTTP_CONTENT_LENGTH"):
            environ[key] = value

    response = Response.from_app(application, environ)

    returndict = {u"statusCode": response.status_code}

    if u"multiValueHeaders" in event:
        returndict["multiValueHeaders"] = group_headers(response.headers)
    else:
        returndict["headers"] = split_headers(response.headers)

    if event.get("requestContext").get("elb"):
        # If the request comes from ALB we need to add a status description
        returndict["statusDescription"] = u"%d %s" % (
            response.status_code,
            HTTP_STATUS_CODES[response.status_code],
        )

    if response.data:
        mimetype = response.mimetype or "text/plain"
        if (
            mimetype.startswith("text/"or mimetype in TEXT_MIME_TYPES
        ) and not response.headers.get("Content-Encoding"""):
            returndict["body"] = response.get_data(as_text=True)
            returndict["isBase64Encoded"] = False
        else:
            returndict["body"] = base64.b64encode(response.data).decode("utf-8")
            returndict["isBase64Encoded"] = True

    return returndict



def main_handler(event, context):
    return handle_request(mydjango.wsgi.application, event, context)

然后我们部署到函数上,看一下效果:
函数信息:

from django.shortcuts import render
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

# Create your views here.
@csrf_exempt
def hello(request):
     标签: serverless framework计算机视觉PythonOCR验证码识别深度学习神经网络

作者简介:刘宇,毕业于浙江大学,硕士学历,目前在腾讯工作,著有《Serverless 架构》一书,是Serverless架构的热衷者,曾做一款叫Anycodes的软件,目前下载超过100万次。

返回列表

上一篇:未命名

下一篇:未命名

相关文章

云函数中使用Python-ORM: Peewee

云函数中使用Python-ORM: Peewee

前言 ORM(Object Ralational Mapping,对象关系映射)用来把对象模型表示的对象映射到基于SQL的关系模型数据库结构中去。这样,我们在具体的操作实体对象的时候,就不需要再去和复...

serverless-git和serverless-cicd

serverless-git和serverless-cicd

前言 传统情况下,我们写完代码,可能面对两个事情:发布到代码仓库以及部署到线上,传统的我们会手动实现这些操作,出现误操作的概率也是蛮高的,相对来说也是比较机械化的工作。CICD的引入,大大改善了持续继...

基于Serverless架构的Git代码统计工具

基于Serverless架构的Git代码统计工具

前言 自己毕业也有一年多了,很想统计一下过去一年自己贡献了多少的代码。想了一下,可能要用git log,简单的做了一下,感觉不是很爽,正直自己想通过Serverless做一个工具合集,就想能不能...

2020年函数计算的冷启动怎么样了

2020年函数计算的冷启动怎么样了

前言 自从Serverless架构被提出,函数计算这个名词变得越发的火热,甚至在很多时候有人会认为Serverless就是函数计算。 作为Serverless架构中的一个重要组成部分,云函数确实值得...

用Serverlss部署一个基于深度学习的古诗词生成API

用Serverlss部署一个基于深度学习的古诗词生成API

第六篇:用Serverlss部署一个基于深度学习的古诗词生成API 前言 古诗词是中国文化殿堂的瑰宝,记得曾经在韩国做Exchange Student的时候,看到他们学习我们的古诗词,有中文的还有翻译...

利与弊-传统框架要不要部署在Serverless架构上

利与弊-传统框架要不要部署在Serverless架构上

Serverless架构发展的速度可以说是非常的迅速,而且Serverless的发展也是有一套自己的独特的打法,这种打法在一定程度上,让很多开发者不适应,尤其是传统的Web框架无法在Serverles...

利与弊-多个接口要分成多个函数还是写到一个函数中

利与弊-多个接口要分成多个函数还是写到一个函数中

我们在做一个项目的时候,会有多个函数/方法,这个是一个很常见的事情,就算不按照函数/方法来划分,也通常会有多个功能,以一个简单的博客为例,可能拥有最基础的: 获取分类列表 获取文章列表(默认/...

评论列表

frank
2020-08-09 20:58:50

在serverLess 上 部署了吗

kris
2020-07-10 23:38:07

点赞

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。