generated from nhcarrigan/template
naomi
a40188413a
All Python cohort scripts now use DATA_DIR = Path(__file__).parent.parent.parent / "data" to correctly resolve the repo-root data/ directory regardless of the working directory set by run.sh. All TypeScript scripts have expanded JSDoc headers documenting data file requirements and environment variables.
266 lines
9.1 KiB
Python
Executable File
266 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Check for team members who have not sent a message in their channel within 36 hours.
|
|
|
|
Scans each team's Discord channel and flags members with no recent activity.
|
|
Optionally sends a direct mention message to inactive members.
|
|
|
|
Data files (place in data/):
|
|
- team_assignments.json Team rosters with leaders and participants per team
|
|
|
|
Outputs (written to data/):
|
|
- discord_activity_report.json Inactive members per team with last-seen timestamps
|
|
|
|
Env vars:
|
|
- DISCORD_BOT_TOKEN Bot token for the Discord API
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import aiohttp
|
|
|
|
DATA_DIR = Path(__file__).parent.parent.parent / "data"
|
|
|
|
# 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(DATA_DIR / "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(DATA_DIR / "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())
|