foxhole/tasks.py

import os
import sys
import tomli_w
import asyncio

from pathlib import Path

from prompt_toolkit import prompt
from prompt_toolkit.key_binding import KeyBindings
from invoke import Context  # type: ignore
from invoke import run  # type: ignore
from invoke import task  # type: ignore

_kb = KeyBindings()


@_kb.add("c-@")
def _(event):
    """Save multi-line buffer on CTRL + space"""
    event.current_buffer.validate_and_handle()

@task
def config(ctx):
    from Crypto.PublicKey import RSA

    _ROOT_DIR = Path().parent.resolve()
    _KEY_PATH = _ROOT_DIR / "data" / "key.pem"

    if _KEY_PATH.exists():
        print("is existing!")
        sys.exit(2)
    else:
        k = RSA.generate(2048)
        privkey_pem = k.exportKey("PEM").decode("utf-8")
        _KEY_PATH.write_text(privkey_pem)

    config_file = Path("data/config.toml")

    if config_file.exists():
        print("is existing!")
        sys.exit(2)

    config_dict: dict = {}
    config_dict["domain"] = prompt("domain: ")
    config_dict["username"] = prompt("username: ")
    config_dict["name"] = prompt("name (e.g. John Doe): ",
                                 default=config_dict["username"])
    config_dict["summary"] = prompt("summary: ")
    config_dict["https"] = True
    proto = "https"
    yn = ""
    while yn not in ["y", "n"]:
        yn = prompt("will the site be served via https? (y/n): ", default="y").lower()
    if yn == "n":
        config_dict["https"] = False
        proto = "http"

    config_dict["icon_url"] = prompt(
        "icon URL: ", default=f'{proto}://{config_dict["domain"]}/static/nopic.png'
    )
    config_dict["secret"] = os.urandom(16).hex()

    with config_file.open("w") as f:
        f.write(tomli_w.dumps(config_dict))

    print("Done")
    sys.exit(0)


@task
def follow(ctx):
    from app.database import async_session
    from app.boxes import send_follow

    acct = prompt("enter want to follow actor id:")

    async def _dodo():
        async with async_session() as db_session: #type: ignore
            await send_follow(db_session,acct)

        print("Done!")


    asyncio.run(_dodo())


@task
def accept_follow(ctx):
    from app.database import async_session
    from sqlalchemy import select
    from sqlalchemy.orm import joinedload
    from app import models
    from app.boxes import _send_accept

    ingress_id = prompt("enter want to accept ingress id:")

    async def _do():
        async with async_session() as db_session: #type: ignore
            try:
                exist_request = (
                    await db_session.scalars(
                        select(models.IncomingActivity).where(
                            models.IncomingActivity.id == int(ingress_id)
                        )
                    )
                ).one_or_none()

                if not exist_request or exist_request.ap_object["type"] != "Follow":
                    raise ValueError("Non-valid id!")

                exist_inbox = (
                    await db_session.scalars(
                        select(models.InboxObject).where(
                            models.InboxObject.ap_id == exist_request.ap_id
                        ).options(
                            joinedload(models.InboxObject.actor)
                        )
                    )
                ).one_or_none()

                await _send_accept(
                    db_session,
                    exist_inbox.actor,
                    exist_inbox,
                )
                await db_session.delete(exist_request)
                await db_session.commit()
            except Exception as e:
                print(f"ERROR: Failed to {e}")
                return

        print("Done!")

    asyncio.run(_do())


@task
def send_note(ctx):
    from app.database import async_session
    from app.activitypub import VisibilityEnum
    from app.boxes import _send_create
    from app.orgpython import to_html


    content = prompt(
        (
            "note contents, in org mode, "
            "use [CTRL] + [SPACE] to submit:\n"
        ),
        key_bindings=_kb,
        multiline=True,
    )
    content = to_html(content).replace("\n", "")

    async def _dodo():
        async with async_session() as db_session: #type: ignore

            await _send_create(
                db_session,
                "Note",
                content,
                VisibilityEnum.PUBLIC
            )
        print("Done!")

    asyncio.run(_dodo())