wolframalpha: move to its own plugin

master
dece 1 year ago
parent 1bef65ea73
commit 84f4eaeec9

@ -200,6 +200,14 @@
"ambiguous_response": "It is ambiguous.",
"empty_response": "I can't find it."
},
"wolframalpha": {
"commands": ["wolfram"],
"aliases": {
"wolfram": ["wolframalpha", "compute"]
},
"api_key": "",
"max_pods": 1
},
"yell": {
"commands": ["yell"],
"target_word": "to",

@ -2,29 +2,12 @@ from edmond.plugin import Plugin
class UnknownCommandPlugin(Plugin):
"""Handle unknown command by sending it to WolframAlpha."""
REQUIRED_CONFIGS = ["api_key", "max_pods"]
MAX_LENGTH = 256
CUT_MARK = "..."
def __init__(self, bot):
super().__init__(bot)
self.priority = -6
self._client = None
@property
def client(self):
if self._client is None and self.has_api_key():
self._client = wolframalpha.Client(self.config["api_key"])
return self._client
def has_api_key(self):
return self.config["api_key"] != ""
def on_pubmsg(self, event):
if not self.has_api_key:
return False
message = self.should_read_message(event.arguments[0])
if not message:
return False
@ -33,51 +16,5 @@ class UnknownCommandPlugin(Plugin):
return False
query = " ".join(words[:-1])
self.process_query(query, event.target)
self.bot.say(event.target, f"UnknownCommand: {query}")
return True
def process_query(self, query, target):
self.bot.log_d(f"Processing '{query}' with WolframAlpha.")
try:
response = self.client.query(query)
except Exception as exc: # unstable lib
self.bot.log_w(f"wolframalpha exception: {exc}")
self.signal_failure(target)
return
if not response["@success"]:
self.bot.log_d("Call to WA succeeded but response is an error.")
self.signal_failure(target)
return
inputs = []
answers = []
num_pods = 0
for pod in response.pods:
for subpod in pod.subpods:
self.bot.log_d(f"WolframAlpha subpod: {subpod}")
if pod["@id"] == "Input":
inputs.append(self.sanitize_text(subpod.plaintext or ""))
else:
answers.append(self.sanitize_text(subpod.plaintext or ""))
num_pods += 1
if num_pods >= self.config["max_pods"]:
break
if num_pods >= self.config["max_pods"]:
break
input_text = ", ".join(inputs)
answer_text = ", ".join(answers)
if input_text:
reply = input_text + " -- " + answer_text
else:
reply = answer_text
if len(reply) > self.MAX_LENGTH - len(self.CUT_MARK):
reply = (
reply[: self.MAX_LENGTH - len(self.CUT_MARK)] + self.CUT_MARK
)
self.bot.say(target, reply)
@staticmethod
def sanitize_text(text):
return text.replace("\n", ", ")

@ -0,0 +1,79 @@
import wolframalpha
from edmond.plugin import Plugin
class WolframAlphaPlugin(Plugin):
"""Handle unknown command by sending it to WolframAlpha."""
REQUIRED_CONFIGS = ["commands", "api_key", "max_pods"]
MAX_LENGTH = 256
CUT_MARK = "..."
def __init__(self, bot):
super().__init__(bot)
self.priority = -6
self._client = None
@property
def client(self):
if self._client is None and self.has_api_key():
self._client = wolframalpha.Client(self.config["api_key"])
return self._client
def on_welcome(self, _):
if not self.config["api_key"]:
self.bot.log_w("API key unavailable.")
self.is_ready = False
def on_pubmsg(self, event):
if not self.should_handle_command(event.arguments[0]):
return False
self.process_query(self.command.content, event.target)
return True
def process_query(self, query, target):
self.bot.log_d(f"Processing '{query}' with WolframAlpha.")
try:
response = self.client.query(query)
except Exception as exc: # unstable lib
self.bot.log_w(f"wolframalpha exception: {exc}")
self.signal_failure(target)
return
if not response["@success"]:
self.bot.log_d("Call to WA succeeded but response is an error.")
self.signal_failure(target)
return
inputs = []
answers = []
num_pods = 0
for pod in response.pods:
for subpod in pod.subpods:
self.bot.log_d(f"WolframAlpha subpod: {subpod}")
if pod["@id"] == "Input":
inputs.append(self.sanitize_text(subpod.plaintext or ""))
else:
answers.append(self.sanitize_text(subpod.plaintext or ""))
num_pods += 1
if num_pods >= self.config["max_pods"]:
break
if num_pods >= self.config["max_pods"]:
break
input_text = ", ".join(inputs)
answer_text = ", ".join(answers)
if input_text:
reply = input_text + " -- " + answer_text
else:
reply = answer_text
if len(reply) > self.MAX_LENGTH - len(self.CUT_MARK):
reply = (
reply[: self.MAX_LENGTH - len(self.CUT_MARK)] + self.CUT_MARK
)
self.bot.say(target, reply)
@staticmethod
def sanitize_text(text):
return text.replace("\n", ", ")
Loading…
Cancel
Save