ticketfrei3/backend/src/kibicara/platforms/mastodon/webapi.py

# Copyright (C) 2020 by Cathy Hu <cathy.hu@fau.de>
# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
#
# SPDX-License-Identifier: 0BSD

from asyncio import get_event_loop
from fastapi import APIRouter, Depends, HTTPException, Response, status
from ormantic.exceptions import NoMatch
from pydantic import BaseModel, validate_email, validator
from sqlite3 import IntegrityError

from kibicara.config import config
from kibicara.platforms.mastodon.bot import spawner
from kibicara.platforms.mastodon.model import MastodonAccount, MastodonInstance
from kibicara.webapi.hoods import get_hood, get_hood_unauthorized

from mastodon import Mastodon, MastodonNetworkError
from mastodon.errors import MastodonIllegalArgumentError

from logging import getLogger

logger = getLogger(__name__)


class BodyMastodonPublic(BaseModel):
    username: str
    instance: str


class BodyMastodonAccount(BaseModel):
    email: str
    instance_url: str
    password: str

    @validator("email")
    def validate_email(cls, value):
        return validate_email(value)


async def get_mastodon(mastodon_id, hood=Depends(get_hood)):
    try:
        return await MastodonAccount.objects.get(id=mastodon_id, hood=hood)
    except NoMatch:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)


async def get_mastodon_instance(instance_url: str) -> MastodonInstance:
    """Return a MastodonInstance ORM object with valid client_id and client_secret.

    :param: instance_url: the API base URL of the mastodon server
    :return the MastodonInstance ORM object
    """
    try:
        return await MastodonInstance.objects.get(name=instance_url)
    except NoMatch:
        app_name = config.get("frontend_url")
        client_id, client_secret = Mastodon.create_app(
            app_name, api_base_url=instance_url
        )
        await MastodonInstance.objects.create(
            name=instance_url, client_id=client_id, client_secret=client_secret
        )
        return await MastodonInstance.objects.get(name=instance_url)


router = APIRouter()
twitter_callback_router = APIRouter()


@router.get(
    "/public",
    # TODO response_model,
    operation_id="get_mastodons_public",
)
async def mastodon_read_all_public(hood=Depends(get_hood_unauthorized)):
    mastodonbots = await MastodonAccount.objects.filter(hood=hood).all()
    mbots = []
    for mbot in mastodonbots:
        if mbot.enabled == 1 and mbot.username:
            instance = await MastodonInstance.objects.get(id=mbot.instance)
            mbots.append(
                BodyMastodonPublic(username=mbot.username, instance=instance.name)
            )
    return mbots


@router.get(
    "/",
    # TODO response_model,
    operation_id="get_mastodons",
)
async def mastodon_read_all(hood=Depends(get_hood)):
    return await MastodonAccount.objects.filter(hood=hood).all()


@router.delete(
    "/{mastodon_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    # TODO response_model
    operation_id="delete_mastodon",
)
async def mastodon_delete(mastodon=Depends(get_mastodon)):
    spawner.stop(mastodon)
    await mastodon.delete()
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@router.get(
    "/{mastodon_id}/status",
    status_code=status.HTTP_200_OK,
    # TODO response_model
    operation_id="status_mastodon",
)
async def mastodon_status(mastodon=Depends(get_mastodon)):
    return {"status": spawner.get(mastodon).status.name}


@router.post(
    "/{mastodon_id}/start",
    status_code=status.HTTP_200_OK,
    # TODO response_model
    operation_id="start_mastodon",
)
async def mastodon_start(mastodon=Depends(get_mastodon)):
    await mastodon.update(enabled=True)
    spawner.get(mastodon).start()
    return {}


@router.post(
    "/{mastodon_id}/stop",
    status_code=status.HTTP_200_OK,
    # TODO response_model
    operation_id="stop_mastodon",
)
async def mastodon_stop(mastodon=Depends(get_mastodon)):
    await mastodon.update(enabled=False)
    spawner.get(mastodon).stop()
    return {}


@router.post(
    "/",
    status_code=status.HTTP_201_CREATED,
    # TODO response_model
    operation_id="create_mastodon",
)
async def mastodon_create(values: BodyMastodonAccount, hood=Depends(get_hood)):
    """Add a Mastodon Account to a Ticketfrei account.

    open questions:
    can the instance_url have different ways of writing?

    :param: values: a BodyMastodonAccount object in json
    :param: hood: the hood ORM object
    """
    try:
        instance = await get_mastodon_instance(values.instance_url)
    except MastodonNetworkError:
        raise HTTPException(422, "Invalid Mastodon Instance")
    account = Mastodon(
        instance.client_id, instance.client_secret, api_base_url=values.instance_url
    )
    try:
        access_token = await get_event_loop().run_in_executor(
            None, account.log_in, values.email, values.password
        )
        logger.debug(f"{access_token}")
        mastodon = await MastodonAccount.objects.create(
            hood=hood, instance=instance, access_token=access_token, enabled=True
        )
        spawner.start(mastodon)
        return mastodon
    except MastodonIllegalArgumentError:
        logger.warning("Login to Mastodon failed.", exc_info=True)
        raise HTTPException(status_code=status.HTTP_422_INVALID_INPUT)
    except IntegrityError:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT)