"""Legendy Baru: the bar's conversation memory condenses into legends. pamiec.json grows forever - every chat appends a user+assistant pair. This cog periodically summarises the oldest slice of that file into a single, in-character "legend" through the active AI backend, replaces those old messages with the summary (bounding the file and keeping continuity for the next startup's context), archives the legend to legendy.json, and announces it on #legendy. The compaction transform is kept as pure functions so it can be unit-tested without Discord or the AI; the cog only wires them to the file, the AI and the channel. The write is done as re-read -> atomic rewrite (no await in between) so a concurrent handle_response append can neither be lost nor corrupt the file. """ import json import logging import os import random import shutil from datetime import datetime import discord from discord.ext import commands, tasks from ai_functions import handle_response from constants import ( ENCODING, GPT_SETTINGS, GUILD_ID, LEGENDS_CHANNEL_ID, LEGENDS_FILE, MEMORY_COMPACT_HOURS, MEMORY_COMPACT_THRESHOLD, MEMORY_FIVE_SIARA, MEMORY_KEEP_RECENT, ) _MAX_MSG = 1900 _TRANSCRIPT_MAX = 10000 # chars of old memory fed to the summariser _LEGEND_MARKER = "[LEGENDA BARU] " # ----------------------------------------------------------- pure transforms def build_transcript(messages, max_chars=_TRANSCRIPT_MAX): """Render messages as 'role: content' lines, head+tail truncated to fit.""" lines = [] for msg in messages: content = str(msg.get("content", "")).strip() if content: lines.append(f"{msg.get('role', '?')}: {content}") text = "\n".join(lines) if len(text) <= max_chars: return text half = max_chars // 2 return text[:half] + "\n[...urwane...]\n" + text[-half:] def apply_compaction(messages, old_count, legend_text): """Replace the first `old_count` messages with one legend system-message. Everything from index old_count onward is preserved verbatim, so any append that happened while the summary was being generated (they only ever go to the end) is kept. """ legend_msg = {"role": "system", "content": _LEGEND_MARKER + legend_text} return [legend_msg] + messages[old_count:] def _persona() -> str: try: if isinstance(GPT_SETTINGS, list) and GPT_SETTINGS and isinstance(GPT_SETTINGS[0], dict): return GPT_SETTINGS[0].get("content", "") or "" except Exception: # pylint: disable=broad-except pass return "Jesteś Conjurerem, barmanem i kronikarzem baru mechawojowników. Mówisz po polsku." def _load_json_list(path) -> list: try: with open(path, "r", encoding=ENCODING) as handle: data = json.load(handle) return data if isinstance(data, list) else [] except (FileNotFoundError, json.JSONDecodeError, OSError): return [] class LoreModule(commands.Cog): """Memory compaction into legends + legend recall.""" def __init__(self, bot): self.bot = bot self.logger = logging.getLogger("discord") async def cog_load(self): if not self.compaction_loop.is_running(): self.compaction_loop.start() async def cog_unload(self): self.compaction_loop.cancel() @tasks.loop(hours=MEMORY_COMPACT_HOURS) async def compaction_loop(self): try: await self._maybe_compact() except Exception as exc: # pylint: disable=broad-except self.logger.exception("Memory compaction tick failed: %s", exc) @compaction_loop.before_loop async def _before_compaction(self): await self.bot.wait_until_ready() # --------------------------------------------------------------- helpers async def _summarise(self, transcript: str) -> str: instructions = ( "Jesteś kronikarzem naszego baru. Poniżej najstarszy fragment historii rozmów. " "Streść go w JEDNĄ zwartą, klimatyczną legendę w swoim stylu: zachowaj kluczowe " "postacie, wydarzenia i wątki (dla ciągłości pamięci), ale skróć drastycznie. " "Po polsku, bez wstępów typu 'oto streszczenie'.\n\n" f"HISTORIA:\n{transcript}" ) prompt = [ {"role": "system", "content": _persona()}, {"role": "user", "content": instructions}, ] result, _ = await handle_response("", True, True, [], "kronikarz", "NONE", none_request=prompt) return (result or "").strip() def _write_memory_atomic(self, old_count: int, legend: str) -> bool: """Re-read, back up, then rewrite pamiec.json - all synchronously. Returns False (and leaves the file untouched) if the file changed shape in a way that makes compaction unsafe. """ messages = _load_json_list(MEMORY_FIVE_SIARA) if len(messages) < old_count: self.logger.warning( "Memory shrank during compaction (%d < %d) - skipping rewrite", len(messages), old_count, ) return False try: if os.path.exists(MEMORY_FIVE_SIARA): shutil.copyfile(MEMORY_FIVE_SIARA, MEMORY_FIVE_SIARA + ".bak") new = apply_compaction(messages, old_count, legend) with open(MEMORY_FIVE_SIARA, "w", encoding=ENCODING) as handle: json.dump(new, handle, indent=4, ensure_ascii=False) except OSError as exc: self.logger.error("Failed to rewrite compacted memory %s: %s", MEMORY_FIVE_SIARA, exc) return False return True def _append_legend(self, legend: str) -> None: legends = _load_json_list(LEGENDS_FILE) legends.append({"kiedy": datetime.now().isoformat(timespec="seconds"), "tekst": legend}) try: with open(LEGENDS_FILE, "w", encoding=ENCODING) as handle: json.dump(legends, handle, indent=2, ensure_ascii=False) except OSError as exc: self.logger.warning("Nie mogę zapisać legendy do %s: %s", LEGENDS_FILE, exc) async def _post(self, channel, text: str) -> None: text = text or "" while text: await channel.send(text[:_MAX_MSG]) text = text[_MAX_MSG:] async def _post_legend(self, legend: str) -> None: channel = self.bot.get_channel(LEGENDS_CHANNEL_ID) if channel is None: self.logger.warning("Legends channel %s not found - legend not announced", LEGENDS_CHANNEL_ID) return await self._post(channel, "📜 *Z mgieł baru wyłania się nowa legenda...*\n\n" + legend) async def _maybe_compact(self, force: bool = False): messages = _load_json_list(MEMORY_FIVE_SIARA) if not force and len(messages) <= MEMORY_COMPACT_THRESHOLD: return None old_count = len(messages) - MEMORY_KEEP_RECENT if old_count <= 1: self.logger.info("Compaction: nothing meaningful to compact (%d msgs)", len(messages)) return None transcript = build_transcript(messages[:old_count]) legend = await self._summarise(transcript) if not legend: self.logger.warning("Compaction: summariser returned nothing - aborting") return None if not self._write_memory_atomic(old_count, legend): return None self._append_legend(legend) self.logger.info( "Compacted %d old messages into a legend (kept last %d)", old_count, MEMORY_KEEP_RECENT ) await self._post_legend(legend) return legend # --------------------------------------------------------------- commands @commands.hybrid_command( name="zapisz_legende", description="Wymuś kompaktowanie pamięci w legendę teraz (tylko Vykidailo)", guild=discord.Object(id=GUILD_ID), ) async def zapisz_legende(self, ctx): is_admin = isinstance(ctx.author, discord.Member) and any( role.name == "Vykidailo" for role in ctx.author.roles ) if not is_admin: await ctx.send("Tylko Vykidailo spisuje legendy.") return async with ctx.channel.typing(): legend = await self._maybe_compact(force=True) if legend is None: await ctx.send("*Conjurer wzrusza ramionami* Za mało jeszcze historii na legendę.") else: await ctx.send("*Conjurer maczkiem spisuje kolejną legendę baru.* Poszła na #legendy.") @commands.hybrid_command( name="legendy", description="Przypomina losową legendę baru", guild=discord.Object(id=GUILD_ID), ) async def legendy(self, ctx): async with ctx.channel.typing(): legends = _load_json_list(LEGENDS_FILE) if not legends: await ctx.send("*Conjurer patrzy w ogień* Legend jeszcze nie spisano. Wszystko przed nami.") return pick = random.choice(legends) # nosec B311 - flavour, not security await self._post(ctx, f"📜 *Conjurer przywołuje legendę ({pick.get('kiedy', '?')}):*\n\n{pick.get('tekst', '')}") async def setup(bot): logger = logging.getLogger("discord") await bot.add_cog(LoreModule(bot)) logger.info("Loading lore commands module done")