-
Notifications
You must be signed in to change notification settings - Fork 2
/
hakcbot_init.py
65 lines (49 loc) · 2.05 KB
/
hakcbot_init.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
#!/usr/bin/python3
import os
import asyncio
from socket import socket
from config import * # pylint: disable=unused-wildcard-import
from hakcbot_utilities import load_from_file, Log as L
class Init:
def __init__(self, Hakcbot):
self.Hakcbot = Hakcbot
self.Hakcbot.sock.setblocking(0)
self.connect_process = [
f'PASS {PASS}',
f'NICK {IDENT}',
f'JOIN #{CHANNEL}',
'CAP REQ :twitch.tv/tags'
]
async def initialize(self):
self._load_json_data()
self._create_tld_set()
await self._join_room()
await self._wait_for_eol()
async def _join_room(self):
loop = asyncio.get_running_loop()
await loop.sock_connect(self.Hakcbot.sock, (self._host, self._port))
for step in self.connect_process:
await loop.sock_sendall(self.Hakcbot.sock, f'{step}\r\n'.encode('utf-8'))
async def _wait_for_eol(self):
loop = asyncio.get_running_loop()
while True:
data = await loop.sock_recv(self.Hakcbot.sock, 1024)
data = data.decode('utf-8', 'ignore').strip()
if ('authentication failed' in data):
L.l0('Authentication failure!')
os._exit(1)
if ('End of /NAMES list' in data):
L.l1('hakcbot: NOW CONNECTED TO INTERWEBS. PREPARE FOR CYBER WARFARE.')
break
def _load_json_data(self):
stored_data = load_from_file('config')
self.Hakcbot.titles = stored_data['titles']
self.Hakcbot.quotes = stored_data['quotes']
self.Hakcbot.url_whitelist = set(stored_data['url_whitelist']) # easier to deal with a set
self.Hakcbot.word_filter = stored_data['word_filter']
self._host = stored_data['twitch']['host']
self._port = stored_data['twitch']['port']
def _create_tld_set(self):
with open('TLDs') as TLDs:
tlds = TLDs.read().splitlines()
self.Hakcbot.domain_tlds = set([t.lower() for t in tlds if len(t) <= 6 and not t.startswith('#')])