factor out common bot|repl code

main
Yorick van Pelt 2024-04-27 10:21:15 +02:00
parent 04ceff8a29
commit c98824ceaa
Signed by: yorick
GPG Key ID: D8D3CC6D951384DE
6 changed files with 59 additions and 49 deletions

View File

@ -1,9 +1,12 @@
[url]
nitter_domain = "https://localhost"
[networks."libera"]
[irc."libera"]
host = "irc.libera.net"
port = 6697
# doesn't really exist
[networks."libera".channels."##yobotdev"]
[irc."libera".channels."##yobotdev"]
plugins = ["url"]
[repl]
plugins = ["url"]

View File

@ -1,20 +1,28 @@
#!/usr/bin/env python
import sys
import yobot.repl
import yobot.bot
from yobot.yobot import Yobot
import asyncio
if __name__ == "__main__":
def main() -> None:
if len(sys.argv) < 2:
print("usage: $0 repl|bot")
sys.exit(1)
bot = Yobot()
match sys.argv[1]:
case "repl":
asyncio.run(yobot.repl.main())
asyncio.run(yobot.repl.main(bot))
case "bot":
asyncio.run(yobot.bot.main())
asyncio.run(yobot.bot.main(bot))
case _:
print("usage: $0 repl|bot")
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,41 +1,31 @@
import tomllib
from pathlib import Path
from __future__ import annotations
from irctokens.line import build, Line
from ircrobots.bot import Bot as BaseBot
from ircrobots.server import Server as BaseServer
from ircrobots.params import ConnectionParams
from importlib import import_module
from .interface import IPlugin, IServer
with Path("config.toml").open("rb") as f:
CONFIG = tomllib.load(f)
PLUGINS: dict[str, IPlugin] = {
"url": import_module("yobot.plugins.url").Plugin(CONFIG["url"]),
}
from .interface import IServer
from .yobot import Yobot
class Server(BaseServer, IServer):
bot: Bot
async def line_read(self, line: Line) -> None:
if line.command not in ("PING", "PONG", "AWAY"):
print(f"{self.name} < {line.format()}")
me = self.nickname_lower
conf = CONFIG["networks"][self.name]
# TODO: config might not match casefold
if (
line.command == "PRIVMSG"
and self.has_channel(line.params[0])
and line.hostmask is not None
and self.casefold(line.hostmask.nickname) != me
and self.casefold(line.params[0]) in conf["channels"]
and self.has_user(line.hostmask.nickname)
):
channel_conf = conf["channels"][self.casefold(line.params[0])]
for plugin in channel_conf["plugins"]:
await PLUGINS[plugin].handle(self, line)
await self.bot.yobot.handle_irc(
self, self.name, self.casefold(line.params[0]), line
)
async def line_send(self, line: Line) -> None:
if line.command not in ("PING", "PONG"):
@ -46,13 +36,17 @@ class Server(BaseServer, IServer):
class Bot(BaseBot):
def __init__(self, yobot: Yobot) -> None:
self.yobot = yobot
super().__init__()
def create_server(self, name: str) -> BaseServer:
return Server(self, name)
async def main() -> None:
bot = Bot()
for name, network in CONFIG["networks"].items():
async def main(yobot: Yobot) -> None:
bot = Bot(yobot)
for name, network in yobot.config["irc"].items():
params = ConnectionParams(
"yobot",
network["host"],

View File

@ -3,22 +3,25 @@
import sys
import yobot.repl
import yobot.bot
from yobot.yobot import Yobot
import asyncio
def main() -> None:
if len(sys.argv) < 2:
print("usage: $0 repl|bot")
sys.exit(1)
bot = Yobot()
match sys.argv[1]:
case "repl":
asyncio.run(yobot.repl.main())
asyncio.run(yobot.repl.main(bot))
case "bot":
asyncio.run(yobot.bot.main())
asyncio.run(yobot.bot.main(bot))
case _:
print("usage: $0 repl|bot")
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,25 +1,19 @@
import tomllib
from aioconsole import ainput
from pathlib import Path
from importlib import import_module
from .interface import IPlugin, IServer, ILine
with Path("config.toml").open("rb") as f:
CONFIG = tomllib.load(f)
PLUGINS: dict[str, IPlugin] = {
"url": import_module("yobot.plugins.url").Plugin(CONFIG["url"]),
}
from .interface import IServer, ILine
from .yobot import Yobot
class Srv(IServer):
async def reply(self, _line: ILine, msg: str) -> None:
def __init__(self, yobot: Yobot) -> None:
self.yobot = yobot
super().__init__()
async def reply(self, line: ILine, msg: str) -> None: # noqa: ARG002
print(msg)
async def line_read(self, msg: ILine) -> None:
for plugin in PLUGINS.values():
await plugin.handle(self, msg)
await self.yobot.handle_repl(self, msg)
async def run(self) -> None:
try:
@ -27,12 +21,15 @@ class Srv(IServer):
s = await ainput("<user> ")
await self.line_read(
ILine(
source="user", tags={}, command="PRIVMSG", params=["#test", s],
source="user",
tags={},
command="PRIVMSG",
params=["#test", s],
),
)
except EOFError:
pass
async def main() -> None:
await Srv().run()
async def main(yobot: Yobot) -> None:
await Srv(yobot).run()

View File

@ -3,8 +3,9 @@ from pathlib import Path
from importlib import import_module
import tomllib
class Yobot:
def __init__(self, config_path: str) -> None:
def __init__(self, config_path: str = "config.toml") -> None:
with Path(config_path).open("rb") as f:
self.config = tomllib.load(f)
# todo: casefold config
@ -17,11 +18,15 @@ class Yobot:
plugin_names.update(self.config["repl"]["plugins"])
self.plugins: dict[str, IPlugin] = {
plugin_name: import_module(f"yobot.plugins.{plugin_name}").Plugin(self.config[plugin_name])
plugin_name: import_module(f"yobot.plugins.{plugin_name}").Plugin(
self.config[plugin_name]
)
for plugin_name in plugin_names
}
async def handle_irc(self, bot: IServer, network: str, channel: str, line: ILine) -> None:
async def handle_irc(
self, bot: IServer, network: str, channel: str, line: ILine
) -> None:
conf = self.config["irc"][network]
if channel in conf["channels"]:
channel_conf = conf["channels"][channel]