太好玩了，爬虫、部署API、加小程序，一条龙玩转知乎热榜！

正文

请到「今天看啥」查看全文

hot_link = db.Column(db.String( 64 ))
hot_cardid = db.Column(db.String( 32 ))


class ZhihuMetrics (db.Model) :
__tablename__ = 'ZhihuMetrics'
id = db.Column(db.Integer, primary_key= True )
hot_metrics = db.Column(db.String( 64 ))
hot_cardid = db.Column(db.String( 32 ), index= True )
update_time = db.Column(db.DateTime)


class ZhihuContent (db.Model) :
__tablename__ = 'ZhihuContent'
id = db.Column(db.Integer, primary_key= True )
answer_id = db.Column(db.Integer, index= True )
author = db.Column(db.String( 32 ), index= True )
voteup_count = db.Column(db.Integer)
comment_count = db.Column(db.Integer)
reward_info = db.Column(db.Integer)
content = db.Column(db.Text)

定时任务

由于我们需要定时查询热榜列表和热榜的热度值，所以这里需要定时运行相关的任务，使用插件 flask_apscheduler 来做定时任务

我们的定时任务，涉及到了网络请求和数据入库的操作，把这部分定时任务代码单独拉出来，在 Flask 项目的根目录下创建一个文件 apschedulerjob.py，由于在运行该文件时，是没有 Flask app 变量的，所以我们需要手动调用 app_context() 方法来创建 app 上下文

def opera_db():
    with scheduler.app.app_context():
...

当然，这里的 scheduler 变量是在 create_app 中初始化过的

from flask_apscheduler import APScheduler

scheduler = APScheduler()


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    db.init_app(app)
    scheduler.init_app(app)
...

接着，我们就可以根据前面的两个爬虫函数，来分别入库数据了

入库热榜热度数据

update_metrics = ZhihuMetrics(hot_metrics=i['target']['metricsArea']['text'],
                                              hot_cardid=i['cardId'],
                                              update_time=datetime.datetime.now())

入库热榜列表数据

new_details = ZhihuDetails(hot_id=i['id'], hot_name=i['target']['titleArea']['text'],
                                           hot_link=i['target']['link']['url'], hot_cardid=i['cardId'])

入库热榜回答数据

new_content = ZhihuContent(answer_id=answer_id, author=answer['author'], voteup_count=answer['voteup_count'],
                                                   comment_count=answer['comment_count'], reward_info=answer['reward_info'],
                                                   content=answer['content'])

最后我们就可以在 Flask 的入口程序中启动定时任务了

import os
from app import create_app, scheduler


app = create_app(os.getenv('FLASK_CONFIG') or 'default')


if __name__ == '__main__':
    scheduler.start()
    app.run(debug=True)

编写 API

热榜列表 API

请到「今天看啥」查看全文