# modules/pirates_list/pirates_list.py import asyncio import discord from discord.ext import commands from discord import app_commands from mod_perms import require_mod_ctx, require_mod_interaction from modules.common.settings import cfg as _cfg class PiratesListCog(commands.Cog): """ Maintains a compact, alphabetized pirates list in the configured channel. Entry format: - **{Character}** (*{Account}*) [{Threat}%] - In group: {never/rarely/...}. Destructive: {never/rarely/...}. Encounters: N. Last: Posts are chunked to stay <2000 chars and previous posts are deleted on refresh. """ def __init__(self, bot: commands.Bot): self.bot = bot c = _cfg(bot) self.list_channel_id = c.int("pirates_list_channel_id") self.group_threshold = c.int("threat_group_threshold", 3) self.min_samples = c.int("threat_min_samples_for_stats", 3) # serialize refreshes per guild self._locks: dict[int, asyncio.Lock] = {} # never ping on posted content self._no_mentions = discord.AllowedMentions.none() # ----------------- utils ----------------- def _lock_for(self, guild_id: int) -> asyncio.Lock: self._locks.setdefault(guild_id, asyncio.Lock()) return self._locks[guild_id] @staticmethod def _esc(s: str) -> str: """Escape markdown and neuter @mentions in user-provided strings.""" safe = discord.utils.escape_markdown(str(s)) return safe.replace("@", "@\u200b") def _encounters_for(self, pirate: dict): """Return encounters matching either char name or account (case-insensitive), defensively.""" acct_l = str(pirate.get("account_name", "")).lower() char_l = str(pirate.get("character_name", "")).lower() out = [] for e in self.bot.data_manager.get("encounters"): try: ident = str(e.get("identifier", "")).lower() if ident in (acct_l, char_l): out.append(e) except Exception: continue return out def _bucket(self, rate: float, samples: int) -> str: if samples < self.min_samples: return "unknown" if rate <= 0: return "never" if rate <= 0.25: return "rarely" if rate <= 0.60: return "sometimes" if rate <= 0.85: return "often" return "always" def _format_entry(self, p: dict) -> str: encs = self._encounters_for(p) total = len(encs) def _safe_int(v, default=0): try: return int(v) except Exception: return default group_rate = ( sum(1 for e in encs if _safe_int(e.get("group_size"), 0) >= self.group_threshold) / total if total else 0.0 ) destroy_rate = ( sum(1 for e in encs if bool(e.get("destruction"))) / total if total else 0.0 ) groups_str = self._bucket(group_rate, total) destr_str = self._bucket(destroy_rate, total) last_ts = None if total: try: last_ts = max(int(float(e.get("timestamp", 0))) for e in encs) except Exception: last_ts = None char = self._esc(p.get("character_name", "Unknown")) acct = self._esc(p.get("account_name", "Unknown#00000")) threat = int(p.get("threat_level", 0)) last_str = f"" if last_ts else "—" line1 = f"- **{char}** (*{acct}*) [{threat}%]" line2 = f" - In group: {groups_str}. Destructive: {destr_str}. Encounters: {total}. Last: {last_str}" return f"{line1}\n{line2}" async def refresh_list(self, guild: discord.Guild): """Edit list messages in place; only send extra messages when we need more chunks (new pirates).""" async with self._lock_for(guild.id): channel = guild.get_channel(self.list_channel_id) if not channel: print("[pirates_list] list channel not found:", self.list_channel_id) return dm = self.bot.data_manager allow = self._no_mentions # ---- load & prune existing posts for this guild/channel ---- records = [r for r in dm.get("pirates_list_posts") if r.get("guild_id") == guild.id and r.get("channel_id") == self.list_channel_id] msgs, kept_records = [], [] for r in records: try: m = await channel.fetch_message(int(r["message_id"])) msgs.append(m) kept_records.append(r) except Exception: dm.remove("pirates_list_posts", lambda x, mid=r["message_id"]: x.get("message_id") == mid) # ---- build fresh, sorted contents ---- pirates = sorted( dm.get("pirates"), key=lambda x: ( str(x.get("character_name", "")).lower(), str(x.get("account_name", "")).lower() ) ) # Empty state if not pirates: placeholder = "_No verified pirates yet._" if msgs: if msgs[0].content != placeholder: try: await msgs[0].edit(content=placeholder, allowed_mentions=allow) except Exception as e: print("[pirates_list] edit placeholder failed:", repr(e)) for extra in msgs[1:]: try: await extra.delete() except Exception: pass dm.remove("pirates_list_posts", lambda r, mid=extra.id: r.get("message_id") == mid) else: try: m = await channel.send(placeholder, allowed_mentions=allow) dm.add("pirates_list_posts", { "guild_id": guild.id, "channel_id": self.list_channel_id, "message_id": m.id }) except Exception as e: print("[pirates_list] send placeholder failed:", repr(e)) return # Chunk into <2000 char blocks (~1900 for margin) chunks, current = [], "" for p in pirates: entry = self._format_entry(p) if len(current) + len(entry) + 1 > 1900: chunks.append(current.rstrip()) current = entry + "\n" else: current += entry + "\n" if current.strip(): chunks.append(current.rstrip()) # ---- edit existing messages with new content ---- common = min(len(msgs), len(chunks)) for i in range(common): if msgs[i].content != chunks[i]: try: await msgs[i].edit(content=chunks[i], allowed_mentions=allow) except Exception as e: print("[pirates_list] edit block failed:", repr(e)) # ---- send additional messages if needed ---- if len(chunks) > len(msgs): for i in range(len(msgs), len(chunks)): try: m = await channel.send(chunks[i], allowed_mentions=allow) dm.add("pirates_list_posts", { "guild_id": guild.id, "channel_id": self.list_channel_id, "message_id": m.id }) except Exception as e: print("[pirates_list] send block failed:", repr(e)) # ---- delete extras if fewer chunks now ---- elif len(chunks) < len(msgs): extras = msgs[len(chunks):] for m in extras: try: await m.delete() except Exception: pass dm.remove("pirates_list_posts", lambda r, mid=m.id: r.get("message_id") == mid) # Manual refresh command (hybrid: works as /pirates_list_refresh and !pirates_list_refresh) @app_commands.command(name="pirates_list_refresh", description="[MOD] Rebuild the compact pirates list") @app_commands.checks.cooldown(1, 10) # guild-scope anti-spam analogue async def pirates_list_refresh(self, interaction: discord.Interaction): if not interaction.guild: return await interaction.response.send_message("Use this in a server.", ephemeral=True) if not await require_mod_interaction(interaction, "This command is restricted to moderators."): return await interaction.response.defer(ephemeral=True) await self.refresh_list(interaction.guild) await interaction.followup.send("Pirates list refreshed.", ephemeral=True) async def setup(bot: commands.Bot): await bot.add_cog(PiratesListCog(bot))