前面写了一个知乎爬虫、API和小程序一条龙第一弹,反响还不错,因而在这些天的空闲时间里,我又优化下代码,而且把服务部署到了云服务器上,开放了 API 供须要的小伙伴使用。
也有不少人要源代码看看,想本身动手实践下,今天就把代码放出来,写的很差,仅供参考,也欢迎一块儿讨论维护!css
由于准备开放 API 接口出来,因此考虑了下,仍是作一些简单的验证,毕竟安全措施作的好,你好我也好!html
首先咱们先来看下总体的请求流程前端
客户端先经过 getToken 接口来获取一个具备时间期限的 token 信息,而后再携带该 token 信息访问对应的数据接口python
我这里使用第三方库 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 中,下次再请求时直接拿数据便可。
如今咱们的详情页面展现以下
最后咱们把已经完成的代码部署到云服务器上,使用的仍是那套 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 信息
获取 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 |
{
"username": "admin",
"pwd": "admin"
}
复制代码
请求示例
{
"token":"eyJhbGciOiJIUzUxMiIsImlhdCI6MTU3NzI0NDE4MywiZXhwIjoxNTc3MjQ1OTgzfQ.eyJ1c2VyaWQiOjEsImlhdCI6MTU3NzI0NDE4My4zMjcwNjY0fQ.FptYNm0KnA8b4G_zcRJn9POrOgkiZxpvfBbzQqxoTTt7q96WeMo7Y6xCLL_oS4ksBP8jMztqopDRRqScXPKowg",
"secret":"周萝卜真帅"}
复制代码
请求示例
{
"token":"eyJhbGciOiJIUzUxMiIsImlhdCI6MTU3NzI0NDE4MywiZXhwIjoxNTc3MjQ1OTgzfQ.eyJ1c2VyaWQiOjEsImlhdCI6MTU3NzI0NDE4My4zMjcwNjY0fQ.FptYNm0KnA8b4G_zcRJn9POrOgkiZxpvfBbzQqxoTTt7q96WeMo7Y6xCLL_oS4ksBP8jMztqopDRRqScXPKowg",
"secret":"周萝卜真帅"}
复制代码
请求示例
最后给出代码地址:https://github.com/zhouwei713/Mini_Flask
欢迎你们来讨论,若是有须要(遇到服务挂死,API 不可用等状况z)能够加我微信,备注“入群”,一块儿解决问题
这里先给出 H5 地址,能够先体验下效果!由于服务等缘由,若是没法展现内容,请屡次刷新!!至于小程序,仍是看腾讯的脸色行事吧