personal-website/dynamic/app.py

"""Run an aiohttp web app."""

# Standard library modules
import asyncio
import datetime
import logging
import os

# Third party modules
from aiohttp import web

# Project modules
from data.config import (
    log_file_name, errors_file_name, app_host, app_port
)
from data.passwords import pull_token

path = os.path.dirname(__file__)
log_file = "{}/data/{}".format(path, log_file_name)
errors_file = "{}/data/{}".format(path, errors_file_name)

# Outputs the log in console, log_file and errors_file
#   Log formatter: datetime, module name (filled with spaces up to 15
#       characters), logging level name (filled to 8), message
log_formatter = logging.Formatter(
    "%(asctime)s [%(module)-15s %(levelname)-8s]     %(message)s",
    style='%'
)
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)

file_handler = logging.FileHandler(log_file, mode="a", encoding="utf-8")
file_handler.setFormatter(log_formatter)
file_handler.setLevel(logging.DEBUG)
root_logger.addHandler(file_handler)

file_handler = logging.FileHandler(errors_file, mode="a", encoding="utf-8")
file_handler.setFormatter(log_formatter)
file_handler.setLevel(logging.ERROR)
root_logger.addHandler(file_handler)

consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(log_formatter)
consoleHandler.setLevel(logging.DEBUG)
root_logger.addHandler(consoleHandler)

app = web.Application()
# davtebot = Bot.get(bot_token)


def page(path='/', headers={'Content-Type': 'text/html'}, prefix=''):
    """Decorator: serve the result of such decorated function at given path."""
    path = "{pr}{pa}".format(
        pr=prefix,
        pa=path
    )

    def decorator(func):
        async def decorated(request):
            response = await func(request)
            if type(response) is str:
                if response.endswith('.html'):
                    with open('vita/homepage.html', 'r') as _file:
                        response = _file.read()
                response = web.Response(
                    text=response,
                    headers=headers
                )
            assert isinstance(
                response, web.Response
            ), "Invalid response returned!"
            return response
        app.router.add_get(path, decorated)
        return
    return decorator


@page('/test')
async def test(request):
    """Test page: return datetime."""
    return str(datetime.datetime.now())


@page('/pull/{t}/image-editor'.format(t=pull_token))
async def pull_image_editor_project(request):
    """Git pull at request."""
    try:
        _subprocess = await asyncio.create_subprocess_shell(
            'bash {path}/update_image_editor.sh'.format(
                path=path
            )
        )
        stdout, stderr = await _subprocess.communicate()
    except Exception as e:
        logging.error(
            "Exception {e}:\n{o}\n{er}".format(
                e=e,
                o=stdout.decode().strip(),
                er=stderr.decode().strip()
            )
        )
    return web.Response(
        status=302,
        headers={
            'location': 'https://michele.davte.it/image-editor/',
        },
    )


web.run_app(app, host=app_host, port=app_port)