ska-referee-test/ref-test/common/data_tools.py

59 lines
2.0 KiB
Python
Raw Normal View History

import os
from shutil import rmtree
import pathlib
from json import dump, loads
from datetime import datetime
from main import app
from werkzeug.utils import secure_filename
def check_data_folder_exists():
if not os.path.exists(app.config['DATA_FILE_DIRECTORY']):
pathlib.Path(app.config['DATA_FILE_DIRECTORY']).mkdir(parents='True', exist_ok='True')
def check_current_indicator():
if not os.path.isfile(os.path.join(app.config['DATA_FILE_DIRECTORY'], '.current.txt')):
open(os.path.join(app.config['DATA_FILE_DIRECTORY'], '.current.txt'),'w').close()
def make_temp_dir(file):
if not os.path.isdir('tmp'):
os.mkdir('tmp')
if os.path.isfile(f'tmp/{file.filename}'):
os.remove(f'tmp/{file.filename}')
file.save(f'tmp/{file.filename}')
def check_json_format(file):
if not '.' in file.filename:
return False
if not file.filename.rsplit('.', 1)[-1] == 'json':
return False
return True
def validate_json_contents(file):
file.stream.seek(0)
data = loads(file.read())
if not type(data) is dict:
return False
elif not all( key in data for key in ['meta', 'questions']):
return False
elif not type(data['meta']) is dict:
return False
elif not type(data['questions']) is list:
return False
return True
def store_data_file(file):
from admin.views import get_id_from_cookie
check_current_indicator()
timestamp = datetime.utcnow()
filename = '.'.join([timestamp.strftime('%Y%m%d%H%M%S'),'json'])
filename = secure_filename(filename)
file_path = os.path.join(app.config['DATA_FILE_DIRECTORY'], filename)
file.stream.seek(0)
data = loads(file.read())
data['meta']['timestamp'] = timestamp.strftime('%Y-%m-%d %H%M%S')
data['meta']['author'] = get_id_from_cookie()
with open(file_path, 'w') as _file:
dump(data, _file, indent=4)
with open(os.path.join(app.config['DATA_FILE_DIRECTORY'], '.current.txt'), 'w') as _file:
_file.write(filename)