fourier-ondemand/ondemand/ondemand.py

from __future__ import print_function, absolute_import
import warnings
warnings.simplefilter("ignore", UserWarning)

from tornado.ioloop import IOLoop
from boxconfig import parse_config
from dejavu.recognize import FilePerSecondRecognizer
from dejavu import Dejavu
from endpoint import setup_endpoint
import logging as log
import requests
import json
import time
import os

from queue import Queue, Empty

log.basicConfig(format='[%(asctime)s] [%(module)s] %(message)s', level=log.INFO)

PATH = '/tmp'
config = parse_config()
queue = Queue()
recognizer = FilePerSecondRecognizer


def obt_siguiente_trabajo():
    log.info("Consultando trabajo")
    url = 'https://api.fourier.audio/na/calendario/pendiente?id=%s' % (config['device_id'],)
    response = requests.get(url)
    return response.json()


def descargar_anuncio(ad_path):
    anuncio = os.path.basename(ad_path)
    path = os.path.join(PATH, 'ads')
    os.makedirs(path, exist_ok=True)
    ruta_anuncio = os.path.join(path, anuncio)

    if os.path.isfile(ruta_anuncio):
        return ruta_anuncio
    cloud_base_url = 'https://storage.googleapis.com/{}' \
        .format(config['bucket'])
    url = '{}/{}'.format(cloud_base_url, ad_path)
    response = requests.get(url)

    # TODO: Agregar alerta cuando la respuesta no sea 200
    if response.status_code == 200:
        with open(ruta_anuncio, "wb") as fp:
            fp.write(response.content)
            return ruta_anuncio

    else:
        log.info("[Anuncio][error] %s" % (response.text))
        return None


def descargar_media(box, station, media):
    ref = '{}/{}/{}'.format(box, station, media)
    file = os.path.basename(ref)
    path = os.path.join(PATH, 'fourier', box, station)
    os.makedirs(path, exist_ok=True)
    out_file = os.path.join(path, file)

    if os.path.isfile(out_file):
        return out_file

    filename = ref.replace("/","%2F") \
        .replace("+","%2B")
    cloud_base_url = '%s%s' % (
        'https://firebasestorage.googleapis.com',
        '/v0/b/fourier-6e14d.appspot.com/o'
    )
    url = '{}/{}?alt=media'.format(cloud_base_url, filename)
    response = requests.get(url)

    if response.status_code == 200:
        with open(out_file, "wb") as fp:
            fp.write(response.content)
            return out_file
    else:
        log.info("[Media][url] %s" % (response.text))
        log.info("[Media][error] %s" % (response.text))
        return None


def enviar_resultados(trabajo):
    log.info('[Pendiente] %s' % (json.dumps(trabajo),))
    url = 'https://api.fourier.audio/na/calendario/resultado'
    response = requests.post(url, json=trabajo)
    log.info('[Response] %s' % (response.text))
    return response


def llenar_pila():
    """ Search for pending scheduled work in
    server and add them to a memory queue. """
    try:
        response = obt_siguiente_trabajo()
        if len(response["elementos"]) > 0:
            queue.put(response)

        if queue.qsize() > 0:
            loop.add_callback(procesar_siguiente_pila)
        else:
            loop.add_timeout(time.time() + 30, llenar_pila)

    except Exception as ex:
        """ Errores desconocidos """
        log.error('[feed_queue] {}'.format(ex))
        loop.add_timeout(time.time() + 60, llenar_pila)
        raise ex


def procesar_siguiente_pila():
    """ Try to the next item in a queue and start
    processing it accordingly. If success, repeat
    the function or go to feed if no more items. """
    try:
        item = queue.get(False)
        procesar_trabajo(item)
        loop.add_callback(procesar_siguiente_pila)
    except Empty:
        loop.add_callback(llenar_pila)
    except Exception as ex:
        log.error(ex)
        loop.add_callback(procesar_siguiente_pila)


def procesar_trabajo(pendiente):
    ciudad = pendiente['origen']
    estacion = pendiente['estacion']
    confianza = 35
    segmento = 5
    if "segmento" in pendiente:
        segmento = int(pendiente["segmento"])

    if "confianza" in pendiente:
        confianza = int(pendiente["confianza"])

    # Descarga de anuncios
    log.info("Descargando anuncios")
    try:
        anuncios = []
        id_by_ad = {}
        item_ids = []
        for i in pendiente["elementos"]:
            id_by_ad[i['anuncio']] = i['id']
            if i['id'] not in item_ids:
                item_ids.append(i['id'])

            anuncio = descargar_anuncio(i["ruta"])
            if anuncio is not None:
                anuncios.append(anuncio)

    except Exception as err:
        log.info('[process_segment] [{}] {}'.format(estacion, err))

    # Descarga de media
    log.info("Descargando media")
    try:
        media = []
        for i in pendiente["media"]:
            archivo = descargar_media(ciudad, estacion, i["ruta"])
            if archivo is not None:
                media.append((archivo, i["fecha"], i["timestamp"]))

    except Exception as err:
        log.info(err)

    log.info("Inicia la comparacion, tamaño de segmento %s" % (segmento,))
    try:
        dejavu = None
        resultados = {}
        if len(media) > 0 and len(anuncio) > 0:
            dejavu = Dejavu({"database_type": "mem"})
            try:
                x = 0
                for ruta, fecha, ts in media:
                    log.info("Huellando %s" % (ruta,))
                    dejavu.fingerprint_file(ruta, ts)
            except Exception as ex:
                log.info(ex)

            for anuncio in anuncios:
                for i in dejavu.recognize(recognizer, anuncio, 5):
                    if not "id" in i:
                        continue

                    nombre_anuncio = os.path.split(anuncio)[-1]
                    id = id_by_ad[nombre_anuncio]
                    if id not in resultados:
                        resultados[id] = []

                    obj = i
                    obj["match_time"] = None
                    dict = {
                        "id": id,
                        "anuncio": anuncio,
                        "fecha": obj["name"],
                        "timestamp": obj["name"] + int(obj['offset_seconds']),
                        "confianza": obj["confidence"],
                        "longitud": obj["length"],
                        "desfase_segundos": obj["offset_seconds"]
                    }
                    resultados[id].append(dict)

            aux = {}
            for k in resultados.keys():
                lista = sorted(resultados[k], key=lambda d: d['timestamp'])
                lista_nueva = []
                ult = None
                for x in range(0, len(lista)):
                    if x == 0:
                        ult = x
                        lista_nueva.append(lista[ult])
                    else:
                        dif = lista[x]['timestamp'] - lista[x - 1]['timestamp']
                        if dif <= 30:
                            lista_nueva[ult]['confianza'] = int(lista_nueva[ult]['confianza']) + int(lista[x]['confianza'])
                            lista_nueva[ult]['longitud'] = int(lista_nueva[ult]['longitud']) +int(lista[x]['longitud'])
                        else:
                            lista_nueva.append(lista[x])
                            ult = len(lista_nueva) - 1

                aux[k] = lista_nueva

        else:
            for i in pendiente['elementos']:
                i['comentario'] = 'Problemas técnicos'

        for id in aux:
            for e in aux[id]:
                for i in pendiente['elementos']:
                    i['comentario'] = ''
                    anuncio = e['anuncio'].replace('/tmp/ads/', '')
                    if i['id'] == e['id'] and i['anuncio'] == anuncio:
                        if 'encontrados' not in i:
                            i['encontrados'] = []
                        obj = {
                            "fecha": e["fecha"],
                            "anuncio": anuncio,
                            "longitud": int(e["longitud"] / 1000),
                            "confianza": e["confianza"],
                            "timestamp": e["timestamp"],
                            "desfase_segundos": e["desfase_segundos"]
                        }
                        i['encontrados'].append(obj)
                        break

        # log.info(json.dumps(extras))
        log.info("[Resultado] %s" % (json.dumps(resultados)))
        pendiente["media"] = None
        enviar_resultados(pendiente)

    except Exception as ex:
        log.info(ex)


app = setup_endpoint(queue=queue)
loop = IOLoop.current()
loop.add_callback(llenar_pila)

if __name__ == '__main__':
    try:
        log.info('Starting ondemand service')
        loop.start()
    except KeyboardInterrupt:
        log.error('Process killed')