feat: add remaining cohort scripts

python/cohort/discord_activity_checker.py

@@ -0,0 +1,223 @@
+#!/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 aiohttp
+from datetime import datetime, timezone, timedelta
+from typing import Dict, List, Set
+import json
+import time
+import sys
+import os
+
+# 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", "r") 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}: {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}: {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 {INACTIVE_THRESHOLD_HOURS} hours")
+    print(f"Cutoff time: {datetime.now(timezone.utc) - timedelta(hours=INACTIVE_THRESHOLD_HOURS)}")
+    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 = f"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{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())