krosh/tg-keeper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix small bugs

Browse Source
This commit is contained in:
kr0sh512
2024-11-04 00:57:42 +03:00
parent f83d72a436
commit 9959ee74d5
5 changed files with 165 additions and 853 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -1,2 +1,3 @@
config.yaml
database/*
__pycache__/*
for_db.py
+73 -9
View File
@@ -1,10 +1,10 @@
import yaml
from datetime import datetime
import yaml, os
from datetime import datetime, timedelta
from telebot import types
import errno, os
users_path = "database/users.yaml"
user_path = "database/users/{}.yaml"
config = yaml.safe_load(open("config.yaml"))
users_path = config["users_path"]
user_path = config["user_path"]
def check_user(user_id: int) -> dict:
@@ -30,9 +30,11 @@ def new_user(message: types.Message) -> bool:
"first_name": message.chat.first_name,
"last_name": message.chat.last_name,
"time_created": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"last_message": None,
"remind_delta": 12 * 60 * 60, # 12 часов по умолчанию
}
users = yaml.safe_load(open(users_path, "r"))
users = yaml.safe_load(open(users_path, "r"), encoding="utf-8")
if not users:
users = {}
@@ -54,7 +56,9 @@ def new_user(message: types.Message) -> bool:
return True
def add_note(id: int, text: str, time_notif: datetime = None, list: str = "Default"):
def add_note(
id: int, text: str, time_notif: datetime = None, list: str = "Default"
) -> bool:
if not check_user(id):
return False
@@ -127,7 +131,7 @@ def add_note(id: int, text: str, time_notif: datetime = None, list: str = "Defau
return True
def get_notes(id, list="Default"):
def get_notes(id: int, list="Default") -> dict:
if not check_user(id):
return None
@@ -139,7 +143,7 @@ def get_notes(id, list="Default"):
return notes[list] if list in notes else None
def delete_note(id, note_ind, list="Default"):
def delete_note(id, note_ind, list="Default") -> bool:
if not check_user(id):
return False
@@ -169,3 +173,63 @@ def delete_note(id, note_ind, list="Default"):
file.write(yaml_data)
return True
def new_message(message: types.Message) -> int:
id = message.chat.id
new_msg_id = message.message_id
if not check_user(id):
return None
users = yaml.safe_load(open(users_path, "r", encoding="utf-8"))
user = users[id]
if "last_message" not in user:
user["last_message"] = None
last_id = user["last_message"]
user["last_message"] = new_msg_id
users[id] = user
yaml_data = yaml.dump(
users,
default_flow_style=False,
encoding="utf-8",
allow_unicode=True,
width=float("inf"),
sort_keys=False,
)
with open(users_path, "wb") as file:
file.write(yaml_data)
return last_id
def check_old_notes() -> list[int, str, int]: # возвращает первую устаревшую заметку
users = yaml.safe_load(open(users_path, "r", encoding="utf-8"))
for user_id in users.keys():
notes = yaml.safe_load(open(user_path.format(user_id), "r", encoding="utf-8"))
if "remind_delta" not in users[user_id]:
users[user_id]["remind_delta"] = 12 * 60 * 60 # не сохранится!
timedt = timedelta(seconds=users[user_id]["remind_delta"])
for list in notes.keys():
for ind in range(len(notes[list]["notes"])):
if (
notes[list]["notes"][ind]["time_notif"]
and datetime.strptime(
notes[list]["notes"][ind]["time_notif"], "%Y-%m-%d %H:%M:%S"
)
- timedt
< datetime.now()
):
return user_id, list, ind
return None, None, None
lang.yaml
+14
View File
@@ -0,0 +1,14 @@
"ru":
welcome_message:
"Добро пожаловать! Я Keeper: бот, который поможет вам отслеживать напоминания\
\n\nИспользуйте команду /help, чтобы узнать, как мной пользоваться."
help_message:
"Для создания напоминания пришлите сообщение с датой формата 1.10|1 октябрь|1ое октября и текстом напоминания.\
\nПример: 01.01 Встреча с друзьями\
\n(В беседах используйте {} для обращения к боту)\
\n\nИспользуйте команду /list, чтобы посмотреть все активные напоминания.\
\n/source - Страница бота на Github\
\n\nИли же можно всегда написать напрямую: @Kr0sH_512"
error_message: "Произошла ошибка. \
\nПожалуйста, попробуйте снова или свяжитесь с @Kr0sH_512"
no_reminders: "У вас нет напоминаний."
schedules.py
-763
View File
@@ -1,763 +0,0 @@
#!/usr/bin/python3.3
import threading, telebot, schedule, time, random, yaml
from datetime import datetime
import os, sys
from telebot import types
from telegram.constants import ParseMode
import for_json
from admin import admin_command, is_admin, send_admin_message, send_admin_document
config = yaml.safe_load(open("config.yaml"))
bot = telebot.TeleBot(config["test_token"])
PARITY_FIRST = 0
@bot.message_handler(commands=["help", "faq"])
def help(message):
help_msg = "Мои команды:\
\n/start - используй, чтобы сменить номер группы.\
\n/schedule - используй, чтобы получить расписание на сегодня.\
\n/random - сделай очередь из людей\
\n/info - используй, чтобы узнать твои настройки бота\
\n/pause - используй, чтобы прекратить получать сообщения от бота\
\n/thread - используй в нужном чате канала, чтобы бот отправлял сообщения именно туда\
\n/timeout - настрой время, когда бот будет присылать тебе сообщение\
\n/request - используй, чтобы отправить разработчику какой-то запрос\
\n/source - Страница бота на Github\
\n\nИли же можно всегда написать напрямую: @Kr0sH_512"
send_message(message.chat.id, help_msg)
if is_admin(message):
admin_help_msg = "И команды только для админа:\
\n/restart - перезапуск бота.\
\n/update - обновление schedule задач\
\n/stats - получание статистики\
\n/json - получить файл пользователей и расписания\
\n/info <i>id_пользователя</i> - узнать настройки пользователя\
\n/spam - сделать рассылку\
\n/pause_all - приостановить бота для всех (каникулы/выходные)\
\n/stop <i>id_пользователя</i> - приостановить отправку сообщений для пользователя"
send_admin_message(admin_help_msg)
return
### --//--
@bot.message_handler(commands=["test", "t"])
@admin_command
def test(message):
text = '<b>Жирный текст</b>\
\n<i>Курсивный текст</i>\
\n<u>Подчёркнутый</u>\
\n<s>Перечёркнутый текст</s>\
\n<a href="google.com">Ссылка</a>\
\n<code>Моноширинный текст</code>\
\n<pre>Форматированный с сохранением пробелов</pre>\
\n<blockquote>Цитата</blockquote>'
send_admin_message(text)
return
@bot.message_handler(commands=["restart", "r"])
@admin_command
def restart_bot(message):
send_admin_message("bye")
os.execv(sys.executable, ["python3", "-u"] + sys.argv)
@bot.message_handler(commands=["update"])
@admin_command
def update_schedules(message):
for_json.create_schedule_tasks(manual=True)
send_admin_message("Произошёл update")
return
@bot.message_handler(commands=["stats"])
@admin_command
def send_stat(message):
text = "<u>Статистика:</u>\
\n(группа: всего/откл. увед./беседы)"
sum_users = 0
sum_dis_not = 0
sum_chats = 0
groups = list(for_json.groups_in_json())
groups.append("other")
for i in groups:
users = for_json.students_in_group(i)
sum_users += len(users)
dis_not = len(
[
i
for i, x in enumerate(users)
if for_json.return_infos(x)["allow_message"] == "no"
]
)
sum_dis_not += dis_not
chats = len([i for i, x in enumerate(users) if x[0] == "-"])
sum_chats += chats
text += "\n{}: {} / {} / {}".format(i, len(users), dis_not, chats)
text += "\n\nВсего: {} / {} / {}".format(sum_users, sum_dis_not, sum_chats)
send_admin_message(text)
return
@bot.message_handler(commands=["json"])
@admin_command
def send_json(message):
with open(for_json.path_users, "rb") as json_file:
send_admin_document(json_file)
with open(for_json.path_schedule, "rb") as json_file:
send_admin_document(json_file)
with open(for_json.path_students, "rb") as json_file:
send_admin_document(json_file)
return
# info user_id
@bot.message_handler(commands=["spam"])
@admin_command
def spam(message):
send_admin_message(
"Пришли мне id, номер группы или <i>all</i> для рассылки сообщения этим пользователям"
)
bot.register_next_step_handler(message, spam_msg)
return
def spam_msg(message):
if message.text == "stop" or message.text == "стоп" or message.text[0] == "/":
send_admin_message("Отмена отправки")
return
send_admin_message("Пришли мне текст для рассылки сообщения этим пользователям")
bot.register_next_step_handler(message, spam_cnf, message.text)
return
def spam_cnf(message, data):
if message.text == "stop" or message.text == "стоп" or message.text[0] == "/":
send_admin_message("Отмена отправки")
return
global text_spam
text_spam = message.text
text = "Пожалуйста, проверь итоговое сообщение и подтверди отправку:\
\n\nКому: "
if data == "all":
text += "<u>всем</u>"
elif data in for_json.groups_in_json():
text += "{} группе".format(data)
elif for_json.return_infos(data) != None:
text += "пользователю @{}".format(for_json.return_infos(data)["username"])
else:
send_admin_message("Ошибка при выборе пользователей")
return
text += "\n\nСообщение: {}".format(message.text)
markup = types.InlineKeyboardMarkup()
markup.add(
types.InlineKeyboardButton(text="✅", callback_data="spam#{}".format(data)),
types.InlineKeyboardButton(text="❌", callback_data="spam_no"),
)
bot.send_message(
message.from_user.id, text, parse_mode=ParseMode.HTML, reply_markup=markup
)
return
@bot.callback_query_handler(func=lambda call: "spam" in call.data)
def spam_send(call):
if call.data == "spam_no":
send_admin_message("Отмена отправки")
return
tmp, data = call.data.split("#")
global text_spam
text = text_spam
users = []
if data == "all":
groups = for_json.groups_in_json()
for group in groups:
users += for_json.students_in_group(group)
users += for_json.students_in_group("other")
elif data in for_json.groups_in_json():
users = for_json.students_in_group(data)
elif for_json.return_infos(data) != None:
users = [data]
for i in users:
send_message(i, text)
bot.edit_message_text(
chat_id=call.message.chat.id,
message_id=call.message.message_id,
text="Успешно!",
parse_mode=ParseMode.HTML,
)
return
@bot.message_handler(commands=["pause_all"])
@admin_command
def pause_bot(message):
for_json.pause_bot()
return
@bot.message_handler(commands=["stop"])
@admin_command
def stop_msg(message):
if len(str(message.text).split(" ")) != 2:
send_admin_message("Эта команда вида /stop <i>id_пользователя</i>")
return
for_json.change_user_param(str(message.text).split(" ")[1], "allow_message", "no")
send_admin_message("Успешно")
return
### --//--
@bot.message_handler(commands=["start"])
def start(message):
start_txt = "Привет! Это бот, который будет кидать тебе сообщения перед нужной парой с номером кабинета/фамилией препода\
\n\nТы можешь настроить отправку сообщений в лс или добавить меня в группу, куда я буду присылать расписание\
\n\nCreated by: @Kr0sH_512"
if message.chat.type == "supergroup":
if len(message.text.split(" ")) != 2:
send_message(message.chat.id, start_txt)
send_message(
message.chat.id,
"Похоже, что этот бот добавлен в группу. Чтобы он присылал расписание в чат, \
укажите вместе с /start@vmk_schedule_bot номер своей группы.\
\nПример:\n/start@vmk_schedule_bot 102",
)
else:
temp = message.text.split(" ")[1]
if temp.isdigit():
infos = [
message.chat.id,
message.chat.title,
"",
message.from_user.username,
temp,
]
if temp in for_json.groups_in_json():
send_message(
message.chat.id,
"Отлично! Теперь я буду присылать вам расписание {} группы".format(
temp
),
)
else:
send_message(
message.chat.id,
"Похоже, что расписания для этой группы ещё не существует. \
\nРазработчик уже пинается, но можете дополнительно написать ему: @Kr0sH_512",
)
send_admin_message(
"В группе {} был запрос на {} группу.\
\nid: {}".format(
message.chat.title, temp, message.chat.id
)
)
temp = "other"
for_json.save_user(infos)
else:
send_message(
message.chat.id,
"Введено не число. Попробуйте снова.\
\nПример:\n/start@vmk_schedule_bot 102",
)
return
send_message(message.chat.id, start_txt)
markup = types.InlineKeyboardMarkup()
# for i in for_json.groups_in_json():
# markup.add(types.InlineKeyboardButton(text=i, callback_data=i))
markup.add(types.InlineKeyboardButton(text="1 курс", callback_data="1course"))
markup.add(types.InlineKeyboardButton(text="2 курс", callback_data="2course"))
bot.send_message(
message.chat.id,
"Пожалуйста, выбери свой курс",
parse_mode=ParseMode.HTML,
reply_markup=markup,
)
return
@bot.callback_query_handler(
func=lambda call: call.data in for_json.groups_in_json()
or call.data in ["other", "1course", "2course"]
)
def callback_inline(call):
if "course" in call.data:
course = call.data[0]
markup = types.InlineKeyboardMarkup()
for i in [i for i in for_json.groups_in_json() if i[0] == course]:
markup.add(types.InlineKeyboardButton(text=i, callback_data=i))
markup.add(
types.InlineKeyboardButton(
text="Моей группы нет в этом списке", callback_data="other"
)
)
bot.edit_message_text(
text="Пожалуйста, выбери свою группу",
chat_id=call.message.chat.id,
message_id=call.message.message_id,
reply_markup=markup,
)
# bot.edit_message_reply_markup(
# call.message.chat.id, call.message.message_id, reply_markup=markup
# )
return
infos = [
call.from_user.id,
call.from_user.first_name,
call.from_user.last_name,
call.from_user.username,
call.data,
]
for_json.save_user(infos)
if call.data == "other":
bot.edit_message_text(
chat_id=call.message.chat.id,
message_id=call.message.message_id,
text="Используй команду /request и напиши номер группы, которую хочешь добавить\
\nПосле создания расписания для твоей группы, я пришлю тебе сообщение",
parse_mode=ParseMode.HTML,
)
else:
bot.edit_message_text(
chat_id=call.message.chat.id,
message_id=call.message.message_id,
text="Отлично, теперь тебе будут приходить сообщения!",
parse_mode=ParseMode.HTML,
)
return
@bot.message_handler(commands=["schedule"])
def send_schedule(message):
text = (
"К сожалению, бот не может отправить тебе расписание твоей группы на сегодня :("
)
try:
day = datetime.today().strftime("%A").lower()[:3] # mon tue ...
text = for_json.get_schedule(message.chat.id, day)
except Exception as e:
# send_admin_message(
# "Schedule error from: {} \n\n {}".format(message.chat.id, str(e))
# )
bot.send_message(
message.chat.id,
"Пожалуйста, проверь, что ты зарегестрировался с помощью команды /start",
ParseMode.HTML,
message_thread_id=message.message_thread_id,
)
markup = types.InlineKeyboardMarkup()
markup.add(
types.InlineKeyboardButton(text="◀️", callback_data="left"),
types.InlineKeyboardButton(text="▶️", callback_data="right"),
)
bot.send_message(
message.chat.id,
text,
ParseMode.HTML,
reply_markup=markup,
message_thread_id=message.message_thread_id,
)
return
@bot.callback_query_handler(func=lambda call: call.data in ["left", "right"])
def change_schedule(call):
days = ["mon", "tue", "wed", "thu", "fri", "sat"]
russian_days = ["понедельник", "вторник", "среду", "четверг", "пятницу", "субботу"]
i = 6 # выходной
for day in russian_days:
if day in call.message.text:
i = russian_days.index(day)
break
delta = (1) if (call.data == "right") else (-1)
ind = ((i + delta) % 6) if (i + delta > 0) else (i + delta)
text = for_json.get_schedule(call.message.chat.id, days[ind])
markup = types.InlineKeyboardMarkup()
markup.add(
types.InlineKeyboardButton(text="◀️", callback_data="left"),
types.InlineKeyboardButton(text="▶️", callback_data="right"),
)
bot.edit_message_text(
text,
call.message.chat.id,
call.message.message_id,
parse_mode=ParseMode.HTML,
reply_markup=markup,
)
return
@bot.message_handler(commands=["random"])
def random_people(message):
list_of_num = ["1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6", "7", "8", "9"]
list_of_var = for_json.students_in_json(message.chat.id)
text = "Для вашей группы эта команда не настроена.\n\nНапиши запрос через /request"
if list_of_var == None:
send_message(message.chat.id, text)
send_admin_message(
"Запрос на random команду от {}".format(
for_json.return_infos(message.chat.id)["group"]
)
)
return
text = "• Выбери как именно мне перемешать вас:\n"
l = []
for i, val in enumerate(list_of_var):
text += "\n{} {}".format(list_of_num[i], val)
l.append(
types.InlineKeyboardButton(text=list_of_num[i], callback_data="rand#" + val)
)
l = [l]
markup = types.InlineKeyboardMarkup(l)
bot.send_message(message.chat.id, text, ParseMode.HTML, reply_markup=markup)
return
@bot.callback_query_handler(func=lambda call: "rand#" in call.data)
def make_random(call):
list_of_stud = for_json.students_in_json(
call.message.chat.id, call.data.split("#")[1]
)
random.shuffle(list_of_stud)
text = "<b><u>{}</u></b>:\n\n".format(call.data.split("#")[1])
for i, name in enumerate(list_of_stud):
text += "<code>{}{}</code>. {}\n".format(" " if i + 1 < 10 else "", i + 1, name)
bot.edit_message_text(
"Хорошо, вот ваше распределение!",
call.message.chat.id,
call.message.message_id,
parse_mode=ParseMode.HTML,
)
send_message(call.message.chat.id, text)
return
@bot.message_handler(commands=["info"])
def send_info(message):
if len(message.text.split(" ")) == 2 and is_admin(message):
send_admin_message(
"Окей, держи настройки <code>{}</code>".format(message.text.split(" ")[1])
)
infos = for_json.return_infos(message.text.split(" ")[1])
text = "<i>Никнейм</i>: <b>@{}</b>\
\n<i>Имя</i>: <b>{} {}</b>\
\n<i>Выбранная группа</i>: <b>{}</b>\
\n<i>Время напоминания до урока</i>: <b>{}</b>\
\n<i>Разрешены ли напоминания</i>: <b>{}</b>".format(
infos["username"],
infos["first_name"],
infos["last_name"],
(infos["group"] if infos["group"] != "other" else "не выбрана"),
infos["timeout"],
("да" if infos["allow_message"] == "yes" else "нет"),
)
send_admin_message(text)
return
send_message(message.chat.id, "Хорошо, вот твои настройки:")
infos = for_json.return_infos(message.chat.id)
text = "<i>Выбранная группа</i>: <b>{}</b>\
\n<i>Время напоминания до урока</i>: <b>{}</b>\
\n<i>Разрешены ли напоминания</i>: <b>{}</b>".format(
(infos["group"] if infos["group"] != "other" else "не выбрана"),
infos["timeout"],
("да" if infos["allow_message"] == "yes" else "нет"),
)
send_message(message.chat.id, text)
return
@bot.message_handler(commands=["pause"])
def pause_schedule(message):
infos = for_json.return_infos(message.chat.id)["allow_message"]
if infos == "yes":
for_json.change_user_param(str(message.chat.id), "allow_message", "no")
send_message(
message.chat.id,
"Рассылка сообщений прекращена. \
\nДля возобновления воспользуйтесь командой\n/pause",
)
else:
for_json.change_user_param(str(message.chat.id), "allow_message", "yes")
send_message(message.chat.id, "Рассылка сообщений возоблена!")
return
@bot.message_handler(commands=["thread"])
def change_thread(message):
thread_id = ""
try:
thread_id = message.reply_to_message.message_thread_id
except AttributeError:
thread_id = "General"
for_json.change_user_param(str(message.chat.id), "thread", thread_id)
send_message(
message.chat.id,
"Хорошо, теперь я буду отправлять сообщения в этот чат",
thread_id,
)
return
@bot.message_handler(commands=["timeout"])
def set_timeout(message):
send_message(
message.chat.id,
"Пришли мне число от 1 до 59. \
\nЭто будет количество минут, за которое я буду присылать тебе напоминание о паре (По умолчанию: 10)",
)
bot.register_next_step_handler(message, save_timeout)
return
def save_timeout(message):
if (
str(message.text).isdigit()
and int(message.text) <= 59
and int(message.text) >= 1
):
for_json.change_user_param(message.chat.id, "timeout", str(message.text))
send_message(
message.chat.id,
"Хорошо, теперь ты будешь получать напоминания за {} минут до урока".format(
str(message.text)
),
)
else:
send_message(
message.chat.id,
"Пожалуйста, пришли <u>число</u> от 1 до 59. \
\nПо умолчанию: 10",
)
send_message(message.chat.id, "Изменения не были внесены")
return
@bot.message_handler(commands=["request"])
def request(message):
send_message(
message.from_user.id,
"Отправь мне сообщение и я перешлю его разработчику\
\n(или напиши stop, чтобы отменить отправку)",
)
bot.register_next_step_handler(message, send_request)
return
def send_request(message):
if message.text == "stop" or message.text == "стоп" or message.text[0] == "/":
send_message(message.from_user.id, "Отмена отправки")
return
infos = [
message.from_user.id,
message.from_user.first_name,
message.from_user.last_name,
message.from_user.username,
message.text,
]
for i in range(len(infos)):
if type(infos[i]) == type(None):
infos[i] = ""
admin_text = "Сообщение от {} {} (@{}):\n\n{}\nid: <code>{}</code>".format(
infos[1], infos[2], infos[3], infos[4], infos[0]
)
send_admin_message(admin_text)
send_message(message.from_user.id, "Ваше сообщение успешно доставлено!")
return
@bot.message_handler(commands=["source"])
def send_source(message):
bot.send_message(
message.chat.id,
"https://github.com/kr0sh512/Telegram-bot-schedule",
parse_mode=ParseMode.HTML,
)
return
@bot.message_handler(content_types=["text"])
def text_message(message):
if message.chat.type == "supergroup" or is_admin(message):
return
send_message(
message.from_user.id,
"К сожалению, я ещё не умею обрабатывать сообщения.\
\nИспользуй команды из меню или напиши /help.",
)
return
def send_message(id, text, thread_id="General", parity=None):
if thread_id == "General":
thread_id = None
if parity != None:
parity = int(parity)
count_of_weeks = (datetime.now() - datetime(2024, 2, 5)).days // 7
is_odd = (count_of_weeks + 1) % 2 == PARITY_FIRST
if is_odd != bool(parity):
return
try:
bot.send_message(
chat_id=id,
text=text,
parse_mode=ParseMode.HTML,
message_thread_id=thread_id,
disable_web_page_preview=True,
)
except Exception as e:
time.sleep(5)
try:
bot.send_message(
chat_id=id,
text=text,
parse_mode=ParseMode.HTML,
message_thread_id=thread_id,
disable_web_page_preview=True,
)
except Exception as e:
text_error = "Error from user: @{} <code>{}</code>\n{}".format(
for_json.return_infos(id)["username"], id, str(e)
)
# send_admin_message(text_error)
print(f"--- {text_error} ---")
return
def send_document(id, file, text=""):
try:
bot.send_document(id, file, caption=text)
except Exception as e:
time.sleep(5)
try:
bot.send_document(id, file, caption=text)
except Exception as e:
text_error = "Error from user: <code>{}</code>\n{}".format(id, str(e))
send_admin_message(text_error)
print(id, e)
return
if __name__ == "__main__":
for_json.create_schedule_tasks(True)
send_admin_message("Я перезапустился!")
print("-------------------------")
threading.Thread(
target=bot.infinity_polling, name="bot_infinity_polling", daemon=True
).start()
while True:
try:
schedule.run_pending()
time.sleep(5)
except Exception as e:
print(e)
time.sleep(10)
tg.py
+73 -77
View File
@@ -1,14 +1,14 @@
#!/usr/bin/python3.3
import threading, telebot, schedule, time, random, yaml
import threading, telebot, schedule, time, yaml
from datetime import datetime
import os, sys
from telebot import types
import os, sys, inspect
import re, locale
from telegram.constants import ParseMode
from telebot import types
import for_db as db
import re
import locale
config = yaml.safe_load(open("config.yaml"))
lang = yaml.safe_load(open("lang.yaml"))["ru"]
bot = telebot.TeleBot(
config["api_token"],
colorful_logs=True,
@@ -19,12 +19,13 @@ bot = telebot.TeleBot(
admin_id = config["admin_id"]
bot_username = config["bot_username"]
locale.setlocale(locale.LC_ALL, "ru_RU.UTF-8")
def is_admin(message):
def is_admin(message: types.Message):
if (
str(message.from_user.id) == admin_id
str(message.from_user.id) == str(admin_id)
and message.from_user.id == message.chat.id
):
return True
@@ -33,37 +34,33 @@ def is_admin(message):
@bot.message_handler(commands=["restart", "r"])
def restart_bot(message):
def restart_bot(message: types.Message):
if is_admin(message):
send_admin_message("bye")
os.execv(sys.executable, ["python3", "-u"] + sys.argv)
@bot.message_handler(commands=["start"])
def start(message):
def start(message: types.Message):
print(db.new_user(message))
send_message(message, lang["start_message"])
return
@bot.message_handler(commands=["help", "faq"])
def help(message):
help_msg = "Мои команды:\
\n/start - используй, чтобы сменить номер группы.\
\n/list - список напоминаний.\
\n/source - Страница бота на Github\
\n\nИли же можно всегда написать напрямую: @Kr0sH_512"
def help(message: types.Message):
help_msg = lang["help_message"].format(config["bot_username"])
send_message(message.chat.id, help_msg)
send_message(message, help_msg)
return
@bot.callback_query_handler(func=lambda call: "#return" in call.data)
def send_notes_callback(call):
list_notes(call.message, call.data.split("#")[0])
bot.delete_message(call.message.chat.id, call.message.message_id)
list_notes(call.message, call.data.split("#")[0], edit=True)
return
@@ -77,13 +74,9 @@ def delete_notes_callback(call):
db.delete_note(call.message.chat.id, note_ind, prefix)
bot.edit_message_text(
f"❌ Удалено:\n\n{note['text']}",
call.message.chat.id,
call.message.message_id,
)
list_notes(call.message, prefix, edit=True)
list_notes(call.message, prefix)
bot.send_message(call.message.chat.id, f"❌ Удалено:\n\n{note['text']}")
return
@@ -132,18 +125,15 @@ def choose_note_callback(call):
page = int(call.data.split("#")[2])
prefix = call.data.split("#")[0]
markup = types.InlineKeyboardMarkup()
num_on_page = 8 # Лимит телеграмма
num_on_page = 6 # Лимит телеграмма - 8
notes = db.get_notes(call.message.chat.id, prefix)["notes"]
max_page = (len(notes) - 1) // num_on_page
lst_button = []
for i in range(num_on_page):
# Calculate the index of the current program
ind = i + page * num_on_page
# Create an inline keyboard button for the program
lst_button.append(
types.InlineKeyboardButton(
ind + 1,
@@ -154,11 +144,9 @@ def choose_note_callback(call):
if ind == len(notes) - 1:
break
# Create the inline keyboard markup from the list of buttons
markup = types.InlineKeyboardMarkup([lst_button])
if len(notes) > num_on_page:
# Create the back button (only visible if we're not on the first page)
back_btn = (
types.InlineKeyboardButton(
"Назад", callback_data=f"{prefix}#select#{page - 1}"
@@ -167,13 +155,11 @@ def choose_note_callback(call):
else None
)
# Create the current page button
now_btn = types.InlineKeyboardButton(
f"{page + 1}/{max_page + 1}",
callback_data=f"{prefix}#select#{0 if page > 0 else max_page}",
)
# Create the next button (only visible if we're not on the last page)
next_btn = (
types.InlineKeyboardButton(
"Вперед", callback_data=f"{prefix}#select#{page + 1}"
@@ -182,7 +168,6 @@ def choose_note_callback(call):
else None
)
# Add the buttons to the inline keyboard markup
markup.row(*[i for i in [back_btn, now_btn, next_btn] if i])
markup.add(
@@ -202,34 +187,18 @@ def choose_note_callback(call):
return
@bot.message_handler(commands=["list"])
def list_notes(message, list="Default"):
notes = db.get_notes(message.chat.id, list)
if not notes or not notes["notes"]:
send_message(message.chat.id, "У вас нет напоминаний.")
@bot.message_handler(commands=["settings", "setting", "edit"])
def display_settings(message: types.Message):
return
list_notes = notes["notes"]
for i in range(len(list_notes)):
if list_notes[i]["time_notif"]:
if (
datetime.strptime(list_notes[i]["time_notif"], "%Y-%m-%d %H:%M:%S")
- datetime.now()
).days < 0:
send_message(
message.chat.id,
f"❌ {list_notes[i]['text']} (просрочено: {list_notes[i]['time_notif']})",
)
db.delete_note(message.chat.id, i)
@bot.message_handler(commands=["list"])
def list_notes(message: types.Message, list: str = "Default", edit: bool = False):
notes = db.get_notes(message.chat.id, list)
if not notes:
send_message(message.chat.id, "У вас нет напоминаний.")
send_message(message, lang["no_reminders"])
return
@@ -250,15 +219,27 @@ def list_notes(message, list="Default"):
)
)
send_message(message.chat.id, notes_msg, markup)
if edit:
bot.edit_message_text(
notes_msg, message.chat.id, message.message_id, reply_markup=markup
)
return
msg = send_message(message, notes_msg, markup)
old_msg_id = db.new_message(msg)
if old_msg_id:
bot.delete_message(message.chat.id, old_msg_id)
return
@bot.message_handler(content_types=["text"])
def text_message(message):
def text_message(message: types.Message):
if message.chat.type == "supergroup" and bot_username not in message.text:
# print(message.text)
return # ignore messages from supergroups
if db.check_user(message.chat.id) is None:
@@ -293,10 +274,10 @@ def text_message(message):
}
def translate_date_to_datetime(date_str):
for ru_month, en_month in month_mapping.items():
if ru_month in date_str:
date_str = date_str.replace(ru_month, en_month)
# break
for month_from, month_to in month_mapping.items():
if month_from in date_str:
date_str = date_str.replace(month_from, month_to)
break
try:
return datetime.strptime(date_str, "%d %B").replace(
year=datetime.now().year
@@ -319,7 +300,6 @@ def text_message(message):
for pattern in date_patterns:
match = re.search(pattern, message.text, re.IGNORECASE)
if match:
# print(match.group())
date_found_dt = translate_date_to_datetime(match.group())
break
@@ -334,18 +314,22 @@ def text_message(message):
date_found_dt,
)
send_message(message.chat.id, "📝 Напоминание добавлено:")
send_message(message, "📝 Напоминание добавлено:")
list_notes(message)
return
def send_message(id, text, markup=None, thread_id="General"):
if thread_id == "General":
thread_id = None
def send_message(
message: types.Message, text, markup=None, thread_id=None
) -> types.Message:
id = message.chat.id
if not thread_id:
thread_id = message.message_thread_id
try:
bot.send_message(
msg = bot.send_message(
chat_id=id,
text=text,
message_thread_id=thread_id,
@@ -354,22 +338,22 @@ def send_message(id, text, markup=None, thread_id="General"):
except Exception as e:
time.sleep(5)
try:
bot.send_message(
msg = bot.send_message(
chat_id=id,
text=text,
message_thread_id=thread_id,
reply_markup=markup,
)
except Exception as e:
text_error = "Error from user: <code>{}</code>\n{}".format(id, str(e))
# send_admin_message(text_error)
text_error = "\tError from user {}: \n{}\n\t---\n".format(id, str(e))
print(f"--- {text_error} ---")
return
return msg
def send_admin_message(text):
send_message(admin_id, "🛑 " + text)
text += f"\n\ncalled by:{inspect.stack()[1][3]}"
bot.send_message(admin_id, f"🛑 {text}")
return
@@ -381,10 +365,22 @@ if __name__ == "__main__":
target=bot.infinity_polling,
name="bot_infinity_polling",
daemon=True,
# kwargs={"restart_on_change": True},
).start()
while True: # keep alive
# добавить проверку живых напоминаний, удалять, если они не нужны
print("I'm alive!")
time.sleep(36000) # 10 hours
chat_id, list, ind_note = db.check_old_notes()
if chat_id:
note = db.get_notes(chat_id, list)["notes"][ind_note]
db.delete_note(chat_id, ind_note, list)
msg = bot.send_message(
chat_id,
f"❌ {note['text']} \n(Уведомление: {datetime.strptime(note['time_notif'], '%Y-%m-%d %H:%M:%S').strftime('%d %B %Y')})",
timeout=1,
)
list_notes(msg, list)
time.sleep(15)