API: upload a zip archive containing gpx files

This commit is contained in:
Sam 2018-05-29 16:28:59 +02:00
parent 276a886a61
commit d5b79e3c57
7 changed files with 170 additions and 15 deletions

View File

@ -9,7 +9,7 @@ from ..users.utils import authenticate, verify_extension
from .models import Activity
from .utils import (
ActivityException, create_activity, edit_activity, get_chart_data,
handle_one_activity
process_files
)
activities_blueprint = Blueprint('activities', __name__)
@ -146,16 +146,27 @@ def post_activity(auth_user_id):
activity_file = request.files['file']
try:
new_activity = handle_one_activity(
auth_user_id, activity_file, activity_data
new_activities = process_files(
auth_user_id, activity_data, activity_file
)
response_object = {
'status': 'created',
'data': {
'activities': [new_activity.serialize()]
if len(new_activities) > 0:
response_object = {
'status': 'created',
'data': {
'activities': [new_activity.serialize()
for new_activity in new_activities]
}
}
}
return jsonify(response_object), 201
code = 201
else:
response_object = {
'status': 'fail',
'data': {
'activities': []
}
}
code = 400
return jsonify(response_object), code
except ActivityException as e:
db.session.rollback()
if e.e:

View File

@ -1,5 +1,6 @@
import os
import tempfile
import zipfile
from datetime import datetime, timedelta
import gpxpy.gpx
@ -248,12 +249,8 @@ def get_new_file_path(auth_user_id, activity_date, old_filename, sport):
return file_path
def handle_one_activity(auth_user_id, activity_file, activity_data):
filename = secure_filename(activity_file.filename)
file_path = get_file_path(auth_user_id, filename)
def process_one_gpx_file(auth_user_id, activity_data, file_path, filename):
try:
activity_file.save(file_path)
gpx_data = get_gpx_info(file_path)
sport = Sport.query.filter_by(id=activity_data.get('sport_id')).first()
@ -285,3 +282,43 @@ def handle_one_activity(auth_user_id, activity_file, activity_data):
raise ActivityException(
'fail', 'Error during activity save.', e
)
def process_zip_archive(auth_user_id, activity_data, zip_path):
extract_dir = os.path.join(
current_app.config['UPLOAD_FOLDER'],
'activities',
str(auth_user_id),
'extract')
with zipfile.ZipFile(zip_path, "r") as zip_ref:
zip_ref.extractall(extract_dir)
new_activities = []
for gpx_file in os.listdir(extract_dir):
if ('.' in gpx_file and gpx_file.rsplit('.', 1)[1].lower()
in current_app.config.get('ACTIVITY_ALLOWED_EXTENSIONS')):
file_path = os.path.join(extract_dir, gpx_file)
new_activity = process_one_gpx_file(auth_user_id, activity_data,
file_path, gpx_file)
new_activities.append(new_activity)
return new_activities
def process_files(auth_user_id, activity_data, activity_file):
filename = secure_filename(activity_file.filename)
extension = f".{filename.rsplit('.', 1)[1].lower()}"
file_path = get_file_path(auth_user_id, filename)
try:
activity_file.save(file_path)
except Exception as e:
raise ActivityException('error', 'Error during activity file save.', e)
if extension == ".gpx":
return [process_one_gpx_file(
auth_user_id, activity_data, file_path, filename
)]
else:
return process_zip_archive(auth_user_id, activity_data, file_path)

View File

@ -15,7 +15,7 @@ class BaseConfig:
current_app.root_path, 'uploads'
)
PICTURE_ALLOWED_EXTENSIONS = {'jpg', 'png', 'gif'}
ACTIVITY_ALLOWED_EXTENSIONS = {'gpx'}
ACTIVITY_ALLOWED_EXTENSIONS = {'gpx', 'zip'}
class DevelopmentConfig(BaseConfig):

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@ -1,4 +1,5 @@
import json
import os
from io import BytesIO
@ -645,3 +646,109 @@ def test_add_activity_zero_value(
assert len(data['data']['activities'][0]['segments']) == 0
assert len(data['data']['activities'][0]['records']) == 0
def test_get_an_activity_with_zip(app, user_1, sport_1_cycling):
file_path = os.path.join(app.root_path, 'tests/files/gpx_test.zip')
# 'gpx_test.zip' contains 3 gpx files (same data) and 1 non-gpx file
with open(file_path, 'rb') as zip_file:
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(zip_file, 'gpx_test.zip'),
data='{"sport_id": 1}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 201
assert 'created' in data['status']
assert len(data['data']['activities']) == 3
assert 'just an activity' == data['data']['activities'][0]['title']
assert_activity_data_with_gpx(data)
def test_get_an_activity_with_zip_folder(app, user_1, sport_1_cycling):
file_path = os.path.join(app.root_path, 'tests/files/gpx_test_folder.zip')
# 'gpx_test_folder.zip' contains 3 gpx files (same data) and 1 non-gpx file
# in a folder
with open(file_path, 'rb') as zip_file:
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(zip_file, 'gpx_test_folder.zip'),
data='{"sport_id": 1}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 400
assert 'fail' in data['status']
assert len(data['data']['activities']) == 0
def test_get_an_activity_with_zip_incorrect_file(app, user_1, sport_1_cycling):
file_path = os.path.join(app.root_path, 'tests/files/gpx_test_incorrect.zip') # noqa
# 'gpx_test_incorrect.zip' contains 2 gpx files, one is incorrect
with open(file_path, 'rb') as zip_file:
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.post(
'/api/activities',
data=dict(
file=(zip_file, 'gpx_test_incorrect.zip'),
data='{"sport_id": 1}'
),
headers=dict(
content_type='multipart/form-data',
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 500
assert 'error' in data['status']
assert 'Error during gpx file parsing.' in data['message']
assert 'data' not in data