MySQL+Flask，在本地实现一个API接口。

小F

发布于 2020-10-09 15:45:25

1K0

发布于 2020-10-09 15:45:25

文章被收录于专栏：用户6291251的专栏用户6291251的专栏

微信小程序中历史长河的数据，是通过调用一位大佬提供的API接口获得到的。

既然小F已经拥有了一台云服务器，所以就想自己来实现这个接口。

这样就不用担心大佬服务器过期，亦或接口发生变化。

首先本地调试一下，最后部署到服务器上。

毕竟服务器上各种环境各种头大，还是比较难搞的。

一步一个脚印慢慢来～

01 MySQL数据

MySQL老早就安装在我的Mac上了。

可视化工具则是使用「Sequel Pro」，一个小黄油桶。

对于8.0以上的MySQL版本，必须使用「Sequel Pro」的测试版，否则使用时会报错。

安装什么的就不说了，网上一大堆教程，这里记录一下数据的获取。

首先创建数据库。

import pymysql

db = pymysql.connect(host='127.0.0.1', user='root', password='20191126', port=3306)
cursor = db.cursor()
cursor.execute("CREATE DATABASE history DEFAULT CHARACTER SET utf8mb4")
db.close()

期间忘记了MySQL的密码，所以重置了一下。

网上也有教程，挺简单的。

然后创建表格。

import pymysql

db = pymysql.connect(host='127.0.0.1', user='root', password='20191126', port=3306, db='history')
cursor = db.cursor()
sql = 'CREATE TABLE IF NOT EXISTS messages (date VARCHAR(255) NOT NULL, id INT NOT NULL, title VARCHAR(255) NOT NULL, year VARCHAR(255) NOT NULL, year_num INT NOT NULL, PRIMARY KEY (id))'
cursor.execute(sql)
db.close()

最后数据爬取。

import requests
import operator
import pymysql
import json
import time

headers = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36'
}

db = pymysql.connect(host='127.0.0.1', user='root', password='20191126', port=3306,  db='history', charset='utf8mb4')
cursor = db.cursor()


def to_mysql(data):
    """
    写入数据库
    """
    table = 'messages'
    keys = ', '.join(data.keys())
    values = ', '.join(['%s'] * len(data))
    sql = 'INSERT INTO {table}({keys}) VALUES ({values}) ON DUPLICATE KEY UPDATE'.format(table=table, keys=keys, values=values)
    update = ','.join([" {key} = %s".format(key=key) for key in data])
    sql += update
    try:
        if cursor.execute(sql, tuple(data.values())*2):
            print('Successful')
    except:
        print('Failed')
        db.rollback()
    db.commit()


# 每月 
for i in range(1, 13):
    time.sleep(10)
    # 每日
    for j in range(1, 31):
        url = 'http://www.jiahengfei.cn:33550/port/history?dispose=easy&key=jiahengfei&month=' + str(i) + '&day=' + str(j)
        response = requests.get(url=url, headers=headers)
        result = json.loads(response.text)
        list1 = []
        for k in result['data']:
            if len(k['title']) < 100:
                if int(k['year']) > 0:
                    item = {}
                    item['date'] = str(i) + '/' + str(j)
                    item['id'] = k['id']
                    item['title'] = k['title']
                    item['year'] = k['year'] + '年' + str(i) + '月' + str(i) + '号'
                    item['year_num'] = int(k['year'])
                    # print(item)
            list1.append(item)
        # 按照年份排序
        sorted_x = sorted(list1, key=operator.itemgetter('year_num'))
        for x in sorted_x:
            # print(x)
            to_mysql(x)

这里有一个对列表中的字典进行排序，使用到了「operator」这个库。

如此，历史长河中的数据就获取到了。

02 Flask接口

主要使用json、pymysql、flask这三个库。

Flask中连接MySQL数据库有好几种办法，这里采用pymysql连接。

import json
import pymysql
from flask import Flask, request
from pymysql.cursors import DictCursor

app = Flask(__name__)


# 只接受get方法访问
@app.route("/port/history/", methods=["GET"])
def check():
    # 默认返回内容
    return_dict = {'code': 1, 'result': False, 'msg': '请求成功'}
    # 判断入参是否为空
    if request.args is None:
        return_dict['return_code'] = '504'
        return_dict['return_info'] = '请求参数为空'
        return json.dumps(return_dict, ensure_ascii=False)
    # 获取传入的参数
    get_data = request.args.to_dict()
    month = get_data.get('month')
    day = get_data.get('day')
    date = str(month) + '/' + str(day)
    # 对参数进行操作
    return_dict['result'] = sql_result(date)
    return json.dumps(return_dict, ensure_ascii=False)


# 功能函数
def sql_result(date):
    conn = pymysql.connect(host='127.0.0.1', database='history', user='root', password='20191126')
    cursor = conn.cursor(DictCursor)
    cursor.execute("SELECT * FROM messages WHERE date= '%s'" % str(date))
    result = cursor.fetchall()
    conn.close()
    return result


if __name__ == "__main__":
    app.run(host='127.0.0.1', port=5000)

通过SQL语句查询，并获取全部结果。