#!/usr/bin/env python3 import re # thank you claude import base64 import hashlib import asyncio import json import time import random import traceback import aiohttp # already installed, discord.py needs it import discord import websockets from pathlib import Path from urllib.parse import urlparse 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 PROMPT_FILE.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" DM_DIR = LOG_DIR / "dm" # one memory file per person so dms never end up in the shared memory ERROR_FILE = LOG_DIR / "errorlog.txt" MEMORY_FILE = LOG_DIR / "memory.txt" MEDIA_FILE = MEDIA_DIR / "media.txt" THREADS_FILE = CONFIG_DIR / "threads.txt" PROMPT_FILE = CONFIG_DIR / "system_prompt.txt" BLOCKED_FILE = CONFIG_DIR / "blocked.txt" # edited with admin.py, re-read on every message SETTINGS_FILE = CONFIG_DIR / "settings.json" REACTIONS_FILE = CONFIG_DIR / "reactions.json" # sticker/emoji/media descriptions she picks from DEFAULT_SETTINGS = {"admin_id": "1000000000000001", "moralfag_id": "1000000000000002", "faglimit": 2} # faglimit: ignore every x reply def init_files(): for d in (LOG_DIR, MEDIA_DIR, CONFIG_DIR, DM_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") if not PROMPT_FILE.exists(): # otherwise every single message dies on a missing file PROMPT_FILE.write_text("you are faggot-chan, a chaotic discord bot. edit files/config/system_prompt.txt\n", encoding="utf-8") if not BLOCKED_FILE.exists(): BLOCKED_FILE.write_text("# one discord user id per line, edit with admin.py edit blocked\n1000000000000003 # fuck these two in particular\n1000000000000004\n", encoding="utf-8") if not SETTINGS_FILE.exists(): SETTINGS_FILE.write_text(json.dumps(DEFAULT_SETTINGS, indent=2) + "\n", encoding="utf-8") 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 FALLBACK_URI = _conn_lines[2] if len(_conn_lines) > 2 else None # optional, local main.py (ollama) that takes over when gemini runs out of quota, same token intents = discord.Intents.default() intents.message_content = True intents.members = True client = discord.Client(intents=intents) moralfag_counter = 0 moralfag_skip = False last_server_message = 0.0 # unix timestamp of last message anyone sent in the server dm_sent = False interaction_count = 0 QUOTA_MSG = "usage limit reached, see you in a month" QUOTA_COOLDOWN = 3600 # after gemini says it's out of quota go straight to the fallback for this long (seconds), then try it again quota_until = 0.0 # unix timestamp MAX_REACTION_OPTIONS = 25 # per kind (stickers/emojis/media), random sample so the prompt stays small REACTION_TAG = re.compile(r'\[(sticker|emoji|media):\s*([^\]]+?)\s*\]', re.I) QUIET_COOLDOWN = 600 # seconds of silence before she's allowed to send a random message (10 mins) MEDIA_MIMES = ("image/png", "image/jpeg", "image/webp", "image/heic", "image/heif", "audio/", "video/", "application/pdf", "text/plain") # what gemini can actually read MEDIA_MAX_BYTES = 10 * 1024 * 1024 # total per message, gemini caps inline request size (~20mb and base64 adds a third) MEDIA_SAVE_MAX_BYTES = 8 * 1024 * 1024 # per file we keep, stays under discord's 10mb upload limit so she can send it back def log_error(text: str): with ERROR_FILE.open("a", encoding="utf-8") as f: f.write(text + "\n") def load_blocked() -> set[str]: # re-read on every message so admin.py edits apply without restarting try: lines = [l.split("#")[0].strip() for l in BLOCKED_FILE.read_text(encoding="utf-8").splitlines()] return {l for l in lines if l} except Exception as e: log_error(f"[{now_str()}] error reading blocked list: {e}") return set() def load_settings() -> dict: try: return {**DEFAULT_SETTINGS, **json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))} except Exception as e: log_error(f"[{now_str()}] error reading settings: {e}") return dict(DEFAULT_SETTINGS) def log_message(text: str, file: Path = MEMORY_FILE): text = " ".join(text.splitlines()) # one message = one line, otherwise multiline messages break the history parsing with file.open("a", encoding="utf-8") as f: f.write(text + "\n") return text.strip() # what shortterm will read back for this line, build_payload uses it to find the message she's answering def load_shortterm(file: Path = MEMORY_FILE) -> list[str]: try: return tail_lines(file, 67) # 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, media: list = None, logged: str = None, dm_file: Path = None, tools: str = "") -> str: global vector_db_ready shortterm = load_shortterm(dm_file or MEMORY_FILE) if logged and logged in shortterm: # cut at the message she's answering (it goes in the final turn instead), found by content and not by position: # people keep talking while she's busy so it's not always the last line, and whatever got logged after it isn't part of this reply shortterm = shortterm[:len(shortterm) - 1 - shortterm[::-1].index(logged)] if dm_file: longterm = [] # dms stay out of the shared rag so private stuff can't come up in a public channel should_rebuild = False else: should_rebuild = rebuild or not vector_db_ready # always rebuild on first ever message longterm = load_longterm() if should_rebuild else [] # only sent when it's needed, the backend asks for it (need_longterm) if it lost it if should_rebuild: vector_db_ready = True return json.dumps({ "system_prompt": load_system_prompt() + tools, "user_input": user_input, "shortterm": shortterm, "longterm": longterm, "rebuild_vectors": should_rebuild, "media": media or [], # [{"mime_type", "data" (base64)}], gemini.py reads everything, main.py only the images "auth_token": MAIN_TOKEN, }) _main_ws = {} # one connection per backend uri (gemini / fallback), kept open across messages instead of reconnecting every time _main_lock = None # one send/recv at a time on the shared sockets, made on first use so it lives in discord's loop async def call_backend(uri: str, payload: str) -> tuple[bool, str, bool]: # last bool is True when the backend said it's out of quota resent = False for attempt in range(2): # retry once with a fresh connection if the cached one turned out dead try: ws = _main_ws.get(uri) if ws is None: ws = _main_ws[uri] = await websockets.connect(uri, open_timeout=None, ping_timeout=None) await ws.send(payload) data = json.loads(await ws.recv()) if data.get("need_longterm") and not resent: # backend has no vector db (restarted / first time as fallback), send it the longterm once resent = True p = json.loads(payload) p["longterm"] = load_longterm() p["rebuild_vectors"] = True payload = json.dumps(p) await ws.send(payload) data = json.loads(await ws.recv()) if "error" in data: return False, data["error"], bool(data.get("quota")) return True, data["ok"], False except Exception as e: _main_ws.pop(uri, None) if attempt == 0: continue # one retry with a fresh connection before giving up return False, f"[{now_str()}] websocket error: {e}", False async def call_main(payload: str) -> tuple[bool, str]: global _main_lock, quota_until if _main_lock is None: _main_lock = asyncio.Lock() async with _main_lock: if not (FALLBACK_URI and time.time() < quota_until): # skip gemini while it's known to be out of quota ok, result, quota = await call_backend(MAIN_URI, payload) if ok: return ok, result if quota: quota_until = time.time() + QUOTA_COOLDOWN if not FALLBACK_URI: return False, QUOTA_MSG print(f"[{now_str()}] gemini is out of quota, using the fallback for the next {QUOTA_COOLDOWN // 60} min") elif not FALLBACK_URI: return False, result else: print(f"[{now_str()}] main backend failed ({result}), trying the fallback") # dead server / 502 / any other error ok, result, _ = await call_backend(FALLBACK_URI, payload) return ok, result async def handle_error(error_msg: str, notify_channel=None, notify_user=None): print(error_msg) log_error(error_msg) notice = QUOTA_MSG if error_msg == QUOTA_MSG else f"someone tell <@{load_settings()['admin_id']}> 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 def dm_file_for(user) -> Path: f = DM_DIR / f"{user.id}.txt" f.touch(exist_ok=True) return f def split_reply(text: str, limit: int = 2000) -> list[str]: # discord rejects anything over 2000 chars (and empty messages), split on a newline/space when possible chunks = [] text = text.strip() while len(text) > limit: cut = max(text.rfind("\n", 0, limit), text.rfind(" ", 0, limit)) if cut <= 0: cut = limit chunks.append(text[:cut].strip()) text = text[cut:].strip() if text: chunks.append(text) return chunks async def send_reply(message, text: str): # first chunk is a real reply, the rest just follow in the channel chunks = split_reply(text) await message.reply(chunks[0]) for chunk in chunks[1:]: await message.channel.send(chunk) vector_db_ready = False # gets set to True after first rebuild so we don't keep rebuilding saved_media = set() # file names inside MEDIA_DIR with MEDIA_FILE.open("r", encoding="utf-8") as f: for line in f: if "saved media:" in line: name = line.split("saved media:", 1)[1].strip() if (MEDIA_DIR / name).is_file(): # old entries were urls (those expire), they don't match a file so they get dropped saved_media.add(name) async def steal_media(url: str, name: str): # discord cdn links are signed and die after a day or so, so we keep our own copy if name in saved_media: return saved_media.add(name) # claim it first so two messages can't download the same thing at once try: data = b"" async with aiohttp.ClientSession() as session: async with session.get(url) as r: r.raise_for_status() async for chunk in r.content.iter_chunked(65536): data += chunk if len(data) > MEDIA_SAVE_MAX_BYTES: break if len(data) > MEDIA_SAVE_MAX_BYTES: # too big for her to send back anyway saved_media.discard(name) return (MEDIA_DIR / name).write_bytes(data) except Exception as e: saved_media.discard(name) log_error(f"[{now_str()}] failed to save media {url}: {e}") return with MEDIA_FILE.open("a", encoding="utf-8") as f: f.write(f"[{now_str()}] saved media: {name}\n") async def steal_attachments(message): #steal images and gifs and stuff (always do this regardless of what else fires) media = [ (a.url, f"{a.id}_{Path(a.filename).name}") for a in message.attachments if a.content_type and a.content_type.startswith(("image/", "video/")) and a.size <= MEDIA_SAVE_MAX_BYTES ] for embed in message.embeds: for part in (embed.image, embed.video): if part and part.url: ext = Path(urlparse(part.url).path).suffix # no extension, no preview on discord, not worth keeping if ext: media.append((part.url, hashlib.sha1(part.url.encode()).hexdigest()[:16] + ext)) for url, name in media: await steal_media(url, name) async def grab_media(message) -> list[dict]: # downloads supported attachments to send to gemini, only call this when the bot was directly mentioned media = [] total = 0 for a in message.attachments: mime = (a.content_type or "").split(";")[0].strip().lower() if not mime.startswith(MEDIA_MIMES): continue if total + a.size > MEDIA_MAX_BYTES: continue try: data = await a.read() except Exception as e: log_error(f"[{now_str()}] failed to download attachment {a.filename}: {e}") continue total += len(data) media.append({"mime_type": mime, "data": base64.b64encode(data).decode("ascii")}) return media 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", dm_file=dm_file_for(victim) ) 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) # just log it, the victim has nothing to do with this continue if not result.strip(): continue try: #sends the dm dm = await victim.create_dm() for chunk in split_reply(result): await dm.send(chunk) dm_sent = True log_message(f"[{now_str()}] Faggot-chan said in DM: {result}", dm_file_for(victim)) 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}") 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 short, 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 if not result.strip(): continue for chunk in split_reply(result): await channel.send(chunk) log_message(f"[{now_str()}] random message: {result}") print(f"[{now_str()}] random message: {result}") @client.event async def on_ready(): asyncio.get_running_loop().set_exception_handler(ignore_connection_closed_ok) for guild in client.guilds: sync_reactions(guild) client.loop.create_task(random_message()) client.loop.create_task(random_dm()) print(f"faggot-chan is active as {client.user}") def load_reactions() -> dict: # re-read every time so admin.py edit reactions applies without a restart try: data = json.loads(REACTIONS_FILE.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): data = {} for k in ("stickers", "emojis", "media"): data.setdefault(k, {}) # name -> short description return data def sync_reactions(guild): # anything new gets added with an empty description, fill them in with admin.py edit reactions data = load_reactions() changed = False for key, names in (("stickers", [s.name for s in guild.stickers]), ("emojis", [e.name for e in guild.emojis]), ("media", sorted(saved_media))): for n in names: if n not in data[key]: data[key][n] = "" changed = True if changed: REACTIONS_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def reaction_tool_prompt(guild) -> str: # the "tool": instead of (or along with) words she can write a [sticker: name] / [emoji: name] / [media: name] tag and # split_reactions picks it back out of the reply. same call as the reply, so no extra latency or gemini quota. sync_reactions(guild) data = load_reactions() have = { "stickers": {s.name for s in guild.stickers if s.available}, # unavailable = locked after the boost level dropped "emojis": {e.name for e in guild.emojis if e.available}, "media": {n for n in saved_media if (MEDIA_DIR / n).is_file()}, } sections = [] for key in ("stickers", "emojis", "media"): options = [(n, d) for n, d in data[key].items() if n in have[key] and (d or key != "media")] # media without a description tells her nothing if not options: continue options = random.sample(options, min(len(options), MAX_REACTION_OPTIONS)) sections.append(f"{key}:\n" + "\n".join(f"- {n}: {d}" if d else f"- {n}" for n, d in options)) if not sections: return "" return ("\n\nREACTIONS: you can send a sticker, an emoji reaction or a media file instead of (or along with) words, but only when it really fits the moment, most of the time just reply normally. " "to do it put ONE tag at the end of your message: [sticker: name], [emoji: name] or [media: name], using exactly one of the names below. " "the tag alone is fine if you have nothing to say. never invent names.\n" + "\n".join(sections)) def split_reactions(text: str) -> tuple[str, list]: return REACTION_TAG.sub("", text).strip(), [(k.lower(), n) for k, n in REACTION_TAG.findall(text)] async def send_reaction(message, kind: str, name: str): # looks the name up for real, a made up name just does nothing name = name.strip().strip(":").lower() try: if kind == "sticker": chosen = next((s for s in message.guild.stickers if s.available and s.name.lower() == name), None) if chosen: 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}") elif kind == "emoji": chosen = next((e for e in message.guild.emojis if e.available and e.name.lower() == name), None) if chosen: await message.add_reaction(chosen) elif kind == "media": chosen = next((n for n in saved_media if n.lower() == name and (MEDIA_DIR / n).is_file()), None) # only names we saved ourselves if chosen: await message.channel.send(file=discord.File(MEDIA_DIR / chosen)) log_message(f"[{now_str()}] Faggot-chan sent media: {chosen}") print(f"[{now_str()}] Faggot-chan sent media: {chosen}") except Exception as e: await handle_error(f"[{now_str()}] failed to send {kind} {name}: {e}", notify_channel=message.channel) @client.event async def on_message(message): global dm_sent, interaction_count, last_server_message if message.author == client.user or message.author.bot: # other bots too, otherwise two bots can loop forever 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 load_blocked(): 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: await steal_attachments(message) content = resolve_mentions(message.content, message.guild) mem_file = MEMORY_FILE if message.guild else dm_file_for(message.author) # dms get their own memory # always detect replies replied_message = None replied_to_bot = False replied_content = "" if message.reference: try: replied_message = message.reference.resolved if not isinstance(replied_message, discord.Message): # deleted or not cached, resolved is a placeholder with no author replied_message = 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) logged_line = log_message(f'[{now_str()}] {message.author} replied to {replied_message.author} saying: "{replied_content}" with reply: "{content}"', mem_file) else: logged_line = log_message(f'[{now_str()}] {message.author} said: {content}', mem_file) except Exception as e: replied_message = None # fetch failed (deleted), treat it like a normal message log_error(f"[{now_str()}] failed to resolve reply: {e}") logged_line = log_message(f'[{now_str()}] {message.author} said: {content}', mem_file) else: logged_line = log_message(f'[{now_str()}] {message.author} said: {content}', mem_file) 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, await grab_media(message), logged_line, mem_file) # a dm is always aimed at her 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 if not result.strip(): return try: await send_reply(message, 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}", mem_file) except Exception as e: await handle_error(f"[{now_str()}] failed to send DM reply: {e}", notify_user=message.author) return mentioned = client.user in message.mentions # mentioned_in() also fires on @everyone/@here #if it replied to bot, treat it as a mention automatically if replied_to_bot: mentioned = True if not mentioned and random.random() > 0.03: #3% chance of replying if not mentiond return settings = load_settings() if str(message.author.id) == settings["moralfag_id"]: # only checked once she was actually going to reply, otherwise the skip gets eaten by chatter global moralfag_skip, moralfag_counter if moralfag_skip: moralfag_skip = False 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}" media = await grab_media(message) if client.user in message.mentions else [] # only direct @mentions, not replies or random 3% ones payload = build_payload(user_input, rebuild, media, logged_line, tools=reaction_tool_prompt(message.guild)) 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 result, reactions = split_reactions(result) # she can answer with a [sticker: x] style tag instead of words if not result.strip() and not reactions: return try: if result.strip(): await send_reply(message, 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) == settings["moralfag_id"]: moralfag_counter += 1 if moralfag_counter >= settings["faglimit"]: moralfag_skip = True moralfag_counter = 0 for kind, name in reactions[:1]: # one is enough await send_reaction(message, kind, name) except Exception as e: await handle_error(f"[{now_str()}] failed to send reply: {e}", notify_channel=message.channel) client.run(TOKEN)