foxhole/tasks.py

import os
import sys
import tomli_w
import asyncio

from pathlib import Path

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

@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
            try:
                await send_follow(db_session,acct)
            except Exception as e:
                print(f"ERROR: Failed to {e}")
                return

        print("Done!")


    asyncio.run(_dodo())