137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
import asyncio
|
|
import random
|
|
import time
|
|
import argparse
|
|
|
|
from db import DuckDB
|
|
from db import DuckEvent
|
|
from db import DuckStats
|
|
|
|
from channels import ChannelDB
|
|
|
|
from lang import lang
|
|
|
|
from admin import Admin
|
|
|
|
# fill in with password = "password"
|
|
from secret import password
|
|
|
|
from irctokens import build, Line
|
|
from ircrobots import Bot as BaseBot
|
|
from ircrobots import Server as BaseServer
|
|
from ircrobots import ConnectionParams
|
|
|
|
class DuckLogic:
|
|
async def new_duck(self):
|
|
self.messages = 0
|
|
self.duckactive = True
|
|
self.duckactivetime = time.time()
|
|
await self.msgall(lang["duck"])
|
|
|
|
async def duck_test(self):
|
|
if self.messages > 100 and random.randint(0, 99) < 10: await self.new_duck()
|
|
|
|
async def misstime(self):
|
|
return format(time.time() - self.lastduck, '.2f')
|
|
|
|
async def coughttime(self):
|
|
return format(self.lastduck - self.duckactivetime, '.2f')
|
|
|
|
async def duck_action(self, user, chan):
|
|
db = DuckDB(self.db)
|
|
if self.duckactive:
|
|
self.duckactive = False
|
|
self.messages = 0
|
|
self.lastduck = time.time()
|
|
await self.msgall(lang["duckcought"].format(user, chan, await self.coughttime()))
|
|
db.add("B", user, time.time(), float(await self.coughttime()), chan)
|
|
elif self.lastduck != 0:
|
|
await self.msg(chan, lang["noduck"].format(await self.misstime()), user)
|
|
db.add("M", user, time.time(), float(await self.misstime()), chan)
|
|
else:
|
|
await self.msg(chan, lang["noduckstart"], user)
|
|
db.add("M", user, time.time(), -1, chan)
|
|
db.write(self.db)
|
|
|
|
class Server(BaseServer, DuckLogic):
|
|
messages = 0
|
|
duckactive = False
|
|
duckactivetime = 0
|
|
lastduck = 0
|
|
db = "duckdb"
|
|
chandb = "chandb"
|
|
|
|
async def msg(self, chan, msg, usr=None):
|
|
await self.send(build("PRIVMSG", ["ChanServ", "identify {}".format(password)]))
|
|
if usr != None:
|
|
await self.send(build("PRIVMSG", [chan, usr + ": " + msg]))
|
|
else: await self.send(build("PRIVMSG", [chan, msg]))
|
|
|
|
async def msgall(self, msg):
|
|
[await self.msg(channel, msg) for channel in self.channels]
|
|
|
|
async def line_read(self, line: Line):
|
|
print(f"{self.name} < {line.format()}")
|
|
if line.command == "001":
|
|
chans = ChannelDB(self.chandb)
|
|
for i in chans.list():
|
|
await self.send(build("JOIN", [i]))
|
|
elif line.command == "PRIVMSG":
|
|
if line.params[1][0] == '%':
|
|
cmd = line.params[1].split(' ')[0][1:]
|
|
chan = line.params[0]
|
|
user = line.hostmask.nickname
|
|
args = line.params[1].split(' ')[1:]
|
|
if cmd == "bef": await self.duck_action(user, chan)
|
|
# elif cmd == "trigger": await self.new_duck()
|
|
elif cmd == "stats":
|
|
db = DuckDB(self.db)
|
|
stats = DuckStats(db)
|
|
await self.msg(chan, lang["stats"].format(
|
|
args[0],
|
|
stats.cought(args[0]),
|
|
stats.channels(args[0]),
|
|
format(stats.ratio(args[0]), ".3f")
|
|
), user)
|
|
return
|
|
|
|
self.messages += 1
|
|
await self.duck_test()
|
|
elif line.command == "INVITE":
|
|
chans = ChannelDB(self.chandb)
|
|
await self.send(build("JOIN", [line.params[1]]))
|
|
chans.add(line.params[1])
|
|
chans.write(self.chandb)
|
|
elif line.command == "KICK":
|
|
chans = ChannelDB(self.chandb)
|
|
chans.remove(line.params[0])
|
|
chans.write(self.chandb)
|
|
|
|
async def line_send(self, line: Line):
|
|
print(f"{self.name} > {line.format()}")
|
|
|
|
class Bot(BaseBot):
|
|
def create_server(self, name: str):
|
|
return Server(self, name)
|
|
|
|
async def main(srv, port, tls, nick):
|
|
print(srv, port, tls, nick)
|
|
bot = Bot()
|
|
params = ConnectionParams(nick, srv, port, tls)
|
|
await bot.add_server("main", params)
|
|
await bot.run()
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="cross-channel duck bot")
|
|
parser.add_argument('-s', '--host', default="beepboop.systems")
|
|
parser.add_argument('-p', '--port', default=6667, type=int)
|
|
parser.add_argument('-t', '--tls', action="store_true")
|
|
parser.add_argument('-n', '--nick', default="test")
|
|
parser.add_argument('-a', '--password', default=password)
|
|
return parser.parse_args()
|
|
|
|
if __name__ == "__main__":
|
|
args = parse_args()
|
|
password = args.password
|
|
asyncio.run(main(args.host, args.port, args.tls, args.nick))
|