def get_topic_id(message: telebot.types.Message) -> str:
"""
Get the topic ID from a Telegram message.
Parameters:
message (telebot.types.Message): The Telegram message object.
Returns:
str: '[chat.id] [topic.id]'
"""
chat_id = message.chat.id
# topic_id = 'not topic'
topic_id = 0
if message.reply_to_message and message.reply_to_message.is_topic_message:
topic_id = message.reply_to_message.message_thread_id
elif message.is_topic_message:
topic_id = message.message_thread_id
# bot.reply_to(message, f'DEBUG: [{chat_id}] [{topic_id}]')
return f'[{chat_id}] [{topic_id}]'>>> a=(300).to_bytes(length=1)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
OverflowError: int too big to convert
>>> a=(300).to_bytes(length=2)
>>> a
b'\x01,'
>>>#!/usr/bin/env python3
import threading
import telebot
import cfg
bot = telebot.TeleBot(cfg.token, skip_pending=True)
# до 40 одновременных потоков для чата с гпт и бингом
semaphore_talks = threading.Semaphore(40)
class ShowAction(threading.Thread):
"""Поток который можно остановить. Беспрерывно отправляет в чат уведомление об активности.
Телеграм автоматически гасит уведомление через 5 секунд, по-этому его надо повторять.
Использовать в коде надо как то так
with ShowAction(chat_id, 'typing'):
делаем что-нибудь и пока делаем уведомление не гаснет
"""
def __init__(self, chat_id, action):
"""_summary_
Args:
chat_id (_type_): id чата в котором будет отображаться уведомление
action (_type_): "typing", "upload_photo", "record_video", "upload_video", "record_audio",
"upload_audio", "upload_document", "find_location", "record_video_note", "upload_video_note"
"""
super().__init__()
self.actions = [ "typing", "upload_photo", "record_video", "upload_video", "record_audio",
"upload_audio", "upload_document", "find_location", "record_video_note", "upload_video_note"]
assert action in self.actions, f'Допустимые actions = {self.actions}'
self.chat_id = chat_id
self.action = action
self.is_running = True
#self.start()
self.timerseconds = 1
def run(self):
while self.is_running:
bot.send_chat_action(self.chat_id, self.action)
n = 50
while n > 0:
time.sleep(0.1)
n = n - self.timerseconds
def stop(self):
self.timerseconds = 50
self.is_running = False
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
@bot.message_handler(content_types = ['voice'])
def handle_voice(message: telebot.types.Message):
"""Автоматическое распознавание текст из голосовых сообщений"""
thread = threading.Thread(target=handle_voice_thread, args=(message,))
thread.start()
def handle_voice_thread(message: telebot.types.Message):
"""Автоматическое распознавание текст из голосовых сообщений"""
my_log.log_media(message)
with semaphore_talks:
# Создание временного файла
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
file_path = temp_file.name
# Скачиваем аудиофайл во временный файл
file_info = bot.get_file(message.voice.file_id)
downloaded_file = bot.download_file(file_info.file_path)
with open(file_path, 'wb') as new_file:
new_file.write(downloaded_file)
# Распознаем текст из аудио
# если мы не в привате и в этом чате нет блокировки автораспознавания то показываем активность
if not (message.chat.id in BLOCKS and BLOCKS[message.chat.id] == 1) or message.chat.type == 'private':
with ShowAction(message.chat.id, 'typing'):
if cfg.stt == 'vosk':
text = my_stt.stt(file_path)
elif cfg.stt == 'whisper':
text = my_whisper.get_text(file_path)
else:
if cfg.stt == 'vosk':
text = my_stt.stt(file_path)
elif cfg.stt == 'whisper':
text = my_whisper.get_text(file_path)
os.remove(file_path)
# если мы не в привате и в этом чате нет блокировки автораспознавания
if not (message.chat.id in BLOCKS and BLOCKS[message.chat.id] == 1) or message.chat.type == 'private':
# Отправляем распознанный текст
if text.strip() != '':
bot.reply_to(message, text, reply_markup=get_keyboard('hide'))
my_log.log_echo(message, f'[ASR] {text}')
else:
bot.reply_to(message, 'Очень интересно, но ничего не понятно.', reply_markup=get_keyboard('hide'))
my_log.log_echo(message, '[ASR] no results')
# и при любом раскладе отправляем текст в обработчик текстовых сообщений, возможно бот отреагирует на него если там есть кодовые слова
if text:
message.text = text
echo_all(message)
def set_default_commands():
"""
Reads a file containing a list of commands and their descriptions,
and sets the default commands for the bot.
"""
commands = []
with open('commands.txt', encoding='utf-8') as file:
for line in file:
try:
command, description = line[1:].strip().split(' - ', 1)
if command and description:
commands.append(telebot.types.BotCommand(command, description))
except Exception as error:
print(error)
bot.set_my_commands(commands)
def main():
"""
Runs the main function, which sets default commands and starts polling the bot.
"""
set_default_commands()
bot.polling()
if __name__ == '__main__':
main()>>> a = list (range(100))
>>> b = [range(100)]
>>> c = [list(range(100))]
>>> a
[0, 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]
>>> b
[range(0, 100)]
>>> c
[[0, 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]]
>>> type(b[0])
<class 'range'>
>>> d = [x for x in range(100)]
>>> d
[0, 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 pprint
data = {
'level_1_key_1': 'value_1',
'level_1_key_2': {
'level_2_key_1': 'value_2',
'level_2_key_2': {
'level_3_key_1': 'value_3',
'level_3_key_2': {
'level_4_key_1': 'value_4',
'level_4_key_2': 'value_5'
}
}
}
}
pprint.pprint(data)
{'level_1_key_1': 'value_1',
'level_1_key_2': {'level_2_key_1': 'value_2',
'level_2_key_2': {'level_3_key_1': 'value_3',
'level_3_key_2': {'level_4_key_1': 'value_4',
'level_4_key_2': 'value_5'}}}}