feat(commands): implement split_trigger_and_args method for improved command parsing

Added a new method `split_trigger_and_args` to the `BaseCommand` class, which splits message content into a matched keyword and arguments. This enhancement allows for better handling of command triggers and arguments across various command classes, ensuring that multi-word triggers are prioritized and leading characters are stripped appropriately. Updated the `ChannelsCommand`, `DiceCommand`, `HackerCommand`, `MultitestCommand`, `RollCommand`, and `TraceCommand` classes to utilize this new method for cleaner and more consistent command execution logic.
This commit is contained in:
agessaman
2026-08-24 18:51:24 -07:00
parent 3b5e95fb93
commit 537657c211
9 changed files with 238 additions and 178 deletions
+30
View File
@@ -954,6 +954,36 @@ class BaseCommand(ABC):
message.content_lower = content.lower()
return message.content_lower
def split_trigger_and_args(self, content: str) -> tuple[Optional[str], str]:
"""Split message content into ``(matched_keyword, args)``.
Matches against ``self.keywords`` (built-in stems plus config ``aliases``),
preferring the longest keyword so multi-word triggers win. Leading ``!``
is stripped for execute paths that still see raw command-style text.
Args:
content: Raw or partially cleaned message text.
Returns:
``(keyword, args)`` when a keyword matches as the first token(s);
``(None, content)`` (after optional ``!`` strip) otherwise.
"""
text = content.strip()
if text.startswith('!'):
text = text[1:].strip()
lower = text.lower()
if not lower or not self.keywords:
return None, text
# Longest first so "dad joke" wins over a hypothetical shorter stem
for keyword in sorted(self.keywords, key=lambda k: len(k), reverse=True):
kw = keyword.lower()
if lower == kw:
return kw, ""
if lower.startswith(kw + " "):
return kw, text[len(kw):].strip()
return None, text
def matches_keyword(self, message: MeshMessage) -> bool:
"""Check if this command matches the message content based on keywords.
+27 -29
View File
@@ -93,9 +93,11 @@ class ChannelsCommand(BaseCommand):
# Don't match if this looks like a subcommand of another command
# (e.g., "stats channels" should not match "channels" command)
# First word must be one of our keywords (including config aliases).
if ' ' in content_lower:
parts = content_lower.split()
if len(parts) > 1 and parts[0] not in ['channels', 'channel']:
keyword_stems = {k.lower() for k in self.keywords}
if len(parts) > 1 and parts[0] not in keyword_stems:
return False
for keyword in self.keywords:
@@ -127,40 +129,36 @@ class ChannelsCommand(BaseCommand):
bool: True if execution was successful.
"""
try:
# Parse the command to check for sub-commands
content = message.content.strip()
if content.startswith('!'):
content = content[1:].strip()
# Remainder after trigger (built-in stem or config alias), e.g.
# "channels seattle", "channel seahawks", "ch list", "channels #bot"
_trigger, args = self.split_trigger_and_args(message.content)
# Check for sub-command (e.g., "channels seattle", "channel seahawks", "channels list", "channels #bot")
sub_command = None
specific_channel = None
if content.lower().startswith('channels ') or content.lower().startswith('channel '):
parts = content.split(' ', 1)
if len(parts) > 1:
sub_command = parts[1].strip().lower()
if args:
sub_command = args.lower()
# Handle special "list" command to show all categories
if sub_command == 'list':
await self._show_all_categories(message)
return True
# Handle special "list" command to show all categories
if sub_command == 'list':
await self._show_all_categories(message)
return True
# Check if user is asking for a specific channel (starts with #)
if sub_command.startswith('#'):
specific_channel = sub_command
sub_command = None
# Check if user is asking for a specific channel (starts with #)
if sub_command.startswith('#'):
specific_channel = sub_command
sub_command = None
else:
# First check if this is a valid category
if self._is_valid_category(sub_command):
# It's a category, keep it as sub_command
pass
else:
# First check if this is a valid category
if self._is_valid_category(sub_command):
# It's a category, keep it as sub_command
pass
else:
# Check if this might be a channel search (not a category)
# Try to find a channel that matches this name across all categories
found_channel = self._find_channel_by_name(sub_command)
if found_channel:
specific_channel = '#' + found_channel
sub_command = None
# Check if this might be a channel search (not a category)
# Try to find a channel that matches this name across all categories
found_channel = self._find_channel_by_name(sub_command)
if found_channel:
specific_channel = '#' + found_channel
sub_command = None
# Handle specific channel request
if specific_channel:
+3 -32
View File
@@ -76,30 +76,6 @@ class DiceCommand(BaseCommand):
"""
return self.translate('commands.dice.help')
def matches_keyword(self, message: MeshMessage) -> bool:
"""Override to handle dice-specific matching.
Args:
message: The received message.
Returns:
bool: True if message is a dice command, False otherwise.
"""
content_lower = self.cleanup_message_for_matching(message)
# Check for exact "dice" match
if content_lower == "dice":
return True
# Check for dice with parameters (dice d20, dice 20, dice d6, etc.)
# Match any message starting with "dice " - validation happens in execute()
if content_lower.startswith("dice "):
words = content_lower.split()
if len(words) >= 2 and words[0] == "dice":
return True # Match any dice command, validation in execute()
return False
def parse_dice_notation(self, dice_input: str) -> tuple:
"""Parse dice notation and return (sides, count, is_decade).
@@ -281,22 +257,17 @@ class DiceCommand(BaseCommand):
Returns:
bool: True if executed successfully, False otherwise.
"""
content = message.content.strip()
# Handle command-style messages
if content.startswith('!'):
content = content[1:].strip()
_trigger, dice_part = self.split_trigger_and_args(message.content)
# Default to d6 if no specification
if content.lower() == "dice":
if not dice_part:
sides = 6
count = 1
results = self.roll_dice(sides, count)
response = self.format_dice_result(sides, count, results)
return await self.send_response(message, response)
# Parse dice specification
dice_part = content[5:].strip() # Get everything after "dice "
# Parse dice specification after the trigger keyword / alias
# Try parsing as mixed dice first (multiple dice types)
mixed_dice = self.parse_mixed_dice(dice_part)
+18 -1
View File
@@ -66,8 +66,16 @@ class HackerCommand(BaseCommand):
if content.startswith('!'):
content = content[1:].strip()
# Config aliases are extra stems; strip them so "hack sudo ls" routes like "sudo ls".
# Built-in comedy keywords stay in the routed text (get_hacker_error matches on prefix).
trigger, args = self.split_trigger_and_args(message.content)
builtin_stems = {k.lower() for k in type(self).keywords}
routed = content
if trigger is not None and trigger not in builtin_stems:
routed = args
# Get the appropriate error message
error_msg = self.get_hacker_error(content)
error_msg = self.get_hacker_error(routed)
# Send the response
return await self.send_response(message, error_msg)
@@ -510,4 +518,13 @@ class HackerCommand(BaseCommand):
if len(content_lower) == len(keyword.lower()) or content_lower[len(keyword.lower())] == ' ':
return True
# Config aliases (keywords not already covered by the comedy lists above)
known = {k.lower() for k in exact_match_commands + prefix_match_commands}
for keyword in self.keywords:
kw = keyword.lower()
if kw in known:
continue
if content_lower == kw or content_lower.startswith(kw + ' '):
return True
return False
+9 -39
View File
@@ -654,25 +654,6 @@ class MultitestCommand(BaseCommand):
def get_help_text(self) -> str:
return self.translate('commands.multitest.help', fallback="Listens for 6 seconds and collects all unique paths from incoming messages")
def matches_keyword(self, message: MeshMessage) -> bool:
"""Check if message matches multitest keyword"""
content_lower = self.cleanup_message_for_matching(message)
# Check for exact match or keyword followed by space
for keyword in self.keywords:
if content_lower == keyword or content_lower.startswith(keyword + ' '):
return True
# Check for variants: "mt long", "mt xlong", "multitest long", "multitest xlong"
if content_lower.startswith('mt ') or content_lower.startswith('multitest '):
parts = content_lower.split()
if len(parts) >= 2 and parts[0] in ['mt', 'multitest']:
variant = parts[1]
if variant in ['long', 'xlong']:
return True
return False
def extract_path_from_rf_data(self, rf_data: dict) -> Optional[str]:
"""Extract path in prefix string format from RF data routing_info.
Supports 1-, 2-, and 3-byte-per-hop (2, 4, or 6 hex chars per node).
@@ -996,27 +977,16 @@ class MultitestCommand(BaseCommand):
self.record_execution(user_id)
# Determine listening duration based on command variant
content = message.content.strip()
if content.startswith('!'):
content = content[1:].strip()
content_lower = content.lower()
_trigger, args = self.split_trigger_and_args(message.content)
listening_duration = 6.0 # Default
# Check for variants: "mt long", "mt xlong", "multitest long", "multitest xlong"
if content_lower.startswith('mt ') or content_lower.startswith('multitest '):
parts = content_lower.split()
if len(parts) >= 2 and parts[0] in ['mt', 'multitest']:
variant = parts[1]
if variant == 'long':
listening_duration = 10.0
self.logger.info(f"Multitest command (long) executed by {user_id} - starting 10 second listening window")
elif variant == 'xlong':
listening_duration = 14.0
self.logger.info(f"Multitest command (xlong) executed by {user_id} - starting 14 second listening window")
else:
self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window")
else:
self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window")
# Variants: "<trigger> long" / "<trigger> xlong" (trigger = multitest, mt, or alias)
variant = args.split()[0].lower() if args else ""
if variant == 'long':
listening_duration = 10.0
self.logger.info(f"Multitest command (long) executed by {user_id} - starting 10 second listening window")
elif variant == 'xlong':
listening_duration = 14.0
self.logger.info(f"Multitest command (xlong) executed by {user_id} - starting 14 second listening window")
else:
self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window")
+1 -6
View File
@@ -26,7 +26,7 @@ class PrefixCommand(BaseCommand):
# Read-only informational output; safe for scheduled {cmd:...} rendering.
render_safe = True
name = "prefix"
keywords = ['prefix', 'repeater', 'lookup']
keywords = ['prefix', 'lookup']
description = "Look up repeaters by prefix (2, 4, or 6 hex chars = 13 bytes; longer input truncated)"
category = "meshcore_info"
requires_dm = False
@@ -178,11 +178,6 @@ class PrefixCommand(BaseCommand):
return self.translate('commands.prefix.help_no_api', location_note=location_note)
return self.translate('commands.prefix.help_api', location_note=location_note)
def matches_keyword(self, message: MeshMessage) -> bool:
"""Check if message starts with 'prefix' keyword"""
content_lower = self.cleanup_message_for_matching(message)
return content_lower == 'prefix' or content_lower.startswith('prefix ')
async def _parse_location_to_lat_lon(self, location: str) -> tuple[Optional[float], Optional[float], Optional[str]]:
"""Parse location string to latitude/longitude coordinates.
+12 -31
View File
@@ -65,31 +65,18 @@ class RollCommand(BaseCommand):
return self.translate('commands.roll.help')
def matches_keyword(self, message: MeshMessage) -> bool:
"""Override to handle roll-specific matching.
Custom matching logic to support variable maximums (e.g., "roll 50").
Args:
message: The message to check for a match.
Returns:
bool: True if the message matches the roll command syntax, False otherwise.
"""
"""Match ``roll`` / aliases; with args, only when the arg is a valid max."""
content_lower = self.cleanup_message_for_matching(message)
if not content_lower:
return False
# Check for exact "roll" match
if content_lower == "roll":
return True
# Check for roll with parameters (roll 50, roll 1000, etc.)
# Ensure "roll" is the first word and followed by valid number
if content_lower.startswith("roll "):
words = content_lower.split()
if len(words) >= 2 and words[0] == "roll":
roll_part = content_lower[5:].strip() # Get everything after "roll "
# Check if the roll part is valid number notation (not just any word)
max_num = self.parse_roll_notation(roll_part)
return max_num is not None # Only match if it's valid number notation
for keyword in self.keywords:
kw = keyword.lower()
if content_lower == kw:
return True
if content_lower.startswith(kw + " "):
roll_part = content_lower[len(kw):].strip()
return self.parse_roll_notation(roll_part) is not None
return False
@@ -151,18 +138,12 @@ class RollCommand(BaseCommand):
Returns:
bool: True if executed successfully, False otherwise.
"""
content = message.content.strip()
# Handle command-style messages
if content.startswith('!'):
content = content[1:].strip()
_trigger, roll_part = self.split_trigger_and_args(message.content)
# Default to 1-100 if no specification
if content.lower() == "roll":
if not roll_part:
max_num: Optional[int] = 100
else:
# Parse roll specification
roll_part = content[5:].strip() # Get everything after "roll "
max_num = self.parse_roll_notation(roll_part)
if max_num is None:
+5 -20
View File
@@ -93,12 +93,6 @@ class TraceCommand(BaseCommand):
"No path = use your message path (round-trip)."
)
def matches_keyword(self, message: MeshMessage) -> bool:
content_lower = self.cleanup_message_for_matching(message)
if content_lower == "trace" or content_lower == "tracer":
return True
return bool(content_lower.startswith("trace ") or content_lower.startswith("tracer "))
def _extract_path_from_message(self, message: MeshMessage) -> list[str]:
"""Extract path node IDs from message.path (supports 1-byte, 2-byte, and 3-byte hashes)."""
if not message.path:
@@ -133,21 +127,14 @@ class TraceCommand(BaseCommand):
return valid
def _parse_path_arg(self, content: str) -> Optional[list[str]]:
"""Parse path from command content after 'trace ' or 'tracer '.
"""Parse path from command content after the matched trigger keyword.
Accepts comma-separated hex nodes where each segment is the same length:
2-char = 1-byte (e.g. 01,7a,55), 4-char = 2-byte (e.g. feed,6ddf),
6-char = 3-byte (e.g. feedca,6ddf01).
Without commas, treats contiguous hex as 2-char (1-byte) nodes.
Returns list of hex node IDs, or None if no path args / invalid.
"""
content = content.strip()
if content.startswith("!"):
content = content[1:].strip()
rest = ""
for kw in ["tracer ", "trace "]:
if content.lower().startswith(kw):
rest = content[len(kw) :].strip()
break
_trigger, rest = self.split_trigger_and_args(content)
if not rest:
return None
# Comma-separated: each segment is one node; preserves multibyte groupings
@@ -224,11 +211,9 @@ class TraceCommand(BaseCommand):
return "\n".join(lines)
async def execute(self, message: MeshMessage) -> bool:
content = message.content.strip()
if content.startswith("!"):
content = content[1:].strip()
is_tracer = content.lower().startswith("tracer")
trigger, _args = self.split_trigger_and_args(message.content)
# Reciprocal round-trip only when the tracer stem is used (not aliases of trace)
is_tracer = trigger == "tracer"
path_arg = self._parse_path_arg(message.content)
if path_arg is not None:
path_nodes = path_arg[: self.maximum_hops]
@@ -1,36 +1,150 @@
#!/usr/bin/env python3
"""Regression: disabled built-in must not block another command's alias.
Reproduces the report where [Test_Command] aliases = path, p with
[Path_Command] enabled = false still claimed !path and sent no reply.
"""
"""Regression: config aliases work across commands that had hardcoded matchers."""
from unittest.mock import AsyncMock, MagicMock, Mock
import pytest
from modules.commands.channels_command import ChannelsCommand
from modules.commands.dice_command import DiceCommand
from modules.commands.hacker_command import HackerCommand
from modules.commands.multitest_command import MultitestCommand
from modules.commands.path_command import PathCommand
from modules.commands.prefix_command import PrefixCommand
from modules.commands.roll_command import RollCommand
from modules.commands.test_command import TestCommand as MeshTestCommand
from modules.commands.trace_command import TraceCommand
from tests.conftest import mock_message
from tests.test_command_manager import make_manager
from tests.unit.test_command_path_byte_gating import _base_bot
@pytest.mark.unit
def test_test_command_aliases_match_path_and_p():
bot = _base_bot()
bot.config.add_section("Test_Command")
bot.config.set("Test_Command", "enabled", "true")
bot.config.set("Test_Command", "aliases", "path, p")
def _with_aliases(bot, section: str, aliases: str):
if not bot.config.has_section(section):
bot.config.add_section(section)
bot.config.set(section, "enabled", "true")
bot.config.set(section, "aliases", aliases)
return bot
cmd = MeshTestCommand(bot)
assert "path" in cmd.keywords
assert "p" in cmd.keywords
assert cmd.matches_keyword(mock_message(content="!path", is_dm=True)) is True
assert cmd.matches_keyword(mock_message(content="path", is_dm=True)) is True
assert cmd.matches_keyword(mock_message(content="p", is_dm=True)) is True
assert cmd.matches_keyword(mock_message(content="test", is_dm=True)) is True
assert cmd.matches_keyword(mock_message(content="ping", is_dm=True)) is False
@pytest.mark.unit
@pytest.mark.parametrize(
"factory,section,alias,content",
[
(MeshTestCommand, "Test_Command", "path", "!path"),
(DiceCommand, "Dice_Command", "d", "d d20"),
(DiceCommand, "Dice_Command", "d", "d"),
(RollCommand, "Roll_Command", "r", "r"),
(RollCommand, "Roll_Command", "r", "r 50"),
(TraceCommand, "Trace_Command", "tr", "tr"),
(TraceCommand, "Trace_Command", "tr", "tr 01,7a"),
(PrefixCommand, "Prefix_Command", "pfx", "pfx"),
(PrefixCommand, "Prefix_Command", "pfx", "pfx free"),
(MultitestCommand, "Multitest_Command", "mtest", "mtest"),
(MultitestCommand, "Multitest_Command", "mtest", "mtest long"),
(ChannelsCommand, "Channels_Command", "ch", "ch"),
(PathCommand, "Path_Command", "routehex", "routehex"),
],
)
def test_config_alias_matches(factory, section, alias, content):
bot = _with_aliases(_base_bot(), section, alias)
cmd = factory(bot)
assert alias in [k.lower() for k in cmd.keywords]
assert cmd.matches_keyword(mock_message(content=content, is_dm=True)) is True
@pytest.mark.unit
def test_roll_alias_still_rejects_non_numeric_args():
bot = _with_aliases(_base_bot(), "Roll_Command", "r")
cmd = RollCommand(bot)
assert cmd.matches_keyword(mock_message(content="r abc", is_dm=True)) is False
@pytest.mark.unit
def test_channels_alias_does_not_match_as_subcommand():
bot = _with_aliases(_base_bot(), "Channels_Command", "ch")
cmd = ChannelsCommand(bot)
# "stats channels" style: first word is not a channels keyword/alias
assert cmd.matches_keyword(mock_message(content="stats ch", is_dm=True)) is False
@pytest.mark.asyncio
@pytest.mark.unit
async def test_dice_alias_execute_uses_args():
bot = _with_aliases(_base_bot(), "Dice_Command", "d")
cmd = DiceCommand(bot)
cmd.send_response = AsyncMock(return_value=True)
cmd.roll_dice = Mock(return_value=[4])
cmd.format_dice_result = Mock(return_value="ok")
await cmd.execute(mock_message(content="!d d20", is_dm=True))
cmd.roll_dice.assert_called()
cmd.send_response.assert_awaited()
@pytest.mark.asyncio
@pytest.mark.unit
async def test_roll_alias_execute_parses_max():
bot = _with_aliases(_base_bot(), "Roll_Command", "r")
cmd = RollCommand(bot)
cmd.send_response = AsyncMock(return_value=True)
cmd.roll_number = Mock(return_value=7)
cmd.format_roll_result = Mock(return_value="rolled")
await cmd.execute(mock_message(content="!r 50", is_dm=True))
cmd.roll_number.assert_called_once_with(50)
@pytest.mark.unit
def test_trace_parse_path_arg_honors_alias():
bot = _with_aliases(_base_bot(), "Trace_Command", "tr")
cmd = TraceCommand(bot)
assert cmd._parse_path_arg("!tr 01,7a") == ["01", "7a"]
@pytest.mark.unit
def test_hacker_config_alias_matches():
bot = _with_aliases(_base_bot(), "Hacker_Command", "hack")
# Hacker may use a different enabled key; force on
bot.config.set("Hacker_Command", "enabled", "true")
cmd = HackerCommand(bot)
cmd.enabled = True
assert cmd.matches_keyword(mock_message(content="hack", is_dm=True)) is True
@pytest.mark.asyncio
@pytest.mark.unit
async def test_channels_alias_execute_honors_list_subcommand():
bot = _with_aliases(_base_bot(), "Channels_Command", "ch")
cmd = ChannelsCommand(bot)
cmd._show_all_categories = AsyncMock()
cmd._show_specific_channel = AsyncMock()
cmd.send_response = AsyncMock(return_value=True)
await cmd.execute(mock_message(content="!ch list", is_dm=True))
cmd._show_all_categories.assert_awaited_once()
cmd._show_specific_channel.assert_not_called()
@pytest.mark.asyncio
@pytest.mark.unit
async def test_hacker_alias_execute_routes_inner_command():
bot = _with_aliases(_base_bot(), "Hacker_Command", "hack")
bot.config.set("Hacker_Command", "enabled", "true")
cmd = HackerCommand(bot)
cmd.enabled = True
cmd.get_hacker_error = Mock(return_value="denied")
cmd.send_response = AsyncMock(return_value=True)
await cmd.execute(mock_message(content="!hack sudo ls", is_dm=True))
cmd.get_hacker_error.assert_called_once_with("sudo ls")
cmd.get_hacker_error.reset_mock()
await cmd.execute(mock_message(content="!sudo ls", is_dm=True))
cmd.get_hacker_error.assert_called_once_with("sudo ls")
@pytest.mark.unit
@@ -45,7 +159,6 @@ def test_check_keywords_prefers_test_alias_when_path_disabled():
path_cmd = PathCommand(bot)
test_cmd = MeshTestCommand(bot)
# path before test — same claim order as the user report
manager = make_manager(bot, commands={"path": path_cmd, "test": test_cmd})
matches = manager.check_keywords(mock_message(content="!path", is_dm=True))