diff --git a/ghostwipe/ghostwipe.py b/ghostwipe/ghostwipe.py index ae04f73..c8746d2 100644 --- a/ghostwipe/ghostwipe.py +++ b/ghostwipe/ghostwipe.py @@ -10,6 +10,7 @@ classified via the audit log purely for reporting/toggle purposes. from __future__ import annotations import asyncio +import base64 import json import logging import time @@ -47,6 +48,8 @@ COLOUR_ERR = discord.Colour.red() MAX_HISTORY_ENTRIES = 50 DISCORD_ATTACHMENT_LIMIT = 8 * 1024 * 1024 # 8MB, safe default (no boost assumptions) IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp") +AVATAR_MAX_BYTES = 2 * 1024 * 1024 # avatars are always small; bail if something's wrong +AVATAR_FETCH_TIMEOUT = 5 def _humanize_size(num_bytes: int) -> str: @@ -176,6 +179,39 @@ class GhostWipe(commands.Cog): "stickers": stickers, } + async def _fetch_avatar_data_uri(self, url: str) -> Optional[str]: + """Download the member's avatar and inline it as a base64 data URI. + + The report is a standalone file that may be opened offline or in a + viewer that blocks remote network requests, so a plain https:// URL + isn't reliable there. Embedding the bytes directly guarantees the + avatar always renders. Best-effort: returns None on any failure and + the report falls back to a plain initial-letter avatar. + """ + if not url: + return None + try: + import aiohttp + + async with aiohttp.ClientSession() as session: + async with session.get( + url, timeout=aiohttp.ClientTimeout(total=AVATAR_FETCH_TIMEOUT) + ) as resp: + if resp.status != 200: + return None + if resp.content_length and resp.content_length > AVATAR_MAX_BYTES: + return None + data = await resp.content.read(AVATAR_MAX_BYTES + 1) + if len(data) > AVATAR_MAX_BYTES: + return None + content_type = resp.content_type or "image/png" + except Exception: + LOG.debug("GhostWipe failed to download avatar for report, using fallback.", exc_info=True) + return None + + encoded = base64.b64encode(data).decode("ascii") + return f"data:{content_type};base64,{encoded}" + async def _gather_target_channels( self, guild: discord.Guild, conf: Dict[str, Any] ) -> List[Tuple[str, discord.abc.Messageable, str]]: @@ -200,16 +236,19 @@ class GhostWipe(commands.Cog): targets.append(("thread", th, th.name)) if conf["include_archived_threads"]: - for ch in guild.text_channels: - if ch.id in ignored: - continue + async def _archived_for(ch: discord.TextChannel) -> List[discord.Thread]: try: - async for th in ch.archived_threads(limit=100): - if th.id in ignored: - continue - targets.append(("thread", th, th.name)) + return [th async for th in ch.archived_threads(limit=100)] except (discord.Forbidden, discord.HTTPException): - continue + return [] + + candidates = [ch for ch in guild.text_channels if ch.id not in ignored] + results = await asyncio.gather(*(_archived_for(ch) for ch in candidates)) + for threads in results: + for th in threads: + if th.id in ignored: + continue + targets.append(("thread", th, th.name)) return targets @@ -292,6 +331,15 @@ class GhostWipe(commands.Cog): moderator: Optional[str] = None, mod_reason: Optional[str] = None, ) -> None: + event_time = datetime.now(timezone.utc) + placeholder_message = await self._send_placeholder_log( + guild, conf, member_display, member.id, member_avatar_url, reason, conf["dry_run"] + ) + + # Kick off the avatar download alongside the channel scan instead of after + # it β€” they're independent, so there's no reason to pay for both in series. + avatar_task = asyncio.create_task(self._fetch_avatar_data_uri(member_avatar_url)) + targets = await self._gather_target_channels(guild, conf) history_limit = conf["history_limit"] dry_run = conf["dry_run"] @@ -303,7 +351,11 @@ class GhostWipe(commands.Cog): async def _worker(kind: str, channel: discord.abc.Messageable): async with semaphore: result = await self._scan_channel(kind, channel, member, history_limit, dry_run) - if rate_delay: + # Only pace ourselves when a channel actually had deletions to make β€” + # there's no delete-rate-limit to protect against on a channel with + # zero matches, and most channels a departing member never posted in + # would otherwise add pure dead time to every single purge. + if rate_delay and not result["skipped"] and result["message_count"] > 0: await asyncio.sleep(rate_delay) return result @@ -312,13 +364,13 @@ class GhostWipe(commands.Cog): total_deleted = sum(c["message_count"] for c in channels_data if not c["skipped"]) event_id = f"{int(time.time())}_{member.id}" - event_time = datetime.now(timezone.utc) + avatar_data_uri = await avatar_task html = generate_report_html( guild_name=guild.name, member_name=member_display, member_id=member.id, - avatar_url=member_avatar_url, + avatar_data_uri=avatar_data_uri, reason=reason, event_time_str=event_time.strftime("%Y-%m-%d %H:%M:%S UTC"), moderator=moderator, @@ -388,7 +440,7 @@ class GhostWipe(commands.Cog): await self._send_log( guild, conf, member_display, member.id, member_avatar_url, reason, moderator, mod_reason, total_deleted, channels_affected, channels_skipped, - dry_run, html_path, channels_data, + dry_run, html_path, channels_data, placeholder_message, ) def _prune_old_reports(self, guild_id: int, keep_days: int) -> None: @@ -403,6 +455,46 @@ class GhostWipe(commands.Cog): except OSError: continue + async def _send_placeholder_log( + self, + guild: discord.Guild, + conf: Dict[str, Any], + member_display: str, + member_id: int, + avatar_url: str, + reason: str, + dry_run: bool, + ) -> Optional[discord.Message]: + """Post an immediate "in progress" notice, edited later with the real results. + + Scanning + purging a whole server can take a few seconds; without this, + the log channel sits silent that whole time with nothing to show for it. + """ + channel_id = conf["log_channel"] + if not channel_id: + return None + log_channel = guild.get_channel(channel_id) + if not isinstance(log_channel, discord.TextChannel): + return None + + title = f"πŸ‘» GhostWipe β€” {REASON_LABEL.get(reason, reason.title())}" + if dry_run: + title += " (Dry Run)" + embed = discord.Embed( + title=title, + description="πŸ”Ž Scanning the server and purging messages β€” results incoming...", + colour=REASON_COLOUR.get(reason, COLOUR_INFO), + ) + embed.set_thumbnail(url=avatar_url) + embed.add_field(name="Member", value=f"{member_display} (`{member_id}`)", inline=True) + try: + return await log_channel.send(embed=embed) + except discord.HTTPException: + LOG.exception( + "Failed to send GhostWipe placeholder log to channel %s in guild %s", channel_id, guild.id + ) + return None + async def _send_log( self, guild: discord.Guild, @@ -419,6 +511,7 @@ class GhostWipe(commands.Cog): dry_run: bool, html_path: Path, channels_data: List[Dict[str, Any]], + placeholder_message: Optional[discord.Message] = None, ) -> None: channel_id = conf["log_channel"] if not channel_id: @@ -464,6 +557,19 @@ class GhostWipe(commands.Cog): except OSError: pass + if placeholder_message is not None: + try: + if file_obj: + await placeholder_message.edit(embed=embed, attachments=[file_obj]) + else: + await placeholder_message.edit(embed=embed) + return + except discord.HTTPException: + LOG.warning( + "GhostWipe failed to edit its placeholder log message in guild %s, sending fresh.", + guild.id, + ) + try: if file_obj: await log_channel.send(embed=embed, file=file_obj) diff --git a/ghostwipe/report.py b/ghostwipe/report.py index e311206..bd96d98 100644 --- a/ghostwipe/report.py +++ b/ghostwipe/report.py @@ -1,24 +1,24 @@ """ GhostWipe HTML report generator. -Builds a single self-contained HTML file (no external resources) that lets a -moderator browse every message GhostWipe deleted for a departing member, -switching between channels, searching content, and inspecting attachments. +Builds a single self-contained HTML file (no external resources, no required +JavaScript) that lets a moderator browse every message GhostWipe deleted for +a departing member. All content β€” stats, avatar, channel list, and every +message β€” is rendered server-side directly into the HTML/CSS, using a pure +CSS radio/label technique for channel switching. This way the report +displays correctly even in minimal viewers that don't execute scripts or +load remote network resources (a common failure mode for local file:// +HTML viewers). A small script is layered on top purely as a progressive +enhancement for live search filtering; if it never runs, everything is +still fully visible via the CSS tab switcher. """ from __future__ import annotations -import html -import json -from typing import Any, Dict, List +import html as html_lib +from typing import Any, Dict, List, Optional -TEMPLATE = r""" - - - - -__PAGE_TITLE__ - - - -
- -
- -
-
-

Select a channel

- -
-
-
-
- -
+""" - - - - """ @@ -387,6 +204,7 @@ def _badge_class(reason: str) -> str: "left": "badge-left", "kicked": "badge-kicked", "banned": "badge-banned", + "manual": "badge-manual", }.get(reason, "badge-left") @@ -395,28 +213,125 @@ def _badge_label(reason: str) -> str: "left": "Left", "kicked": "Kicked", "banned": "Banned", + "manual": "Manual", }.get(reason, reason.title()) +CHANNEL_ICON = {"text": "#", "thread": "\U0001F9F5", "voice": "\U0001F50A"} + + +def _esc(value: Any) -> str: + return html_lib.escape(str(value), quote=True) + + +def _render_message(msg: Dict[str, Any], redact: bool) -> str: + content = msg.get("content") or "" + search_key = _esc(content.lower()) if not redact else "" + parts = [f'
'] + parts.append( + f'
{_esc(msg.get("timestamp", ""))}' + f'#{_esc(msg.get("id", ""))}
' + ) + if redact: + parts.append('
[content redacted by server configuration]
') + else: + text = content if content else ("" if msg.get("attachments") else "[no text content]") + parts.append(f'
{_esc(text)}
') + + attachments = msg.get("attachments") or [] + if attachments: + parts.append('
') + for att in attachments: + if att.get("is_image"): + parts.append( + f'' + ) + else: + size = att.get("size", "") + parts.append( + f'' + f'\U0001F4CE {_esc(att.get("filename", "file"))}' + f'{f" ({_esc(size)})" if size else ""}' + ) + parts.append("
") + + for sticker in msg.get("stickers") or []: + parts.append(f'✨ {_esc(sticker)}') + + parts.append("
") + return "".join(parts) + + +def _render_channel_panel(channel: Dict[str, Any], redact: bool) -> str: + cid = _esc(channel["id"]) + icon = CHANNEL_ICON.get(channel.get("type"), "#") + name = _esc(channel.get("name", cid)) + count = channel.get("message_count", 0) + + body_parts = [] + if channel.get("skipped"): + reason = _esc(channel.get("skip_reason") or "unknown reason") + body_parts.append(f'
This channel was skipped: {reason}
') + else: + messages = channel.get("messages") or [] + if not messages: + body_parts.append('
No messages recorded for this channel.
') + else: + for msg in messages: + body_parts.append(_render_message(msg, redact)) + + return ( + f'
' + f'

{icon} {name}

' + f'{count} message{"s" if count != 1 else ""}
' + f'
{"".join(body_parts)}
' + f"
" + ) + + +def _render_sidebar_item(channel: Dict[str, Any]) -> str: + cid = _esc(channel["id"]) + icon = CHANNEL_ICON.get(channel.get("type"), "#") + name = _esc(channel.get("name", cid)) + if channel.get("skipped"): + right = 'skipped' + cls = "ch-item skipped" + else: + count = channel.get("message_count", 0) + right = f'{count}' + cls = "ch-item" + return ( + f'" + ) + + def generate_report_html( *, guild_name: str, member_name: str, member_id: int, - avatar_url: str, reason: str, event_time_str: str, + channels: List[Dict[str, Any]], + avatar_data_uri: Optional[str] = None, moderator: str = None, mod_reason: str = None, dry_run: bool = False, reveal_content: bool = True, - channels: List[Dict[str, Any]], version: str = "1.0.0", generated_at: str = "", ) -> str: """Build the standalone HTML report string. - ``channels`` is a list of dicts shaped like: + Everything (stats, avatar, channel list, messages) is rendered directly + into the markup here in Python β€” there is no client-side rendering step, + so the report displays fully correctly even with JavaScript disabled or + blocked. ``channels`` is a list of dicts shaped like: { "id": str, "name": str, "type": "text"|"thread"|"voice", "skipped": bool, "skip_reason": str | None, @@ -433,42 +348,105 @@ def generate_report_html( channels_affected = sum(1 for c in channels if not c.get("skipped") and c["message_count"] > 0) channels_skipped = sum(1 for c in channels if c.get("skipped")) - if not reveal_content: - for ch in channels: - for m in ch.get("messages", []): - m["redacted"] = True + # Only list channels that are actually relevant to this event β€” a full + # server-wide sidebar of untouched channels adds nothing but clutter. + # Affected channels (by message count, descending) come first; skipped ones last. + relevant = [c for c in channels if c.get("skipped") or c.get("message_count", 0) > 0] + relevant.sort(key=lambda c: (c.get("skipped", False), -c.get("message_count", 0))) - meta_bits = [f"Server: {guild_name}"] + meta_bits = [f"Server: {html_lib.escape(guild_name)}"] if moderator: - meta_bits.append(f"By: {moderator}") + meta_bits.append(f"By: {html_lib.escape(moderator)}") if mod_reason: - meta_bits.append(f"Reason: {mod_reason}") + meta_bits.append(f"Reason: {html_lib.escape(mod_reason)}") meta_line = " β€’ ".join(meta_bits) - payload = { - "guild_name": guild_name, - "member_name": member_name, - "member_id": str(member_id), - "reason": reason, - "dry_run": dry_run, - "meta_line": meta_line, - "totals": { - "messages_deleted": total_deleted, - "channels_affected": channels_affected, - "channels_skipped": channels_skipped, - }, - "channels": channels, - } + if avatar_data_uri: + avatar_html = f'avatar' + else: + initial = (member_name.strip()[:1] or "?").upper() + avatar_html = f'
{_esc(initial)}
' - out = TEMPLATE - out = out.replace("__PAGE_TITLE__", html.escape(f"GhostWipe Report β€” {member_name}")) - out = out.replace("__AVATAR_URL__", html.escape(avatar_url or "", quote=True)) - out = out.replace("__USERNAME__", html.escape(member_name)) - out = out.replace("__REASON_BADGE_CLASS__", _badge_class(reason)) - out = out.replace("__REASON_LABEL__", _badge_label(reason)) - out = out.replace("__USER_ID__", str(member_id)) - out = out.replace("__GHOSTWIPE_VERSION__", html.escape(version)) - out = out.replace("__GENERATED_AT__", html.escape(generated_at or event_time_str)) - out = out.replace("__REPORT_DATA_JSON__", json.dumps(payload)) - return out + dryrun_badge = ' Dry Run' if dry_run else "" + + radios = [] + sidebar_items = [] + panels = [] + for i, channel in enumerate(relevant): + cid = _esc(channel["id"]) + checked = " checked" if i == 0 else "" + radios.append(f'') + sidebar_items.append(_render_sidebar_item(channel)) + panels.append(_render_channel_panel(channel, not reveal_content)) + + # The radio inputs are siblings of #app (both direct children of ) β€” + # #body/#main/#sidebar are all nested *inside* #app, not siblings of the + # radios themselves, so the `~` combinator has to anchor on #app. + tab_css_rules = [] + for channel in relevant: + cid = _esc(channel["id"]) + tab_css_rules.append( + f'#tab-{cid}:checked ~ #app #main #panel-{cid} {{ display: flex; }}\n' + f'#tab-{cid}:checked ~ #app #sidebar label[for="tab-{cid}"] {{ ' + f'background: var(--accent-soft); color: var(--text); }}' + ) + + if relevant: + main_content = "".join(panels) + sidebar_content = "".join(sidebar_items) + else: + main_content = '
Nothing was deleted for this event.
' + sidebar_content = '
No channels affected.
' + + title = f"GhostWipe Report β€” {html_lib.escape(member_name)}" + + doc = f""" + + + + +{title} + + + +{"".join(radios)} +
+ +
+ +
+ {main_content} +
+
+ +
+ + + +""" + return doc