asr_processor.py

import os
import uuid
from abc import ABC, abstractmethod
from pathlib import Path

from flask import Flask, Response, jsonify, request

from sziszapangma.integration.base_asr_service.asr_result import AsrResult

_TEMP_DIRECTORY = "asr_processing"


class AsrProcessor(ABC):

    @abstractmethod
    def process_asr(self, audio_file_path: str) -> AsrResult:
        """Method to call for ASR results."""

    def process_request(self) -> Response:
        file_tag = str(uuid.uuid4())
        f = request.files["file"]
        if f is not None and f.filename is not None:
            file_extension = f.filename.split(".")[-1]
            file_name = f"{file_tag}.{file_extension}"
            file_path = f"{_TEMP_DIRECTORY}/{file_name}"
            f.save(file_path)
            try:
                transcription = self.process_asr(file_path)
                os.remove(file_path)
                result_object = jsonify(
                    {"transcription": transcription.words, "full_text": transcription.full_text}
                )
            except:
                result_object = jsonify({"error": "Error on asr processing"})
        else:
            result_object = jsonify({"error": "Error on asr processing"})
        return result_object

    def start_processor(self):
        app = Flask(__name__)
        Path(_TEMP_DIRECTORY).mkdir(parents=True, exist_ok=True)
        app.route("/process_asr", methods=["POST"])(self.process_request)
        app.run(debug=True, host="0.0.0.0")