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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
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 .utils import generate_voting_blocks, get_random_channel
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 botnet-like social experiment in 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 in the Slack.
Messages will be posted on behalf of MANY USERS simultaneously.
The message must make sense coming from ANYONE, not just the submitter.
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.)
- Self-promotion, advertising, or promotional content
- Links to personal websites, products, or services
- Strong opinions or polarizing statements (politics, religion, software, licensing, etc.)
- Tech flame wars (X is better than Y, Z sucks, etc.)
- Negative statements about people, projects, or technologies
- Complaints or rants about specific tools, languages, or frameworks
- Anything that could be interpreted as speaking for others in a controversial way
- 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)
blocks = generate_voting_blocks(deployment_id, message, 0, 0, True)
client = AsyncWebClient(token=selfbot_token)
response = await client.chat_postMessage(
channel="C0BFEDTUFCP",
blocks=blocks,
text=f"Vote on Deployment #deployment_id"
)
message_ts = response["ts"]
# 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 submitted for voting!"
}
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"])
channel = await get_random_channel(user["oauth_token"])
if not channel:
print(f"No channels found for {user['slack_id']}, using #botnet")
channel_id = "C0BGMV1CAQG"
else:
channel_id = channel["id"]
response = await client.chat_postMessage(
channel=channel_id, # 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")
|
