Migrate built-in commands from hand-crafted per-backend strings to FormattedMessage

timkpaine · timkpaine · commit fdc3e2d4dc97 · 2026-05-27T17:24:58.000-04:00
executor.py — New _extract_attachments() helper converts FormattedImage/FormattedAttachment from a FormattedMessage into base Attachment objects; _coerce_response() now populates Message.attachments when returning a FormattedMessage
echo.py — Replaced manual mention_users() calls with FormattedMessage + UserMention nodes
help.py — Replaced per-backend if/elif rendering (Symphony &lt;expandable-card&gt;, Slack mrkdwn, Discord markdown table) with FormattedMessage + Heading + Table.from_dict_list() that auto-renders for each backend
status.py — Same migration: builds a Table.from_dict_list() of metrics instead of manual **bold** markdown lines
schedule.py — Same migration for the schedule listing
test_command_framework.py — 3 new tests for _coerce_response() covering images in content, file attachments, and mixed content+attachments

Signed-off-by: Tim Paine &lt;3105306+timkpaine@users.noreply.github.com&gt;
diff --git a/csp_bot/commands/echo.py b/csp_bot/commands/echo.py
@@ -1,15 +1,15 @@
 """Echo command for csp-bot.
 
-A simple command that echoes back messages.
+A simple command that echoes back messages using FormattedMessage.
 """
 
 from logging import getLogger
 from typing import Optional, Type
 
 from chatom import Message
+from chatom.format import FormattedMessage, Text, UserMention
 
 from csp_bot.structs import BotCommand
-from csp_bot.utils import mention_users
 
 from .base import BaseCommand, BaseCommandModel, ReplyToOtherCommand
 
@@ -31,21 +31,27 @@ def help(self) -> str:
     def execute(self, command: BotCommand) -> Optional[Message]:
         log.info(f"Echo command: {command.command}")
 
-        # Build content from args
-        content = " ".join(command.args) if command.args else ""
+        text = " ".join(command.args) if command.args else ""
+        has_targets = bool(command.targets)
 
-        # Add mentions for any tagged users
-        if command.targets:
-            mentions = mention_users(list(command.targets), command.backend)
-            if mentions:
-                content = f"{content} {mentions}".strip()
-
-        # If nothing to echo (no args and no targets), return None
-        if not content:
+        if not text and not has_targets:
             return None
 
+        msg = FormattedMessage(metadata={"backend": command.backend})
+        if text:
+            msg.content.append(Text(content=text))
+        for target in command.targets:
+            if msg.content:
+                msg.content.append(Text(content=" "))
+            msg.content.append(
+                UserMention(
+                    user_id=target.id,
+                    display_name=getattr(target, "display_name", "") or getattr(target, "name", "") or "",
+                )
+            )
+
         return Message(
-            content=content,
+            content=msg.render_for(command.backend),
             channel=command.channel,
             metadata={"backend": command.backend},
         )
diff --git a/csp_bot/commands/executor.py b/csp_bot/commands/executor.py
@@ -15,7 +15,9 @@
 from typing import Any, List, Optional, Union
 
 from chatom import Message
+from chatom.base.attachment import Attachment, AttachmentType, Image as BaseImage
 from chatom.format import FormattedMessage
+from chatom.format.attachment import FormattedAttachment, FormattedImage
 
 from csp_bot.structs import BotCommand
 
@@ -43,6 +45,49 @@ def _get_event_loop() -> asyncio.AbstractEventLoop:
     return _loop
 
 
+def _extract_attachments(fm: FormattedMessage) -> list:
+    """Extract base Attachment objects from a FormattedMessage.
+
+    Converts FormattedAttachment/FormattedImage from both the content list
+    and the attachments list into chatom base Attachment/Image objects
+    suitable for Message.attachments.
+    """
+    result = []
+    for node in fm.content:
+        if isinstance(node, FormattedImage):
+            result.append(
+                BaseImage(
+                    url=node.url,
+                    alt_text=node.alt_text,
+                    content_type="image/png",
+                    attachment_type=AttachmentType.IMAGE,
+                )
+            )
+        elif isinstance(node, FormattedAttachment):
+            att_type = Attachment.from_content_type(node.content_type) if node.content_type else AttachmentType.FILE
+            result.append(
+                Attachment(
+                    filename=node.filename,
+                    url=node.url,
+                    size=node.size,
+                    content_type=node.content_type,
+                    attachment_type=att_type,
+                )
+            )
+    for fa in fm.attachments:
+        att_type = Attachment.from_content_type(fa.content_type) if fa.content_type else AttachmentType.FILE
+        result.append(
+            Attachment(
+                filename=fa.filename,
+                url=fa.url,
+                size=fa.size,
+                content_type=fa.content_type,
+                attachment_type=att_type,
+            )
+        )
+    return result
+
+
 def _coerce_response(item: Any, backend: str) -> Optional[Union[Message, BotCommand]]:
     """Coerce a command return value into a chatom Message.
 
@@ -71,8 +116,10 @@ def _coerce_response(item: Any, backend: str) -> Optional[Union[Message, BotComm
 
     if isinstance(item, FormattedMessage):
         rendered = item.render_for(backend)
+        attachments = _extract_attachments(item)
         msg = Message(
             content=rendered,
+            attachments=attachments,
             metadata={"backend": backend, "formatted": item},
         )
         return msg
diff --git a/csp_bot/commands/help.py b/csp_bot/commands/help.py
@@ -1,13 +1,13 @@
 """Help command for csp-bot.
 
-Uses chatom's formatting utilities for cross-platform output.
+Uses chatom's FormattedMessage for automatic cross-platform rendering.
 """
 
-from html import escape
 from logging import getLogger
-from typing import Mapping, Optional, Type
+from typing import Any, Mapping, Optional, Type
 
 from chatom import Message
+from chatom.format import FormattedMessage, Heading, Table, Text
 
 from csp_bot.structs import BotCommand
 
@@ -16,6 +16,25 @@
 log = getLogger(__name__)
 
 
+def _command_backends(runner: Any) -> list:
+    """Get backends list from either legacy or new command types."""
+    if isinstance(runner, BaseCommand):
+        return runner.backends()
+    return getattr(runner, "backends", []) or []
+
+
+def _command_name(runner: Any) -> str:
+    if isinstance(runner, BaseCommand):
+        return runner.name()
+    return getattr(runner, "name", "") or ""
+
+
+def _command_help(runner: Any) -> str:
+    if isinstance(runner, BaseCommand):
+        return runner.help()
+    return getattr(runner, "help", "") or ""
+
+
 class HelpCommand(ReplyCommand):
     """Display help for available commands."""
 
@@ -31,69 +50,39 @@ def help(self) -> str:
     def execute(
         self,
         command: BotCommand,
-        commands: Mapping[str, "BaseCommand"] = None,
+        commands: Mapping[str, Any] = None,
     ) -> Optional[Message]:
         log.info(f"Help command: {command.command}")
 
         # Collect help for each command
-        helps = []
+        rows = []
         for cmd_key, cmd_inst in commands.items():
-            # Skip hidden commands
             if cmd_key.startswith("_"):
                 continue
 
-            # Skip commands not supported on this backend
-            if cmd_inst.backends() and command.backend not in cmd_inst.backends():
+            backends = _command_backends(cmd_inst)
+            if backends and command.backend not in backends:
                 continue
 
-            # Filter by requested commands if args provided
             if command.args and cmd_key not in command.args:
                 continue
 
-            helps.append(
-                (
-                    escape(cmd_key, quote=True),
-                    escape(cmd_inst.name(), quote=True),
-                    escape(cmd_inst.help(), quote=True),
-                )
+            rows.append(
+                {
+                    "Command": f"/{cmd_key}",
+                    "Name": _command_name(cmd_inst),
+                    "Info": _command_help(cmd_inst),
+                }
             )
 
-        helps = sorted(helps, key=lambda x: x[0])
-
-        # Build response using chatom's FormattedMessage
-        if command.backend == "symphony":
-            # Symphony supports expandable cards
-            table = "<table><thead><tr><td>Command</td><td>Name</td><td>Info</td></tr></thead><tbody>"
-            for cmd_key, name, help_text in helps:
-                table += f"<tr><td>/{cmd_key}</td><td>{name}</td><td>{help_text}</td></tr>"
-            table += "</tbody></table>"
-            content = f'<expandable-card state="collapsed"><header>Bot Commands Help</header><body variant="default">{table}</body></expandable-card>'
-        elif command.backend == "slack":
-            # Slack uses mrkdwn code blocks for tables
-            lines = ["*Bot Commands Help*", "```"]
-            lines.append(f"{'Command':<15} {'Name':<15} {'Info'}")
-            lines.append("-" * 60)
-            for cmd_key, name, help_text in helps:
-                lines.append(f"/{cmd_key:<14} {name:<15} {help_text}")
-            lines.append("```")
-            content = "\n".join(lines)
-        elif command.backend == "discord":
-            # Discord uses markdown tables
-            lines = ["# Bot Commands Help", ""]
-            lines.append("| Command | Name | Info |")
-            lines.append("|---------|------|------|")
-            for cmd_key, name, help_text in helps:
-                lines.append(f"| /{cmd_key} | {name} | {help_text} |")
-            content = "\n".join(lines)
-        else:
-            # Plain text fallback
-            lines = ["Bot Commands Help", ""]
-            for cmd_key, name, help_text in helps:
-                lines.append(f"/{cmd_key}: {name} - {help_text}")
-            content = "\n".join(lines)
+        rows.sort(key=lambda r: r["Command"])
+
+        msg = FormattedMessage(metadata={"backend": command.backend})
+        msg.content.append(Heading(child=Text(content="Bot Commands Help"), level=3))
+        msg.content.append(Table.from_dict_list(rows, columns=["Command", "Name", "Info"]))
 
         return Message(
-            content=content,
+            content=msg.render_for(command.backend),
             channel=command.channel,
             metadata={"backend": command.backend},
         )
diff --git a/csp_bot/commands/schedule.py b/csp_bot/commands/schedule.py
@@ -3,11 +3,11 @@
 Allows scheduling commands for delayed or recurring execution.
 """
 
-from html import escape
 from logging import getLogger
 from typing import TYPE_CHECKING, List, Mapping, Optional, Type
 
 from chatom import Message
+from chatom.format import Bold, FormattedMessage, Table, Text
 from croniter import CroniterBadCronError, croniter
 from dateparser import parse
 
@@ -111,21 +111,14 @@ def preexecute(
     def execute(self, command: BotCommand, schedule: Mapping[str, BotCommand]) -> Message:
         log.info("Schedule list command")
 
-        # Build schedule listing
-        if command.backend == "symphony":
-            table = "<table><thead><tr><td>ID</td><td>Command</td></tr></thead><tbody>"
-            for scheduled_cmd in schedule.values():
-                table += f"<tr><td>{id(scheduled_cmd)}</td><td>{escape(scheduled_cmd.command)}</td></tr>"
-            table += "</tbody></table>"
-            content = f'<expandable-card state="collapsed"><header>Bot Schedule</header><body variant="default">{table}</body></expandable-card>'
-        else:
-            lines = ["*Scheduled Commands*", ""]
-            for scheduled_cmd in schedule.values():
-                lines.append(f"- {id(scheduled_cmd)}: /{scheduled_cmd.command}")
-            content = "\n".join(lines)
+        rows = [{"ID": str(id(scheduled_cmd)), "Command": f"/{scheduled_cmd.command}"} for scheduled_cmd in schedule.values()]
+
+        msg = FormattedMessage(metadata={"backend": command.backend})
+        msg.content.append(Bold(child=Text(content="Scheduled Commands")))
+        msg.content.append(Table.from_dict_list(rows, columns=["ID", "Command"]))
 
         return Message(
-            content=content,
+            content=msg.render_for(command.backend),
             channel=command.channel,
             metadata={"backend": command.backend},
         )
diff --git a/csp_bot/commands/status.py b/csp_bot/commands/status.py
@@ -1,6 +1,6 @@
 """Status command for csp-bot.
 
-Displays system and bot status information.
+Displays system and bot status information using FormattedMessage.
 """
 
 from datetime import datetime
@@ -12,6 +12,7 @@
 
 import psutil
 from chatom import Message
+from chatom.format import Bold, FormattedMessage, Table, Text
 
 from csp_bot.structs import BotCommand
 
@@ -47,25 +48,27 @@ def preexecute(self, command: BotCommand, bot_instance: "Bot") -> BotCommand:
     def execute(self, command: BotCommand) -> Optional[Message]:
         log.info("Status command")
 
-        # Build status information
-        lines = []
-
-        lines.append(f"**Now**: {datetime.utcnow()}")
-        lines.append(f"**Backends**: {', '.join(self._adapters)}")
-        lines.append(f"**CPU**: {psutil.cpu_percent()}%")
-        lines.append(f"**Memory**: {psutil.virtual_memory().percent}%")
-        lines.append(f"**Memory Available**: {round(psutil.virtual_memory().available * 100 / psutil.virtual_memory().total, 2)}%")
-        lines.append(f"**Host**: {_HOSTNAME}")
-        lines.append(f"**User**: {_USER}")
-
-        current_process = psutil.Process()
-        lines.append(f"**PID**: {current_process.pid}")
-        lines.append(f"**Active Threads**: {active_count()}")
-
-        content = "\n".join(lines)
+        mem = psutil.virtual_memory()
+        proc = psutil.Process()
+
+        rows = [
+            {"Metric": "Now", "Value": str(datetime.utcnow())},
+            {"Metric": "Backends", "Value": ", ".join(self._adapters)},
+            {"Metric": "CPU", "Value": f"{psutil.cpu_percent()}%"},
+            {"Metric": "Memory", "Value": f"{mem.percent}%"},
+            {"Metric": "Memory Available", "Value": f"{round(mem.available * 100 / mem.total, 2)}%"},
+            {"Metric": "Host", "Value": _HOSTNAME},
+            {"Metric": "User", "Value": _USER},
+            {"Metric": "PID", "Value": str(proc.pid)},
+            {"Metric": "Active Threads", "Value": str(active_count())},
+        ]
+
+        msg = FormattedMessage(metadata={"backend": command.backend})
+        msg.content.append(Bold(child=Text(content="Bot Status")))
+        msg.content.append(Table.from_dict_list(rows, columns=["Metric", "Value"]))
 
         return Message(
-            content=content,
+            content=msg.render_for(command.backend),
             channel=command.channel,
             metadata={"backend": command.backend},
         )
diff --git a/csp_bot/tests/test_command_framework.py b/csp_bot/tests/test_command_framework.py
