Skype Morse code bot

morse_code_bot.py

#!/usr/bin/env python3
"""Skype bot that translates its input messages to International Morse code and back.

The standard:
https://www.itu.int/dms_pubrec/itu-r/rec/m/R-REC-M.1677-1-200910-I!!PDF-E.pdf

# XXX 3.3.2, 3.5.1, 4.3, Part II, no abbreviations
"""
import asyncio
import json
import logging
import os
import re
import sys
import urllib.parse
import aiohttp.web  # $ pip install aiohttp
import werkzeug.contrib.cache  # $ pip install werkzeug

__version__ = '0.2.2'

log = logging.getLogger(__name__).debug

letter_sep = ' '
word_sep = ' \n'
morse_code = {
    'a': '.-',     'b': '-...',   'c': '-.-.',
    'd': '-..',    'e': '.',      'f': '..-.',
    'g': '--.',    'h': '....',   'i': '..',
    'j': '.---',   'k': '-.-',    'l': '.-..',
    'm': '--',     'n': '-.',     'o': '---',
    'p': '.--.',   'q': '--.-',   'r': '.-.',
    's': '...',    't': '-',      'u': '..-',
    'v': '...-',   'w': '.--',    'x': '-..-',
    'y': '-.--',   'z': '--..',

    'é': '..-..',

    '1': '.----',  '2': '..---',  '3': '...--',
    '4': '....-',  '5': '.....',  '6': '-....',
    '7': '--...',  '8': '---..',  '9': '----.',
    '0': '-----',

    '.': '.-.-.-', ',': '--..--', ':': '---...',
    '?': '..--..', "'": '.----.', '-': '-....-',
    '/': '-..-.', '(': '-.--.', ')': '-.--.-',
    '"': '.-..-.', '=': '-...-',

    'understood': '...-.',
    'error': '.' * 8,
    '+': '.-.-.',
    'invitation to transmit': '-.-',
    'wait': '.-...',
    'end of work': '...-.-',
    'starting signal': '-.-.-',
    '*': '-..-', '@': '.--.-.',
}
morse_code['%'] = ' '.join(map(morse_code.get, '0/0'))
morse_code['‰'] = ' '.join(map(morse_code.get, '0/00'))
english_from_morse = dict(zip(morse_code.values(), morse_code.keys()))


# cache access token on disk
cache = werkzeug.contrib.cache.FileSystemCache('.cachedir', threshold=86400)
common_http_headers = {'User-Agent': f'morse-code-bot/{__version__}'}


def is_morse_code(text):
    """
    >>> is_morse_code('-.-')
    True
    >>> is_morse_code('........ ........ ........')
    True
    >>> is_morse_code('abc')
    False
    """
    return frozenset(''.join(text.split())) <= {'-', '.'}


def morse2english(code):
    r"""
    >>> morse2english('........ ........ ........')
    'errorerrorerror'
    >>> morse2english('.... .. \n.--.-. .-.-.-')
    'hi @.'
    """
    return ' '.join(''.join(english_from_morse.get(morse_char_code, '\ufffd')
                            for morse_char_code in word.split())
                    for word in filter(None, re.split(r'\s{2,}', code)))


def english2morse(text):
    r"""
    >>> english2morse('hi @.')
    '.... .. \n.--.-. .-.-.-'
    """
    words = text.split()
    return word_sep.join(letter_sep.join(
        morse_code.get(c, c if c.isspace() else morse_code['error'])
        for c in word)
        for word in words)


async def get_access_token():
    token = cache.get(key='token')
    if not token:
        # request access token
        url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
        data = dict(client_id=os.environ['APP_ID'],
                    scope='https://graph.microsoft.com/.default',
                    grant_type='client_credentials',
                    client_secret=os.environ['APP_SECRET'])
        headers = {**common_http_headers, **{'Cache-Control': 'no-cache'}}
        async with aiohttp.post(url, data=data, headers=headers) as r:
            assert 200 <= r.status < 300
            token = await r.json()
        cache.set('token', token, timeout=token['expires_in'])
    return token['access_token']


async def send_message(message, skypeid):
    """
    POST / v2 / conversations / 8: alice / activities HTTP / 1.1
    Host: apis.skype.com
    Authorization: Bearer < redacted oauth2 token >
    {
      "message": {"content": "Hi! (wave)"}
    }
    """
    url = f'https://apis.skype.com/v2/conversations/8:{skypeid}/activities'
    token = await get_access_token()
    headers = dict(common_http_headers, Authorization=f'Bearer {token}')
    headers['Content-Type'] = headers['Accept'] = 'application/json'
    data = json.dumps(dict(message=dict(content=message)))
    async with aiohttp.post(url, data=data, headers=headers) as r:
        assert 200 <= r.status < 300


async def handle(request):
    messages = await request.json()
    log('got %r', messages)
    for msg in messages:
        if msg['activity'] != 'message':
            continue
        assert msg['to'] == ('28:' + os.environ['APP_ID'])  # ours
        assert msg['from'][:2] == '8:'
        skypeid = msg['from'][2:]
        assert urllib.parse.quote(skypeid, safe='') == skypeid
        text = msg['content'].casefold()
        translated = [english2morse, morse2english][is_morse_code(text)](text)
        assert {'<', '>', '&'}.isdisjoint(translated)
        asyncio.ensure_future(send_message(translated, skypeid))  # don't wait
    return aiohttp.web.HTTPCreated()  # 201

app = aiohttp.web.Application()
app.router.add_route('POST', '/v1/chat', handle)

if __name__ == "__main__":
    logging.basicConfig(format="%(asctime)-15s %(message)s",
                        datefmt="%F %T",
                        level=logging.DEBUG)
    aiohttp.web.run_app(app,
                        host='localhost',
                        ssl_context=None,
                        port=int(sys.argv[1]) if len(sys.argv) > 1 else None)