hotpocket/services/extension/tasks.py

# -*- coding: utf-8 -*-
# type: ignore

from __future__ import annotations

import json
import os
import typing

from invoke import Context, task
from invoke.exceptions import UnexpectedExit
import jinja2
import werkzeug
import werkzeug.routing

from hotpocket_workspace_tools import get_workspace_mode
from hotpocket_workspace_tools.tasks import *  # noqa: F401,F403

WORKSPACE_MODE = get_workspace_mode()


class PreviewApp:
    ROUTES = werkzeug.routing.Map([
        werkzeug.routing.Rule('/preview/popup.html', endpoint='preview_popup'),
    ])

    def __init__(self):
        here = os.path.dirname(__file__)
        self.jinja_env = jinja2.Environment(
            loader=jinja2.FileSystemLoader(
                [
                    os.path.join(here, 'src/content/templates'),
                ],
            ),
            auto_reload=True,
            autoescape=True,
        )

    @classmethod
    def create_app(cls) -> typing.Self:
        return cls()

    def render_template(self,
                        template_name: str,
                        **context,
                        ) -> werkzeug.Response:
        template = self.jinja_env.get_template(template_name)
        return werkzeug.Response(
            template.render(context), mimetype='text/html',
        )

    def dispatch_request(self,
                         request: werkzeug.Request,
                         ) -> werkzeug.Response | Exception:
        adapter = self.ROUTES.bind_to_environ(request.environ)
        try:
            endpoint, values = adapter.match()
            return getattr(self, f'on_{endpoint}')(request, **values)
        except werkzeug.exceptions.HTTPException as exception:
            return exception

    def on_preview_popup(self, request: werkzeug.Request) -> werkzeug.Response:
        return self.render_template('popup_preview.html')

    def wsgi_app(self,
                 environ: werkzeug.wsgi.WSGIEnvironment,
                 start_response: typing.Any,
                 ) -> werkzeug.Response:
        request = werkzeug.Request(environ)
        response = self.dispatch_request(request)
        return response(environ, start_response)

    def __call__(self, environ, start_response):
        return self.wsgi_app(environ, start_response)


@task
def clean(ctx: Context):
    ctx.run('rm -rf dist/')


@task
def test(ctx: Context):
    # ctx.run('pytest -v --disable-warnings')
    print('NOOP')


@task
def flake8(ctx: Context):
    ctx.run('flake8')


@task
def isort(ctx, check=False, diff=False):
    command_parts = [
        'isort',
    ]

    if check is True:
        command_parts.append('--check')

    if diff is True:
        command_parts.append('--diff')

    command_parts.append('.')

    ctx.run(' '.join(command_parts))


@task
def eslint(ctx: Context):
    ctx.run('yarn run eslint')


@task
def lint(ctx: Context):
    ihazsuccess = True

    try:
        flake8(ctx)
    except UnexpectedExit:
        ihazsuccess = False

    try:
        isort(ctx, check=True)
    except UnexpectedExit:
        ihazsuccess = False

    try:
        eslint(ctx)
    except UnexpectedExit:
        ihazsuccess = False

    if ihazsuccess is False:
        raise RuntimeError('FIAL')


@task
def typecheck(ctx: Context):
    ctx.run('mypy .')


@task
def django_shell(ctx: Context):
    raise NotImplementedError()


@task
def ci(ctx: Context):
    ihazsuccess = True

    ci_tasks = [test, lint, typecheck]
    for ci_task in ci_tasks:
        try:
            ci_task(ctx)
        except UnexpectedExit:
            ihazsuccess = False

    if ihazsuccess is False:
        raise RuntimeError('FIAL')


@task
def setup(ctx: Context):
    print('NOOP')


@task
def start_web(ctx: Context):
    from werkzeug.serving import run_simple
    app = PreviewApp.create_app()

    bind = os.getenv('HOTPOCKET_EXTENSION_PREVIEW_BIND', '127.0.0.1:6524')
    host, port = bind.split(':', maxsplit=1)
    run_simple(host, int(port), app, use_debugger=True, use_reloader=True)


@task
def start_safari(ctx: Context):
    ctx.run('yarn watch:safari')


@task(pre=[clean])
def start_chrome(ctx: Context):
    ctx.run('yarn watch:chrome')


@task(pre=[clean])
def start_firefox(ctx: Context):
    ctx.run('yarn watch:firefox')


@task
def build_safari(ctx: Context):
    ctx.run('yarn build:safari')


@task(pre=[clean])
def build_chrome(ctx: Context):
    ctx.run('yarn build:chrome')
    ctx.run(' '.join([
        r'/Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome',
        '--pack-extension=dist/chrome-production/',
        '--pack-extension-key=secrets/chrome.pem',
    ]))


@task(pre=[clean])
def build_firefox(ctx: Context):
    ctx.run('yarn build:firefox')

    firefox_secrets = None
    with open('secrets/firefox.json', 'r', encoding='utf-8') as firefox_secrets_f:
        firefox_secrets = json.load(firefox_secrets_f)

    with ctx.cd('dist/firefox-production'):
        ctx.run(' '.join([
            'web-ext',
            'sign',
            '--channel=unlisted',
            f'--api-key={firefox_secrets["api_key"]}',
            f'--api-secret={firefox_secrets["api_secret"]}',
        ]))