pubspace/pubspace.py

import logging
from os import environ

import owncloud
from fastapi import FastAPI, Request
from mastodon import Mastodon, StreamListener

MASTODON_ACCESS_TOKEN = environ.get("MASTODON_ACCESS_TOKEN")
MASTODON_API_BASE_URL = environ.get("MASTODON_API_BASE_URL")

NEXTCLOUD_API_BASE_URL = environ.get("NEXTCLOUD_API_BASE_URL")
NEXTCLOUD_USER = environ.get("NEXTCLOUD_USER")
NEXTCLOUD_APP_PASSWORD = environ.get("NEXTCLOUD_APP_PASSWORD")

APP_LOG_LEVEL = environ.get("APP_LOG_LEVEL")
if APP_LOG_LEVEL == "info":
    APP_LOG_LEVEL = logging.INFO
elif APP_LOG_LEVEL == "debug":
    APP_LOG_LEVEL = logging.DEBUG
else:
    APP_LOG_LEVEL = logging.INFO

mastodon = Mastodon(
    access_token=MASTODON_ACCESS_TOKEN,
    api_base_url=f"https://{MASTODON_API_BASE_URL}",
)
nextcloud = owncloud.Client(f"https://{NEXTCLOUD_API_BASE_URL}")

log = logging.getLogger("uvicorn")
log.setLevel(APP_LOG_LEVEL)

app = FastAPI()
app.state.log = log


class PubspaceListener(StreamListener):
    def on_update(self, status):
        log.info(status)


mastodon.stream_hashtag("pubspace", PubspaceListener(), run_async=True)
nextcloud.login(NEXTCLOUD_USER, NEXTCLOUD_APP_PASSWORD)


def create_share(fpath):
    if not nextcloud.is_shared(fpath):
        info = nextcloud.share_file_with_link(fpath)
        return info.get_link()
    return None


@app.get("/")
async def home(request: Request):
    try:
        payload = await request.json()
        file = payload["file"]
        link = create_share(file)
        if link:
            app.state.log(f"Shared {file} on {link}")
        else:
            app.state.log(f"{file} has already been shared!")
    except Exception:
        return {}


@app.get("/healthz")
async def healthz(request: Request):
    return {"detail": "ALL ENGINES FIRING"}