EDeev/y.calendarkin
1
0
Fork 0
mirror of https://github.com/EDeev/y.calendarkin.git synced 2026-10-07 16:37:50 +03:00
Code Releases 2 Activity

Версия на aiogram 3 — та, что работает на сервере

Browse source 
В репозитории оставался код на aiogram 2 (executor, message_handler), а requirements.txt
уже требовал aiogram 3 — по README бот не запускался (issue #1). Переписанная версия с
разделением на bot.py / handlers.py / init.py давно работала на сервере, теперь она и здесь.
Переводы строк приведены к LF.
This commit is contained in:
Деев Егор Викторович 2026-10-05 19:22:10 +00:00
parent 31bf093d94
commit 18e79e24a5
6 changed files with 354 additions and 356 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitattributes vendored Normal file
View file

@ -0,0 +1,3 @@
* text=auto eol=lf
*.jpg binary
*.db binary

269
code/bot.py
View file

@ -1,235 +1,8 @@
import logging, config, sql, asyncio, wget, os
import asyncio, logging, time, wget, os
from aiogram import Bot, Dispatcher, executor, types
from aiogram.types import ParseMode
from datetime import datetime, time
from script import text_ical, message_form, dt_now, delta_time
# инициализируем токен
logging.basicConfig(level=logging.INFO)
bot = Bot(token=config.TOKEN)
dp = Dispatcher(bot)
# инициализируем соединение с БД
du = sql.Users('../db/users.db')
dc = sql.Clock('../db/clock.db')
# ПРИВЕТСТВЕННОЕ СООБЩЕНИЕ
@dp.message_handler(commands=['start', 'help'])
async def helps(message: types.Message):
tg_id = int(message.chat.id)
if not du.user_exists(tg_id):
du.add_user(tg_id)
buttons = [types.InlineKeyboardButton(text="КОМАНДЫ", callback_data="com"),
types.InlineKeyboardButton(text="АВТОР", callback_data="auth")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await message.answer(text="<b>Я.Календаркин</b> - бот для оповещения о событиях из <b>Яндекс.Календаря</b>. "
"Для начала работы вам нужно всего лишь прислать в чат ссылку экспорта календаря в "
"<b>формате ICal</b>. После получения ссылки, бот начнёт оповещать о всех новых событиях "
"и появится возможность настройки оповещений. О том, какие команды есть для настройки, "
"вы можете ознакомиться по кнопке <b>КОМАНДЫ</b>",
parse_mode=ParseMode.HTML, reply_markup=keyboard)
@dp.callback_query_handler(text="auth")
async def author(call: types.CallbackQuery):
await call.message.answer(text='*| АВТОР |*\n\n*>>* Этот бот не коммерческий проект, для упрощенного получения '
'уведомлений о событиях в Яндекс.Календаре. Не многим этот бот будет полезен, но '
'людям, чья работа подразумевает его использование, он станет лишь удобным '
'инструментом. Я же пишу подобные небольшие проекты, о которых вы можете узнать '
'больше на моём [GitHub](https://github.com/IGlek).',
parse_mode=ParseMode.MARKDOWN)
@dp.callback_query_handler(text="com")
async def commands(call: types.CallbackQuery):
await call.message.answer(text='<b>| КОМАНДЫ |</b>\n\n'
'<b>/help</b> - вспомогательная функция для уточнения работы команд\n'
'<b>/list</b> - список событий календаря, запланированных на сегодняшний день\n'
'<b>/notif</b> - команда, отключающая рассылку уведомлений, даже при наличии событий в календаре\n'
'<b>/daily</b> - оповещение в 8 утра по вашему часовому поясу со списком событий на день\n'
'<b>/moment</b> - напоминание, приходящее в момент начала события\n\n'
'<b>/get_alarm</b> - информация о времени на которое настроены оповещения\n'
'<b>/edit_alarm</b> - изменение времени оповещений\n'
'<b>/stop_alarm</b> - команда, отключающая второе оповещение о событии',
parse_mode=types.ParseMode.HTML)
# КОМАНДЫ
@dp.message_handler(commands=['list'])
async def check_list(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if du.url_exists(user_id):
txt = "<b>Имеющиеся события на сегодня</b>\n\n"
lst_events = sorted(text_ical(user_id, du.get_tz(user_id)))
today = dt_now(du.get_tz(user_id)).date()
counter = 0
for event in lst_events:
if event[0] == today:
counter += 1
txt += message_form(counter, event[3])
if counter:
await message.answer(text=txt, parse_mode=ParseMode.HTML)
else:
await message.answer(text="<b>В данный момент</b> событий на сегодня найдено не было!",
parse_mode=ParseMode.HTML)
else:
await message.answer("Для отображения событий вы должны прислать ical-ссылку на календарь!")
@dp.message_handler(commands=['notif'])
async def notif_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if du.url_exists(user_id):
if du.get_status(user_id):
await message.answer("Уведомления о событиях выключены!")
else:
await message.answer("Уведомления о событиях включены!")
du.update_status(user_id)
else:
await message.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@dp.message_handler(commands=['daily'])
async def daily_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if dc.clock_exists(user_id):
if dc.get_daily(user_id):
await message.answer("Ежедневные утренние уведомления выключены!")
else:
await message.answer("Ежедневные утренние уведомления включены!")
dc.update_daily(user_id)
else:
await message.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@dp.message_handler(commands=['moment'])
async def start_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if dc.clock_exists(user_id):
if dc.get_start(user_id):
await message.answer("Уведомления в момент события выключены!")
else:
await message.answer("Уведомления в момент события включены!")
dc.update_start(user_id)
else:
await message.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@dp.message_handler(commands=['get_alarm'])
async def start_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if dc.clock_exists(user_id):
alarms = dc.get_alarm(user_id)
start = dc.get_start(user_id)
status2 = dc.get_status2(user_id)
txt = ""
if status2:
txt += "<b>У вас работает два оповещения"
else:
txt += "<b>У вас работает лишь первое оповещение"
if start:
txt += " и сообщение в момент начала события!</b>"
else:
txt += "!</b>"
await message.answer(text=(txt + f"\n\n<b>Первое оповещение</b> приходит за {alarms[0]} минут\n"
f"<b>Второе оповещение</b> приходит за {alarms[1]} минут"),
parse_mode=ParseMode.HTML)
else:
await message.answer("Для того, чтобы получить таймеры, вы должны прислать ical-ссылку на свой календарь!")
@dp.message_handler(commands=['edit_alarm'])
async def start_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if dc.clock_exists(user_id):
await message.bot.send_photo(chat_id=message.chat.id, photo=open("../data/photo_edit_alarm.jpg", "rb"),
caption="Для изменения времени вам надо в ответ на это сообщение прислать два "
"числа через пробел: разница времени первого и второго таймера по ходу "
"времени соответственно")
else:
await message.answer("Для того, чтобы изменить таймеры, вы должны прислать ical-ссылку на свой календарь!")
@dp.message_handler(commands=['stop_alarm'])
async def daily_up(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if dc.clock_exists(user_id):
if dc.get_status2(user_id):
await message.answer("Второе уведомление выключено!")
else:
await message.answer("Второе уведомление включено!")
dc.update_status2(user_id)
else:
await message.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
# ЗАГРУЗКА ССЫЛКИ
@dp.message_handler(content_types=['text'])
async def downloading_file_ics(message: types.Message):
user_id = du.get_user_id(int(message.chat.id))
if message.text[:5] == "https":
try:
wget.download(message.text, f'../data/icals/{str(user_id)}_new.ics')
try:
os.remove(f'../data/icals/{str(user_id)}.ics')
except FileNotFoundError:
pass
os.rename(f'../data/icals/{str(user_id)}_new.ics', f'../data/icals/{str(user_id)}.ics')
time_zone = message.text.split("=")[-1]
if not du.url_exists(user_id):
du.add_url(user_id, message.text, time_zone)
dc.add_clock(user_id)
else:
du.update_url(user_id, message.text, time_zone)
await message.answer("Ссылка успешно добавлена! Уведомления уже включены!")
except Exception:
await message.answer("Ошибка скачивания! Проверьте правильность ссылки и пришлите ещё раз")
if 'reply_to_message' in message and dc.clock_exists(user_id):
text = "Для изменения времени вам надо в ответ на это сообщение прислать два числа через " \
"пробел: разница времени первого и второго таймера по ходу времени соответственно"
if message.reply_to_message.caption == text:
alarm_new = message.text.split()
if int(alarm_new[0]) < 60 and int(alarm_new[1]) < 60:
dc.update_alarm1(user_id, int(alarm_new[0]))
dc.update_alarm2(user_id, int(alarm_new[1]))
await message.answer("Время отправки уведомлений успешно обновлено!")
else:
await message.answer("Время отправки уведомлений должно быть меньше 60 минут!")
from init import *
from script import *
from handlers import router
# ПРОВЕРКА НА СОБЫТИЕ
@ -273,29 +46,27 @@ async def alarm(wait_for):
delta_start = delta_time(d_event, 1, 0)
if start and delta_start[0] < time_check <= delta_start[1]:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True,
text=f"<b>Событие начинается!</b>\n\n" + message_form(0, event[3]))
delta_alarm1 = delta_time(d_event, alarm[0], alarm[0] - 1)
if delta_alarm1[0] < time_check <= delta_alarm1[1]:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True,
text=f"<b>Напоминаю!</b>\n<i>Через {alarm[0]} минут "
f"будет событие:</i>\n\n{message_form(0, event[3])}")
if alarm2_status:
delta_alarm2 = delta_time(d_event, alarm[1], alarm[1] - 1)
if delta_alarm2[0] < time_check <= delta_alarm2[1]:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True,
text=f"<b>Напоминаю!</b>\n<i>Через {alarm[1]} минут "
f"будет событие:</i>\n\n{message_form(0, event[3])}")
if daily and delta_daily1 <= time_check < delta_daily2:
if counter:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
text=txt)
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True, text=txt)
else:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
text=f"Сегодня событий <b>нет</b>")
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True, text=f"Сегодня событий <b>нет</b>")
async def update(wait_for):
@ -314,8 +85,20 @@ async def update(wait_for):
os.rename(f'../data/icals/{str(user_id)}_new.ics', f'../data/icals/{str(user_id)}.ics')
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.create_task(alarm(60)) # ПРОВЕРКА КАЖДУЮ 1 МИНУТУ
loop.create_task(update(780)) # ПРОВЕРКА КАЖДУЮ 13 МИНУТУ
executor.start_polling(dp, skip_updates=True)
async def main() -> None:
dp.include_router(router)
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types(), skipUpdates=True)
async def process() -> None:
await asyncio.gather(main(), # ЗАПУСК ОСНОВНОЙ ПРОГРАММЫ
alarm(60), # ПРОВЕРКА КАЖДУЮ 1 МИНУТУ
update(780)) # ПРОВЕРКА КАЖДУЮ 13 МИНУТУ
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
try: asyncio.run(process())
except KeyboardInterrupt: pass

2
code/config.py
View file

@ -1 +1 @@
TOKEN = "**********:***************************"
TOKEN = "**********:***************************"

197
code/handlers.py Normal file
View file

@ -0,0 +1,197 @@
from aiogram import F, Router
from aiogram.types import Message, CallbackQuery, ContentType, InlineKeyboardMarkup, InlineKeyboardButton, FSInputFile
from aiogram.filters import Command
import wget, os
from init import *
from script import *
router = Router()
# ПРИВЕТСТВЕННОЕ СООБЩЕНИЕ
@router.message(Command('start', 'help'))
async def helps(msg: Message) -> None:
tg_id = int(msg.chat.id)
if not du.user_exists(tg_id):
du.add_user(tg_id)
buttons = [[InlineKeyboardButton(text="КОМАНДЫ", callback_data="com"),
InlineKeyboardButton(text="АВТОР", callback_data="auth")]]
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons, row_width=2)
await msg.answer(text="<b>Я.Календаркин</b> - бот для оповещения о событиях из <b>Яндекс.Календаря</b>. "
"Для начала работы вам нужно всего лишь прислать в чат ссылку экспорта календаря в "
"<b>формате ICal</b>. После получения ссылки, бот начнёт оповещать о всех новых событиях "
"и появится возможность настройки оповещений. О том, какие команды есть для настройки, "
"вы можете ознакомиться по кнопке <b>КОМАНДЫ</b>", reply_markup=keyboard)
@router.callback_query(F.data == "auth")
async def author(call: CallbackQuery) -> None:
await call.message.answer(text='<b>| АВТОР |</b>\n\n<b>>></b> Этот бот не коммерческий проект, для упрощенного '
'получения уведомлений о событиях в Яндекс.Календаре. Не многим этот бот будет '
'полезен, но людям, чья работа подразумевает его использование, он станет лишь '
'удобным инструментом. Я же пишу подобные небольшие проекты, о которых вы можете '
'узнать больше на моём <a href="https://github.com/IGlek">GitHub</a>.')
@router.callback_query(F.data == "com")
async def commands(call: CallbackQuery) -> None:
await call.message.answer(text='<b>| КОМАНДЫ |</b>\n\n'
'<b>/help</b> - вспомогательная функция для уточнения работы команд\n'
'<b>/list</b> - список событий календаря, запланированных на сегодняшний день\n'
'<b>/notif</b> - команда, отключающая рассылку уведомлений, даже при наличии событий в календаре\n'
'<b>/daily</b> - оповещение в 8 утра по вашему часовому поясу со списком событий на день\n'
'<b>/moment</b> - напоминание, приходящее в момент начала события\n\n'
'<b>/get_alarm</b> - информация о времени на которое настроены оповещения\n'
'<b>/edit_alarm</b> - изменение времени оповещений\n'
'<b>/stop_alarm</b> - команда, отключающая второе оповещение о событии')
# КОМАНДЫ
@router.message(Command('list'))
async def check_list(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if du.url_exists(user_id):
txt = "<b>Имеющиеся события на сегодня</b>\n\n"
lst_events = sorted(text_ical(user_id, du.get_tz(user_id)))
today = dt_now(du.get_tz(user_id)).date()
counter = 0
for event in lst_events:
if event[0] == today:
counter += 1
txt += message_form(counter, event[3])
if counter: await msg.answer(text=txt, disable_web_page_preview=True)
else: await msg.answer("<b>В данный момент</b> событий на сегодня найдено не было!")
else: await msg.answer("Для отображения событий вы должны прислать ical-ссылку на календарь!")
@router.message(Command('notif'))
async def notif_up(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if du.url_exists(user_id):
if du.get_status(user_id): await msg.answer("Уведомления о событиях выключены!")
else: await msg.answer("Уведомления о событиях включены!")
du.update_status(user_id)
else: await msg.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@router.message(Command('daily'))
async def daily_up(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if dc.clock_exists(user_id):
if dc.get_daily(user_id): await msg.answer("Ежедневные утренние уведомления выключены!")
else: await msg.answer("Ежедневные утренние уведомления включены!")
dc.update_daily(user_id)
else: await msg.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@router.message(Command('moment'))
async def moment_up(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if dc.clock_exists(user_id):
if dc.get_start(user_id): await msg.answer("Уведомления в момент события выключены!")
else: await msg.answer("Уведомления в момент события включены!")
dc.update_start(user_id)
else: await msg.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
@router.message(Command('get_alarm'))
async def alarm_get(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if dc.clock_exists(user_id):
alarms = dc.get_alarm(user_id)
start = dc.get_start(user_id)
status2 = dc.get_status2(user_id)
txt = ""
if status2: txt += "<b>У вас работает два оповещения"
else: txt += "<b>У вас работает лишь первое оповещение"
if start: txt += " и сообщение в момент начала события!</b>"
else: txt += "!</b>"
await msg.answer(txt + f"\n\n<b>Первое оповещение</b> приходит за {alarms[0]} минут\n"
f"<b>Второе оповещение</b> приходит за {alarms[1]} минут")
else:
await msg.answer("Для того, чтобы получить таймеры, вы должны прислать ical-ссылку на свой календарь!")
@router.message(Command('edit_alarm'))
async def alarm_edit(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if dc.clock_exists(user_id):
await msg.answer_photo(photo=FSInputFile(path="../data/photo_edit_alarm.jpg"),
caption="Для изменения времени вам надо в ответ на это сообщение прислать два "
"числа через пробел: разница времени первого и второго таймера по ходу "
"времени соответственно")
else:
await msg.answer("Для того, чтобы изменить таймеры, вы должны прислать ical-ссылку на свой календарь!")
@router.message(Command('stop_alarm'))
async def alarm_stop(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if dc.clock_exists(user_id):
if dc.get_status2(user_id): await msg.answer("Второе уведомление выключено!")
else: await msg.answer("Второе уведомление включено!")
dc.update_status2(user_id)
else: await msg.answer("Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь!")
# ЗАГРУЗКА ССЫЛКИ
@router.message(F.content_type == ContentType.TEXT)
async def downloading_file_ics(msg: Message) -> None:
user_id = du.get_user_id(int(msg.chat.id))
if msg.text[:5] == "https":
try:
wget.download(msg.text, f'../data/icals/{str(user_id)}_new.ics')
try: os.remove(f'../data/icals/{str(user_id)}.ics')
except FileNotFoundError: pass
os.rename(f'../data/icals/{str(user_id)}_new.ics', f'../data/icals/{str(user_id)}.ics')
time_zone = msg.text.split("=")[-1]
if not du.url_exists(user_id):
du.add_url(user_id, msg.text, time_zone)
dc.add_clock(user_id)
else:
du.update_url(user_id, msg.text, time_zone)
await msg.answer("Ссылка успешно добавлена! Уведомления уже включены!")
except Exception:
await msg.answer("Ошибка скачивания! Проверьте правильность ссылки и пришлите ещё раз")
if (msg.reply_to_message is not None) and dc.clock_exists(user_id):
text = "Для изменения времени вам надо в ответ на это сообщение прислать два числа через " \
"пробел: разница времени первого и второго таймера по ходу времени соответственно"
if msg.reply_to_message.caption == text:
alarm_new = msg.text.split()
if int(alarm_new[0]) < 60 and int(alarm_new[1]) < 60:
dc.update_alarm1(user_id, int(alarm_new[0]))
dc.update_alarm2(user_id, int(alarm_new[1]))
await msg.answer("Время отправки уведомлений успешно обновлено!")
else:
await msg.answer("Время отправки уведомлений должно быть меньше 60 минут!")

15
code/init.py Normal file
View file

@ -0,0 +1,15 @@
from aiogram import Bot, Dispatcher
from aiogram.enums.parse_mode import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.client.bot import DefaultBotProperties
from config import botToken
bot = Bot(token=botToken, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())
from sql import Users, Clock
du = Users('../db/users.db')
dc = Clock('../db/clock.db')

224
code/script.py
View file

@ -1,112 +1,112 @@
from datetime import datetime, time
import icalendar
from dateutil.rrule import *
from dateutil.parser import *
def text_ical(user_id, tz):
date = dt_now(tz)
path = f'../data/icals/{user_id}.ics'
e = open(path, 'rb')
ecal = icalendar.Calendar.from_ical(e.read())
events = []
for i, component in enumerate(ecal.walk()):
if component.name == "VEVENT":
# НАЧАЛО
dt = icalendar.vDDDTypes.to_ical(component.get('dtstart')).decode('utf-8')
dt = dt.split("T")
dt_start = component.decoded("dtstart")
dt_end = component.decoded("dtend")
if len(dt) == 1:
dt_start = datetime.combine(dt_start, time(minute=1))
dt_end = datetime.combine(dt_end, time(minute=1))
else:
dt_start.replace(tzinfo=None)
dt_start = dt_start.astimezone(tz)
dt_end.replace(tzinfo=None)
dt_end = dt_end.astimezone(tz)
# НАСТРОЙКА
r_rule = component.get('rrule')
if r_rule:
list_rrule = icalendar.vRecur.to_ical(r_rule).decode('utf-8').split(';')
until, u_flag = [elem for elem in list_rrule if 'UNTIL' in elem], True
list_rrule = [elem for elem in list_rrule if 'UNTIL' not in elem]
if until:
until_old = until[0].split('=')[1]
until = isoparse(until_old).replace(tzinfo=None)
if until.replace(tzinfo=None) <= date.replace(tzinfo=None):
until, u_flag = "UNTIL=" + until_old, False
if u_flag:
until = "UNTIL=" + "".join(date.date().isoformat().split("-")) + "T235900Z"
list_rrule.append(until)
list_rrule = ";".join(list_rrule)
date_iso = "".join(dt_start.isoformat().split("-"))
iso = "".join(date_iso.split(":")).split("+")[0] + "Z"
dates = list(map(lambda x: x.replace(tzinfo=None), list(rrulestr(list_rrule, dtstart=parse(iso)))))
if dates:
dif = dt_end - dt_start
dt_start = dates[-1]
dt_end = dt_start + dif
if date.date() == dt_start.date():
org = component.get("organizer")
desc = component.get("description")
event = {"name": component.get('summary'), "desc": desc if desc else "отсутствует",
"org": org if org else "не назначен", "datetime": [dt_start, dt_end]}
events.append([dt_start.date(), dt_start.time(), i, event])
e.close()
return events
def message_form(k, event):
txt = ""
if k:
txt += f"-------\n"
txt += f"<b>{k}. {event['name']}</b>\n"
else:
txt += f"<b>{event['name']}</b>\n"
txt += f"<b>Описание:</b> {event['desc']}\n"
txt += f"<b>Организатор:</b> {event['org']}\n\n"
txt += f"<b>Начало: {event['datetime'][0].strftime('%H:%M - %d.%m.%Y года')}</b>\n"
txt += f"<b>Конец: {event['datetime'][1].strftime('%H:%M - %d.%m.%Y года')}</b>\n"
return txt
def delta_time(d_event, start, end):
d_start = datetime.combine(d_event.date(), time(hour=0, minute=start))
d_end = datetime.combine(d_event.date(), time(hour=0, minute=end))
zero = datetime.combine(d_event.date(), time(0, 0, 0, 0))
dt_start = zero + (d_event - d_start)
dt_end = zero + (d_event - d_end)
return [dt_start.time(), dt_end.time()]
def dt_now(tz):
return datetime.now(tz=tz)
from datetime import datetime, time
import icalendar
from dateutil.rrule import *
from dateutil.parser import *
def text_ical(user_id, tz):
date = dt_now(tz)
path = f'../data/icals/{user_id}.ics'
e = open(path, 'rb')
ecal = icalendar.Calendar.from_ical(e.read())
events = []
for i, component in enumerate(ecal.walk()):
if component.name == "VEVENT":
# НАЧАЛО
dt = icalendar.vDDDTypes.to_ical(component.get('dtstart')).decode('utf-8')
dt = dt.split("T")
dt_start = component.decoded("dtstart")
dt_end = component.decoded("dtend")
if len(dt) == 1:
dt_start = datetime.combine(dt_start, time(minute=1))
dt_end = datetime.combine(dt_end, time(minute=1))
else:
dt_start.replace(tzinfo=None)
dt_start = dt_start.astimezone(tz)
dt_end.replace(tzinfo=None)
dt_end = dt_end.astimezone(tz)
# НАСТРОЙКА
r_rule = component.get('rrule')
if r_rule:
list_rrule = icalendar.vRecur.to_ical(r_rule).decode('utf-8').split(';')
until, u_flag = [elem for elem in list_rrule if 'UNTIL' in elem], True
list_rrule = [elem for elem in list_rrule if 'UNTIL' not in elem]
if until:
until_old = until[0].split('=')[1]
until = isoparse(until_old).replace(tzinfo=None)
if until.replace(tzinfo=None) <= date.replace(tzinfo=None):
until, u_flag = "UNTIL=" + until_old, False
if u_flag:
until = "UNTIL=" + "".join(date.date().isoformat().split("-")) + "T235900Z"
list_rrule.append(until)
list_rrule = ";".join(list_rrule)
date_iso = "".join(dt_start.isoformat().split("-"))
iso = "".join(date_iso.split(":")).split("+")[0] + "Z"
dates = list(map(lambda x: x.replace(tzinfo=None), list(rrulestr(list_rrule, dtstart=parse(iso)))))
if dates:
dif = dt_end - dt_start
dt_start = dates[-1]
dt_end = dt_start + dif
if date.date() == dt_start.date():
org = component.get("organizer")
desc = component.get("description")
event = {"name": component.get('summary'), "desc": desc if desc else "отсутствует",
"org": org if org else "не назначен", "datetime": [dt_start, dt_end]}
events.append([dt_start.date(), dt_start.time(), i, event])
e.close()
return events
def message_form(k, event):
txt = ""
if k:
txt += f"-------\n"
txt += f"<b>{k}. {event['name']}</b>\n"
else:
txt += f"<b>{event['name']}</b>\n"
txt += f"<b>Описание:</b> {event['desc']}\n"
txt += f"<b>Организатор:</b> {event['org']}\n\n"
txt += f"<b>Начало: {event['datetime'][0].strftime('%H:%M - %d.%m.%Y года')}</b>\n"
txt += f"<b>Конец: {event['datetime'][1].strftime('%H:%M - %d.%m.%Y года')}</b>\n"
return txt
def delta_time(d_event, start, end):
d_start = datetime.combine(d_event.date(), time(hour=0, minute=start))
d_end = datetime.combine(d_event.date(), time(hour=0, minute=end))
zero = datetime.combine(d_event.date(), time(0, 0, 0, 0))
dt_start = zero + (d_event - d_start)
dt_end = zero + (d_event - d_end)
return [dt_start.time(), dt_end.time()]
def dt_now(tz):
return datetime.now(tz=tz)