diff --git a/src/crewai/utilities/events/utils/console_formatter.py b/src/crewai/utilities/events/utils/console_formatter.py index 5cb894b66..0572af11e 100644 --- a/src/crewai/utilities/events/utils/console_formatter.py +++ b/src/crewai/utilities/events/utils/console_formatter.py @@ -1,5 +1,5 @@ import os -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Union from rich.console import Console from rich.panel import Panel @@ -8,6 +8,39 @@ from rich.tree import Tree from rich.live import Live from rich.syntax import Syntax +DEFAULT_EMOJI_MAP = { + "โœ…": "[DONE]", + "โŒ": "[FAILED]", + "๐Ÿš€": "[CREW]", + "๐Ÿ”„": "[RUNNING]", + "๐Ÿ“‹": "[TASK]", + "๐Ÿ”ง": "[TOOL]", + "๐Ÿง ": "[THINKING]", + "๐ŸŒŠ": "[FLOW]", + "โœจ": "[CREATED]", + "๐Ÿงช": "[TEST]", + "๐Ÿ“š": "[KNOWLEDGE]", + "๐Ÿ”": "[SEARCH]", + "๐Ÿ”Ž": "[QUERY]", + "๐Ÿค–": "[AGENT]", + "๐Ÿ“Š": "[METRICS]", + "โšก": "[QUICK]", + "๐ŸŽฏ": "[TARGET]", + "๐Ÿ”—": "[LINK]", + "๐Ÿ’ก": "[IDEA]", + "โš ๏ธ": "[WARNING]", + "๐ŸŽ‰": "[SUCCESS]", + "๐Ÿ”ฅ": "[HOT]", + "๐Ÿ’พ": "[SAVE]", + "๐Ÿ”’": "[SECURE]", + "๐ŸŒŸ": "[STAR]", +} + + +def _parse_bool_env(val: str) -> bool: + """Parse environment variable value to boolean.""" + return val.lower() in ("true", "1", "yes") if val else False + class ConsoleFormatter: current_crew_tree: Optional[Tree] = None @@ -25,7 +58,13 @@ class ConsoleFormatter: def __init__(self, verbose: bool = False): self.console = Console(width=None) self.verbose = verbose - self.disable_emojis = os.getenv("CREWAI_DISABLE_EMOJIS", "").lower() in ("true", "1", "yes") + self.disable_emojis = _parse_bool_env(os.getenv("CREWAI_DISABLE_EMOJIS", "")) + + self.emoji_map = DEFAULT_EMOJI_MAP.copy() + self.icon_cache: Dict[str, str] = {} + if self.disable_emojis: + self.icon_cache = {emoji: text for emoji, text in self.emoji_map.items()} + # Live instance to dynamically update a Tree renderable (e.g. the Crew tree) # When multiple Tree objects are printed sequentially we reuse this Live # instance so the previous render is replaced instead of writing a new one. @@ -33,37 +72,15 @@ class ConsoleFormatter: # final Tree persists on the terminal. self._live: Optional[Live] = None - EMOJI_MAP = { - "โœ…": "[DONE]", - "โŒ": "[FAILED]", - "๐Ÿš€": "[CREW]", - "๐Ÿ”„": "[RUNNING]", - "๐Ÿ“‹": "[TASK]", - "๐Ÿ”ง": "[TOOL]", - "๐Ÿง ": "[THINKING]", - "๐ŸŒŠ": "[FLOW]", - "โœจ": "[CREATED]", - "๐Ÿงช": "[TEST]", - "๐Ÿ“š": "[KNOWLEDGE]", - "๐Ÿ”": "[SEARCH]", - "๐Ÿ”Ž": "[QUERY]", - "๐Ÿค–": "[AGENT]", - "๐Ÿ“Š": "[METRICS]", - "โšก": "[QUICK]", - "๐ŸŽฏ": "[TARGET]", - "๐Ÿ”—": "[LINK]", - "๐Ÿ’ก": "[IDEA]", - "โš ๏ธ": "[WARNING]", - "๐ŸŽ‰": "[SUCCESS]", - "๐Ÿ”ฅ": "[HOT]", - "๐Ÿ’พ": "[SAVE]", - "๐Ÿ”’": "[SECURE]", - "๐ŸŒŸ": "[STAR]", - } - def _get_icon(self, emoji: str) -> str: + """Get emoji or text alternative based on disable_emojis setting.""" if self.disable_emojis: - return self.EMOJI_MAP.get(emoji, emoji.encode('ascii', 'ignore').decode('ascii')) + if emoji in self.icon_cache: + return self.icon_cache[emoji] + ascii_fallback = emoji.encode('ascii', 'ignore').decode('ascii') + fallback = f"[ICON:{ascii_fallback or 'UNKNOWN'}]" + self.icon_cache[emoji] = fallback + return fallback return emoji def create_panel(self, content: Text, title: str, style: str = "blue") -> Panel: diff --git a/tests/utilities/events/utils/test_console_formatter_emoji_disable.py b/tests/utilities/events/utils/test_console_formatter_emoji_disable.py index 8f71eddd6..fe02c94dd 100644 --- a/tests/utilities/events/utils/test_console_formatter_emoji_disable.py +++ b/tests/utilities/events/utils/test_console_formatter_emoji_disable.py @@ -96,11 +96,11 @@ class TestConsoleFormatterEmojiDisable: assert formatter._get_icon(emoji) == expected_text def test_unknown_emoji_fallback(self): - """Test that unknown emojis fall back to ASCII-only representation.""" + """Test that unknown emojis fall back to proper representation.""" with patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"}): formatter = ConsoleFormatter(verbose=True) result = formatter._get_icon("๐Ÿฆ„") - assert result == "" + assert result == "[ICON:UNKNOWN]" @patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"}) def test_crew_tree_creation_without_emojis(self):