ephemere/python/cohort/fix_channel_permissions.py

#!/usr/bin/env python3
"""Fix Send Messages / Send Messages in Threads permissions for a Discord channel.

Denies Send Messages and Send Messages in Threads for both @everyone and the
@cohort role on the target channel. Update CHANNEL_ID and COHORT_ROLE_ID before
running.

Data files (place in data/):
  - None

Env vars:
  - DISCORD_BOT_TOKEN  Bot token for the Discord API
"""

import asyncio
import os

import aiohttp

DISCORD_BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
BASE_URL = "https://discord.com/api/v10"
GUILD_ID = "739845668582981683"
CHANNEL_ID = "1467964405646885237"

SEND_MESSAGES = 0x0000000000000800
SEND_MESSAGES_IN_THREADS = 0x0000004000000000

EVERYONE_ROLE_ID = GUILD_ID
COHORT_ROLE_ID = "1390925253102010521"


async def fix_permissions() -> None:
    """Fix the channel permissions."""
    headers = {
        "Authorization": f"Bot {DISCORD_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    async with aiohttp.ClientSession() as session:
        print("Fetching current channel permissions...")
        async with session.get(
            f"{BASE_URL}/channels/{CHANNEL_ID}", headers=headers
        ) as resp:
            if resp.status != 200:
                error = await resp.text()
                print(f"Error fetching channel: {resp.status} - {error}")
                return

            channel = await resp.json()
            print(f"Channel: {channel['name']}")

        permission_overwrites = channel.get("permission_overwrites", [])

        everyone_overwrite = None
        cohort_overwrite = None

        for overwrite in permission_overwrites:
            if overwrite["id"] == EVERYONE_ROLE_ID:
                everyone_overwrite = overwrite
            elif overwrite["id"] == COHORT_ROLE_ID:
                cohort_overwrite = overwrite

        print("\nFixing @everyone permissions...")
        if everyone_overwrite:
            current_deny = int(everyone_overwrite.get("deny", "0"))
            current_allow = int(everyone_overwrite.get("allow", "0"))

            new_deny = current_deny | SEND_MESSAGES | SEND_MESSAGES_IN_THREADS
            new_allow = current_allow & ~SEND_MESSAGES & ~SEND_MESSAGES_IN_THREADS

            payload = {
                "type": 0,
                "deny": str(new_deny),
                "allow": str(new_allow),
            }
        else:
            payload = {
                "type": 0,
                "deny": str(SEND_MESSAGES | SEND_MESSAGES_IN_THREADS),
                "allow": "0",
            }

        async with session.put(
            f"{BASE_URL}/channels/{CHANNEL_ID}/permissions/{EVERYONE_ROLE_ID}",
            headers=headers,
            json=payload,
        ) as resp:
            if resp.status == 204:
                print("✅ @everyone permissions fixed!")
            else:
                error = await resp.text()
                print(f"❌ Error fixing @everyone: {resp.status} - {error}")

        print("\nFixing @cohort permissions...")
        if cohort_overwrite:
            current_deny = int(cohort_overwrite.get("deny", "0"))
            current_allow = int(cohort_overwrite.get("allow", "0"))

            new_deny = current_deny | SEND_MESSAGES | SEND_MESSAGES_IN_THREADS
            new_allow = current_allow & ~SEND_MESSAGES & ~SEND_MESSAGES_IN_THREADS

            payload = {
                "type": 0,
                "deny": str(new_deny),
                "allow": str(new_allow),
            }
        else:
            payload = {
                "type": 0,
                "deny": str(SEND_MESSAGES | SEND_MESSAGES_IN_THREADS),
                "allow": "0",
            }

        async with session.put(
            f"{BASE_URL}/channels/{CHANNEL_ID}/permissions/{COHORT_ROLE_ID}",
            headers=headers,
            json=payload,
        ) as resp:
            if resp.status == 204:
                print("✅ @cohort permissions fixed!")
            else:
                error = await resp.text()
                print(f"❌ Error fixing @cohort: {resp.status} - {error}")

        print("\n✨ Done! Permissions have been fixed.")


if __name__ == "__main__":
    asyncio.run(fix_permissions())