acapela.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. import logging
  2. import re
  3. import requests
  4. from kalliope.core import FileManager
  5. from kalliope.core.TTS.TTSModule import TTSModule, FailToLoadSoundFile, MissingTTSParameter
  6. logging.basicConfig()
  7. logger = logging.getLogger("kalliope")
  8. TTS_URL = "https://acapela-box.com/AcaBox/dovaas.php"
  9. TTS_CONTENT_TYPE = "audio/mpeg"
  10. TTS_TIMEOUT_SEC = 30
  11. class TCPTimeOutError(Exception):
  12. """
  13. This error is raised when the TCP connection has been lost. Probably due to a low internet
  14. connection while trying to access the remote API.
  15. """
  16. pass
  17. class Acapela(TTSModule):
  18. def __init__(self, **kwargs):
  19. super(Acapela, self).__init__(**kwargs)
  20. self.voice = kwargs.get('voice', None)
  21. if self.voice is None:
  22. raise MissingTTSParameter("voice parameter is required by the Acapela TTS")
  23. # speech rate
  24. self.spd = kwargs.get('spd', 180)
  25. # VOICE SHAPING
  26. self.vct = kwargs.get('vct', 100)
  27. self.words = None
  28. def say(self, words):
  29. """
  30. :param words: The sentence to say
  31. """
  32. self.words = words
  33. self.generate_and_play(words, self._generate_audio_file)
  34. def _generate_audio_file(self):
  35. """
  36. Generic method used as a Callback in TTSModule
  37. - must provided the audio file and write it on the disk
  38. .. raises:: FailToLoadSoundFile, TCPTimeOutError
  39. """
  40. # Prepare payload
  41. payload = self.get_payload()
  42. cookie = Acapela._get_cookie()
  43. # Get the mp3 URL from the page
  44. mp3_url = Acapela.get_audio_link(TTS_URL, payload, cookie)
  45. # getting the mp3
  46. headers = {
  47. "Cookie": "%s" % cookie
  48. }
  49. r = requests.get(mp3_url, headers=headers, stream=True, timeout=TTS_TIMEOUT_SEC)
  50. content_type = r.headers['Content-Type']
  51. logger.debug("Acapela : Trying to get url: %s response code: %s and content-type: %s",
  52. r.url,
  53. r.status_code,
  54. content_type)
  55. # Verify the response status code and the response content type
  56. if r.status_code != requests.codes.ok or content_type != TTS_CONTENT_TYPE:
  57. raise FailToLoadSoundFile("Acapela : Fail while trying to remotely access the audio file")
  58. # OK we get the audio we can write the sound file
  59. FileManager.write_in_file(self.file_path, r.content)
  60. def get_payload(self):
  61. """
  62. Generic method used load the payload used to access the remote api
  63. :return: Payload to use to access the remote api
  64. """
  65. return {
  66. "text": "%s" % self.words,
  67. "voice": "%s22k" % self.voice,
  68. "spd": "%s" % self.spd,
  69. "vct": "%s" % self.vct,
  70. "codecMP3": "1",
  71. "format": "WAV 22kHz",
  72. "listen": 1
  73. }
  74. @staticmethod
  75. def get_audio_link(url, payload, cookie, timeout_expected=TTS_TIMEOUT_SEC):
  76. """
  77. Return the audio link
  78. :param url: the url to access
  79. :param payload: the payload to use to access the remote api
  80. :param timeout_expected: timeout before the post request is cancel
  81. :param cookie: cookie used for authentication
  82. :return: the audio link
  83. :rtype: String
  84. """
  85. headers = {
  86. "Cookie": "%s" % cookie
  87. }
  88. r = requests.post(url, data=payload, headers=headers, timeout=timeout_expected)
  89. data = r.json()
  90. return data["snd_url"]
  91. @staticmethod
  92. def _get_cookie():
  93. """
  94. Get a cookie that is used to authenticate post request
  95. :return: the str cookie
  96. """
  97. returned_cookie = ""
  98. index_url = "https://acapela-box.com/AcaBox/index.php"
  99. r = requests.get(index_url)
  100. regex = "(acabox=\w+)"
  101. cookie_match = re.match(regex, r.headers["Set-Cookie"])
  102. if cookie_match:
  103. returned_cookie = cookie_match.group(1)
  104. return returned_cookie