generated from nhcarrigan/template
naomi
4fdb5d06f1
- Port 19 cohort scripts from /home/naomi/docs/cohort/ - Replace all hardcoded tokens and dotenv usage with os.environ - Add pandas==3.0.1 dependency - Add E501 to ruff ignore list for Discord message string content - Make remove_resigned_members.py reusable (empty RESIGNED_IDS constant) - Make update_roster_messages.py reusable (iterates all teams from JSON) - Exclude 12 one-off/event-specific scripts as non-reusable
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Get all members who currently have the Cohort role."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
|
|
import aiohttp
|
|
|
|
DISCORD_BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
|
|
GUILD_ID = "692816967895220344"
|
|
COHORT_ROLE_ID = "1464314780935258112"
|
|
|
|
|
|
async def get_all_members_with_role(
|
|
session: aiohttp.ClientSession,
|
|
) -> list[dict[str, str]]:
|
|
"""Get all members who have the Cohort role."""
|
|
members = []
|
|
headers = {"Authorization": f"Bot {DISCORD_BOT_TOKEN}"}
|
|
|
|
after = None
|
|
|
|
while True:
|
|
url = f"https://discord.com/api/v10/guilds/{GUILD_ID}/members"
|
|
params: dict[str, str | int] = {"limit": 1000}
|
|
if after:
|
|
params["after"] = after
|
|
|
|
async with session.get(url, headers=headers, params=params) as resp:
|
|
if resp.status != 200:
|
|
error_text = await resp.text()
|
|
raise RuntimeError(
|
|
f"Failed to fetch members: {resp.status} - {error_text}"
|
|
)
|
|
|
|
data = await resp.json()
|
|
|
|
if not data:
|
|
break
|
|
|
|
for member in data:
|
|
if COHORT_ROLE_ID in member.get("roles", []):
|
|
members.append(
|
|
{
|
|
"id": member["user"]["id"],
|
|
"username": member["user"]["username"],
|
|
"display_name": member.get("nick")
|
|
or member["user"]["username"],
|
|
}
|
|
)
|
|
|
|
if len(data) < 1000:
|
|
break
|
|
|
|
after = data[-1]["user"]["id"]
|
|
|
|
return members
|
|
|
|
|
|
async def main() -> None:
|
|
"""Get all cohort members with the Cohort role."""
|
|
async with aiohttp.ClientSession() as session:
|
|
print("Fetching all members with Cohort role...")
|
|
|
|
cohort_members = await get_all_members_with_role(session)
|
|
|
|
print(f"\n✨ Found {len(cohort_members)} members with the Cohort role:\n")
|
|
|
|
for i, member in enumerate(cohort_members, 1):
|
|
print(
|
|
f"{i}. {member['display_name']} (@{member['username']}) - ID: {member['id']}" # noqa: E501
|
|
)
|
|
|
|
with open("active_cohort_members.json", "w") as f:
|
|
json.dump(cohort_members, f, indent=2)
|
|
|
|
print("\nSaved to active_cohort_members.json")
|
|
|
|
print("\nMember IDs only:")
|
|
print(json.dumps([m["id"] for m in cohort_members], indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|