知乎一条龙第二弹，API 部署开放、H5线上展示与源码共享

周萝卜

发布于 2020-05-22 10:19:31

4510

发布于 2020-05-22 10:19:31

文章被收录于专栏：萝卜大杂烩萝卜大杂烩

面写了一个知乎爬虫、API 和小程序一条龙第一弹，反响还不错，于是在这些天的空闲时间里，我又优化下代码，并且把服务部署到了云服务器上，开放了 API 供需要的小伙伴使用。

也有很多人要源代码看看，想自己动手实践下，今天就把代码放出来，写的不好，仅供参考，也欢迎一起讨论维护！

功能增强之token

因为准备开放 API 接口出来，所以考虑了下，还是做一些简单的验证，毕竟安全措施做的好，你好我也好！

首先我们先来看下整体的请求流程

客户端先通过 getToken 接口来获取一个具有时间期限的 token 信息，然后再携带该 token 信息访问对应的数据接口

token 实现

我这里使用第三方库 itsdangerous 来做 token 签名

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

itsdangerous 提供了多种生成签名令牌的方式，我这里选择的 TimedJSONWebSignatureSerializer 可以生成一种具有过期时间的 JSON Web 签名，这样我们也就可以控制我们所签发的 token 是具有时效性的。

生成签名并加密成 token

access_token_gen = Serializer(secret_key=secret_key, salt=salt, expires_in=access_token_expires_in)
timtstamp = time.time()
access_token = access_token_gen.dumps({
        "userid": userid,
        "iat": timtstamp
    })

然后在需要解析 token 时，只要调用 loads 即可

s = Serializer(secret_key=secret_key, salt=salt)
data = s.loads(token)

访问限制装饰器

装饰器是 Python 语言的一大利器，我们当然要好好利用起来了。

在最开始的设计中，我们的路由都是可以直接访问的，没有任何限制

@api.route('/api/zhihu/hot/', methods=['GET', 'POST'])
def zhihu_api_data():
    pass

现在我们想达到一种效果，就是不改变当前视图函数的写法，还要增加访问限制，只有携带了正确 token 的请求才能够正确访问对应的路由

@api.route('/api/zhihu/hot/', methods=['GET', 'POST'])
@token.tokenRequired
def zhihu_api_data():
    pass

毫无疑问，这个功能交给装饰器真是再好不过了

def tokenRequired(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        pass
    return decorated_function

下面的工作就是编写 decorated_function 函数的内容了，只需要加上我们需要的判断即可

if request.method == 'POST':
    post_data = json.loads(request.data)
    if 'token' in post_data and 'secret' in post_data and post_data['secret'] == '周萝卜真帅':
        token = post_data['token']
        check_result = check_token(token)
        if check_result is True:
            return f(*args, **kwargs)
        else:
            return jsonify(check_result), 401
    return jsonify({'code': 422, 'message': '按套路出牌啊'}), 422

当请求方法是 POST 时，如果 token 字段不在请求体内或者请求体的 secret 字段没有按照套路出牌的话，都会返回错误响应的（这里请牢记暗号啊，夸我就对了！）

接下来我们再看看 check_token 函数，这就是具体的校验 token 的方法了

def check_token(token):
    token_list = []
    if rd.keys("token*"):
        for t in rd.keys("token*"):
            token_list.append(rd.get(t))
    if token in token_list:
        return {'code': 401, 'message': 'token is blocked'}, 401
    validator = validateToken(token)
    if validator['code'] != 200:
        if validator['message'] == 'toekn expired':
            return validator
        else:
            return validator
    elif validator['code'] == 200:
        return True

留用了 block token 的功能，以便后面使用。而 validateToken 函数就是调用 loads 方法解析加密后的 token。

功能增强之频率限制

所谓的频率限制，就是在指定的时间之内，访问请求的次数不能超过多少次。我这里设置的是一分钟之内，访问次数不能超过20次

REQUEST_NUM = 20

为了实现这个功能，我们需要用到 Flask 程序的全局请求钩子 before_app_request。该钩子的作用就是在任何请求发生之前，都会先调用该函数。这样我们就可以添加自己的判断逻辑，增加访问频率限制

@main.before_app_request
def before_request():
    remote_add = request.remote_addr
    rd_add = rd.get('access_num_%s' % remote_add)
    if rd_add:
        if int(rd_add) <= Config.REQUEST_NUM:
            rd.incr('access_num_%s' % remote_add)
        else:
            return jsonify({'code': 422, 'message': '访问太频繁啦！'}), 422
    else:
        rd.set('access_num_%s' % remote_add, 1, ex=60)

每个 IP 的访问频率都存储在 redis 中，且该 redis key 的过期时间为60秒。当然这种限制属于防君子不防小人的做法，为什么这么说呢，因为如果你想突破这种入门级的限制，实在是太 easy 啦，而且使用手机4G网络的请求，IP 地址还会不停变化，太楠啦！

功能增强之高频词汇

在上一次的文章中，我们在前端（小程序端）只展示了知乎热点随着时间的走势情况，今天再加上每个热点的回答中的高频词汇，通过 jieba 来分词，还是很容易实现的。

将获取到的回答内容分词并统计词频

def cut_word(word):
    word = re.sub('[a-zA-Z0-9]', '', word)
    empty_str = ' '
    with open(stopwords_path, encoding='utf-8') as f:
        stop_words = f.read()
    stop_words = stop_words + empty_str
    counts = {}
    txt = jieba.lcut(word)
    for w in txt:
        if w not in stop_words:
            counts[w] = counts.get(w, 0) + 1
    sort_counts = sorted(counts.items(), key=lambda item: item[1], reverse=True)

    return sort_counts[:20]

在这里我们去掉了英文和数字，并且返回了词频前20的数据

然后我们修改视图函数 zhihu_api_detail

@api.route('/api/zhihu/detail/<id>/', methods=['GET', 'POST'])
@token.tokenRequired
def zhihu_api_detail(id):
    zhihu_detail = zhihudetail(id)
    redis_word = rd.get('wordcloud_%s' %id)
    redis_content = rd.get('content_%s' % id)
    if redis_word:
        count_list = json.loads(redis_word)
        content_list = json.loads(redis_content)
    else:
        count_list = []
        count_word, content_list = zhihucontent(id)  # 获取回答的词频数据和回答内容
        for count in count_word:
            count_list.append({'name': count[0], 'textSize': count[1]})
        rd.set('wordcloud_%s' %id, json.dumps(count_list), ex=604800)
        rd.set('content_%s' %id, json.dumps(content_list), ex=604800)

    if count_list[0]['textSize'] < 10:
        for i in count_list:
            i['textSize'] = i['textSize']*10
    elif count_list[0]['textSize'] > 200:
        for i in count_list:
            i['textSize'] = i['textSize']/10

    return jsonify({'code': 0, 'data': zhihu_detail, 'count_word': count_list, 'content': content_list}), 200

因为每次使用 jieba 分词时还是比较耗费时间的，所以这里把处理好的数据保存到 redis 中，下次再请求时直接拿数据即可。

现在我们的详情页面展示如下

部署 API

最后我们把已经完成的代码部署到云服务器上，使用的还是那套 Nginx + Gunicorn + Flask + MySQL

配置详情

Nginx 配置

server {
    gzip on;
    listen       443;
    server_name  www.luobodazahui.top;
    ssl on;
    root        /home/mini/mini/      ;
    ssl_certificate  cert/luobodazahui.top.crt;
    ssl_certificate_key cert/luobodazahui.top.key;
    ssl_session_timeout 5m;
    ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:ECDHE:ECDH:AES:HIGH:!NULL:!aNULL:!MD5:!ADH:!RC4;
    ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
    ssl_prefer_server_ciphers on;
    location / {
        proxy_pass       http://127.0.0.1:5002;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        index  index.html index.htm;
    }

    proxy_set_header X-Real-IP $remote_addr;

}
server {
    listen 80;
    server_name luobodazahui.top;
    rewrite ^(.*)$ https://$host$1 permanent;
    }

因为 API 后面想给小程序使用，所以应用了域名 + HTTPS

Gunicorn 配置

#from gevent import monkey
#monkey.patch_all()

import multiprocessing

#debug = True
loglevel = 'debug'
bind = '127.0.0.1:5002'
#bind = '0.0.0.0:5000'
#pidfile = 'pid/gunicorn.pid' 
accesslog = '/home/mini/mini/log/ser_access.log'
errorlog = '/home/mini/mini/log/ser_error.log'

workers = 1
#workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'sync' 
#reload = True

同样是比较简单的配置，打印了访问和错误日志，还启用了适量的 workers。

启动脚本 run.sh

/root/miniconda3/bin/gunicorn -D -c /home/mini/mini/gunicorn manage:app

停止脚本 stop.sh

kill -9 $(ps -ef | grep '/home/mini/mini/gunicorn' | grep -v grep | awk '{print $2}') 2>&1 >/dev/null;echo 0

API 信息

我们来看下当前提供的 API 信息

API地址	请求参数	支持方法‍‍‍
https://www.luobodazahui.top/api/auth/token/	table1	POST/GET
https://www.luobodazahui.top/api/zhihu/hot/	table2	POST/GET
https://www.luobodazahui.top/api/zhihu/detail//	table3	POST/GET

table1

{
    "username": "admin",
    "pwd": "admin"
}

请求示例

table2

{
    "token":"eyJhbGciOiJIUzUxMiIsImlhdCI6MTU3NzI0NDE4MywiZXhwIjoxNTc3MjQ1OTgzfQ.eyJ1c2VyaWQiOjEsImlhdCI6MTU3NzI0NDE4My4zMjcwNjY0fQ.FptYNm0KnA8b4G_zcRJn9POrOgkiZxpvfBbzQqxoTTt7q96WeMo7Y6xCLL_oS4ksBP8jMztqopDRRqScXPKowg",
    "secret":"周萝卜真帅"}

请求示例

table3

{
    "token":"eyJhbGciOiJIUzUxMiIsImlhdCI6MTU3NzI0NDE4MywiZXhwIjoxNTc3MjQ1OTgzfQ.eyJ1c2VyaWQiOjEsImlhdCI6MTU3NzI0NDE4My4zMjcwNjY0fQ.FptYNm0KnA8b4G_zcRJn9POrOgkiZxpvfBbzQqxoTTt7q96WeMo7Y6xCLL_oS4ksBP8jMztqopDRRqScXPKowg",
    "secret":"周萝卜真帅"}

请求示例