173 lines
6.8 KiB
Python
173 lines
6.8 KiB
Python
import random, re, threading
|
|
from src import ModuleManager, utils
|
|
|
|
NO_MARKOV = "Markov chains not enabled in this channel"
|
|
|
|
class Module(ModuleManager.BaseModule):
|
|
_load_thread = None
|
|
|
|
def on_load(self):
|
|
if not self.bot.database.has_table("markov"):
|
|
self.bot.database.execute("""CREATE TABLE markov
|
|
(channel_id INTEGER, first_word TEXT, second_word TEXT,
|
|
third_word TEXT, frequency INT,
|
|
FOREIGN KEY (channel_id) REFERENCES channels(channel_id),
|
|
PRIMARY KEY (channel_id, first_word, second_word))""")
|
|
|
|
@utils.hook("command.regex")
|
|
@utils.kwarg("expect_output", False)
|
|
@utils.kwarg("ignore_action", True)
|
|
@utils.kwarg("command", "markov-trigger")
|
|
@utils.kwarg("pattern", re.compile(".+"))
|
|
def channel_message(self, event):
|
|
if event["target"].get_setting("markov", False):
|
|
self._create(event["target"].id, event["match"].group(0))
|
|
|
|
@utils.hook("received.command.markovlog")
|
|
@utils.kwarg("min_args", 1)
|
|
@utils.kwarg("permission", "markovlog")
|
|
@utils.kwarg("help", "Load a message-only newline-delimited log in to this "
|
|
"channel's markov chain")
|
|
def load_log(self, event):
|
|
if not event["target"].get_setting("markov", False):
|
|
raise utils.EventError(NO_MARKOV)
|
|
|
|
if not self._load_thread == None:
|
|
raise utils.EventError("Log loading already in progress")
|
|
|
|
page = utils.http.request(event["args_split"][0])
|
|
if page.code == 200:
|
|
event["stdout"].write("Importing...")
|
|
self._load_thread = threading.Thread(target=self._load_loop,
|
|
args=[event["target"].id, page.data])
|
|
self._load_thread.daemon = True
|
|
self._load_thread.start()
|
|
else:
|
|
event["stderr"].write("Failed to load log (%d)" % page.code)
|
|
|
|
def _load_loop(self, channel_id, data):
|
|
for line in data.decode("utf8", errors="ignore").split("\n"):
|
|
self.bot.trigger(self._create_factory(channel_id, line.strip()))
|
|
self._load_thread = None
|
|
def _create_factory(self, channel_id, line):
|
|
return lambda: self._create(channel_id, line)
|
|
|
|
def _create(self, channel_id, line):
|
|
if utils.http.REGEX_URL.search(line):
|
|
return
|
|
|
|
words = list(filter(None, line.split(" ")))
|
|
words = [word.lower() for word in words]
|
|
words_n = len(words)
|
|
|
|
if not words_n > 2:
|
|
return
|
|
|
|
inserts = []
|
|
inserts.append([None, None, words[0]])
|
|
inserts.append([None, words[0], words[1]])
|
|
|
|
for i in range(words_n-2):
|
|
inserts.append(words[i:i+3])
|
|
|
|
inserts.append([words[-2], words[-1], None])
|
|
|
|
for insert in inserts:
|
|
frequency = self.bot.database.execute_fetchone("""SELECT
|
|
frequency FROM markov WHERE channel_id=? AND first_word=?
|
|
AND second_word=? AND third_word=?""",
|
|
[channel_id]+insert)
|
|
frequency = (frequency or [0])[0]+1
|
|
|
|
self.bot.database.execute(
|
|
"INSERT OR REPLACE INTO markov VALUES (?, ?, ?, ?, ?)",
|
|
[channel_id]+insert+[frequency])
|
|
|
|
def _choose(self, words):
|
|
words, frequencies = list(zip(*words))
|
|
return random.choices(words, weights=frequencies, k=1)[0]
|
|
|
|
@utils.hook("received.command.markov")
|
|
@utils.kwarg("channel_only", True)
|
|
@utils.kwarg("help", "Generate a markov chain for the current channel")
|
|
@utils.kwarg("usage", "[first-word]")
|
|
def markov(self, event):
|
|
self._markov_for(event["target"], event["stdout"], event["stderr"],
|
|
first_words=event["args_split"][:])
|
|
|
|
@utils.hook("received.command.markovfor")
|
|
@utils.kwarg("min_args", 1)
|
|
@utils.kwarg("help", "Generate a markov chain for a given channel")
|
|
@utils.kwarg("usage", "<channel> [first-word]")
|
|
def markov_for(self, event):
|
|
if event["args_split"][0] in event["server"].channels:
|
|
channel = event["server"].channels.get(event["args_split"][0])
|
|
if not channel.has_user(event["user"]):
|
|
event["check_assert"](utils.Check("permission", "markovfor"))
|
|
|
|
self._markov_for(channel, event["stdout"], event["stderr"],
|
|
first_words=event["args_split"][1:])
|
|
else:
|
|
event["stderr"].write("Unknown channel")
|
|
|
|
def _markov_for(self, channel, stdout, stderr, first_words):
|
|
if not channel.get_setting("markov", False):
|
|
stderr.write(NO_MARKOV)
|
|
else:
|
|
out = self._generate(channel.id, first_words)
|
|
if not out == None:
|
|
stdout.write(out)
|
|
else:
|
|
stderr.write("Failed to generate markov chain")
|
|
|
|
def _generate(self, channel_id, first_words):
|
|
if not first_words:
|
|
first_words = self.bot.database.execute_fetchall("""SELECT
|
|
third_word, frequency FROM markov WHERE channel_id=? AND
|
|
first_word IS NULL AND second_word IS NULL AND third_word
|
|
NOT NULL""", [channel_id])
|
|
if not first_words:
|
|
return None
|
|
first_word = self._choose(first_words)
|
|
|
|
second_words = self.bot.database.execute_fetchall("""SELECT
|
|
third_word, frequency FROM markov WHERE channel_id=? AND
|
|
first_word IS NULL AND second_word=? AND third_word NOT NULL""",
|
|
[channel_id, first_word])
|
|
if not second_words:
|
|
return None
|
|
|
|
second_word = self._choose(second_words)
|
|
words = [first_word, second_word]
|
|
elif len(first_words) == 1:
|
|
first_word = first_words[0].lower()
|
|
second_two_words = self.bot.database.execute_fetchall("""SELECT
|
|
second_word, third_word, frequency FROM markov WHERE
|
|
channel_id=? AND first_word=? AND second_word NOT NULL AND
|
|
third_word NOT NULL""", [channel_id, first_word])
|
|
if not second_two_words:
|
|
return None
|
|
|
|
second_word, third_word = self._choose(
|
|
[[[s, t], f] for s, t, f in second_two_words])
|
|
words = [first_word, second_word, third_word]
|
|
else:
|
|
words = [word.lower() for word in first_words]
|
|
|
|
for i in range(30):
|
|
two_words = words[-2:]
|
|
third_words = self.bot.database.execute_fetchall("""SELECT
|
|
third_word, frequency FROM markov WHERE channel_id=? AND
|
|
first_word=? AND second_word=?""", [channel_id]+two_words)
|
|
if not third_words:
|
|
break
|
|
|
|
third_word = self._choose(third_words)
|
|
if third_word == None:
|
|
break
|
|
words.append(third_word)
|
|
|
|
if words == first_words:
|
|
return None
|
|
|
|
return " ".join(words)
|