-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathrtquizzer.py
executable file
·397 lines (331 loc) · 14.3 KB
/
rtquizzer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
#!/usr/bin/env python3
import asyncio
import re
from asyncirc import irc
from enum import IntEnum
import json
import asyncirc.plugins.addressed
import threading, time, os, re
import random
random = random.SystemRandom()
import urllib.parse
import requests
import sys
import collections
from datetime import date
from bs4 import BeautifulSoup
#supybot.ircutils (https://github.com/ProgVal/limnoria/tree/master/src/ircutils.py)
class ircutils(object):
def bold(s):
"""Returns the string s, bolded."""
return '\x02%s\x02' % s
def italic(s):
"""Returns the string s, italicised."""
return '\x1D%s\x1D' % s
# Definition of mircColors dictionary moved below because it became an IrcDict.
def mircColor(s, fg=None, bg=None):
"""Returns s with the appropriate mIRC color codes applied."""
if fg is None and bg is None:
return s
elif bg is None:
if str(fg) in mircColors:
fg = mircColors[str(fg)]
elif len(str(fg)) > 1:
fg = mircColors[str(fg)[:-1]]
else:
# Should not happen
pass
return '\x03%s%s\x03' % (fg.zfill(2), s)
elif fg is None:
bg = mircColors[str(bg)]
# According to the mirc color doc, a fg color MUST be specified if a
# background color is specified. So, we'll specify 00 (white) if the
# user doesn't specify one.
return '\x0300,%s%s\x03' % (bg.zfill(2), s)
else:
fg = mircColors[str(fg)]
bg = mircColors[str(bg)]
# No need to zfill fg because the comma delimits.
return '\x03%s,%s%s\x03' % (fg, bg.zfill(2), s)
def stripColor(s):
"""Returns the string s, with color removed."""
return _stripColorRe.sub('', s)
_stripColorRe = re.compile(r'\x03(?:\d{1,2},\d{1,2}|\d{1,2}|,\d{1,2}|)')
mircColors = {
'white': '0',
'black': '1',
'blue': '2',
'green': '3',
'red': '4',
'brown': '5',
'purple': '6',
'orange': '7',
'yellow': '8',
'light green': '9',
'teal': '10',
'light blue': '11',
'dark blue': '12',
'pink': '13',
'dark grey': '14',
'light grey': '15',
'dark gray': '14',
'light gray': '15',
}
# We'll map integers to their string form so mircColor is simpler.
for (k, v) in list(mircColors.items()):
if k is not None: # Ignore empty string for None.
sv = str(v)
mircColors[sv] = sv
mircColors[sv.zfill(2)] = sv
class State(IntEnum):
Question = 0
Tips = 1
Pause = 2
Answer = 3
class Quizbot(object):
quiz = None
test = None
event = None
questions = {}
current_question = []
current_category = ""
mode = State.Question
tips = 1
winner = None
points = {}
bot = None
channel = "#rt-quiz"
counter = 0
last = None
def __init__(self, bot):
self.bot = bot
self.event = threading.Event()
self.loadStats()
self.last = date.today()
self.quiz = threading.Thread(daemon=True, target=self.quizzing, args=())
self.quiz.start()
self.test = threading.Thread(daemon=True, target=self.checkForQuiz, args=())
def sleep(self, timeout : int):
self.event.wait(timeout)
self.event.clear()
def loadQuestions(self):
with open("questions.json", "r") as fobj:
self.questions = json.load(fobj)
def loadStats(self):
self.points = collections.defaultdict(lambda: 0)
self.daily = collections.defaultdict(lambda: 0)
if os.path.isfile("stats.json"):
with open("stats.json", "r") as fobj:
self.points.update(json.load(fobj))
if os.path.isfile("daily.json"):
with open("daily.json", "r") as fobj:
self.daily.update(json.load(fobj))
def reply(self, *args):
msg = "".join(ircutils.mircColor(i, 2, 0) for i in args)
self.bot.say(self.channel, msg)
def topic(self, *args):
topic = "".join(ircutils.mircColor(i, 2, 0) for i in args)
self.bot.writeln(f'TOPIC {self.channel} :{topic}')
def random(self, r : int):
return int(random.random() * r)
def checkForQuiz(self):
while True:
if not self.quiz.is_alive():
self.reply("Starte Quiz neu...")
del self.quiz
self.quiz = threading.Thread(daemon=True, target=self.quizzing, args=())
self.quiz.start()
self.sleep(60)
def quizzing(self):
while True:
if self.mode == State.Question:
self.loadQuestions()
self.winner = None
self.tips = 1
self.counter = 0
# [category, question, answer]
try:
self.current_question = random.choice(self.questions)
if not (self.current_question and len(self.current_question) >= 3 and self.validQuestion(self.current_question)):
continue
continue
self.reply(f"Kategorie {ircutils.bold(self.current_question[0])}: {self.current_question[1]}")
l = len(self.current_question[2])
self.current_question.append(l * 2 if l < 80 else l)
if not self.random(10):
self.reply(ircutils.mircColor("ACHTUNG: Dies ist eine Superpunkterunde. Der Gewinner bekommt die dreifache Punktezahl!", 4, 1))
self.current_question[3] *= 3
self.mode = State.Tips
except Exception as e: # general ignore
self.reply(f"Frage konnte nicht geladen werden: {str(e)}")
self.sleep(4)
continue
elif self.mode == State.Tips:
if self.counter < 4:
self.counter += 1
self.sleep(5)
continue
self.reply("{}{}{}".format(ircutils.bold("Tipp: "), self.current_question[2][:self.tips], "." * (len(self.current_question[2]) - self.tips)))
self.tips += 1
if self.tips >= len(self.current_question[2]):
self.counter = 0
self.mode = State.Pause
self.sleep(4)
continue
elif self.mode == State.Pause:
if not self.counter:
self.reply(ircutils.mircColor("Achtung, wenn die Frage innerhalb von 30 Sekunden nicht beantwortet wird, werde ich automatisch eine neue Runde starten!", 3, 1))
if self.counter < 6:
self.counter += 1
self.sleep(5)
continue
else:
self.counter = 0
self.mode = State.Answer
elif self.mode == State.Answer:
if self.winner is not None:
x = re.match(r"(.*?).{1}onAir", self.winner, re.IGNORECASE)
if x:
self.winner = x[1]
aliases = {
"l-micha" : "lmichael",
"spunki" : "lmichael"
}
if self.winner in aliases:
self.winner = aliases[self.winner]
if self.winner not in self.points:
for k in self.points:
if k.lower() == self.winner.lower():
self.winner = k
try:
self.current_question[3] = int(self.current_question[3])
except ValueError:
self.current_question[3] = len(self.current_question[2])
self.points[self.winner] += self.current_question[3]
self.daily[self.winner] += self.current_question[3]
self.reply(f"{self.winner} hat die Antwort", ircutils.mircColor(" " + self.current_question[2] + " ", 7, 1), "korrekt erraten, dafür gibt es", ircutils.mircColor(" " + str(self.current_question[3]) + " ", 4, 1), "Punkte!")
else:
self.reply(f"Keiner hat die Antwort", ircutils.mircColor(" " + self.current_question[2] + " ", 7, 1), "korrekt erraten :(")
self.mode = State.Question
self.current_question = None
self.current_category = ""
with open("stats.json", "w") as fobj:
json.dump(dict(self.points), fobj)
with open("daily.json", "w") as fobj:
json.dump(dict(self.daily), fobj)
if date.today() - self.last:
self.last = date.today()
self.daily = collections.defaultdict(lambda: 0)
self.reply(ircutils.mircColor("-------------", 7, 1))
#self.reply(ircutils.mircColor("Nächste Frage in 20s!", 7, 1))
#self.reply(ircutils.mircColor("-------------", 7, 1))
#self.sleep(20)
self.sleep(5)
def validQuestion(self, q : str) -> bool:
for i in ["Tipp", "Top 10", "[email protected]", "Zeit ist vorbei"]:
if i in q:
return False
return True
quiz = None
def git():
cached = os.stat(__file__).st_mtime
while True:
os.system("git pull")
stamp = os.stat(__file__).st_mtime
if stamp != cached:
cached = stamp
print("Restarting")
os._exit(0)
time.sleep(300)
asyncirc.plugins.addressed.register_command_character("!")
bot = irc.connect("irc.euirc.net", 6667, use_ssl=False)
bot.register("RT-Quizzer", "RT-Quizzer", "RT-Quizzer", password="quizzer").join([Quizbot.channel, "#radio-thirty"])
@bot.on("irc-001")
def connected(par=None):
global quiz
quiz = Quizbot(bot)
threading.Thread(target=git, args=(), daemon=True).start()
bot.writeln(f"MODE {bot.nick} +B")
@bot.on("addressed")
def on_addressed(message, user, target, text):
global quiz
def say(target, text):
text = text.replace("\n", "").replace("\r", "")
while text:
bot._writeln(f"PRIVMSG {target} :{text[:400]}")
text = text[400:]
if target == "#radio-thirty":
aliases = {
re.compile("^wedda") : "wetter chieming",
re.compile("^weer") : "wetter 26197",
re.compile("^wetter f.rth") : "wetter fürth"
}
for regex, location in aliases.items():
text = regex.sub(location, text)
if text.startswith("wetter"):
cmd = text.split()
if len(cmd) < 2:
return
cmd[1] = cmd[1].lower().split(".")[0]
if cmd[1] == "moon" or cmd[1].startswith(":"):
return
r = requests.get(f"http://de.wttr.in/{urllib.parse.quote(cmd[1])}?Q0T")
for i, line in enumerate(r.text.splitlines()):
if i and i % 6 == 0:
sleep(2)
say(target, line)
elif text.startswith("sendeplan"):
page = BeautifulSoup(requests.get("http://radio-thirty.de/sendeplan_xl").text, "lxml")
emissions = []
started = False
current_emission = {}
for tr in page.tr.td.table.find_all("tr", recursive=False):
try:
emission = {
"time" : tr.td.table.tr.td.next_sibling.next_sibling.text.replace("°", "").replace("U", " U"),
"moderator" : tr.td.next_sibling.next_sibling.table.tr.td.next_sibling.next_sibling.text,
"title" : tr.td.next_sibling.next_sibling.table.tr.td.next_sibling.next_sibling.next_sibling.next_sibling.text
}
if emission["moderator"] == "":
started = False
if current_emission != {}:
emissions.append(current_emission.copy())
current_emission = {}
elif started:
if current_emission["moderator"] != emission["moderator"]:
emissions.append(current_emission.copy())
current_emission = emission
else:
current_emission["time"] = current_emission["time"].split("-")[0] + "-" + emission["time"].split("-")[1]
current_emission["title"] += emission["title"]
else:
started = True
current_emission = emission
except Exception:
continue
for emission in emissions:
say(target, f"{emission['title']} mit {emission['moderator']} von {emission['time']}")
if target != Quizbot.channel or not quiz:
return
if text in ["punkte", "tag"]:
for i, p in enumerate(sorted((quiz.daily if text == "tag" else quiz.points).items(), key=lambda x: x[1], reverse=True), start=1):
quiz.reply(f"{i}.\t{p[0]} ({p[1]})")
if i >= 10:
break
elif text == "anzahl":
quiz.reply(f"{len(quiz.questions)} Fragen")
elif text == "frage":
quiz.reply(f"Kategorie {ircutils.bold(quiz.current_question[0])}: {quiz.current_question[1]}")
current_category = ""
current_question = []
questions = {}
@bot.on("message")
def on_message(message, user, target, text):
if target == Quizbot.channel and quiz and quiz.current_question and not quiz.winner and text.lower() == quiz.current_question[2].lower():
quiz.winner = user.nick
quiz.mode = State.Answer
quiz.event.set()
@bot.on("connection-lost")
def on_disconnected(*args):
sys.exit(0)
asyncio.get_event_loop().run_forever()