166 lines
4.4 KiB
Python
166 lines
4.4 KiB
Python
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())
|