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

时间 2019-12-28

原文原文链接

前面写了一个知乎爬虫、API和小程序一条龙第一弹，反响还不错，因而在这些天的空闲时间里，我又优化下代码，而且把服务部署到了云服务器上，开放了 API 供须要的小伙伴使用。
也有不少人要源代码看看，想本身动手实践下，今天就把代码放出来，写的很差，仅供参考，也欢迎一块儿讨论维护！css

功能加强之token

由于准备开放 API 接口出来，因此考虑了下，仍是作一些简单的验证，毕竟安全措施作的好，你好我也好！html

首先咱们先来看下总体的请求流程前端

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

token 实现

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

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
复制代码

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

生成签名并加密成 tokengithub

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 便可web

s = Serializer(secret_key=secret_key, salt=salt)
data = s.loads(token)
复制代码

访问限制装饰器

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

在最开始的设计中，咱们的路由都是能够直接访问的，没有任何限制json

@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 信息

获取 token 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":"周萝卜真帅"}
复制代码

请求示例

将来优化

完善日志：当前只在定时任务当中加了日志，其他功能都未打印日志，后续把日志优化进来，方便问题定位
接口完善：当前接口返回数据庞杂，后续将接口拆分，增长更多参数，好比按照时间请求等
其余数据：后续增长微博、金融，票房等相关数据接口和展现

最后给出代码地址：https://github.com/zhouwei713/Mini_Flask
欢迎你们来讨论，若是有须要（遇到服务挂死，API 不可用等状况z）能够加我微信，备注“入群”，一块儿解决问题

这里先给出 H5 地址，能够先体验下效果！由于服务等缘由，若是没法展现内容，请屡次刷新！！至于小程序，仍是看腾讯的脸色行事吧

H5 示例地址