Discord App (message command) to archive message to WriteFreely instance

README.md

This script creates a message command to archive discord messages to a WriteFreely instance.

Blog Post

• Create collections
• Create posts
• Change collection visibility

Star ⭐ this gist if it was useful. Follows to my GitHub Profile are appreciated.

---

TheOnlyWayUp © 2024

cog.py

class WriteFreelyArchive(commands.Cog):
    """WriteFreelyArchive Cog."""

    def __init__(self, bot: commands.Bot):
        """Initialize."""
        self.bot = bot

    @commands.message_command(name="Archive Post")
    async def archive_post(
        self, inter: disnake.MessageCommandInteraction, message: disnake.Message
    ):
        post_body = message.content
        if message.attachments:
            post_body += "\n\n*Attachments**:"
            for attachment in message.attachments:
                post_body += f"\n- {attachment.url}"
        post_body += f"\n\n*Source*: {message.jump_url}"

        collection = await get_or_create_writefreely_collection(
            message.author.id, message.author.name
        )
        post_id = await create_writefreely_collection_post(
            message.author.id,
            post_body,
            created_at=message.created_at,


        await message.add_reaction("🗃️")
        await inter.send(
            embed=disnake.Embed(
                title="Created Post", description=f"Post ID: `{post_id}`"
            ),
            ephemeral=True,
        )

writefreely.py

# --- WriteFreely Utils --- #

WRITEFREELY_ACCESS_TOKEN: str = ""


async def get_writefreely_access_token(break_cache: bool = False) -> str:
    """Log in to the configured WriteFreely instance and return an access token. This response is cached."""
    global WRITEFREELY_ACCESS_TOKEN

    if not break_cache and WRITEFREELY_ACCESS_TOKEN:
        return WRITEFREELY_ACCESS_TOKEN

    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://fa.rambhat.la/api/auth/login",
            json={
                "alias": writefreely_config["username"],
                "pass": writefreely_config["password"],
            },
            headers={"content-type": "application/json"},
        ) as response:
            auth_data = await response.json()

            WRITEFREELY_ACCESS_TOKEN = auth_data["data"][
                "access_token"
            ]  # No error handling. Live by the Sword. Die by the Sword. lol

    return WRITEFREELY_ACCESS_TOKEN


async def get_writefreely_headers() -> dict:
    """Return WriteFreely Headers to be used on API Requests."""

    access_token = await get_writefreely_access_token()

    return {"Authorization": f"Token {access_token}"}


async def get_or_create_writefreely_collection(collection_id: int, title: str) -> dict:
    """Find or Create a WriteFreely Collection provided the collection ID ('alias') and title. The title is used only during creation."""

    async with aiohttp.ClientSession(
        headers=await get_writefreely_headers()
    ) as session:
        async with session.get(
            "https://fa.rambhat.la/api/collections/{collection_id}",
            headers={"content-type": "application/json"},
        ) as response:
            try:
                data = await response.json()
                if data["code"] == 200:
                    return data
            except aiohttp.ContentTypeError:
                pass

        async with session.post(
            "https://fa.rambhat.la/api/collections",
            json={"alias": str(collection_id), "title": title},
            headers={"content-type": "application/json"},
        ) as response:
            data = await response.json()

        return data


async def update_writefreely_collection_visibility(
    collection_id: int, visible: bool
) -> int:
    """Update a Collection's Visibility attribute."""

    async with aiohttp.ClientSession(
        headers=await get_writefreely_headers()
    ) as session:
        async with session.post(
            "https://fa.rambhat.la/api/collections/{collection_id}",
            data={
                "visibility": ("1" if visible else "2")
            },  # 1: Public, 2: Private. Ref: https://developers.write.as/docs/api/?shell=#update-a-collection:~:text=Valid%20visibility%20values
            headers={
                "content-type": "application/x-www-form-urlencoded",
                "Cookie": "...",
            },
        ) as response:
            response.raise_for_status()
            return response.status


async def create_writefreely_collection_post(
    collection_id: int,
    body: str,
    title: Optional[str] = None,
    created_at: Optional[datetime] = None,
):

    async with aiohttp.ClientSession(
        headers=await get_writefreely_headers()
    ) as session:
        async with session.post(
            f"https://fa.rambhat.la/api/collections/{collection_id}/posts",
            json={
                "body": body,
                "title": title or "",
                "created": (created_at or datetime.now())
                .astimezone(pytz.utc)
                .strftime("%Y-%m-%dT%H:%M:%SZ"),
            },
            headers={"content-type": "application/json"},
        ) as response:
            data = await response.json()
            return data["data"]["id"]