tracer/tracer.py

#!/usr/bin/env python3

from irctokens import build, Line
from ircrobots import Bot as BaseBot
from ircrobots import Server as BaseServer
from ircrobots import ConnectionParams, SASLUserPass
from tracery.modifiers import base_english
import asyncio
import configparser
import glob
import json
import os
import random
import traceback
import tracery

HELPTEXT = "helo i'm a tracery bot that makes cool things from tracery grammars in your ~/.tracery. see http://tracery.io for more info"
REPOLINK = "https://tildegit.org/ben/tracer"

DB = {}

# read configuration
config = configparser.ConfigParser(
    converters={"list": lambda x: [i.strip() for i in x.split(",")]}
)
config.read("config.ini")
config = config["irc"]

account = configparser.ConfigParser()
account.read("account.ini")
account = account["nickserv"]


def grammar(rules):
    try:
        res = tracery.Grammar(rules)
        res.add_modifiers(base_english)
        return res
    except Exception as e:
        print(e)


def load_rules(path):
    try:
        with open(path) as f:
            return json.loads(f.read())
    except Exception as e:
        print(e)


def populate():
    global DB
    DB = {}
    for p in glob.glob("/home/*/.tracery/*"):
        name, ext = os.path.splitext(p)
        name = os.path.basename(name)
        if name.startswith(".") or ext not in [".json", ""]:
            continue
        if p in DB:
            DB[name].append(p)
        else:
            DB[name] = [p]


populate()


def generate(rule):
    populate()
    if rule in DB:
        return grammar(load_rules(random.choice(DB[rule]))).flatten("#origin#")


def listify(col):
    return col if type(col) == type([]) else [col]


def shuffle(col):
    a = random.choice(list(col))
    b = random.choice(list(col))
    if "origin" in [a, b]:
        return col
    col[a], col[b] = col[b], col[a]
    return col


def fuse(argv):
    populate()
    raw = {}
    for gk in argv:
        if gk in DB:
            g = grammar(load_rules(random.choice(DB[gk]))).raw
            for k in g:
                if k in raw:
                    raw[k] = listify(raw[k]) + listify(g[k])
                else:
                    raw[k] = g[k]
    for i in range(20):
        raw = shuffle(raw)
    return grammar(raw).flatten("#origin#")


def think(line):
    words = line.params[1].split(" ")
    if words[0] == "!!list":
        return " ".join(DB)
    elif words[0] == "!!source":
        return REPOLINK
    elif words[0] in ["!!help", "!botlist"]:
        return HELPTEXT
    elif words[0] == "!!fuse":
        if "|" in words:
            w = words.index("|")
            res = fuse(words[1:w])
            if res:
                return " ".join(words[w + 1 :]) + " " + res
        else:
            res = fuse(words[1:])
            if res:
                return res
    elif words[0].startswith("!!"):
        res = generate(words[0][2:])
        if res:
            if "|" in words:
                w = words.index("|")
                return " ".join(words[w + 1 :]) + " " + res
            else:
                return res


class Server(BaseServer):
    async def line_send(self, line: Line):
        print(f"{self.name} > {line.format()}")

    async def line_read(self, line: Line):
        print(f"{self.name} < {line.format()}")
        if line.command == "001":
            await self.send(build("JOIN", [",".join(config.getlist("channels"))]))
            await self.send(build("MODE", [self.nickname, "+B"]))

        if line.command == "INVITE":
            await self.send(build("JOIN", [line.params[1]]))

        if (
            line.command == "PRIVMSG"
            and self.has_channel(line.params[0])
            and self.has_user(line.hostmask.nickname)
            and len(line.params[1].split(" ")) > 0
            and not line.hostmask is None
            and not self.casefold(line.hostmask.nickname) == self.nickname_lower
            and not ("batch" in line.tags and line.tags["batch"] == "1")
            and not "inspircd.org/bot" in line.tags
        ):
            try:
                response = think(line)
                if response is not None:
                    await self.send(build("PRIVMSG", [line.params[0], response]))
            except Exception as e:
                print("ERROR", line)
                print(e)
                traceback.print_exc()


class Bot(BaseBot):
    def create_server(self, name: str):
        return Server(self, name)


async def main():
    params = ConnectionParams(
        config["nick"],
        host=config["server"],
        port=config.getint("port"),
        tls=config.getboolean("tls"),
        sasl=SASLUserPass(account["username"], account["password"]),
    )
    bot = Bot()
    await bot.add_server("tilde", params)
    await bot.run()


if __name__ == "__main__":
    asyncio.run(main())