79 lines
1.8 KiB
Python
79 lines
1.8 KiB
Python
import os
|
||
import json
|
||
import functools
|
||
|
||
from flask import request
|
||
|
||
from setting import API_SECRET, APP_NAME
|
||
|
||
|
||
def file_path(param):
|
||
"""
|
||
返回文件真实路径
|
||
避免因为文件路径问题带来的麻烦
|
||
Parameters:
|
||
param: 文件相对路径
|
||
Returns:
|
||
path: 文件真实路径
|
||
"""
|
||
abs_path = os.path.abspath(os.path.dirname(__file__))
|
||
rel_path = abs_path[:abs_path.find(APP_NAME) + len(APP_NAME)]
|
||
path = os.path.abspath(rel_path + param)
|
||
return path
|
||
|
||
|
||
def read_json_file(param):
|
||
"""
|
||
读取json文件
|
||
Parameters:
|
||
param: json文件真实路径
|
||
Returns:
|
||
result: json内容
|
||
"""
|
||
with open(file_path(param), "r", encoding='utf-8') as f:
|
||
result = json.load(f)
|
||
return result
|
||
|
||
|
||
def verify_token(func):
|
||
"""
|
||
校验token
|
||
"""
|
||
@functools.wraps(func)
|
||
def internal(*args, **kwargs):
|
||
try:
|
||
token = request.headers.get('token')
|
||
if token != API_SECRET:
|
||
return {"info": "接口密钥错误"}, 401
|
||
except Exception:
|
||
return {"info": "请求异常"}, 401
|
||
return func(*args, **kwargs)
|
||
return internal
|
||
|
||
|
||
def sub_dict(param1, param2):
|
||
"""
|
||
获取字典的子集
|
||
Parameters:
|
||
param1: 原字典
|
||
param2: 子集字段
|
||
Returns:
|
||
子集
|
||
"""
|
||
return dict((key, value) for key, value in param1.items() if key in param2)
|
||
|
||
|
||
def df_iterrows(param):
|
||
"""
|
||
按行以数组形式返回DataFrame的index、data
|
||
Parameters:
|
||
param: DataFrame 某个df对象
|
||
Returns:
|
||
result: list 遍历df对象每行数据，包括index
|
||
"""
|
||
result = []
|
||
for row in param.iterrows():
|
||
index, data = row
|
||
result.append([index] + data.tolist())
|
||
return result
|