input-service/src/resources.py

#!/usr/bin/env python3
import json
from datetime import datetime
import tzlocal
from xeger import Xeger
from flask_restful import Resource
from flask import request, current_app, abort
import requests
import pika
from db import db
from influxus import influx_db
from models import SampleMetadata
from schemas import SampleSchema, SampleMetadataSchema

"""
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 abort(400, "no file found")
        else:
            soundfile = request.files['file']

        if 'description' not in request.form:
            return abort(400, "no description found")
        else:
            description = request.form.get("description")

        if soundfile.content_type != 'audio/wave':
            current_app.logger.info(f"Input file was not WAV.")
            return abort(415, 'Input file not a wave file.')
        try:
            desc = self.sampleschema.loads(description)
        except Exception as e:
            current_app.logger.exception(e)
            return abort(417, 'Input JSON schema invalid')

        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
        )
        db.session.add(record)

        files = {
            'description': (None, json.dumps({'tag': generated_tag}), 'application/json'),
            'soundFile': (
                'wave.wav',
                soundfile,
                soundfile.content_type,
                {'Content-Length': soundfile_content_length})}

        r = requests.post(
            f"http://{current_app.config.get('STORAGE_HOSTNAME')}/object",
            files=files)

        if r.status_code not in [200, 201]:
            return abort(500, f"Failed to upload sample to storage service. Upstream status: {r.status_code}: {r.text}")

        try:
            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)
            return abort(569, "AMPQ Publish error")

        influx_db.write_points(
            [
                {
                    'time': datetime.now(tz=tzlocal.get_localzone()),
                    'measurement': 'cloudinput',
                    'tags': {
                        'device': desc['device_id']
                    },
                    'fields': {
                        'bruh': 1.0
                    }
                }
            ]
        )

        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