Mac OS Say API Server

say_server.py

"""
Mac OS Say API Server.
http://ip:port/say/something
"""
import os
import random
import asyncio
import urllib.parse
from subprocess import call


class Wrapper:

    """Connection Wrapper."""

    def __init__(self, reader, writer):
        """Init."""
        self.reader = reader
        self.writer = writer

    async def readline(self, strip=True) -> str:
        """Readline from remote."""
        data = await self.reader.readline()
        data = data.decode(errors="ignore")
        if strip:
            data = data.strip('\n')
        return data

    async def writeline(self, content: str) -> None:
        """Writeline to remote."""
        content = content.strip('\n')
        self.writer.write(''.join([content, '\n']).encode())
        await self.writer.drain()


def getAudio(content):
    content.replace("'", '')
    content.replace('"', '')
    content.replace('&', '')  # Security Concern

    content = ''.join(['"', content, '"'])

    filename = ''.join([random.choice('1234568') for i in range(5)] + ['.wav'])
    call(['say', '-o', filename, '--data-format=LEF32@44100', content])
    with open(filename, mode='rb') as reader:
        data = reader.read()
    os.remove(filename)
    return data


async def handle_web(reader, writer):
    """Web handler."""
    conn = Wrapper(reader, writer)
    while True:
        data = await conn.readline(reader)
        if data.startswith('GET'):
            data = data.strip('\n').strip('/')
            data = data.split(' ')[1][1:].split('/')
            break
    if data[0] == 'say':
        content = urllib.parse.unquote(data[1])
        audio = getAudio(content)
        writer.write(b'HTTP/1.0 200 OK\r\n')
        writer.write(b'Content-Type: audio/wav\r\n')
        length = ''.join(['Content-Length: ', str(len(audio)), '\r\n'])
        writer.write(length.encode())
        writer.write(b'Content-Transfer-Encoding: binary\r\n\r\n')
        writer.write(audio)
        writer.write(b'Content-Transfer-Encoding: binary\r\n')
    else:
        writer.write(b'HTTP/1.0 403 Forbidden\r\n')
    writer.close()


def main():
    """Mac OS Say Server Main Function."""
    web_addr = ('0.0.0.0', 8080)
    loop = asyncio.get_event_loop()
    web = asyncio.start_server(handle_web, *web_addr)
    server = loop.run_until_complete(web)
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()


if __name__ == '__main__':
    main()