mirror of
https://git.squeakyrat.moe/faggot-chan/
synced 2026-11-04 18:35:45 +01:00
434 lines
16 KiB
Python
434 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
import os
|
|
import re # thank you claude
|
|
import asyncio
|
|
import json
|
|
import time
|
|
import random
|
|
import traceback
|
|
import discord
|
|
import websockets
|
|
from pathlib import Path
|
|
from utils import nots, now_str, tail_lines
|
|
|
|
def ignore_connection_closed_ok(loop, context): #this doesn't fucking work
|
|
exc = context.get("exception")
|
|
if isinstance(exc, websockets.exceptions.ConnectionClosedOK):
|
|
return
|
|
loop.default_exception_handler(context)
|
|
|
|
|
|
def load_system_prompt() -> str:
|
|
return (CONFIG_DIR / "system_prompt.txt").read_text(encoding="utf-8")
|
|
|
|
# resolve files relative to this script, not whatever dir it happens to get launched from
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
FILES_DIR = SCRIPT_DIR / "files"
|
|
LOG_DIR = FILES_DIR / "log"
|
|
MEDIA_DIR = FILES_DIR / "media"
|
|
CONFIG_DIR = FILES_DIR / "config"
|
|
|
|
ERROR_FILE = LOG_DIR / "errorlog.txt"
|
|
MEMORY_FILE = LOG_DIR / "memory.txt"
|
|
MEDIA_FILE = MEDIA_DIR / "media.txt"
|
|
THREADS_FILE = CONFIG_DIR / "threads.txt"
|
|
|
|
def init_files():
|
|
for d in (LOG_DIR, MEDIA_DIR, CONFIG_DIR):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
for f in (ERROR_FILE, MEMORY_FILE, MEDIA_FILE):
|
|
f.touch(exist_ok=True)
|
|
if not THREADS_FILE.exists():
|
|
THREADS_FILE.write_text("6\n")
|
|
|
|
init_files()
|
|
|
|
_config_lines = [l.strip() for l in (CONFIG_DIR / "tokens.txt").read_text(encoding="utf-8").splitlines() if l.strip() and not l.strip().startswith("#")]
|
|
TOKEN = _config_lines[0]
|
|
server = int(_config_lines[1])
|
|
|
|
_conn_lines = [l.strip() for l in (CONFIG_DIR / "connection.txt").read_text(encoding="utf-8").splitlines() if l.strip() and not l.strip().startswith("#")]
|
|
MAIN_URI = _conn_lines[0] # ws:// or wss:// address of main.py
|
|
MAIN_TOKEN = _conn_lines[1] # shared secret main.py checks on every payload
|
|
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
intents.members = True
|
|
client = discord.Client(intents=intents)
|
|
|
|
block_id = ["1000000000000003", "1000000000000004"] #fuck these two in particular
|
|
|
|
moralfag_id = "1000000000000002"
|
|
moralfag_counter = 0
|
|
moralfag_skip = False
|
|
faglimit = 2 #ignore every x reply
|
|
|
|
last_server_message = 0.0 # unix timestamp of last message anyone sent in the server
|
|
dm_sent = False
|
|
interaction_count = 0
|
|
|
|
QUIET_COOLDOWN = 600 # seconds of silence before she's allowed to send a random message (10 mins)
|
|
|
|
def log_error(text: str):
|
|
with ERROR_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(text + "\n")
|
|
|
|
def log_message(text: str):
|
|
with MEMORY_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(text + "\n")
|
|
|
|
def load_shortterm() -> list[str]:
|
|
try:
|
|
return tail_lines(MEMORY_FILE, 15) # reads bottom 15 lines without loading the whole file
|
|
except Exception as e:
|
|
log_error(f"[{now_str()}] error reading shortterm: {e}")
|
|
return []
|
|
|
|
def load_longterm() -> list[str]:
|
|
try:
|
|
lines = [nots(l) for l in tail_lines(MEMORY_FILE, 515)]
|
|
return lines[:-15] if len(lines) > 15 else []
|
|
except Exception as e:
|
|
log_error(f"[{now_str()}] error reading longterm: {e}")
|
|
return []
|
|
|
|
def build_payload(user_input: str, rebuild: bool = False) -> str:
|
|
global vector_db_ready
|
|
should_rebuild = rebuild or not vector_db_ready # always rebuild on first ever message
|
|
if should_rebuild:
|
|
vector_db_ready = True
|
|
return json.dumps({
|
|
"system_prompt": load_system_prompt(),
|
|
"user_input": user_input,
|
|
"shortterm": load_shortterm(),
|
|
"longterm": load_longterm(),
|
|
"rebuild_vectors": should_rebuild,
|
|
"auth_token": MAIN_TOKEN,
|
|
})
|
|
|
|
_main_ws = None # kept open across messages instead of reconnecting every time
|
|
|
|
async def call_main(payload: str) -> tuple[bool, str]:
|
|
global _main_ws
|
|
try:
|
|
if _main_ws is None or _main_ws.closed:
|
|
_main_ws = await websockets.connect(MAIN_URI, open_timeout=None, ping_timeout=None)
|
|
await _main_ws.send(payload)
|
|
raw = await _main_ws.recv()
|
|
data = json.loads(raw)
|
|
if "error" in data:
|
|
return False, data["error"]
|
|
return True, data["ok"]
|
|
except websockets.exceptions.ConnectionClosedOK:
|
|
_main_ws = None
|
|
return False, f"[{now_str()}] websocket closed cleanly (no reply)"
|
|
except Exception as e:
|
|
_main_ws = None
|
|
return False, f"[{now_str()}] websocket error: {e}"
|
|
|
|
async def handle_error(error_msg: str, notify_channel=None, notify_user=None):
|
|
print(error_msg)
|
|
log_error(error_msg)
|
|
notice = "someone tell <@1000000000000001> there is a problem with my ai"
|
|
try:
|
|
if notify_channel:
|
|
await notify_channel.send(notice)
|
|
elif notify_user:
|
|
dm = await notify_user.create_dm()
|
|
await dm.send(notice)
|
|
except Exception:
|
|
pass
|
|
|
|
async def keep_typing(channel, stop_event):
|
|
while not stop_event.is_set():
|
|
await channel.typing()
|
|
await asyncio.sleep(8) # refresh every 8s before discord's 10s timeout
|
|
|
|
vector_db_ready = False # gets set to True after first rebuild so we don't keep rebuilding
|
|
|
|
saved_media_urls = set()
|
|
|
|
with MEDIA_FILE.open("r", encoding="utf-8") as f:
|
|
for line in f:
|
|
if "saved media:" in line:
|
|
url = line.split("saved media:", 1)[1].strip()
|
|
saved_media_urls.add(url)
|
|
|
|
def steal_media(text: str):
|
|
if "saved media:" not in text:
|
|
return
|
|
url = text.split("saved media:", 1)[1].strip()
|
|
if url in saved_media_urls:
|
|
return
|
|
saved_media_urls.add(url)
|
|
with MEDIA_FILE.open("a", encoding="utf-8") as f:
|
|
f.write(text + "\n")
|
|
|
|
def steal_attachments(message):
|
|
#steal images and gifs and stuff (always do this regardless of what else fires)
|
|
media_urls = [
|
|
a.url for a in message.attachments
|
|
if a.content_type and a.content_type.startswith(("image/", "video/"))
|
|
]
|
|
|
|
for embed in message.embeds:
|
|
if embed.image and embed.image.url:
|
|
media_urls.append(embed.image.url)
|
|
if embed.video and embed.video.url:
|
|
media_urls.append(embed.video.url)
|
|
|
|
for url in media_urls:
|
|
steal_media(f"[{now_str()}] saved media: {url}")
|
|
|
|
def resolve_mentions(content: str, guild) -> str: # thank you claude
|
|
if not guild:
|
|
return content
|
|
def replace(m):
|
|
member = guild.get_member(int(m.group(1)))
|
|
return f"@{member.name}" if member else m.group(0)
|
|
return re.sub(r'<@!?(\d+)>', replace, content)
|
|
|
|
def is_member_in_server(member: discord.Member) -> bool:
|
|
guild = client.get_guild(server)
|
|
return guild and guild.get_member(member.id) is not None
|
|
|
|
async def random_dm():
|
|
global dm_sent
|
|
await client.wait_until_ready()
|
|
while not client.is_closed():
|
|
await asyncio.sleep(600) #every 10 minutes
|
|
for guild in client.guilds:
|
|
if random.random() < 0.001: #should be 0.1%
|
|
members = [m for m in guild.members if not m.bot]
|
|
if dm_sent:
|
|
dm_sent = False
|
|
continue
|
|
if not members:
|
|
continue
|
|
victim = random.choice(members)
|
|
payload = build_payload(
|
|
f"System: you are sending a dm to {victim}, write a long, playful story or an over-the-top pun monologue about anything. "
|
|
"It should sound like something faggot-chan would post when she's bored — chaotic, funny, and full of personality (at least 20 words)."
|
|
"keep in mind this is a dm so make it a bit personal"
|
|
)
|
|
stop_typing = asyncio.Event()
|
|
asyncio.create_task(keep_typing(victim.dm_channel or await victim.create_dm(), stop_typing))
|
|
ok, result = await call_main(payload)
|
|
stop_typing.set()
|
|
if not ok:
|
|
await handle_error(result, notify_user=victim)
|
|
continue
|
|
try: #sends the dm
|
|
dm = await victim.create_dm()
|
|
await dm.send(result)
|
|
dm_sent = True
|
|
log_message(f"[{now_str()}]Faggot-chan sent a random dm to {victim}: {result}")
|
|
print(f"[{now_str()}] random dm to {victim}: {result}")
|
|
except Exception as e:
|
|
await handle_error(f"[{now_str()}] failed to send dm to {victim}: {e}", notify_user=victim)
|
|
async def random_message():
|
|
await client.wait_until_ready()
|
|
while not client.is_closed():
|
|
await asyncio.sleep(600) #every 10 minutes
|
|
if time.time() - last_server_message < QUIET_COOLDOWN: #only send if server has been quiet for 10 mins
|
|
continue
|
|
if random.random() < 0.0001: #0.01%
|
|
for guild in client.guilds:
|
|
channel = discord.utils.get(guild.text_channels, name="general")
|
|
if channel:
|
|
payload = build_payload(
|
|
"System: write a long, playful story or an over-the-top pun monologue about cheese. or anything random"
|
|
"it should sound like something faggot-chan would post when she's bored."
|
|
"chaotic, funny, and full of personality (at least 15 words and no more than 20)"
|
|
)
|
|
ok, result = await call_main(payload)
|
|
if not ok:
|
|
await handle_error(result, notify_channel=channel)
|
|
continue
|
|
await channel.send(result)
|
|
log_message(f"[{now_str()}] random message: {result}")
|
|
print(f"[{now_str()}] random message: {result}")
|
|
|
|
@client.event
|
|
async def on_ready():
|
|
client.loop.create_task(random_message())
|
|
client.loop.create_task(random_dm())
|
|
print(f"faggot-chan is active as {client.user}")
|
|
|
|
|
|
@client.event
|
|
async def on_random(message) -> bool:
|
|
# single weighted pick between sticker/emoji/media, used as a substitute for an actual
|
|
# llm reply — only called once on_message already decided this message gets a response.
|
|
# returns True if it fired (caller should skip the llm call), False if it should proceed normally.
|
|
stickers = list(message.guild.stickers)
|
|
emojis = [str(e) for e in message.guild.emojis]
|
|
|
|
options = []
|
|
if stickers:
|
|
options.append(("sticker", 0.10))
|
|
if emojis:
|
|
options.append(("emoji", 0.13))
|
|
if saved_media_urls:
|
|
options.append(("media", 0.07)) # ~30% combined chance of firing when all three are available
|
|
|
|
if not options:
|
|
return False
|
|
|
|
roll = random.random()
|
|
cumulative = 0.0
|
|
for kind, weight in options:
|
|
cumulative += weight
|
|
if roll >= cumulative:
|
|
continue
|
|
|
|
if kind == "sticker":
|
|
chosen = random.choice(stickers)
|
|
try:
|
|
await message.channel.send(stickers=[chosen])
|
|
log_message(f"[{now_str()}] Faggot-chan sent sticker: {chosen.name}")
|
|
print(f"[{now_str()}] Faggot-chan sent sticker: {chosen.name}")
|
|
except websockets.exceptions.ConnectionClosedOK:
|
|
pass
|
|
except Exception as e:
|
|
log_error(f"[{now_str()}] error: {e}")
|
|
await message.reply("someone tell <@1000000000000001> there is a problem with my ai")
|
|
print(f"Sticker send failed: {e}")
|
|
return True
|
|
|
|
if kind == "emoji":
|
|
try:
|
|
await message.add_reaction(random.choice(emojis))
|
|
except websockets.exceptions.ConnectionClosedOK:
|
|
pass
|
|
except Exception as e:
|
|
log_error(f"[{now_str()}] error: {e}")
|
|
await message.reply("someone tell <@1000000000000001> there is a problem with my ai")
|
|
print(f"Emoji reaction failed: {e}")
|
|
return True
|
|
|
|
if kind == "media":
|
|
chosen = random.choice(list(saved_media_urls))
|
|
await message.channel.send(chosen)
|
|
log_message(f"[{now_str()}] Faggot-chan sent media: {chosen}")
|
|
print(f"[{now_str()}] Faggot-chan sent media: {chosen}")
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
@client.event
|
|
async def on_message(message):
|
|
global dm_sent, interaction_count, last_server_message
|
|
if message.author == client.user:
|
|
return
|
|
|
|
if not is_member_in_server(message.author):
|
|
return
|
|
|
|
# update last activity so random_message knows the server isn't quiet
|
|
last_server_message = time.time()
|
|
|
|
if str(message.author.id) in block_id:
|
|
return
|
|
|
|
# drop guild messages outside allowed channels before doing anything else, thank you claude
|
|
if message.guild and message.channel.name not in ["general", "announcements"]:
|
|
return
|
|
|
|
if message.guild:
|
|
steal_attachments(message)
|
|
|
|
content = resolve_mentions(message.content, message.guild)
|
|
|
|
# always detect replies
|
|
replied_message = None
|
|
replied_to_bot = False
|
|
replied_content = ""
|
|
if message.reference:
|
|
try:
|
|
replied_message = message.reference.resolved or await message.channel.fetch_message(message.reference.message_id)
|
|
if replied_message:
|
|
if replied_message.author == client.user:
|
|
replied_to_bot = True
|
|
replied_content = resolve_mentions(replied_message.content, message.guild)
|
|
log_message(f'[{now_str()}] {message.author} replied to {replied_message.author} saying: "{replied_content}" with reply: "{content}"')
|
|
else:
|
|
log_message(f'[{now_str()}] {message.author} said: {content}')
|
|
except Exception as e:
|
|
log_error(f"[{now_str()}] failed to resolve reply: {e}")
|
|
log_message(f'[{now_str()}] {message.author} said: {content}')
|
|
else:
|
|
log_message(f'[{now_str()}] {message.author} said: {content}')
|
|
|
|
interaction_count += 1
|
|
rebuild = (interaction_count % 150 == 0)
|
|
|
|
if message.guild is None: #dms now should work
|
|
dm_sent = False
|
|
if replied_message:
|
|
user_input = f'{message.author} replied to {replied_message.author} saying: "{replied_content}" with reply: "{content}"'
|
|
else:
|
|
user_input = f"{message.author} said in DM: {content}"
|
|
payload = build_payload(user_input, rebuild)
|
|
stop_typing = asyncio.Event()
|
|
asyncio.create_task(keep_typing(message.channel, stop_typing))
|
|
ok, result = await call_main(payload)
|
|
stop_typing.set()
|
|
if not ok:
|
|
await handle_error(result, notify_user=message.author)
|
|
return
|
|
try:
|
|
await message.reply(result)
|
|
print(f"[{now_str()}] DM from {message.author}: {content} Bot replied in DM: {result}")
|
|
log_message(f"[{now_str()}] Faggot-chan said in DM: {result}")
|
|
except Exception as e:
|
|
await handle_error(f"[{now_str()}] failed to send DM reply: {e}", notify_user=message.author)
|
|
return
|
|
|
|
mentioned = client.user.mentioned_in(message)
|
|
|
|
#if it replied to bot, treat it as a mention automatically
|
|
if replied_to_bot:
|
|
mentioned = True
|
|
|
|
if str(message.author.id) == moralfag_id:
|
|
global moralfag_skip, moralfag_counter
|
|
if moralfag_skip:
|
|
moralfag_skip = False
|
|
return
|
|
|
|
if not mentioned and random.random() > 0.03: #3% chance of replying if not mentiond
|
|
return
|
|
|
|
if await on_random(message): # ~30% chance she just sends a sticker/emoji/media instead
|
|
return
|
|
|
|
if replied_message:
|
|
user_input = f'{message.author} replied to {replied_message.author} saying: "{replied_content}" with reply: "{content}"'
|
|
else:
|
|
user_input = f"{message.author} said: {content}"
|
|
|
|
payload = build_payload(user_input, rebuild)
|
|
stop_typing = asyncio.Event()
|
|
asyncio.create_task(keep_typing(message.channel, stop_typing))
|
|
ok, result = await call_main(payload)
|
|
stop_typing.set()
|
|
if not ok:
|
|
await handle_error(result, notify_channel=message.channel)
|
|
return
|
|
|
|
try:
|
|
await message.reply(result)
|
|
print(f"[{now_str()}] {message.author} said: {content}\n[{now_str()}] Faggot-chan said: {result}")
|
|
log_message(f"[{now_str()}] Faggot-chan said: {result}")
|
|
if str(message.author.id) == moralfag_id:
|
|
moralfag_counter += 1
|
|
if moralfag_counter >= faglimit:
|
|
moralfag_skip = True
|
|
moralfag_counter = 0
|
|
except Exception as e:
|
|
await handle_error(f"[{now_str()}] failed to send reply: {e}", notify_channel=message.channel)
|
|
|
|
client.run(TOKEN)
|