modified: bot.py

This commit is contained in:
SimolZimol
2026-06-24 11:48:14 +02:00
parent ac301f3370
commit db18516193

388
bot.py
View File

@@ -6074,20 +6074,20 @@ async def viewwarn(ctx, warning_id: int):
@client.hybrid_command()
async def viewmute(ctx, identifier: str):
"""View detailed information about a specific mute (Requires Permission Level 5 or higher)
Parameters:
- identifier: Mute ID (e.g. 123) or Process UUID (e.g. abc123def-456...)
"""
is_slash_command = hasattr(ctx, "interaction") and ctx.interaction is not None
is_slash_command = hasattr(ctx, 'interaction') and ctx.interaction
if is_slash_command:
await ctx.defer()
async def send_response(content=None, embed=None, ephemeral=False, file=None, files=None):
try:
if is_slash_command:
if hasattr(ctx, "followup") and ctx.followup:
if hasattr(ctx, 'followup') and ctx.followup:
await ctx.followup.send(content=content, embed=embed, ephemeral=ephemeral, file=file, files=files or [])
elif hasattr(ctx, "response") and not ctx.response.is_done():
elif hasattr(ctx, 'response') and not ctx.response.is_done():
await ctx.response.send_message(content=content, embed=embed, ephemeral=ephemeral, file=file, files=files or [])
else:
await ctx.send(content=content, embed=embed, file=file, files=files or [])
@@ -6103,42 +6103,41 @@ async def viewmute(ctx, identifier: str):
except Exception as fallback_error:
logger.error(f"Fallback send also failed: {fallback_error}")
async def build_attachment_field(attachments_json: str) -> tuple[str, list[discord.File]]:
def build_attachment_field(attachments_json: str) -> tuple[str, list[discord.File]]:
"""
Parses attachment JSON, loads local files, returns (field_value_text, discord_files).
Parses stored attachments JSON, loads local files.
Returns (field_value_text, list_of_discord_Files).
"""
field_text = ""
discord_files = []
try:
atts = json.loads(attachments_json) if isinstance(attachments_json, str) else attachments_json
if not atts:
return field_text, discord_files
local_files = []
field_lines = []
for att in atts:
filename = att.get("filename", "Unknown")
try:
attachments = json.loads(attachments_json) if isinstance(attachments_json, str) else attachments_json
for att in attachments:
local_path = att.get("local_path")
filename = att.get("filename", "unknown_file")
size = att.get("size", 0)
size_kb = round(size / 1024, 1) if size else "?"
content_type = att.get("content_type", "unknown")
size_str = f"{size / 1024:.1f} KB" if size else "?"
if local_path and os.path.exists(local_path):
field_text += f"• ✅ `{filename}` ({size_kb} KB, {content_type})\n"
discord_files.append(discord.File(local_path, filename=filename))
try:
local_files.append(discord.File(local_path, filename=filename))
field_lines.append(f"✅ `{filename}` ({size_str}) — *locally archived*")
except Exception as e:
logger.error(f"Could not load local file {local_path}: {e}")
field_lines.append(f"⚠️ `{filename}` ({size_str}) — *file unreadable*")
else:
# File missing locally, try original URL as fallback info
original_url = att.get("original_url", "")
stored = att.get("stored", False)
if stored:
field_text += f"• ❌ `{filename}` — locally saved file missing\n"
elif original_url:
field_text += f"• ⚠️ `{filename}` — [original CDN link]({original_url}) (may be expired)\n"
if original_url:
field_lines.append(f"❌ `{filename}` ({size_str}) — *local file missing* ([original URL]({original_url}) may be expired)")
else:
field_text += f"• ❓ `{filename}` — not available\n"
field_lines.append(f" `{filename}` ({size_str}) — *local file missing, no URL*")
except Exception as e:
logger.error(f"Error building attachment field: {e}")
field_text = "⚠️ Error loading attachment data."
logger.error(f"Error parsing attachments JSON: {e}")
field_lines.append("⚠️ Could not parse attachment data.")
return field_text.strip(), discord_files
return "\n".join(field_lines) if field_lines else None, local_files
try:
mod_data = await load_user_data(ctx.author.id, ctx.guild.id)
@@ -6160,19 +6159,21 @@ async def viewmute(ctx, identifier: str):
connection = connect_to_database()
cursor = connection.cursor()
# ── Primary lookup: user_mutes table ──────────────────────────────
if is_mute_id:
cursor.execute(
"SELECT * FROM user_mutes WHERE id = %s AND guild_id = %s LIMIT 1",
(int(identifier), ctx.guild.id)
)
select_query = """
SELECT * FROM user_mutes
WHERE id = %s AND guild_id = %s
LIMIT 1
"""
cursor.execute(select_query, (int(identifier), ctx.guild.id))
else:
cursor.execute(
"""SELECT * FROM user_mutes
WHERE process_uuid = %s AND guild_id = %s
ORDER BY created_at DESC LIMIT 1""",
(identifier, ctx.guild.id)
)
select_query = """
SELECT * FROM user_mutes
WHERE process_uuid = %s AND guild_id = %s
ORDER BY created_at DESC
LIMIT 1
"""
cursor.execute(select_query, (identifier, ctx.guild.id))
mute_result = cursor.fetchone()
@@ -6180,26 +6181,26 @@ async def viewmute(ctx, identifier: str):
columns = [desc[0] for desc in cursor.description]
mute_data = dict(zip(columns, mute_result))
muted_user = await client.fetch_user(int(mute_data["user_id"]))
moderator = await client.fetch_user(int(mute_data["moderator_id"]))
channel = ctx.guild.get_channel(int(mute_data["channel_id"])) if mute_data.get("channel_id") else None
muted_user = await client.fetch_user(int(mute_data['user_id']))
moderator = await client.fetch_user(int(mute_data['moderator_id']))
channel = ctx.guild.get_channel(int(mute_data['channel_id'])) if mute_data['channel_id'] else None
embed = discord.Embed(
title=f"🔇 Mute Details — ID: {mute_data['id']}",
color=0xff0000,
timestamp=mute_data["created_at"]
timestamp=mute_data['created_at']
)
embed.add_field(name="👤 Muted User", value=f"{muted_user.mention}\n`{muted_user.id}`", inline=True)
embed.add_field(name="👮 Moderator", value=f"{moderator.mention}\n`{moderator.id}`", inline=True)
embed.add_field(name="📅 Muted At", value=f"<t:{int(mute_data['start_time'].timestamp())}:F>", inline=True)
embed.add_field(name="👤 Muted User", value=f"{muted_user.mention}\n`{muted_user.id}`", inline=True)
embed.add_field(name="👮 Moderator", value=f"{moderator.mention}\n`{moderator.id}`", inline=True)
embed.add_field(name="📅 Muted At", value=f"<t:{int(mute_data['start_time'].timestamp())}:F>", inline=True)
status_emoji = {"active": "🟢", "completed": "", "expired": "", "cancelled": ""}.get(mute_data["status"], "")
aktiv_status = "🟢 Active" if mute_data["aktiv"] else "🔴 Inactive"
status_emoji = {"active": "🟢", "completed": "", "expired": "", "cancelled": ""}.get(mute_data['status'], "")
aktiv_status = "🟢 Active" if mute_data['aktiv'] else "🔴 Inactive"
embed.add_field(name="📊 Status", value=f"{status_emoji} **{mute_data['status'].title()}** ({aktiv_status})", inline=True)
if mute_data.get("end_time"):
if mute_data["aktiv"] and mute_data["status"] == "active":
if mute_data['end_time']:
if mute_data['aktiv'] and mute_data['status'] == 'active':
embed.add_field(
name="⏰ Ends At",
value=f"<t:{int(mute_data['end_time'].timestamp())}:F>\n<t:{int(mute_data['end_time'].timestamp())}:R>",
@@ -6208,202 +6209,213 @@ async def viewmute(ctx, identifier: str):
else:
embed.add_field(name="⏰ Ended At", value=f"<t:{int(mute_data['end_time'].timestamp())}:F>", inline=True)
embed.add_field(name="⏱️ Duration", value=mute_data["duration"], inline=True)
embed.add_field(name="⏱️ Duration", value=mute_data['duration'], inline=True)
# Time remaining (only when active)
if mute_data["aktiv"] and mute_data["status"] == "active" and mute_data.get("end_time"):
# Time remaining if active
if mute_data['aktiv'] and mute_data['status'] == 'active' and mute_data['end_time']:
now = datetime.now()
if mute_data["end_time"] > now:
delta = mute_data["end_time"] - now
days = delta.days
hours, remainder = divmod(delta.seconds, 3600)
if mute_data['end_time'] > now:
time_left = mute_data['end_time'] - now
days = time_left.days
hours, remainder = divmod(time_left.seconds, 3600)
minutes, _ = divmod(remainder, 60)
parts = []
if days: parts.append(f"{days}d")
if hours: parts.append(f"{hours}h")
if minutes: parts.append(f"{minutes}m")
if days > 0:
parts.append(f"{days}d")
if hours > 0:
parts.append(f"{hours}h")
if minutes > 0:
parts.append(f"{minutes}m")
embed.add_field(
name="⏳ Time Remaining",
value=" ".join(parts) if parts else "Less than 1 minute",
inline=True
)
embed.add_field(name="📝 Reason", value=mute_data.get("reason") or "No reason provided", inline=False)
embed.add_field(name="📝 Reason", value=mute_data['reason'] or "No reason provided", inline=False)
if channel:
embed.add_field(name="📍 Channel", value=f"{channel.mention}\n`{channel.id}`", inline=True)
if mute_data.get("mute_role_id"):
mute_role = ctx.guild.get_role(int(mute_data["mute_role_id"]))
if mute_data['mute_role_id']:
mute_role = ctx.guild.get_role(int(mute_data['mute_role_id']))
if mute_role:
embed.add_field(name="🎭 Mute Role", value=f"{mute_role.mention}\n`{mute_role.id}`", inline=True)
else:
embed.add_field(name="🎭 Mute Role", value=f"❌ Deleted Role\n`{mute_data['mute_role_id']}`", inline=True)
if not mute_data["aktiv"] and mute_data.get("unmuted_at"):
if not mute_data['aktiv'] and mute_data['unmuted_at']:
unmute_info = f"<t:{int(mute_data['unmuted_at'].timestamp())}:F>"
if mute_data.get("unmuted_by"):
unmuter = await client.fetch_user(int(mute_data["unmuted_by"]))
if mute_data['unmuted_by']:
unmuter = await client.fetch_user(int(mute_data['unmuted_by']))
unmute_info += f"\nBy: {unmuter.mention}"
if mute_data.get("auto_unmuted"):
if mute_data['auto_unmuted']:
unmute_info += "\n🤖 Automatic unmute"
embed.add_field(name="🔓 Unmuted At", value=unmute_info, inline=True)
# ── Referenced message (from message_data JSON in DB) ──────────
raw_msg_data = mute_data.get("message_data")
local_files = []
# ── Referenced message + locally archived attachments ──────────
local_files = []
if raw_msg_data:
raw_message_data = mute_data.get('message_data')
if raw_message_data:
try:
msg_data = json.loads(raw_msg_data) if isinstance(raw_msg_data, str) else raw_msg_data
msg_data = json.loads(raw_message_data) if isinstance(raw_message_data, str) else raw_message_data
# Determine which format (new context dict or flat dict)
# Determine which format the stored message_data uses
if isinstance(msg_data, dict) and "main_message" in msg_data:
main_msg = msg_data.get("main_message") or {}
atts_json = main_msg.get("attachments", "[]")
msg_id = main_msg.get("id", mute_data.get("message_id", "?"))
msg_ch_id = main_msg.get("channel_id", "?")
msg_author = main_msg.get("author_name", "Unknown")
msg_content = main_msg.get("content", "")
main_msg = msg_data.get("main_message") or {}
attachments_json = main_msg.get("attachments")
# Build referenced message field
msg_info = f"**Message ID:** `{main_msg.get('id', 'N/A')}`\n"
msg_info += f"**Channel:** <#{main_msg.get('channel_id', 'N/A')}>\n"
msg_info += f"**Author:** {main_msg.get('author_name', 'Unknown')}\n"
if main_msg.get("content"):
preview = main_msg["content"][:150] + "..." if len(main_msg["content"]) > 150 else main_msg["content"]
msg_info += f"**Content:** {preview}"
embed.add_field(name="📄 Referenced Message", value=msg_info, inline=False)
else:
atts_json = msg_data.get("attachments", "[]")
msg_id = msg_data.get("id", mute_data.get("message_id", "?"))
msg_ch_id = msg_data.get("channel_id", "?")
msg_author = msg_data.get("author_name") or f"<@{msg_data.get('author_id', '?')}>"
msg_content = msg_data.get("content", "")
# Flat format (honeypot / direct archive)
attachments_json = msg_data.get("attachments")
msg_info = f"**Message ID:** `{msg_id}`\n"
msg_info += f"**Channel:** <#{msg_ch_id}>\n"
msg_info += f"**Author:** {msg_author}\n"
if msg_content:
preview = msg_content[:150] + "..." if len(msg_content) > 150 else msg_content
msg_info += f"**Content:** {preview}"
msg_info = f"**Message ID:** `{msg_data.get('id', 'N/A')}`\n"
msg_info += f"**Channel:** <#{msg_data.get('channel_id', 'N/A')}>\n"
author_id = msg_data.get("author_id")
author_name = msg_data.get("author_name", "Unknown")
msg_info += f"**Author:** {f'<@{author_id}>' if author_id else author_name}\n"
if msg_data.get("content"):
preview = msg_data["content"][:150] + "..." if len(msg_data["content"]) > 150 else msg_data["content"]
msg_info += f"**Content:** {preview}"
embed.add_field(name="📄 Referenced Message", value=msg_info, inline=False)
embed.add_field(name="📄 Referenced Message", value=msg_info, inline=False)
# ── Attachment section ─────────────────────────────────
if atts_json and atts_json != "[]":
att_field_text, local_files = await build_attachment_field(atts_json)
# Load local attachment files
if attachments_json:
att_field_text, local_files = build_attachment_field(attachments_json)
if att_field_text:
stored_count = len(local_files)
total_count = len(json.loads(atts_json) if isinstance(atts_json, str) else atts_json)
field_title = f"📎 Archived Attachments ({stored_count}/{total_count} available locally)"
embed.add_field(name=field_title, value=att_field_text, inline=False)
embed.add_field(
name=f"📎 Archived Attachments ({len(local_files)} file(s) attached)" if local_files else "📎 Archived Attachments",
value=att_field_text,
inline=False
)
except Exception as e:
logger.error(f"Error parsing message_data in viewmute: {e}")
embed.add_field(name="📄 Referenced Message", value="⚠️ Error loading message data.", inline=False)
logger.error(f"Error parsing message_data for viewmute {identifier}: {e}")
embed.add_field(name="📄 Referenced Message", value="⚠️ Could not parse archived message data.", inline=False)
# Fallback: if no message_data but message_id + message_content columns exist
elif mute_data.get("message_id") and mute_data.get("message_content"):
preview = mute_data["message_content"][:100] + "..." if len(mute_data["message_content"]) > 100 else mute_data["message_content"]
elif mute_data.get('message_id') and mute_data.get('message_content'):
# Legacy fallback: separate columns (no attachment support)
content_preview = mute_data['message_content'][:100] + "..." if len(mute_data['message_content']) > 100 else mute_data['message_content']
embed.add_field(
name="📄 Referenced Message",
value=f"ID: `{mute_data['message_id']}`\nContent: {preview}",
value=f"ID: `{mute_data['message_id']}`\nContent: {content_preview}",
inline=False
)
# ────────────────────────────────────────────────────────────────
embed.add_field(name="🆔 Process UUID", value=f"`{mute_data['process_uuid']}`", inline=False)
embed.add_field(name="🆔 Mute Record ID", value=f"`{mute_data['id']}`", inline=True)
embed.add_field(name="🆔 Process UUID", value=f"`{mute_data['process_uuid']}`", inline=False)
embed.add_field(name="🆔 Mute Record ID", value=f"`{mute_data['id']}`", inline=True)
embed.set_thumbnail(url=muted_user.display_avatar.url)
embed.set_footer(text=f"Mute Record from Database | Server: {ctx.guild.name}")
# Send embed + local files (if any)
if local_files:
# Send embed + files together
await send_response(embed=embed, files=local_files[:10])
if len(local_files) > 10:
# Discord allows max 10 files per message
await send_response(
content=f"📎 +{len(local_files) - 10} more attachment(s) exceeded the Discord file limit.",
ephemeral=True
)
else:
await send_response(embed=embed)
return
# ── Fallback lookup: active_processes table ────────────────────────
if not is_mute_id:
cursor.execute(
"""SELECT uuid, process_type, guild_id, channel_id, user_id, target_id,
created_at, end_time, status, data
FROM active_processes
WHERE uuid = %s AND guild_id = %s AND process_type = 'mute'""",
(identifier, ctx.guild.id)
)
result = cursor.fetchone()
else:
# Fallback: active_processes table (UUID only)
if not is_mute_id:
select_query = """
SELECT uuid, process_type, guild_id, channel_id, user_id, target_id,
created_at, end_time, status, data
FROM active_processes
WHERE uuid = %s AND guild_id = %s AND process_type = 'mute'
"""
cursor.execute(select_query, (identifier, ctx.guild.id))
result = cursor.fetchone()
if result:
uuid, process_type, guild_id, channel_id, user_id, target_id, created_at, end_time, status, data = result
proc_data = json.loads(data) if data else {}
muted_user = await client.fetch_user(target_id)
moderator_id = proc_data.get("moderator_id", user_id)
moderator = await client.fetch_user(moderator_id)
channel = ctx.guild.get_channel(channel_id) if channel_id else None
if result:
uuid, process_type, guild_id, channel_id, user_id, target_id, created_at, end_time, status, data = result
proc_data = json.loads(data) if data else {}
embed = discord.Embed(
title=f"🔇 Mute Details — UUID: {str(uuid)[:8]}",
color=0xff0000,
timestamp=created_at
)
embed.add_field(name="👤 Muted User", value=f"{muted_user.mention}\n`{muted_user.id}`", inline=True)
embed.add_field(name="👮 Moderator", value=f"{moderator.mention}\n`{moderator_id}`", inline=True)
embed.add_field(name="📅 Muted At", value=f"<t:{int(created_at.timestamp())}:F>", inline=True)
muted_user = await client.fetch_user(target_id)
moderator_id = proc_data.get('moderator_id', user_id)
moderator = await client.fetch_user(moderator_id)
channel = ctx.guild.get_channel(channel_id) if channel_id else None
status_emoji = {"active": "🟢", "completed": "", "expired": "", "cancelled": ""}.get(status, "")
embed.add_field(name="📊 Status", value=f"{status_emoji} **{status.title()}**", inline=True)
if end_time:
if status == "active":
embed.add_field(
name="⏰ Ends At",
value=f"<t:{int(end_time.timestamp())}:F>\n<t:{int(end_time.timestamp())}:R>",
inline=True
)
else:
embed.add_field(name="⏰ Ended At", value=f"<t:{int(end_time.timestamp())}:F>", inline=True)
if status == "active" and end_time and end_time > datetime.now():
delta = end_time - datetime.now()
days = delta.days
hours, remainder = divmod(delta.seconds, 3600)
minutes, _ = divmod(remainder, 60)
parts = []
if days: parts.append(f"{days}d")
if hours: parts.append(f"{hours}h")
if minutes: parts.append(f"{minutes}m")
embed.add_field(
name="⏳ Time Remaining",
value=" ".join(parts) if parts else "Less than 1 minute",
inline=True
embed = discord.Embed(
title=f"🔇 Mute Details — UUID: {str(uuid)[:8]}",
color=0xff0000,
timestamp=created_at
)
embed.add_field(name="👤 Muted User", value=f"{muted_user.mention}\n`{muted_user.id}`", inline=True)
embed.add_field(name="👮 Moderator", value=f"{moderator.mention}\n`{moderator_id}`", inline=True)
embed.add_field(name="📅 Muted At", value=f"<t:{int(created_at.timestamp())}:F>", inline=True)
embed.add_field(name="📝 Reason", value=proc_data.get("reason") or "No reason provided", inline=False)
status_emoji = {"active": "🟢", "completed": "", "expired": "", "cancelled": ""}.get(status, "")
embed.add_field(name="📊 Status", value=f"{status_emoji} **{status.title()}**", inline=True)
if channel:
embed.add_field(name="📍 Channel", value=f"{channel.mention}\n`{channel.id}`", inline=True)
if end_time:
if status == "active":
embed.add_field(
name="⏰ Ends At",
value=f"<t:{int(end_time.timestamp())}:F>\n<t:{int(end_time.timestamp())}:R>",
inline=True
)
else:
embed.add_field(name="⏰ Ended At", value=f"<t:{int(end_time.timestamp())}:F>", inline=True)
mute_role_id = proc_data.get("mute_role_id")
if mute_role_id:
mute_role = ctx.guild.get_role(int(mute_role_id))
if mute_role:
embed.add_field(name="🎭 Mute Role", value=f"{mute_role.mention}\n`{mute_role.id}`", inline=True)
else:
embed.add_field(name="🎭 Mute Role", value=f"❌ Deleted Role\n`{mute_role_id}`", inline=True)
reason = proc_data.get('reason', 'No reason provided')
embed.add_field(name="📝 Reason", value=reason, inline=False)
embed.add_field(name="🆔 Process UUID", value=f"`{uuid}`", inline=False)
embed.set_thumbnail(url=muted_user.display_avatar.url)
embed.set_footer(text=f"Process Type: {process_type.title()} | Server: {ctx.guild.name}")
if channel:
embed.add_field(name="📍 Channel", value=f"{channel.mention}\n`{channel.id}`", inline=True)
await send_response(embed=embed)
return
mute_role_id = proc_data.get('mute_role_id')
if mute_role_id:
mute_role = ctx.guild.get_role(mute_role_id)
if mute_role:
embed.add_field(name="🎭 Mute Role", value=f"{mute_role.mention}\n`{mute_role.id}`", inline=True)
else:
embed.add_field(name="🎭 Mute Role", value=f"❌ Deleted Role\n`{mute_role_id}`", inline=True)
# ── Nothing found ──────────────────────────────────────────────────
embed = discord.Embed(
title="❌ Mute Not Found",
description=f"No mute with {'ID' if is_mute_id else 'UUID'} `{identifier}` found in this server.",
color=0xff0000
)
await send_response(embed=embed, ephemeral=True)
if status == "active" and end_time:
now = datetime.now()
if end_time > now:
time_left = end_time - now
days = time_left.days
hours, remainder = divmod(time_left.seconds, 3600)
minutes, _ = divmod(remainder, 60)
parts = []
if days > 0:
parts.append(f"{days}d")
if hours > 0:
parts.append(f"{hours}h")
if minutes > 0:
parts.append(f"{minutes}m")
embed.add_field(
name="⏳ Time Remaining",
value=" ".join(parts) if parts else "Less than 1 minute",
inline=True
)
embed.add_field(name="🆔 Process UUID", value=f"`{uuid}`", inline=False)
embed.set_thumbnail(url=muted_user.display_avatar.url)
embed.set_footer(text=f"Process Type: {process_type.title()} | Server: {ctx.guild.name}")
await send_response(embed=embed)
return
embed = discord.Embed(
title="❌ Mute Not Found",
description=f"No mute with {'ID' if is_mute_id else 'UUID'} `{identifier}` found in this server.",
color=0xff0000
)
await send_response(embed=embed, ephemeral=True)
return
finally:
if cursor:
@@ -6418,7 +6430,7 @@ async def viewmute(ctx, identifier: str):
description="An error occurred while retrieving mute details. Please try again.",
color=0xff0000
)
await send_response(embed=embed, ephemeral=True)
await send_response(embed=embed)
@client.hybrid_command()
async def removewarn(ctx, warning_id: int):