input-service/src/resources.py

143 lines
4.9 KiB
Python

#!/usr/bin/env python3
import json
from xeger import Xeger
from flask_restful import Resource
from flask import request, current_app
import requests
import pika
from db import db
from models import SampleMetadata
from schemas import SampleSchema, SampleMetadataSchema
from config import *
"""
Flask Restful endpoints
"""
__author__ = '@tormakris'
__copyright__ = "Copyright 2020, Birbnetes Team"
__module_name__ = "endpoints"
__version__text__ = "1"
class SampleResource(Resource):
"""
Sample endpoint
See: https://swagger.kmlabz.com/?urls.primaryName=Input%20Service
"""
sampleschema = SampleSchema(many=False)
samplemetadataschema = SampleMetadataSchema(many=True)
def post(self):
"""
Post request send to the endpoint
:return:
"""
if 'file' not in request.files:
return {"err_msg": "no file found"}, 469
else:
soundfile = request.files['file']
if 'description' not in request.form:
return {"err_msg": "no description found"}, 470
else:
description = request.form.get("description")
if soundfile.content_type != 'audio/wave':
current_app.logger.info(
f"Input file was not WAV.")
return {'err_msg': 'Input file not a wave file.'}, 415
try:
desc = self.sampleschema.loads(description)
except Exception as e:
current_app.logger.exception(e)
return {'err_msg': 'Input JSON schema invalid'}, 417
xeger = Xeger(limit=30)
while True:
generated_tag = xeger.xeger(r'^[a-zA-Z]+[0-9a-zA-Z_]*$')[:32]
if len(generated_tag) > 2: # Ensure minimum length
break
# Handle mega-autismo-cliento
soundfile_content_length = soundfile.content_length
if soundfile_content_length <= 0: # BRUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUH
current_app.logger.debug(
"The uploader did not provide content-length for the sound file... Calculating manually..."
)
# So, this is a seekable stream, so we just seek to the end
old_ptr = soundfile.tell()
soundfile.seek(0, 2)
# Check where is the end (= content length)
soundfile_content_length = soundfile.tell()
# Seek back to where the stream was
soundfile.seek(old_ptr, 0)
# It's insane, that you can not set this field in curl
record = SampleMetadata(
device_id=desc['device_id'],
device_date=desc['date'],
tag=generated_tag)
try:
db.session.add(record)
requests.post(
f"http://{STORAGE_HOSTNAME}/object",
files={
'description': (None, json.dumps({'tag': generated_tag}), 'application/json'),
'soundFile': (
'wave.wav',
soundfile,
soundfile.content_type,
{'Content-Length': soundfile_content_length})}).raise_for_status() # Anyádat curl am
credentials = pika.PlainCredentials(current_app.config['FLASK_PIKA_PARAMS']['username'],
current_app.config['FLASK_PIKA_PARAMS']['password'])
connection = pika.BlockingConnection(
pika.ConnectionParameters(host=current_app.config['FLASK_PIKA_PARAMS']['host'],
credentials=credentials,
heartbeat=0,
socket_timeout=5))
channel = connection.channel()
channel.exchange_declare(exchange=current_app.config['EXCHANGE_NAME'],
exchange_type='direct')
channel.basic_publish(exchange=current_app.config['EXCHANGE_NAME'],
routing_key='feature',
body=json.dumps({'tag': generated_tag}).encode('UTF-8'))
connection.close()
except Exception as e:
current_app.logger.exception(e)
db.session.rollback()
return {"err_msg": str(
e), "hint": "DB or downstream service error"}, 569
db.session.commit()
return {"tag": generated_tag}, 200
def get(self):
"""
Get all stored items
:return:
"""
samples = SampleMetadata.query.all()
return self.samplemetadataschema.dump(list(samples)), 200
class SampleParameterResource(Resource):
"""
Sample endpoint with parameters
"""
samplemetadataschema = SampleMetadataSchema(many=False)
def get(self, tag: str):
"""
Get a specific item
:param tag:
:return:
"""
sample = SampleMetadata.query.filter_by(tag=tag).first_or_404()
return self.samplemetadataschema.dump(sample), 200