1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
from .db import create_deployment, get_daily_user_deployment_count, get_deployment, update_deployment_status, get_active_users, get_user, add_deployment_message, get_deployment_messages
from slack_sdk.web.async_client import AsyncWebClient
import os
from dotenv import load_dotenv
from google import genai
from google.genai import types
import json
DAILY_LIMIT = 2
load_dotenv()
selfbot_token = os.environ.get("SLACK_SELFBOT_OAUTH_XOXP")
gemini_key = os.environ.get("GEMINI_API_KEY")
system_prompt = """
You are a strict content moderator for a professional Slack workspace called Hack Club.
Your job is to analyze messages and determine if they are SAFE or UNSAFE to be posted by a bot on behalf of multiple users.
A message is UNSAFE if it contains ANY of the following:
- Swear words, profanity, or offensive language
- Harassment, bullying, or personal attacks
- Hate speech or discrimination (race, gender, sexuality, religion, etc.)
- NSFW, sexual, or inappropriate content
- Spam, excessive gibberish, or repetitive messages
- Phishing, scams, or malicious links
- Mentions of illegal activities (hacking, drugs, violence, etc.)
- Anything that would make Slack admins (FD/Fire Dept.) upset or require them to intervene
A message is SAFE if it is:
- Friendly, positive, or neutral
- On-topic for a tech/developer community
- A harmless joke or meme (as long as it's not offensive)
- Helpful, constructive, or collaborative
- Literally just "chicken nugget" or similar harmless nonsense
Respond with JSON only. Do not add any other text.
Output format:
{'safe': true/false, 'reasoning': '<short reasoning here>'}
"""
async def check_ai_safety(message: str):
try:
client = genai.Client(api_key=gemini_key)
response = await client.aio.models.generate_content(
model="gemini-3.1-flash-lite",
contents=f"Message to analyse: {message}",
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json"
)
)
resp_json = json.loads(response.text)
return resp_json.get("safe", False), resp_json.get("reasoning", "No reasoning provided")
except Exception as e:
print(f"ai safety check might be cooked :fear: {e}")
return False, f"AI safety check failed, please try again later.\nError: {e}"
async def deploy_message(user_id: str, message: str):
user = await get_user(user_id)
if not user:
return {
"success": False,
"error": "You must be part of the botnet to deploy messages!"
}
daily_count = await get_daily_user_deployment_count(user_id)
if daily_count >= DAILY_LIMIT and user_id not in ["U0AKCBZHHMH"]:
return {
"success": False,
"error": f"You've reached your daily limit of {DAILY_LIMIT} messages!"
}
safe, reasoning = await check_ai_safety(message)
if not safe:
return {
"success": False,
"error": f"Message flagged by AI: {reasoning}"
}
deployment_id = await create_deployment(user_id, message)
print(f"DEBUG: Executing the deployment {deployment_id} on arrival - REMOVE THIS LATER!")
await execute_deployment(deployment_id)
return {
"success": True,
"deployment_id": deployment_id,
"message": "Your message has been deployed!"
}
async def execute_deployment(deployment_id: int):
deployment = await get_deployment(deployment_id)
if not deployment:
return
users = await get_active_users()
message = deployment["message"]
for user in users:
try:
client = AsyncWebClient(token=user["oauth_token"])
response = await client.chat_postMessage(
channel="C0BGMV1CAQG", # put it in #botnet for now, make it detect recent thread with CM approval yada yada later
text=message,
as_user=True
)
await add_deployment_message(deployment_id=deployment_id, user_id=user["slack_id"], channel_id="C0BGMV1CAQG", message_id=response["ts"])
except Exception as e:
print(f"Failed to send #{deployment_id} on behalf of {user['slack_id']}: {e}")
try:
client = AsyncWebClient(token=selfbot_token)
await client.chat_postMessage(
channel="C0BGMV1CAQG",
text=f"Deployment #{deployment_id} deployed.\n\nMessage: {message}",
as_user=True
)
except Exception as e:
print(f"Failed to send deployment status message: {e}")
await update_deployment_status(deployment_id, "deployed")
|
