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
|
import time
import logging
from enum import Enum
from velascobot2.utils import ChatData, GraphicType, Response
from velascobot2.parrot import Parrot
logger = logging.getLogger(__name__)
class ReplyMode(Enum):
FIXED = 1
CHANCE = 2
class Messenger:
def __init__(
self,
bot_name: str,
bosses: list[str] = [],
nicknames: list[str] = [],
reply: float = 0.1,
repeat: float = 0.05,
wake_up: bool = False,
mode: ReplyMode = ReplyMode.FIXED,
mute_time: int = 60,
max_len: int = 50,
min_period: int = 1,
max_period: int = 100000,
):
# List of nicknames (other than the username) that the bot can react to
self.names = nicknames
# Mute time for Telegram network errors
self.mute_time = mute_time
# Last mute timestamp
self.mute_timer = None
# The bot's username, "@" included. Updated on Bot creation
self.username = "@" + bot_name
# Wakeup flag that determines if it should send a wakeup message to
# stored groupchats
self.wake_up = wake_up
# Chance of sending messages as replies
self.reply = reply
# Chance of sending 2 messages in a row
self.repeat = repeat
# Bot bosses ID
self.bosses = bosses
# Max word length for a message
self.max_len = max_len
# Whether a message is guaranteed after period_messages, or if it's a
# random chance
self.mode = mode
# minimum value allowed for message_period
self.min_period = min_period
# maximum value allowed for message_period
self.max_period = max_period
self._parrot = Parrot(bot_name)
def get_all_chats(self) -> list[ChatData]:
return self._parrot.get_all_chats()
def mentioned(self, text: str) -> bool:
"""Returns True if the bot's username or any of its nicknames is
mentioned in the text, as long as it's not some other user's
username"""
if self.username in text:
return True
for name in self.names:
if name in text and "@{}".format(name) not in text:
return True
return False
def is_mute(self) -> bool:
"""Returns True if not enough time has passed since the last time the
bot got muted."""
current_time = int(time.perf_counter())
return (
self.mute_timer is not None
and (current_time - self.mute_timer) < self.mute_time
)
def give_response(
self, chat_id: str
) -> tuple[Response, str | tuple[GraphicType, str]]:
response = self._parrot.choose_response_type(chat_id)
if response == Response.TEXT:
return response, self._parrot.give_text(chat_id)
elif response == Response.GRAPHIC:
return response, self._parrot.give_graphic(chat_id)
raise Exception("Response type not implemented")
def set_mute_timer(self, counter: int):
self.mute_timer = counter
def process_text(self, chat_id: str, text: str):
logger.debug(f"{chat_id} - Processing text: {text}")
for name in self.names:
if name.casefold() in text.casefold() and len(text.split()) <= 3:
# If it's less than 3 words and one of the bot's names is in
# the message, ignore it as it's most probably just a summon
return
self._parrot.parse_text(chat_id, text)
def process_graphic(
self, chat_id: str, graph_type: GraphicType, graph_id: str
):
logger.debug(f"{chat_id} - Processing graphic {graph_type}: {graph_id}")
self._parrot.parse_graphic(chat_id, graph_type, graph_id)
def get_chat_data(self, chat_id: str) -> ChatData | None:
return self._parrot.get_chat_data(chat_id)
def ensure_chat_exists(self, chat_id: str, chat_type: str, chat_name: str):
self._parrot.ensure_chat_exists(chat_id, chat_type, chat_name)
def update_chat_name(self, chat_id: str, chat_name: str):
self._parrot.update_chat_name(chat_id, chat_name)
def update_message_period(self, chat_id: str, period: int):
if period < self.min_period or period > self.max_period:
raise Exception(
f"The period must be between {self.min_period} and"
f" {self.max_period}, but got {period}"
)
self._parrot.update_chat_message_period(chat_id, period)
def update_answer_chance(self, chat_id: str, chance: float):
if chance < 0 or chance > 1:
raise Exception(
f"The period must be between 0 and 1, but got {chance}"
)
self._parrot.update_chat_answer_chance(chat_id, chance)
def update_set_restricted(self, chat_id: str, restricted: bool):
self._parrot.update_chat_set_restricted(chat_id, restricted)
def update_set_silenced(self, chat_id: str, silenced: bool):
self._parrot.update_chat_set_silenced(chat_id, silenced)
def get_processed_messages(self, chat_id: str) -> int:
return self._parrot.get_processed_messages(chat_id)
|
