feat: add remaining cohort scripts (#4)
CI / dependency-pin-check-typescript (push) Successful in 5s
CI / dependency-pin-check-python (push) Successful in 4s
Security Scan and Upload / Security & DefectDojo Upload (push) Successful in 54s
CI / python (push) Successful in 9m24s
CI / typescript (push) Successful in 9m41s

### Explanation

_No response_

### Issue

_No response_

### Attestations

- [ ] I have read and agree to the [Code of Conduct](https://docs.nhcarrigan.com/community/coc/)
- [ ] I have read and agree to the [Community Guidelines](https://docs.nhcarrigan.com/community/guide/).
- [ ] My contribution complies with the [Contributor Covenant](https://docs.nhcarrigan.com/dev/covenant/).

### Dependencies

- [ ] I have pinned the dependencies to a specific patch version.

### Style

- [ ] I have run the linter and resolved any errors.
- [ ] My pull request uses an appropriate title, matching the conventional commit standards.
- [ ] My scope of feat/fix/chore/etc. correctly matches the nature of changes in my pull request.

### Tests

- [ ] My contribution adds new code, and I have added tests to cover it.
- [ ] My contribution modifies existing code, and I have updated the tests to reflect these changes.
- [ ] All new and existing tests pass locally with my changes.
- [ ] Code coverage remains at or above the configured threshold.

### Documentation

_No response_

### Versioning

_No response_

Reviewed-on: #4
Co-authored-by: Naomi Carrigan <commits@nhcarrigan.com>
Co-committed-by: Naomi Carrigan <commits@nhcarrigan.com>
This commit was merged in pull request #4.
This commit is contained in:
2026-02-02 12:36:44 -08:00
committed by Naomi Carrigan
parent 6169eb4577
commit f5e8deca59
5 changed files with 694 additions and 1 deletions
+133
View File
@@ -0,0 +1,133 @@
#!/usr/bin/env python3
"""Add GitHub users to their appropriate teams in nhcarrigan-spring-2026-cohort org"""
import json
import subprocess
import time
# Load team assignments and Discord to GitHub mappings
with open("team_assignments.json") as f:
teams = json.load(f)
with open("discord_to_github.json") as f:
discord_to_github = json.load(f)
# Map team names to GitHub team slugs
team_name_to_slug = {
"Jade Jasmine": "jade-jasmine",
"Crimson Dahlia": "crimson-dahlia",
"Rose Camellia": "rose-camellia",
"Amber Wisteria": "amber-wisteria",
"Ivory Orchid": "ivory-orchid",
"Teal Iris": "teal-iris",
"Peach Gardenia": "peach-gardenia",
"Violet Carnation": "violet-carnation",
"Azure Lotus": "azure-lotus",
"Coral Sunflower": "coral-sunflower",
"Indigo Tulip": "indigo-tulip",
"Scarlet Hydrangea": "scarlet-hydrangea",
"Mint Narcissus": "mint-narcissus",
"Sage Marigold": "sage-marigold",
}
org = "nhcarrigan-spring-2026-cohort"
total_added = 0
total_skipped = 0
total_errors = 0
def add_user_to_team(username, team_slug, role="member"):
"""Add a user to a GitHub team"""
try:
# Check if user is already a member
check_cmd = (
f"gh api orgs/{org}/teams/{team_slug}/memberships/{username} 2>/dev/null"
)
result = subprocess.run(
check_cmd, shell=True, capture_output=True, text=True, check=False
)
if result.returncode == 0:
print(f"{username} is already in {team_slug}")
return "already_member"
# Add user to team
add_cmd = (
f"gh api -X PUT orgs/{org}/teams/{team_slug}/memberships/{username} "
f"-f role={role}"
)
result = subprocess.run(
add_cmd, shell=True, capture_output=True, text=True, check=False
)
if result.returncode == 0:
print(f" ✓ Added {username} to {team_slug} as {role}")
return "added"
else:
print(f" ✗ Failed to add {username} to {team_slug}: {result.stderr}")
return "error"
except Exception as e:
print(f" ✗ Error adding {username} to {team_slug}: {str(e)}")
return "error"
# Process each team
for team_data in teams:
team_name = team_data["name"]
team_slug = team_name_to_slug[team_name]
print(f"\n{'=' * 60}")
print(f"Processing Team {team_data['team_id']}: {team_name}")
print(f"{'=' * 60}")
# Add leaders to leaders team
leaders_team_slug = f"{team_slug}-leaders"
print(f"\nAdding leaders to {leaders_team_slug}:")
for discord_id in team_data["leaders"]:
github_username = discord_to_github.get(discord_id)
if not github_username or github_username == "nhcarrigan-2025-hackathon":
print(
f" ⚠ Skipping Discord ID {discord_id} - "
"Missing/invalid GitHub username"
)
total_skipped += 1
continue
result = add_user_to_team(github_username, leaders_team_slug, "member")
if result == "added":
total_added += 1
elif result == "error":
total_errors += 1
# Rate limiting
time.sleep(0.5)
# Add participants to main team
print(f"\nAdding participants to {team_slug}:")
for discord_id in team_data["participants"]:
github_username = discord_to_github.get(discord_id)
if not github_username or github_username == "nhcarrigan-2025-hackathon":
print(
f" ⚠ Skipping Discord ID {discord_id} - "
"Missing/invalid GitHub username"
)
total_skipped += 1
continue
result = add_user_to_team(github_username, team_slug, "member")
if result == "added":
total_added += 1
elif result == "error":
total_errors += 1
# Rate limiting
time.sleep(0.5)
print(f"\n{'=' * 60}")
print("Summary:")
print(f"- Total users added: {total_added}")
print(f"- Total users skipped (missing GitHub): {total_skipped}")
print(f"- Total errors: {total_errors}")
print(f"{'=' * 60}")
+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""Create private voice channels for each team in the specified category.
Each channel will be visible and joinable only by the team's role.
"""
import os
import time
import requests
# Discord configuration
BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
GUILD_ID = "692816967895220344"
CATEGORY_ID = "1464311813620502638"
BASE_URL = "https://discord.com/api/v10"
# Team role IDs from send_team_messages.py
TEAMS = {
"Jade Jasmine": {"role_id": "1464314923780931677"},
"Crimson Dahlia": {"role_id": "1464315093402784015"},
"Rose Camellia": {"role_id": "1464315098452726106"},
"Amber Wisteria": {"role_id": "1464315105264275600"},
"Ivory Orchid": {"role_id": "1464315109873684593"},
"Teal Iris": {"role_id": "1464315114378498152"},
"Peach Gardenia": {"role_id": "1464315118904152107"},
"Violet Carnation": {"role_id": "1464315124251754559"},
"Azure Lotus": {"role_id": "1464315128437801177"},
"Coral Sunflower": {"role_id": "1464315132896088168"},
"Indigo Tulip": {"role_id": "1464315138428633241"},
"Scarlet Hydrangea": {"role_id": "1464315142710890520"},
"Mint Narcissus": {"role_id": "1464315149203804405"},
"Sage Marigold": {"role_id": "1464315153599299803"},
}
HEADERS = {"Authorization": f"Bot {BOT_TOKEN}", "Content-Type": "application/json"}
def create_voice_channel(team_name, role_id):
"""Create a voice channel with specific permissions"""
url = f"{BASE_URL}/guilds/{GUILD_ID}/channels"
# Permission overwrites:
# - Deny @everyone from viewing and connecting
# - Allow the team role to view and connect
permission_overwrites = [
{
"id": GUILD_ID, # @everyone role ID is same as guild ID
"type": 0, # Role type
"deny": "1049600", # VIEW_CHANNEL (1 << 10) + CONNECT (1 << 20) = 1049600
"allow": "0",
},
{
"id": role_id, # Team role
"type": 0, # Role type
"allow": "1049600", # VIEW_CHANNEL + CONNECT
"deny": "0",
},
]
data = {
"name": f"{team_name} Voice",
"type": 2, # Voice channel
"parent_id": CATEGORY_ID,
"permission_overwrites": permission_overwrites,
"user_limit": 0, # No user limit
"bitrate": 64000, # 64 kbps
"rtc_region": None, # Auto-select region
}
response = requests.post(url, headers=HEADERS, json=data)
if response.status_code == 201:
channel_data = response.json()
print(f"✓ Created voice channel for {team_name} (ID: {channel_data['id']})")
return True
elif response.status_code == 429:
retry_after = float(response.headers.get("Retry-After", 1))
print(f" Rate limited! Waiting {retry_after:.2f}s...")
time.sleep(retry_after)
return create_voice_channel(team_name, role_id)
else:
print(
f"✗ Failed to create channel for {team_name}: "
f"{response.status_code} - {response.text}"
)
return False
def main():
print(f"Creating private voice channels for {len(TEAMS)} teams...")
print(f"Category ID: {CATEGORY_ID}")
print("-" * 50)
success_count = 0
fail_count = 0
created_channels = []
for team_name, team_data in TEAMS.items():
if create_voice_channel(team_name, team_data["role_id"]):
success_count += 1
created_channels.append(team_name)
else:
fail_count += 1
# Small delay between requests
time.sleep(0.1)
print("-" * 50)
print(f"Complete! Success: {success_count}, Failed: {fail_count}")
if success_count == len(TEAMS):
print("\n✅ All team voice channels have been created!")
print("\nEach voice channel:")
print(" - Is named '[Team Name] Voice'")
print(" - Is only visible to members with that team's role")
print(" - Can only be joined by members with that team's role")
print(" - Has no user limit")
print(" - Uses 64 kbps bitrate")
if __name__ == "__main__":
main()
+251
View File
@@ -0,0 +1,251 @@
#!/usr/bin/env python3
"""Discord Team Activity Checker
Checks for team members who haven't sent messages in their channels within 36 hours
"""
import asyncio
import json
import os
import sys
from datetime import datetime, timedelta, timezone
import aiohttp
# Configuration
DISCORD_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
DISCORD_API_BASE = "https://discord.com/api/v10"
INACTIVE_THRESHOLD_HOURS = 36
# Load team assignments from file
with open("team_assignments.json") as f:
team_data = json.load(f)
# Build TEAMS dictionary with channel IDs and member lists
TEAMS = {}
CHANNEL_IDS = {
"Jade Jasmine": "1464316501573107886",
"Crimson Dahlia": "1464316744909852682",
"Rose Camellia": "1464316751268286611",
"Amber Wisteria": "1464316761410113641",
"Ivory Orchid": "1464316770889240730",
"Teal Iris": "1464316776459407448",
"Peach Gardenia": "1464316785040953543",
"Violet Carnation": "1464316805261824032",
"Azure Lotus": "1464316814455472139",
"Coral Sunflower": "1464316819711066263",
"Indigo Tulip": "1464316826384072925",
"Scarlet Hydrangea": "1464316839306985506",
"Mint Narcissus": "1464316844251807952",
"Sage Marigold": "1464316850669093040",
}
for team in team_data:
team_name = team["name"]
if team_name in CHANNEL_IDS:
TEAMS[team_name] = {
"channel_id": CHANNEL_IDS[team_name],
"member_ids": team["leaders"] + team["participants"],
}
class DiscordActivityChecker:
def __init__(self, token: str):
self.token = token
self.headers = {
"Authorization": f"Bot {token}",
"Content-Type": "application/json",
}
self.session = None
self.cutoff_time = datetime.now(timezone.utc) - timedelta(
hours=INACTIVE_THRESHOLD_HOURS
)
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def get_user_info(self, user_id: str) -> dict:
"""Get information about a specific user"""
url = f"{DISCORD_API_BASE}/users/{user_id}"
async with self.session.get(url, headers=self.headers) as response:
if response.status == 429:
# Rate limited, wait a bit
await asyncio.sleep(1)
return None
elif response.status != 200:
print(f"Failed to get user {user_id}: {response.status}")
return None
user_data = await response.json()
return user_data
async def get_recent_message_authors(self, channel_id: str) -> set[str]:
"""Get user IDs of everyone who sent a message in the last 36 hours"""
url = f"{DISCORD_API_BASE}/channels/{channel_id}/messages?limit=100"
active_users = set()
async with self.session.get(url, headers=self.headers) as response:
if response.status != 200:
print(
f"Failed to get messages for channel {channel_id}: "
f"{response.status}"
)
return active_users
messages = await response.json()
for message in messages:
# Parse message timestamp
timestamp = datetime.fromisoformat(
message["timestamp"].replace("Z", "+00:00")
)
# If message is within our threshold, add the author
if timestamp > self.cutoff_time:
active_users.add(message["author"]["id"])
else:
# Messages are returned newest first, so we can break early
break
return active_users
async def send_message_to_channel(self, channel_id: str, content: str) -> bool:
"""Send a message to a Discord channel"""
url = f"{DISCORD_API_BASE}/channels/{channel_id}/messages"
data = {"content": content}
async with self.session.post(url, headers=self.headers, json=data) as response:
if response.status == 200:
print(f"✅ Message sent to channel {channel_id}")
return True
else:
print(
f"❌ Failed to send message to channel {channel_id}: "
f"{response.status}"
)
return False
async def check_team_activity(
self, team_name: str, channel_id: str, member_ids: list[str]
) -> dict:
"""Check activity for a specific team"""
print(f"Checking {team_name}...")
# Get active users in the channel
active_users = await self.get_recent_message_authors(channel_id)
# Find inactive members
inactive_members = []
total_members = len(member_ids)
for member_id in member_ids:
if member_id not in active_users:
# Get user information
user_info = await self.get_user_info(member_id)
if user_info:
# Skip bots
if user_info.get("bot", False):
total_members -= 1
continue
inactive_members.append(
{"id": member_id, "mention": f"<@{member_id}>"}
)
else:
# If we can't get user info, still track them as inactive
inactive_members.append(
{"id": member_id, "mention": f"<@{member_id}>"}
)
return {
"team": team_name,
"total_members": total_members,
"inactive_members": inactive_members,
"inactive_count": len(inactive_members),
}
async def main():
"""Main function to check all teams"""
print(
f"Discord Activity Checker - Checking for inactivity over "
f"{INACTIVE_THRESHOLD_HOURS} hours"
)
cutoff = datetime.now(timezone.utc) - timedelta(hours=INACTIVE_THRESHOLD_HOURS)
print(f"Cutoff time: {cutoff}")
print("-" * 80)
async with DiscordActivityChecker(DISCORD_TOKEN) as checker:
results = []
for team_name, team_info in TEAMS.items():
channel_id = team_info["channel_id"]
member_ids = team_info["member_ids"]
result = await checker.check_team_activity(
team_name, channel_id, member_ids
)
results.append(result)
# Display results
print("\n" + "=" * 80)
print("INACTIVE MEMBERS REPORT")
print("=" * 80)
# Check if user wants to send messages (via command line arg)
send_messages = "--send" in sys.argv
for team_result in results:
print(f"\n📋 {team_result['team']}")
print(f" Total Members: {team_result['total_members']}")
print(f" Inactive: {team_result['inactive_count']}")
if team_result["inactive_members"]:
print(" Inactive Members:")
for member in team_result["inactive_members"]:
print(f" - {member['mention']}")
# Send message to team channel if requested
if send_messages and team_result["inactive_count"] > 0:
channel_id = TEAMS[team_result["team"]]["channel_id"]
# Build message
mentions = "\n".join(
[m["mention"] for m in team_result["inactive_members"]]
)
message = (
"Good morning, the following people have not sent a "
"message here in the last 36 hours. If you are on this "
"list, please confirm you are still participating.\n\n"
f"{mentions}"
)
await checker.send_message_to_channel(channel_id, message)
await asyncio.sleep(1) # Small delay between messages
else:
print(" ✅ All members are active!")
print("\n" + "=" * 80)
# Save results to JSON
with open("discord_activity_report.json", "w") as f:
json.dump(
{
"generated_at": datetime.now(timezone.utc).isoformat(),
"threshold_hours": INACTIVE_THRESHOLD_HOURS,
"results": results,
},
f,
indent=2,
)
print("\n📄 Detailed report saved to discord_activity_report.json")
if __name__ == "__main__":
asyncio.run(main())
@@ -0,0 +1,183 @@
#!/usr/bin/env python3
"""Update Cohort Leads role permissions to allow pinging in team channels."""
import os
import time
import requests
# Discord configuration
BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
GUILD_ID = "692816967895220344"
BASE_URL = "https://discord.com/api/v10"
# Team channel IDs from send_team_messages.py
TEAM_CHANNELS = {
"Jade Jasmine": {
"channel_id": "1464316501573107886",
"role_id": "1464314923780931677",
},
"Crimson Dahlia": {
"channel_id": "1464316744909852682",
"role_id": "1464315093402784015",
},
"Rose Camellia": {
"channel_id": "1464316751268286611",
"role_id": "1464315098452726106",
},
"Amber Wisteria": {
"channel_id": "1464316761410113641",
"role_id": "1464315105264275600",
},
"Ivory Orchid": {
"channel_id": "1464316770889240730",
"role_id": "1464315109873684593",
},
"Teal Iris": {
"channel_id": "1464316776459407448",
"role_id": "1464315114378498152",
},
"Peach Gardenia": {
"channel_id": "1464316785040953543",
"role_id": "1464315118904152107",
},
"Violet Carnation": {
"channel_id": "1464316805261824032",
"role_id": "1464315124251754559",
},
"Azure Lotus": {
"channel_id": "1464316814455472139",
"role_id": "1464315128437801177",
},
"Coral Sunflower": {
"channel_id": "1464316819711066263",
"role_id": "1464315132896088168",
},
"Indigo Tulip": {
"channel_id": "1464316826384072925",
"role_id": "1464315138428633241",
},
"Scarlet Hydrangea": {
"channel_id": "1464316839306985506",
"role_id": "1464315142710890520",
},
"Mint Narcissus": {
"channel_id": "1464316844251807952",
"role_id": "1464315149203804405",
},
"Sage Marigold": {
"channel_id": "1464316850669093040",
"role_id": "1464315153599299803",
},
}
HEADERS = {"Authorization": f"Bot {BOT_TOKEN}", "Content-Type": "application/json"}
def get_guild_roles():
"""Get all roles in the guild to find Cohort Leads"""
url = f"{BASE_URL}/guilds/{GUILD_ID}/roles"
response = requests.get(url, headers=HEADERS)
if response.status_code == 200:
return response.json()
else:
print(f"Error getting roles: {response.status_code} - {response.text}")
return None
def find_cohort_leads_role(roles):
"""Find the Cohort Leads role from the list"""
for role in roles:
if "cohort" in role["name"].lower() and "lead" in role["name"].lower():
return role
return None
def update_channel_permissions(channel_id, role_id, team_name):
"""Update channel permissions for a specific role"""
url = f"{BASE_URL}/channels/{channel_id}/permissions/{role_id}"
# Permission bits:
# MENTION_EVERYONE = 1 << 17 = 131072
# PIN_MESSAGES = 1 << 51 = 2251799813685248
# Combined: 131072 + 2251799813685248 = 2251799813816320
data = {
"allow": "2251799813816320", # MENTION_EVERYONE + PIN_MESSAGES permissions
"deny": "0",
"type": 0, # Role permission type
}
response = requests.put(url, headers=HEADERS, json=data)
if response.status_code == 204:
print(f"✓ Updated permissions for {team_name} channel")
return True
elif response.status_code == 429:
retry_after = float(response.headers.get("Retry-After", 1))
print(f" Rate limited! Waiting {retry_after:.2f}s...")
time.sleep(retry_after)
return update_channel_permissions(channel_id, role_id, team_name)
else:
print(
f"✗ Failed to update {team_name} channel: "
f"{response.status_code} - {response.text}"
)
return False
def main():
print("Fetching guild roles...")
roles = get_guild_roles()
if not roles:
print("Failed to fetch roles!")
return
cohort_leads_role = find_cohort_leads_role(roles)
if not cohort_leads_role:
print("Could not find Cohort Leads role!")
print("\nAvailable roles:")
for role in roles:
print(f" - {role['name']} (ID: {role['id']})")
return
print(
f"\nFound Cohort Leads role: {cohort_leads_role['name']} "
f"(ID: {cohort_leads_role['id']})"
)
print(f"Updating permissions for {len(TEAM_CHANNELS)} team channels...")
print("-" * 50)
success_count = 0
fail_count = 0
for team_name, team_data in TEAM_CHANNELS.items():
if update_channel_permissions(
team_data["channel_id"], cohort_leads_role["id"], team_name
):
success_count += 1
else:
fail_count += 1
# Small delay between requests
time.sleep(0.1)
print("-" * 50)
print(f"Complete! Success: {success_count}, Failed: {fail_count}")
if success_count == len(TEAM_CHANNELS):
print("\n✅ All team channels have been updated!")
print("Cohort Leads can now:")
print(" - Use @everyone, @here, and mention all roles")
print(" - Pin and unpin messages")
print(
"\nNote: They cannot view these channels unless they have "
"the specific team role."
)
if __name__ == "__main__":
main()
+4
View File
@@ -1,2 +1,6 @@
# Development dependencies # Development dependencies
ruff==0.14.14 ruff==0.14.14
# Runtime dependencies
requests==2.32.3
aiohttp==3.11.12