Further extensions (and there's a lot to be desired):
-* en-/disabling plugins programmatically
-
-* decide how plugin configuration should look like - centralised or not
-
-* decide whether ini is a good choice for configuration, and if runtime storage should use the same
-
-* rate limiting (hard nut, few moments where it's actually needed)
-
-* grep TODO in the repository...
+- en-/disabling plugins programmatically
+- decide how plugin configuration should look like - centralised or not
+- decide whether ini is a good choice for configuration, and if runtime storage should use the same
+- rate limiting (hard nut, few moments where it's actually needed)
+- grep TODO in the repository...
+- eval/use https://dramatiq.io/ for queueing tasks
# License
import logging
# Serializing
import shlex
+from typing import List
import pika
from distbot.common.config import conf_get
logger = logging.getLogger(__name__)
-def get_words(msg):
+def get_words(msg) -> List[str]:
return shlex.split(msg["body"])
-def get_nick_from_message(message_obj):
+def get_nick_from_message(message_obj) -> str:
"""
Extract the actual nick
:type message_obj: Message
import json
import random
+from distbot.common.message import get_nick_from_message, get_words
from distbot.bot.worker import Worker
from distbot.common.action import Action
class DidYouKnow(Worker):
binding_keys = [
"#.wusstet.ihr.#",
- "#.wuĆtet.ihr.#"
+ "Hi.#"
]
description = "smart bot"
usage = "wusstest ihr schon."
]
def parse_body(self, msg):
- with open(self.dynamic_db, "r+") as fd:
+ sender = get_nick_from_message(msg)
+ words = get_words(msg)
+ if "wusstet" in words:
+ fact = self.get_random_fact()
+ self.insert_fact(msg)
+ answer = random.choice(self.answers)
+ return Action(msg="{} {}".format(answer, (fact[0].upper() + fact[1:]) if answer.endswith(".") else fact))
+ elif words[0] == "Hi" and sender == "Q":
+ return Action(msg=self.get_random_fact())
+
+
+ def get_random_fact(self):
+ with open(self.dynamic_db, "r") as fd:
try:
db = json.load(fd)
except Exception as e:
db = {"quotes": []}
fact: str = random.choice(DB + db["quotes"])
+ return fact
+
+
+ def insert_fact(self, msg):
+ with open(self.dynamic_db, "w") as fd:
+ try:
+ db = json.load(fd)
+ except Exception as e:
+ db = {"quotes": []}
db["quotes"].append(msg["body"])
- fd.seek(0)
json.dump(db, fd, indent=2)
- answer = random.choice(self.answers)
- return Action(msg="{} {}".format(answer, (fact[0].upper() + fact[1:]) if answer.endswith(".") else fact))
-
ALL = [DidYouKnow]