Files
ephemere/python/cohort/assign_team_roles.py
T
naomi 6b5fa40599
CI / dependency-pin-check-typescript (push) Successful in 4s
CI / dependency-pin-check-python (push) Successful in 3s
CI / typescript (push) Successful in 9m38s
CI / python (push) Successful in 9m23s
Security Scan and Upload / Security & DefectDojo Upload (push) Successful in 1m6s
feat: add multi-lang support and cohort scripts (#1)
### 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_

Co-authored-by: Hikari <hikari@nhcarrigan.com>
Reviewed-on: #1
Co-authored-by: Naomi Carrigan <commits@nhcarrigan.com>
Co-committed-by: Naomi Carrigan <commits@nhcarrigan.com>
2026-01-23 20:07:16 -08:00

102 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""Assign team-specific roles to all 155 participants.
Respects Discord rate limits with proper backoff and retry logic.
"""
import json
import os
import time
import requests
BOT_TOKEN = os.environ["DISCORD_BOT_TOKEN"]
GUILD_ID = "692816967895220344"
BASE_URL = "https://discord.com/api/v10"
HEADERS = {"Authorization": f"Bot {BOT_TOKEN}", "Content-Length": "0"}
TEAM_ROLE_IDS = {
"Jade Jasmine": "1464314923780931677",
"Crimson Dahlia": "1464315093402784015",
"Rose Camellia": "1464315098452726106",
"Amber Wisteria": "1464315105264275600",
"Ivory Orchid": "1464315109873684593",
"Teal Iris": "1464315114378498152",
"Peach Gardenia": "1464315118904152107",
"Violet Carnation": "1464315124251754559",
"Azure Lotus": "1464315128437801177",
"Coral Sunflower": "1464315132896088168",
"Indigo Tulip": "1464315138428633241",
"Scarlet Hydrangea": "1464315142710890520",
"Mint Narcissus": "1464315149203804405",
"Sage Marigold": "1464315153599299803",
}
def assign_role_with_retry(user_id: str, role_id: str, max_retries: int = 5) -> bool:
url = f"{BASE_URL}/guilds/{GUILD_ID}/members/{user_id}/roles/{role_id}"
for attempt in range(max_retries):
response = requests.put(url, headers=HEADERS)
if response.status_code == 204:
return True
elif response.status_code == 429:
# Check headers first, fall back to JSON body
retry_after = response.headers.get("Retry-After")
if retry_after is None:
retry_after = response.headers.get("X-RateLimit-Reset-After")
if retry_after is None:
try:
retry_after = response.json().get("retry_after", 1)
except Exception:
retry_after = 1
retry_after = float(retry_after)
print(f" Rate limited! Waiting {retry_after:.2f}s before retry...")
time.sleep(retry_after)
else:
print(f" Error {response.status_code}: {response.text}")
backoff_time = (2**attempt) * 0.5
print(f" Retrying in {backoff_time:.2f}s...")
time.sleep(backoff_time)
return False
def main():
with open("team_assignments.json") as f:
teams = json.load(f)
print(f"Assigning team roles to {len(teams)} teams...")
print("-" * 50)
total_success = 0
total_fail = 0
for team in teams:
team_name = team["name"]
role_id = TEAM_ROLE_IDS[team_name]
all_members = team["leaders"] + team["participants"]
print(f"\n[{team_name}] Assigning role to {len(all_members)} members...")
for user_id in all_members:
print(f" {user_id}...", end=" ")
if assign_role_with_retry(user_id, role_id):
print("")
total_success += 1
else:
print("✗ FAILED")
total_fail += 1
# Small delay between requests to be nice to the API
time.sleep(0.1)
print("-" * 50)
print(f"Complete! Success: {total_success}, Failed: {total_fail}")
if __name__ == "__main__":
main()