first version
This commit is contained in:
kr0sh512
@@ -0,0 +1,2 @@
|
||||
config.yaml
|
||||
database/*
|
||||
@@ -0,0 +1,171 @@
|
||||
import yaml
|
||||
from datetime import datetime
|
||||
from telebot import types
|
||||
import errno, os
|
||||
|
||||
users_path = "database/users.yaml"
|
||||
user_path = "database/users/{}.yaml"
|
||||
|
||||
|
||||
def check_user(user_id: int) -> dict:
|
||||
users = yaml.safe_load(open(users_path, "r", encoding="utf-8"))
|
||||
|
||||
if not users:
|
||||
users = {}
|
||||
|
||||
if user_id in users:
|
||||
return users[user_id]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def new_user(message: types.Message) -> bool:
|
||||
if check_user(message.chat.id):
|
||||
return False
|
||||
|
||||
user = {
|
||||
"id": message.chat.id,
|
||||
"type": message.chat.type,
|
||||
"username": message.chat.username,
|
||||
"first_name": message.chat.first_name,
|
||||
"last_name": message.chat.last_name,
|
||||
"time_created": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
|
||||
users = yaml.safe_load(open(users_path, "r"))
|
||||
|
||||
if not users:
|
||||
users = {}
|
||||
|
||||
users[user["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 True
|
||||
|
||||
|
||||
def add_note(id: int, text: str, time_notif: datetime = None, list: str = "Default"):
|
||||
if not check_user(id):
|
||||
return False
|
||||
|
||||
notes = (
|
||||
yaml.safe_load(open(user_path.format(id), "r", encoding="utf-8"))
|
||||
if os.path.exists(user_path.format(id))
|
||||
else {}
|
||||
)
|
||||
|
||||
if list not in notes:
|
||||
notes[list] = {"name": list, "description": "Стандартный список", "notes": []}
|
||||
|
||||
notes[list]["notes"].append(
|
||||
{
|
||||
"text": text,
|
||||
"time_notif": str(time_notif) if time_notif else None,
|
||||
"time_created": str(datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
|
||||
}
|
||||
)
|
||||
|
||||
if True: # TODO: edit sorting
|
||||
import functools
|
||||
|
||||
def comp(x, y):
|
||||
if not x["time_notif"] and not y["time_notif"]:
|
||||
return 0
|
||||
elif not x["time_notif"]:
|
||||
return -1
|
||||
elif not y["time_notif"]:
|
||||
return 1
|
||||
else:
|
||||
x_time = datetime.strptime(
|
||||
x["time_notif"],
|
||||
"%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
y_time = datetime.strptime(
|
||||
y["time_notif"],
|
||||
"%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
if x_time > y_time:
|
||||
return 1
|
||||
elif x_time < y_time:
|
||||
return -1
|
||||
return 0
|
||||
|
||||
notes[list]["notes"].sort(
|
||||
key=lambda x: datetime.strptime(
|
||||
x["time_created"],
|
||||
"%Y-%m-%d %H:%M:%S",
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
notes[list]["notes"].sort(
|
||||
key=functools.cmp_to_key(comp),
|
||||
)
|
||||
|
||||
yaml_data = yaml.dump(
|
||||
notes,
|
||||
default_flow_style=False,
|
||||
encoding="utf-8",
|
||||
allow_unicode=True,
|
||||
width=float("inf"),
|
||||
sort_keys=False,
|
||||
)
|
||||
|
||||
with open(user_path.format(id), "wb") as file:
|
||||
file.write(yaml_data)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def get_notes(id, list="Default"):
|
||||
if not check_user(id):
|
||||
return None
|
||||
|
||||
if not os.path.exists(user_path.format(id)):
|
||||
return None
|
||||
|
||||
notes = yaml.safe_load(open(user_path.format(id), "r", encoding="utf-8"))
|
||||
|
||||
return notes[list] if list in notes else None
|
||||
|
||||
|
||||
def delete_note(id, note_ind, list="Default"):
|
||||
if not check_user(id):
|
||||
return False
|
||||
|
||||
if not os.path.exists(user_path.format(id)):
|
||||
return False
|
||||
|
||||
notes = yaml.safe_load(open(user_path.format(id), "r", encoding="utf-8"))
|
||||
|
||||
if list not in notes:
|
||||
return False
|
||||
|
||||
if note_ind >= len(notes[list]["notes"]):
|
||||
return False
|
||||
|
||||
notes[list]["notes"].pop(note_ind)
|
||||
|
||||
yaml_data = yaml.dump(
|
||||
notes,
|
||||
default_flow_style=False,
|
||||
encoding="utf-8",
|
||||
allow_unicode=True,
|
||||
width=float("inf"),
|
||||
sort_keys=False,
|
||||
)
|
||||
|
||||
with open(user_path.format(id), "wb") as file:
|
||||
file.write(yaml_data)
|
||||
|
||||
return True
|
||||
+763
@@ -0,0 +1,763 @@
|
||||
#!/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)
|
||||
@@ -0,0 +1,390 @@
|
||||
#!/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_db as db
|
||||
import re
|
||||
import locale
|
||||
|
||||
config = yaml.safe_load(open("config.yaml"))
|
||||
bot = telebot.TeleBot(
|
||||
config["api_token"],
|
||||
colorful_logs=True,
|
||||
disable_web_page_preview=True,
|
||||
parse_mode=ParseMode.HTML,
|
||||
)
|
||||
|
||||
admin_id = config["admin_id"]
|
||||
bot_username = config["bot_username"]
|
||||
|
||||
locale.setlocale(locale.LC_ALL, "ru_RU.UTF-8")
|
||||
|
||||
|
||||
def is_admin(message):
|
||||
if (
|
||||
str(message.from_user.id) == admin_id
|
||||
and message.from_user.id == message.chat.id
|
||||
):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@bot.message_handler(commands=["restart", "r"])
|
||||
def restart_bot(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):
|
||||
|
||||
print(db.new_user(message))
|
||||
|
||||
return
|
||||
|
||||
|
||||
@bot.message_handler(commands=["help", "faq"])
|
||||
def help(message):
|
||||
help_msg = "Мои команды:\
|
||||
\n/start - используй, чтобы сменить номер группы.\
|
||||
\n/list - список напоминаний.\
|
||||
\n/source - Страница бота на Github\
|
||||
\n\nИли же можно всегда написать напрямую: @Kr0sH_512"
|
||||
|
||||
send_message(message.chat.id, 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)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@bot.callback_query_handler(func=lambda call: "#del#" in call.data)
|
||||
def delete_notes_callback(call):
|
||||
note_ind = int(call.data.split("#")[2])
|
||||
prefix = call.data.split("#")[0]
|
||||
|
||||
note = db.get_notes(call.message.chat.id, prefix)["notes"][note_ind]
|
||||
|
||||
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)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@bot.callback_query_handler(func=lambda call: "#conf#" in call.data)
|
||||
def edit_notes_callback(call):
|
||||
note_ind = int(call.data.split("#")[2])
|
||||
prefix = call.data.split("#")[0]
|
||||
notes = db.get_notes(call.message.chat.id, prefix)["notes"]
|
||||
|
||||
note = notes[note_ind]
|
||||
|
||||
markup = types.InlineKeyboardMarkup()
|
||||
markup.add(
|
||||
types.InlineKeyboardButton(
|
||||
"Удалить",
|
||||
callback_data=f"{prefix}#del#{note_ind}",
|
||||
)
|
||||
)
|
||||
markup.add(
|
||||
types.InlineKeyboardButton(
|
||||
"Назад",
|
||||
callback_data=f"{prefix}#return",
|
||||
)
|
||||
)
|
||||
|
||||
text = f"📝 <b>Заметка {note_ind + 1}</b>"
|
||||
|
||||
if note["time_notif"]:
|
||||
text += f" ⌚️ <b>{datetime.strptime(note['time_notif'], '%Y-%m-%d %H:%M:%S').strftime('%d %B %Y')}</b>"
|
||||
|
||||
text += f"\n\n{note['text']}"
|
||||
|
||||
bot.edit_message_text(
|
||||
text,
|
||||
call.message.chat.id,
|
||||
call.message.message_id,
|
||||
reply_markup=markup,
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@bot.callback_query_handler(func=lambda call: "#select#" in call.data)
|
||||
def choose_note_callback(call):
|
||||
page = int(call.data.split("#")[2])
|
||||
prefix = call.data.split("#")[0]
|
||||
markup = types.InlineKeyboardMarkup()
|
||||
num_on_page = 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,
|
||||
callback_data=f"{prefix}#conf#{ind}",
|
||||
)
|
||||
)
|
||||
|
||||
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}"
|
||||
)
|
||||
if page > 0
|
||||
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}"
|
||||
)
|
||||
if page < max_page
|
||||
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(
|
||||
types.InlineKeyboardButton(
|
||||
"Назад",
|
||||
callback_data=f"{prefix}#return",
|
||||
)
|
||||
)
|
||||
|
||||
bot.edit_message_text(
|
||||
call.message.text,
|
||||
call.message.chat.id,
|
||||
call.message.message_id,
|
||||
reply_markup=markup,
|
||||
)
|
||||
|
||||
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, "У вас нет напоминаний.")
|
||||
|
||||
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)
|
||||
|
||||
notes = db.get_notes(message.chat.id, list)
|
||||
|
||||
if not notes:
|
||||
send_message(message.chat.id, "У вас нет напоминаний.")
|
||||
|
||||
return
|
||||
|
||||
list_notes = notes["notes"]
|
||||
|
||||
notes_msg = "⚡️ <u>Ваши напоминания</u>:\n\n"
|
||||
for i in range(len(list_notes)):
|
||||
notes_msg += f"<b>{i + 1})</b> "
|
||||
if list_notes[i]["time_notif"]:
|
||||
notes_msg += f" ⌚️ <b>{datetime.strptime(list_notes[i]['time_notif'], '%Y-%m-%d %H:%M:%S').strftime('%d %B %Y')}:</b>\n"
|
||||
|
||||
notes_msg += f"{list_notes[i]['text']}\n\n"
|
||||
|
||||
markup = types.InlineKeyboardMarkup()
|
||||
markup.add(
|
||||
types.InlineKeyboardButton(
|
||||
"Выбрать номер", callback_data=f"{'Default'}#select#0"
|
||||
)
|
||||
)
|
||||
|
||||
send_message(message.chat.id, notes_msg, markup)
|
||||
|
||||
return
|
||||
|
||||
|
||||
@bot.message_handler(content_types=["text"])
|
||||
def text_message(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:
|
||||
db.new_user(message)
|
||||
|
||||
message.text = message.text.replace(f"{bot_username}", "").strip()
|
||||
if not message.text:
|
||||
if not message.reply_to_message:
|
||||
return
|
||||
message.text = message.reply_to_message.text
|
||||
|
||||
date_patterns = [
|
||||
r"\b\d{1,2}\s(?:января|февраля|марта|апреля|мая|июня|июля|августа|сентября|октября|ноября|декабря)\b",
|
||||
r"\b\d{1,2}\s(?:январь|февраль|март|апрель|май|июнь|июль|август|сентябрь|октябрь|ноябрь|декабрь)\b",
|
||||
r"\b\d{1,2}\.\d{1,2}\b",
|
||||
r"\b\d{1,2}ое\s(?:января|февраля|марта|апреля|мая|июня|июля|августа|сентября|октября|ноября|декабря)\b",
|
||||
]
|
||||
|
||||
month_mapping = {
|
||||
"январь": "января",
|
||||
"февраль": "февраля",
|
||||
"март": "марта",
|
||||
"апрель": "апреля",
|
||||
"май": "мая",
|
||||
"июнь": "июня",
|
||||
"июль": "июля",
|
||||
"август": "августа",
|
||||
"сентябрь": "сентября",
|
||||
"октябрь": "октября",
|
||||
"ноябрь": "ноября",
|
||||
"декабрь": "декабря",
|
||||
}
|
||||
|
||||
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
|
||||
try:
|
||||
return datetime.strptime(date_str, "%d %B").replace(
|
||||
year=datetime.now().year
|
||||
)
|
||||
except ValueError:
|
||||
try:
|
||||
return datetime.strptime(date_str, "%dое %B").replace(
|
||||
year=datetime.now().year
|
||||
)
|
||||
except ValueError:
|
||||
try:
|
||||
return datetime.strptime(date_str, "%d.%m").replace(
|
||||
year=datetime.now().year
|
||||
)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
date_found_dt = None
|
||||
|
||||
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
|
||||
|
||||
if date_found_dt:
|
||||
if date_found_dt < datetime.now():
|
||||
date_found_dt = date_found_dt.replace(year=datetime.now().year + 1)
|
||||
|
||||
db.add_note(
|
||||
message.chat.id,
|
||||
message.text,
|
||||
date_found_dt,
|
||||
)
|
||||
|
||||
send_message(message.chat.id, "📝 Напоминание добавлено:")
|
||||
list_notes(message)
|
||||
|
||||
return
|
||||
|
||||
|
||||
def send_message(id, text, markup=None, thread_id="General"):
|
||||
if thread_id == "General":
|
||||
thread_id = None
|
||||
|
||||
try:
|
||||
bot.send_message(
|
||||
chat_id=id,
|
||||
text=text,
|
||||
message_thread_id=thread_id,
|
||||
reply_markup=markup,
|
||||
)
|
||||
except Exception as e:
|
||||
time.sleep(5)
|
||||
try:
|
||||
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)
|
||||
print(f"--- {text_error} ---")
|
||||
|
||||
return
|
||||
|
||||
|
||||
def send_admin_message(text):
|
||||
send_message(admin_id, "🛑 " + text)
|
||||
|
||||
return
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("-------------------------")
|
||||
|
||||
threading.Thread(
|
||||
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
|
||||
Reference in New Issue
Block a user