forked from bossoq/Twitch-Channel-Points-Miner-v2
-
Notifications
You must be signed in to change notification settings - Fork 326
/
Chat.py
99 lines (76 loc) · 2.67 KB
/
Chat.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
import logging
import time
from enum import Enum, auto
from threading import Thread
from irc.bot import SingleServerIRCBot
from TwitchChannelPointsMiner.constants import IRC, IRC_PORT
from TwitchChannelPointsMiner.classes.Settings import Events
logger = logging.getLogger(__name__)
class ChatPresence(Enum):
ALWAYS = auto()
NEVER = auto()
ONLINE = auto()
OFFLINE = auto()
def __str__(self):
return self.name
class ClientIRC(SingleServerIRCBot):
def __init__(self, username, token, channel):
self.token = token
self.channel = "#" + channel
self.__active = False
super(ClientIRC, self).__init__(
[(IRC, IRC_PORT, f"oauth:{token}")], username, username
)
def on_welcome(self, client, event):
client.join(self.channel)
def start(self):
self.__active = True
self._connect()
while self.__active:
try:
self.reactor.process_once(timeout=0.2)
time.sleep(0.01)
except Exception as e:
logger.error(
f"Exception raised: {e}. Thread is active: {self.__active}"
)
def die(self, msg="Bye, cruel world!"):
self.connection.disconnect(msg)
self.__active = False
"""
def on_join(self, connection, event):
logger.info(f"Event: {event}", extra={"emoji": ":speech_balloon:"})
"""
# """
def on_pubmsg(self, connection, event):
msg = event.arguments[0]
# also self._realname
# if msg.startswith(f"@{self._nickname}"):
if f"@{self._nickname.lower()}" in msg.lower():
# nickname!username@nickname.tmi.twitch.tv
nick = event.source.split("!", 1)[0]
# chan = event.target
logger.info(f"{nick} at {self.channel} wrote: {msg}", extra={
"emoji": ":speech_balloon:", "event": Events.CHAT_MENTION})
# """
class ThreadChat(Thread):
def __deepcopy__(self, memo):
return None
def __init__(self, username, token, channel):
super(ThreadChat, self).__init__()
self.username = username
self.token = token
self.channel = channel
self.chat_irc = None
def run(self):
self.chat_irc = ClientIRC(self.username, self.token, self.channel)
logger.info(
f"Join IRC Chat: {self.channel}", extra={"emoji": ":speech_balloon:"}
)
self.chat_irc.start()
def stop(self):
if self.chat_irc is not None:
logger.info(
f"Leave IRC Chat: {self.channel}", extra={"emoji": ":speech_balloon:"}
)
self.chat_irc.die()