shoko/randebu
1
0
Fork 0
Code Issues 6 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: Better error detection for AVE API commands

Browse Source 
- Added _is_error_output helper to detect errors in CLI output
- API errors like 'API error 403' now show proper error message instead of 'No price data available'
- Updated all command execution methods to use the helper
This commit is contained in:
shokollm
2026-04-13 23:55:51 +00:00
parent 2c3b6ef073
commit 6b8912a7eb
1 changed files with 173 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

341
src/backend/app/services/ai_agent/conversational.py
View File

@@ -472,6 +472,15 @@ class ConversationalAgent:
# Track pending command after acknowledgment # Track pending command after acknowledgment
self.pending_command = None self.pending_command = None
def _is_error_output(self, code: int, output: str) -> bool:
"""Check if the command output contains an error."""
if code != 0:
return True
# Check for common error patterns in output
if output.startswith("Error:") or "API error" in output or "api key invalid" in output.lower():
return True
return False
def _handle_slash_command(self, user_message: str) -> Dict[str, Any]: def _handle_slash_command(self, user_message: str) -> Dict[str, Any]:
"""Handle slash command help requests. """Handle slash command help requests.
@@ -664,54 +673,53 @@ class ConversationalAgent:
"trending", "trending",
["--chain", "bsc", "--page-size", "10"], ["--chain", "bsc", "--page-size", "10"],
) )
if code == 0: if self._is_error_output(code, output):
try: return {
data = json.loads(output) "response": f"Failed to get trending tokens: {output}",
# Handle both dict with 'tokens' key and direct list "thinking": None,
data_field = data.get("data", []) "strategy_updated": False,
if isinstance(data_field, list): "strategy_needs_confirmation": False,
tokens = data_field "success": True,
else: }
tokens = data_field.get("tokens", []) try:
if tokens: data = json.loads(output)
token_list = "" # Handle both dict with 'tokens' key and direct list
for t in tokens[:10]: data_field = data.get("data", [])
addr = t.get("token", "") if isinstance(data_field, list):
symbol = t.get("symbol", "") tokens = data_field
name = t.get("name", "") else:
price_change = t.get("token_price_change_24h", "N/A") tokens = data_field.get("tokens", [])
mc = t.get("market_cap", "N/A") if tokens:
try: token_list = ""
mc_str = f"${float(mc):,.0f}" for t in tokens[:10]:
except (ValueError, TypeError): addr = t.get("token", "")
mc_str = str(mc) symbol = t.get("symbol", "")
token_list += f"- **{symbol}** ({name}): `{addr}` - MC: {mc_str} - 24h: {price_change}%\n" name = t.get("name", "")
return { price_change = t.get("token_price_change_24h", "N/A")
"response": f"📈 **Trending Tokens on BSC:**\n\n{token_list}\nWould you like me to set up a strategy for any of these?", mc = t.get("market_cap", "N/A")
"thinking": None, try:
"strategy_updated": False, mc_str = f"${float(mc):,.0f}"
"strategy_needs_confirmation": False, except (ValueError, TypeError):
"success": True, mc_str = str(mc)
} token_list += f"- **{symbol}** ({name}): `{addr}` - MC: {mc_str} - 24h: {price_change}%\n"
else:
return {
"response": "No trending tokens found on BSC right now. Try again later!",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to parse trending data: {output[:200]}", "response": f"📈 **Trending Tokens on BSC:**\n\n{token_list}\nWould you like me to set up a strategy for any of these?",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
"success": True, "success": True,
} }
else: else:
return {
"response": "No trending tokens found on BSC right now. Try again later!",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to get trending tokens: {output}", "response": f"Failed to parse trending data: {output[:200]}",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
@@ -733,54 +741,53 @@ class ConversationalAgent:
"search", "search",
["--keyword", keyword.strip(), "--chain", "bsc", "--limit", "10"], ["--keyword", keyword.strip(), "--chain", "bsc", "--limit", "10"],
) )
if code == 0: if self._is_error_output(code, output):
try: return {
data = json.loads(output) "response": f"Failed to search tokens: {output}",
# Handle both dict with 'tokens' key and direct list "thinking": None,
data_field = data.get("data", []) "strategy_updated": False,
if isinstance(data_field, list): "strategy_needs_confirmation": False,
tokens = data_field "success": True,
else: }
tokens = data_field.get("tokens", []) try:
if tokens: data = json.loads(output)
token_list = "" # Handle both dict with 'tokens' key and direct list
for t in tokens[:10]: data_field = data.get("data", [])
addr = t.get("token", "") if isinstance(data_field, list):
symbol = t.get("symbol", "") tokens = data_field
name = t.get("name", "") else:
price_change = t.get("token_price_change_24h", "N/A") tokens = data_field.get("tokens", [])
mc = t.get("market_cap", "N/A") if tokens:
try: token_list = ""
mc_str = f"${float(mc):,.0f}" for t in tokens[:10]:
except (ValueError, TypeError): addr = t.get("token", "")
mc_str = str(mc) symbol = t.get("symbol", "")
token_list += f"- **{symbol}** ({name}): `{addr}` - MC: {mc_str} - 24h: {price_change}%\n" name = t.get("name", "")
return { price_change = t.get("token_price_change_24h", "N/A")
"response": f"🔍 **Search Results for '{keyword}':**\n\n{token_list}\nWould you like me to set up a strategy for any of these?", mc = t.get("market_cap", "N/A")
"thinking": None, try:
"strategy_updated": False, mc_str = f"${float(mc):,.0f}"
"strategy_needs_confirmation": False, except (ValueError, TypeError):
"success": True, mc_str = str(mc)
} token_list += f"- **{symbol}** ({name}): `{addr}` - MC: {mc_str} - 24h: {price_change}%\n"
else:
return {
"response": f"No tokens found for '{keyword}'. Try a different keyword.",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to parse search results: {output[:200]}", "response": f"🔍 **Search Results for '{keyword}':**\n\n{token_list}\nWould you like me to set up a strategy for any of these?",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
"success": True, "success": True,
} }
else: else:
return {
"response": f"No tokens found for '{keyword}'. Try a different keyword.",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to search tokens: {output}", "response": f"Failed to parse search results: {output[:200]}",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
@@ -802,69 +809,68 @@ class ConversationalAgent:
"risk", "risk",
["--address", address.strip(), "--chain", "bsc"], ["--address", address.strip(), "--chain", "bsc"],
) )
if code == 0: if self._is_error_output(code, output):
try: return {
data = json.loads(output) "response": f"Failed to get risk data: {output}",
data_field = data.get("data") "thinking": None,
risk_data = data_field if isinstance(data_field, dict) else {} "strategy_updated": False,
if risk_data: "strategy_needs_confirmation": False,
is_honeypot = risk_data.get("is_honeypot", "unknown") "success": True,
buy_tax = risk_data.get("buy_tax", 0) }
sell_tax = risk_data.get("sell_tax", 0) try:
status = risk_data.get("status", "unknown") data = json.loads(output)
# Convert is_honeypot to string data_field = data.get("data")
if isinstance(is_honeypot, bool): risk_data = data_field if isinstance(data_field, dict) else {}
is_honeypot_str = str(is_honeypot).lower() if risk_data:
elif isinstance(is_honeypot, int): is_honeypot = risk_data.get("is_honeypot", "unknown")
is_honeypot_str = "true" if is_honeypot == 1 else "false" if is_honeypot == 0 else "unknown" buy_tax = risk_data.get("buy_tax", 0)
else: sell_tax = risk_data.get("sell_tax", 0)
is_honeypot_str = str(is_honeypot).lower() if is_honeypot else "unknown" status = risk_data.get("status", "unknown")
# Convert tax values # Convert is_honeypot to string
try: if isinstance(is_honeypot, bool):
buy_tax_val = float(buy_tax) if buy_tax not in (None, "N/A") else 0 is_honeypot_str = str(is_honeypot).lower()
except (ValueError, TypeError): elif isinstance(is_honeypot, int):
buy_tax_val = 0 is_honeypot_str = "true" if is_honeypot == 1 else "false" if is_honeypot == 0 else "unknown"
try:
sell_tax_val = float(sell_tax) if sell_tax not in (None, "N/A") else 0
except (ValueError, TypeError):
sell_tax_val = 0
risk_text = f"🛡️ **Risk Analysis for `{address}`**\n\n"
risk_text += f"- Status: {status}\n"
risk_text += f"- Honeypot: {is_honeypot_str}\n"
risk_text += f"- Buy Tax: {buy_tax}%\n"
risk_text += f"- Sell Tax: {sell_tax}%\n"
if is_honeypot_str == "true":
risk_text += "\n⚠️ **Warning: This token appears to be a honeypot. Do not buy!**"
elif buy_tax_val > 10 or sell_tax_val > 10:
risk_text += "\n⚠️ **Warning: High tax detected. Trade with caution!**"
else:
risk_text += "\n✅ This token appears safe to trade."
return {
"response": risk_text,
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
else: else:
return { is_honeypot_str = str(is_honeypot).lower() if is_honeypot else "unknown"
"response": f"No risk data available for `{address}`", # Convert tax values
"thinking": None, try:
"strategy_updated": False, buy_tax_val = float(buy_tax) if buy_tax not in (None, "N/A") else 0
"strategy_needs_confirmation": False, except (ValueError, TypeError):
"success": True, buy_tax_val = 0
} try:
except json.JSONDecodeError: sell_tax_val = float(sell_tax) if sell_tax not in (None, "N/A") else 0
except (ValueError, TypeError):
sell_tax_val = 0
risk_text = f"🛡️ **Risk Analysis for `{address}`**\n\n"
risk_text += f"- Status: {status}\n"
risk_text += f"- Honeypot: {is_honeypot_str}\n"
risk_text += f"- Buy Tax: {buy_tax}%\n"
risk_text += f"- Sell Tax: {sell_tax}%\n"
if is_honeypot_str == "true":
risk_text += "\n⚠️ **Warning: This token appears to be a honeypot. Do not buy!**"
elif buy_tax_val > 10 or sell_tax_val > 10:
risk_text += "\n⚠️ **Warning: High tax detected. Trade with caution!**"
else:
risk_text += "\n✅ This token appears safe to trade."
return { return {
"response": f"Failed to parse risk data: {output[:200]}", "response": risk_text,
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
"success": True, "success": True,
} }
else: else:
return {
"response": f"No risk data available for `{address}`",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to get risk data: {output}", "response": f"Failed to parse risk data: {output[:200]}",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
@@ -994,44 +1000,43 @@ class ConversationalAgent:
"price", "price",
["--tokens"] + tokens_list, ["--tokens"] + tokens_list,
) )
if code == 0: if self._is_error_output(code, output):
try: return {
data = json.loads(output) "response": f"Failed to get prices: {output}",
prices = data.get("data", {}) "thinking": None,
if not isinstance(prices, dict): "strategy_updated": False,
prices = {} "strategy_needs_confirmation": False,
if prices: "success": True,
price_text = "💰 **Token Prices:**\n" }
for token_id, price_data in prices.items(): try:
price = price_data.get("price", "N/A") if isinstance(price_data, dict) else "N/A" data = json.loads(output)
change_24h = price_data.get("token_price_change_24h", "N/A") if isinstance(price_data, dict) else "N/A" prices = data.get("data", {})
price_text += f"- {token_id}: ${price} (24h: {change_24h}%)\n" if not isinstance(prices, dict):
return { prices = {}
"response": price_text, if prices:
"thinking": None, price_text = "💰 **Token Prices:**\n"
"strategy_updated": False, for token_id, price_data in prices.items():
"strategy_needs_confirmation": False, price = price_data.get("price", "N/A") if isinstance(price_data, dict) else "N/A"
"success": True, change_24h = price_data.get("token_price_change_24h", "N/A") if isinstance(price_data, dict) else "N/A"
} price_text += f"- {token_id}: ${price} (24h: {change_24h}%)\n"
else:
return {
"response": "No price data available.",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to parse price data: {output[:200]}", "response": price_text,
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,
"success": True, "success": True,
} }
else: else:
return {
"response": "No price data available.",
"thinking": None,
"strategy_updated": False,
"strategy_needs_confirmation": False,
"success": True,
}
except json.JSONDecodeError:
return { return {
"response": f"Failed to get prices: {output}", "response": f"Failed to parse price data: {output[:200]}",
"thinking": None, "thinking": None,
"strategy_updated": False, "strategy_updated": False,
"strategy_needs_confirmation": False, "strategy_needs_confirmation": False,