shaiwatcher/modules/pirate_report/pirate_report.py
Franz Rolfsvaag 5368d21be4 0.3.9.2.a5
performance improvements, stability, and primarily settings-handling improvements.
  - Due to the container transition, some settings handling became quietly broken or defunct.
2025-08-10 20:23:09 +02:00

586 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# modules/pirate_report/pirate_report.py
import re
import time
from datetime import datetime, timezone
import discord
from discord.ext import commands
from discord import app_commands
from modules.common.settings import cfg
from mod_perms import (
is_moderator_member,
is_moderator_userid,
require_mod_ctx,
require_mod_interaction,
)
CHECK = ''
CROSS = ''
# --------------------- Helpers ---------------------
_ACCT_RE = re.compile(r'.+#\d{5}$')
def _acct_ok(s: str) -> bool:
return bool(_ACCT_RE.fullmatch(s.strip()))
def _now_utc_str() -> str:
return datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')
def _parse_bool(s: str) -> bool:
v = s.strip().lower()
if v in ('y', 'yes', 'true', 't', '1'):
return True
if v in ('n', 'no', 'false', 'f', '0'):
return False
raise ValueError("Please enter yes or no")
# --------------------- Modals ----------------------
class ReportModal(discord.ui.Modal, title="Submit Pirate Report"):
def __init__(self, cog: "PirateReportCog"):
super().__init__()
self.cog = cog
self.character_name = discord.ui.TextInput(
label="In-game nickname",
placeholder="e.g., SandStalker",
max_length=64,
required=True
)
self.account_name = discord.ui.TextInput(
label="Account (Name#12345)",
placeholder="Must end with # and 5 digits",
max_length=64,
required=True
)
self.add_item(self.character_name)
self.add_item(self.account_name)
async def on_submit(self, interaction: discord.Interaction):
if not interaction.guild:
return await interaction.response.send_message("Use this in a server.", ephemeral=True)
char = self.character_name.value.strip()
acct = self.account_name.value.strip()
if not _acct_ok(acct):
return await interaction.response.send_message(
"❌ Invalid account format. It must end with `#` and **five digits** (e.g. `SomeUser#12345`).",
ephemeral=True
)
dm = self.cog.bot.data_manager
char_l = char.lower()
acct_l = acct.lower()
# Anti-spam: 60s per reporter
now = time.time()
recent = [r for r in dm.get('reports')
if r.get('submitter_id') == interaction.user.id and (now - r.get('ts', now)) < 60]
if recent:
return await interaction.response.send_message("⏱️ Please wait 60 seconds before submitting another report.", ephemeral=True)
# Already approved?
if any(p['account_name'].lower() == acct_l for p in dm.get('pirates')):
return await interaction.response.send_message("❌ This player is already in the pirate list.", ephemeral=True)
# Pending duplicate?
if any(r['character_name'].lower() == char_l and r['account_name'].lower() == acct_l for r in dm.get('reports')):
return await interaction.response.send_message("⏳ A report for this player is already pending moderation.", ephemeral=True)
# Acknowledge in channel so we can edit later when approved/rejected
try:
ack = await interaction.channel.send(f"{interaction.user.mention} thanks — your report was sent to moderators for review.")
except Exception as e:
print("[pirate_report] ack send failed:", repr(e))
ack = None
# Send to mod channel with ✅/❌
mod_ch = interaction.guild.get_channel(self.cog.mod_channel)
if not mod_ch:
return await interaction.response.send_message("❌ Mod channel not configured.", ephemeral=True)
try:
mod_msg = await mod_ch.send(
f"🚩 **Pirate Report**\n"
f"**Character:** {char}\n"
f"**Account:** {acct}\n"
f"**Submitted by:** {interaction.user.mention}"
)
await mod_msg.add_reaction(CHECK)
await mod_msg.add_reaction(CROSS)
except Exception as e:
print("[pirate_report] mod message failed:", repr(e))
return await interaction.response.send_message("❌ Failed to send to mod channel.", ephemeral=True)
dm.add('reports', {
'report_id': mod_msg.id,
'guild_id': interaction.guild.id,
'character_name': char,
'account_name': acct,
'submitter_id': interaction.user.id,
'origin_channel_id': interaction.channel.id if interaction.channel else 0,
'ack_message_id': ack.id if ack else 0,
'status': 'pending',
'ts': now
})
await interaction.response.send_message("✅ Report submitted to moderators.", ephemeral=True)
class EditPirateModal(discord.ui.Modal, title="Edit Pirate Entry"):
def __init__(self, cog: "PirateReportCog"):
super().__init__()
self.cog = cog
self.old_account = discord.ui.TextInput(
label="Current account (Name#12345)",
placeholder="Exact current account in the list",
max_length=64,
required=True
)
self.new_character = discord.ui.TextInput(
label="New in-game nickname (optional)",
placeholder="Leave blank to keep existing",
max_length=64,
required=False
)
self.new_account = discord.ui.TextInput(
label="New account (optional, Name#12345)",
placeholder="Leave blank to keep existing",
max_length=64,
required=False
)
self.add_item(self.old_account)
self.add_item(self.new_character)
self.add_item(self.new_account)
async def on_submit(self, interaction: discord.Interaction):
if not interaction.guild:
return await interaction.response.send_message("Use this in a server.", ephemeral=True)
if not is_moderator_member(interaction.user, self.cog.bot):
return await interaction.response.send_message("This action is restricted to moderators.", ephemeral=True)
old_acct = self.old_account.value.strip()
new_char = (self.new_character.value or "").strip()
new_acct = (self.new_account.value or "").strip()
if not new_char and not new_acct:
return await interaction.response.send_message(
"❌ Provide **at least one** of: new nickname or new account.",
ephemeral=True
)
dm = self.cog.bot.data_manager
pirates = dm.get('pirates')
rec = next((p for p in pirates if p['account_name'].lower() == old_acct.lower()), None)
if not rec:
return await interaction.response.send_message("❌ Pirate not found.", ephemeral=True)
if new_acct:
if not _acct_ok(new_acct):
return await interaction.response.send_message("❌ Invalid new account format (Name#12345).", ephemeral=True)
if new_acct.lower() != old_acct.lower() and any(p['account_name'].lower() == new_acct.lower() for p in pirates):
return await interaction.response.send_message("❌ Another pirate already has that account.", ephemeral=True)
before = rec.copy()
dm.remove('pirates', lambda p: p['account_name'].lower() == old_acct.lower())
if new_char:
rec['character_name'] = new_char
if new_acct:
rec['account_name'] = new_acct
dm.add('pirates', rec)
await self.cog._modlog(
interaction.guild,
f"✏️ Edited pirate by {interaction.user.mention}: "
f"{'name ' + before['character_name'] + '' + rec['character_name'] if new_char else ''} "
f"{'/ account ' + before['account_name'] + '' + rec['account_name'] if new_acct else ''}".strip()
)
await interaction.response.send_message("✅ Pirate updated.", ephemeral=True)
await self.cog._refresh_pirates_list(interaction.guild)
class EncounterModal(discord.ui.Modal, title="Log Pirate Encounter"):
def __init__(self, cog: "PirateReportCog"):
super().__init__()
self.cog = cog
self.identifier = discord.ui.TextInput(
label="Pirate (name or account)",
placeholder="MuadDib or MuadDib#12345",
required=True,
max_length=64
)
self.group_size = discord.ui.TextInput(
label="Group size (including this pirate)",
placeholder="How many pirates were in the group?",
required=True
)
self.kills = discord.ui.TextInput(
label="Kills (integer ≥ 0; 0 = none/unknown)",
placeholder="How many did the pirate kill during the encounter",
required=True
)
self.destruction = discord.ui.TextInput(
label="Destructive? (yes/no)",
placeholder="Did they destroy a base/ornithopter? yes or no",
required=True
)
self.skill = discord.ui.TextInput(
label="Perceived Skill (05, 0 = unknown)",
placeholder="0..5",
required=True
)
self.add_item(self.identifier)
self.add_item(self.group_size)
self.add_item(self.kills)
self.add_item(self.destruction)
self.add_item(self.skill)
async def on_submit(self, interaction: discord.Interaction):
ident_raw = self.identifier.value.strip()
guild = interaction.guild
if guild is None:
return await interaction.response.send_message("This can only be used in a server.", ephemeral=True)
# Parse numeric fields
try:
group_size = int(self.group_size.value.strip()); assert group_size >= 1
except Exception:
return await interaction.response.send_message("❌ **Group size** must be an integer ≥ 1.", ephemeral=True)
try:
kills = int(self.kills.value.strip()); assert kills >= 0
except Exception:
return await interaction.response.send_message("❌ **Kills** must be an integer ≥ 0.", ephemeral=True)
try:
destruction = _parse_bool(self.destruction.value)
except ValueError:
return await interaction.response.send_message("❌ **Destructive** must be **yes** or **no**.", ephemeral=True)
try:
skill = int(self.skill.value.strip()); assert 0 <= skill <= 5
except Exception:
return await interaction.response.send_message("❌ **Skill** must be an integer between **0** and **5**.", ephemeral=True)
# Resolve to a single pirate record and canonical account identifier
pirate, error = self.cog.resolve_pirate(ident_raw)
if error:
return await interaction.response.send_message(error, ephemeral=True)
if not pirate:
return await interaction.response.send_message(
"❌ No such pirate registered. Ask a moderator to add them, or submit a **/report** for review.",
ephemeral=True
)
canonical_acct = pirate['account_name'] # <- always store account here
acct_l = canonical_acct.lower()
char_l = pirate['character_name'].lower()
dm = self.cog.bot.data_manager
now = time.time()
# Rate-limit by canonical account
recent = [
e for e in dm.get('encounters')
if e['reporter_id'] == interaction.user.id
and e.get('identifier','').lower() in (acct_l, char_l) # respect legacy
and (now - e['timestamp']) < 600
]
if recent:
return await interaction.response.send_message(
"⏱️ You can only report the same pirate once every **10 minutes**.",
ephemeral=True
)
# Store with canonical account as identifier
rec = {
'identifier': canonical_acct,
'reporter_id': interaction.user.id,
'timestamp': now,
'group_size': group_size,
'kills': kills,
'destruction': destruction,
'skill': skill
}
dm.add('encounters', rec)
# Recalculate threat (count both new account-based and legacy name-based entries)
encs = [e for e in dm.get('encounters') if e.get('identifier','').lower() in (acct_l, char_l)]
total = len(encs)
if total:
group_threshold = self.cog.group_threshold
kill_rate = sum(e['kills'] > 0 for e in encs) / total
group_rate = sum(e['group_size'] >= group_threshold for e in encs) / total
destroy_rate = sum(bool(e['destruction']) for e in encs) / total
skill_samples = [e['skill'] / 5 for e in encs if e['skill'] > 0]
skill_rate = (sum(skill_samples) / len(skill_samples)) if skill_samples else 0.0
w_k, w_d, w_g, w_s = self.cog.w_kill, self.cog.w_destruction, self.cog.w_group, self.cog.w_skill
weighted = (w_k * kill_rate) + (w_d * destroy_rate) + (w_g * group_rate) + (w_s * skill_rate)
threat_level = int(round(100 * min(1.0, max(0.0, weighted))))
# Update pirate record
pirates = dm.get('pirates')
def match(p): return p['account_name'].lower() == acct_l
rec0 = next((p for p in pirates if match(p)), None)
if rec0:
dm.remove('pirates', match)
rec0.update({'threat_level': threat_level, 'encounter_count': total})
dm.add('pirates', rec0)
await interaction.response.send_message(
f"✅ Encounter recorded for **{canonical_acct}**.",
ephemeral=True
)
await self.cog._refresh_pirates_list(interaction.guild)
# -------------- Cog: commands + listeners ---------------
class PirateReportCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
r = cfg(bot)
# Channels
self.mod_channel = r.int('mod_channel_id', 0)
self.modlog_channel_id = r.int('modlog_channel_id', 0)
# Threat config
self.group_threshold = r.int('threat_group_threshold', 3)
def _f(key: str, default: float) -> float:
try:
return float(r.get(key, str(default)))
except Exception:
return default
self.w_kill = _f('threat_w_kill', 0.35)
self.w_destruction = _f('threat_w_destruction', 0.30)
self.w_group = _f('threat_w_group', 0.20)
self.w_skill = _f('threat_w_skill', 0.15)
async def _refresh_pirates_list(self, guild: discord.Guild):
plist = self.bot.get_cog('PiratesListCog')
if plist:
await plist.refresh_list(guild)
async def _modlog(self, guild: discord.Guild, content: str):
if not self.modlog_channel_id:
return
ch = guild.get_channel(self.modlog_channel_id)
if ch:
try:
await ch.send(content)
except Exception as e:
print("[pirate_report] modlog send failed:", repr(e))
self.bot.data_manager.add('modlog', {'ts': time.time(), 'guild_id': guild.id, 'content': content})
# --- Resolver: name/account -> pirate record (handles ambiguity) ---
def resolve_pirate(self, ident: str):
"""Return (pirate_record or None, error_message or None)."""
dm = self.bot.data_manager
pirates = dm.get('pirates')
s = ident.strip()
s_l = s.lower()
if '#' in s and _acct_ok(s):
rec = next((p for p in pirates if p['account_name'].lower() == s_l), None)
return (rec, None)
# resolve by character; must be unambiguous
matches = [p for p in pirates if p['character_name'].lower() == s_l]
if not matches:
return (None, None)
if len(matches) > 1:
return (None, "❌ That character name is **ambiguous**. Please use the account (e.g. `Name#12345`).")
return (matches[0], None)
# Remove pirate (mod-only)
@commands.hybrid_command(name='remove_pirate', description='Remove an approved pirate entry')
async def remove_pirate(self, ctx, account_name: str):
if not await require_mod_ctx(ctx, "This command is restricted to moderators."):
return
acct_lower = account_name.strip().lower()
dm = self.bot.data_manager
if not any(p['account_name'].lower() == acct_lower for p in dm.get('pirates')):
return await ctx.reply("Pirate not found.")
dm.remove('pirates', lambda p: p['account_name'].lower() == acct_lower)
await self._modlog(ctx.guild, f"🗑️ Removed pirate {account_name} by {ctx.author.mention}")
await self._refresh_pirates_list(ctx.guild)
await ctx.reply("Removed.", ephemeral=hasattr(ctx, "interaction") and ctx.interaction is not None)
# Modal launchers
@app_commands.command(name="report", description="Submit a pirate report for moderator review (opens a form)")
async def report(self, interaction: discord.Interaction):
if not interaction.guild:
return await interaction.response.send_message("Use this in a server.", ephemeral=True)
await interaction.response.send_modal(ReportModal(self))
@app_commands.command(name="edit_pirate", description="Edit a pirate entry (opens a form)")
async def edit_pirate(self, interaction: discord.Interaction):
if not await require_mod_interaction(interaction):
return
await interaction.response.send_modal(EditPirateModal(self))
@app_commands.command(name="encounter", description="Log a pirate encounter (opens a form)")
async def encounter(self, interaction: discord.Interaction):
if not interaction.guild:
return await interaction.response.send_message("Use this in a server.", ephemeral=True)
await interaction.response.send_modal(EncounterModal(self))
# ---- Migration: convert encounter identifiers to accounts (mod-only) ----
@commands.hybrid_command(name='encounters_migrate_ids', description='Migrate encounter identifiers to account names')
async def encounters_migrate_ids(self, ctx: commands.Context):
if not await require_mod_ctx(ctx, "This command is restricted to moderators."):
return
dm = self.bot.data_manager
pirates = dm.get('pirates')
by_char = {}
for p in pirates:
by_char.setdefault(p['character_name'].lower(), []).append(p)
by_acct = {p['account_name'].lower(): p for p in pirates}
changed = 0
ambiguous = 0
missing = 0
already = 0
for e in dm.get('encounters'):
ident = e.get('identifier', '')
if not ident:
continue
if '#' in ident:
already += 1
continue
key = ident.lower()
matches = by_char.get(key, [])
if not matches:
missing += 1
continue
if len(matches) > 1:
ambiguous += 1
continue
acct = matches[0]['account_name']
# update this one entry atomically
def pred(x, ts=e['timestamp'], rid=e['reporter_id'], ident_old=ident):
return x.get('timestamp') == ts and x.get('reporter_id') == rid and x.get('identifier') == ident_old
def upd(x, acct_new=acct):
x['identifier'] = acct_new
return x
ok = dm.update('encounters', pred, upd)
if ok:
changed += 1
await ctx.reply(
f"Migration complete.\n"
f"- Updated to accounts: **{changed}**\n"
f"- Already accounts: **{already}**\n"
f"- Ambiguous character names (skipped): **{ambiguous}**\n"
f"- Not found in pirates list (skipped): **{missing}**",
ephemeral=hasattr(ctx, "interaction") and ctx.interaction is not None
)
await self._refresh_pirates_list(ctx.guild)
# Moderator reaction handling (atomic claim)
@commands.Cog.listener()
async def on_raw_reaction_add(self, payload: discord.RawReactionActionEvent):
if (
payload.channel_id != self.mod_channel
or str(payload.emoji) not in (CHECK, CROSS)
or payload.user_id == self.bot.user.id
):
return
channel = self.bot.get_channel(payload.channel_id)
if not channel or not channel.guild:
return
if not is_moderator_userid(channel.guild, payload.user_id, self.bot):
return
try:
msg = await channel.fetch_message(payload.message_id)
except Exception as e:
print("[pirate_report] fetch mod msg failed:", repr(e))
return
dm = self.bot.data_manager
# Atomically claim this report
claimed = dm.update(
'reports',
lambda r: r.get('report_id') == msg.id and r.get('status') == 'pending',
lambda r: (r.update({'status': 'deciding'}), r)[1]
)
if not claimed:
return
report = next((r for r in dm.get('reports') if r.get('report_id') == msg.id), None)
if not report:
return
approved = (str(payload.emoji) == CHECK)
guild = channel.guild
stamp = _now_utc_str()
header_emoji = CHECK if approved else CROSS
new_content = (
f"{header_emoji} **Pirate Report**\n"
f"**Character:** {report['character_name']}\n"
f"**Account:** {report['account_name']}\n"
f"**Submitted by:** <@{report['submitter_id']}>\n\n"
f"**Status:** {'Approved' if approved else 'Rejected'} by <@{payload.user_id}> on {stamp}"
)
try:
await msg.clear_reactions()
except Exception as e:
print("[pirate_report] clear reactions failed:", repr(e))
try:
await msg.edit(content=new_content)
except Exception as e:
print("[pirate_report] edit mod msg failed:", repr(e))
if approved:
acct_lower = report['account_name'].lower()
if not any(p['account_name'].lower() == acct_lower for p in dm.get('pirates')):
dm.add('pirates', {
'character_name': report['character_name'],
'account_name': report['account_name'],
'submitter': str(report['submitter_id']),
'threat_level': 0,
'encounter_count': 0,
'added_ts': time.time()
})
await self._refresh_pirates_list(guild)
await self._modlog(guild, f"{'✅ Approved' if approved else '❌ Rejected'} report for {report['character_name']} ({report['account_name']}) by <@{payload.user_id}>.")
try:
if report.get('origin_channel_id') and report.get('ack_message_id'):
ch = guild.get_channel(report['origin_channel_id'])
if ch:
ack_msg = await ch.fetch_message(report['ack_message_id'])
await ack_msg.edit(content=f"{('✅ Approved' if approved else '❌ Rejected')} pirate report of {report['character_name']} ({report['account_name']}) by <@{payload.user_id}> on {stamp}.")
except Exception as e:
print("[pirate_report] edit ack msg failed:", repr(e))
dm.remove('reports', lambda r: r.get('report_id') == msg.id)
async def setup(bot):
cog = PirateReportCog(bot)
await bot.add_cog(cog)
# Register commands either globally or to a specific guild if configured
home_gid = cfg(bot).int('home_guild_id', 0)
if home_gid:
guild_obj = discord.Object(id=home_gid)
bot.tree.add_command(cog.report, guild=guild_obj)
bot.tree.add_command(cog.edit_pirate, guild=guild_obj)
bot.tree.add_command(cog.encounter, guild=guild_obj)
else:
bot.tree.add_command(cog.report)
bot.tree.add_command(cog.edit_pirate)
bot.tree.add_command(cog.encounter)