1
0
Fork 0
mirror of https://github.com/EDeev/y.calendarkin.git synced 2026-10-08 00:39:27 +03:00

Compare commits

...

4 commits

Author SHA1 Message Date
5fdf39e0b5 README на русском и английском 2026-10-05 19:28:43 +00:00
73690afd8f Зависимости по рабочему окружению, тесты, CI, Docker и лицензия MIT
Тесты разбора календаря (события на сегодня, повторения, UNTIL, проверка ссылки, текст
уведомления). Образ по тегу v* публикуется в GHCR и dcr.deev.su.
2026-10-05 19:28:43 +00:00
15bd36580e Исправлены ошибки и уязвимость, улучшены уведомления
Ошибки:
- исключение в фоновых циклах (сбой сети при обновлении календаря, заблокировавший бота
  пользователь, битый календарь) останавливало весь бот — теперь ошибка у одного
  пользователя логируется и не мешает остальным;
- повторяющиеся события с UNTIL местным временем или датой роняли разбор календаря
  (ValueError из dateutil) — UNTIL приводится к UTC;
- названия и описания событий подставлялись в HTML без экранирования: «<» или «&» ломали
  отправку уведомления;
- сообщение до /start и кривой ввод времени напоминаний роняли обработчик;
- при отсутствии файла календаря разбор падал.

Безопасность: ссылка скачивалась без проверки (wget), бота можно было заставить ходить по
любым адресам, включая внутренние. Теперь принимается только https://calendar.yandex.*,
скачивание — через aiohttp с таймаутом и без редиректов, содержимое проверяется как iCal
до замены старого файла.

Уведомления (по issue #3): без даты и года — только время, пустые описание и организатор
не показываются, организатор без «mailto:», для событий на весь день — «Весь день» и без
напоминаний в 00:01; новые команды /link и /unsubscribe.

Прочее: токен из BOT_TOKEN, таблицы SQLite создаются при первом запуске (пустые базы
убраны из репозитория), ссылка «АВТОР» — на EDeev вместо чужого теперь профиля IGlek.
2026-10-05 19:28:43 +00:00
18e79e24a5 Версия на aiogram 3 — та, что работает на сервере
В репозитории оставался код на aiogram 2 (executor, message_handler), а requirements.txt
уже требовал aiogram 3 — по README бот не запускался (issue #1). Переписанная версия с
разделением на bot.py / handlers.py / init.py давно работала на сервере, теперь она и здесь.
Переводы строк приведены к LF.
2026-10-05 19:22:10 +00:00
24 changed files with 974 additions and 577 deletions

2
.env.example Normal file
View file

@ -0,0 +1,2 @@
# Токен бота от @BotFather
BOT_TOKEN=123456:your-token

3
.gitattributes vendored Normal file
View file

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

23
.github/workflows/ci.yml vendored Normal file
View file

@ -0,0 +1,23 @@
name: CI
on:
push:
branches: [main]
pull_request:
jobs:
check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements-dev.txt
- run: ruff check --select E9,F code tests
- run: pytest -q
- name: Бот импортируется
working-directory: code
env:
BOT_TOKEN: "123456:TEST"
run: python -c "import bot"

42
.github/workflows/docker.yml vendored Normal file
View file

@ -0,0 +1,42 @@
name: Docker
on:
push:
tags: ["v*"]
workflow_dispatch:
jobs:
image:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
steps:
- uses: actions/checkout@v4
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- uses: docker/login-action@v3
with:
registry: dcr.deev.su
username: ${{ secrets.ZOT_USERNAME }}
password: ${{ secrets.ZOT_PASSWORD }}
- id: meta
uses: docker/metadata-action@v5
with:
images: |
ghcr.io/edeev/y.calendarkin
dcr.deev.su/edeev/y.calendarkin
tags: |
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=raw,value=latest
- uses: docker/build-push-action@v6
with:
context: .
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}

6
.gitignore vendored Normal file
View file

@ -0,0 +1,6 @@
.env
__pycache__/
*.pyc
.pytest_cache/
db/*.db
data/icals/*.ics

20
Dockerfile Normal file
View file

@ -0,0 +1,20 @@
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY code/ code/
COPY data/photo_edit_alarm.jpg data/
RUN useradd --create-home --uid 1000 app \
&& mkdir -p data/icals db \
&& chown -R app:app /app
USER app
VOLUME ["/app/data/icals", "/app/db"]
# пути к данным в коде — относительно папки code/ (../data, ../db)
WORKDIR /app/code
CMD ["python", "bot.py"]

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2024 Egor Deev
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

88
README.en.md Normal file
View file

@ -0,0 +1,88 @@
# Y.Calendarkin
[Русский](README.md) · **English**
[![CI](https://github.com/EDeev/y.calendarkin/actions/workflows/ci.yml/badge.svg)](https://github.com/EDeev/y.calendarkin/actions/workflows/ci.yml)
[![Docker](https://github.com/EDeev/y.calendarkin/actions/workflows/docker.yml/badge.svg)](https://github.com/EDeev/y.calendarkin/actions/workflows/docker.yml)
[![License](https://img.shields.io/github/license/EDeev/y.calendarkin)](LICENSE)
A Telegram bot that sends notifications about Yandex Calendar events: a morning summary for the day, two
reminders before an event and a message when it starts. All it needs is the calendar's export link. The
bot speaks Russian.
**Status:** personal project, maintained · bot [@calendarkin_ybot](https://t.me/calendarkin_ybot)
**Stack:** Python 3.12 · aiogram 3 · aiohttp · icalendar · python-dateutil · pytz · SQLite · Docker
## Features
- Subscription by iCal export link; the time zone is taken from the link
- Recurring events (`RRULE`) are expanded for today
- A day summary at 8:00 in the calendar's time zone (`/daily`)
- Two reminders before an event: 15 and 5 minutes by default, changed with `/edit_alarm`
- A message when an event starts (`/moment`)
- The calendar is refreshed every 13 minutes; if a refresh fails, the previous version stays
## Commands
| Command | What it does |
|---|---|
| `/link` | how to get the calendar link |
| `/list` | today's events |
| `/daily` | toggle the morning summary |
| `/moment` | toggle the message at event start |
| `/get_alarm`, `/edit_alarm` | view and change reminder times |
| `/stop_alarm` | toggle the second reminder |
| `/notif` | pause or resume all notifications |
| `/unsubscribe` | remove the calendar subscription |
## Running
```bash
git clone https://github.com/EDeev/y.calendarkin.git && cd y.calendarkin
cp .env.example .env # BOT_TOKEN from @BotFather
docker compose up -d
```
Prebuilt image: `docker pull ghcr.io/edeev/y.calendarkin` or `docker pull dcr.deev.su/edeev/y.calendarkin`.
The SQLite databases are created on first start.
Without Docker: Python 3.12, `pip install -r requirements.txt`, then `cd code && BOT_TOKEN=… python bot.py`.
## How it works
```
code/bot.py entry point and two background loops: event checks every minute, calendar refresh
code/handlers.py commands and receiving the link
code/script.py link validation, download, iCal and recurrence parsing, notification text
code/sql.py users, subscriptions and reminder settings (SQLite)
```
Only `https://calendar.yandex.*` links are accepted, so the bot never downloads from arbitrary addresses.
An error for one user (a broken calendar or a blocked bot) doesn't stop notifications for others.
## Development
```bash
pip install -r requirements-dev.txt
ruff check --select E9,F code tests && pytest
```
The tests cover calendar parsing: today's events, recurrences, `UNTIL` in different formats, link
validation and the notification text. The Docker image is built on `v*` tags and published to GitHub
Packages and `dcr.deev.su`.
## License
MIT — see [LICENSE](LICENSE).
## Author
**Egor Deev** — [GitHub](https://github.com/EDeev) · [Telegram](https://t.me/DeevEgor) · [egor@deev.space](mailto:egor@deev.space)
---
<div align="center">
<sub>⭐ If you find this project useful, give it a star on GitHub!</sub>
<p><sub>Made with ❤️ — <a href="https://deev.space">deev.space</a></sub></p>
</div>

174
README.md
View file

@ -1,136 +1,100 @@
# 🗓️ Y.Calendarkin
# Я.Календаркин
Telegram-бот для упрощенного получения уведомлений о событиях из Яндекс.Календаря.
**Русский** · [English](README.en.md)
[![Python](https://img.shields.io/badge/Python-3.8+-blue.svg)](https://python.org)
[![Aiogram](https://img.shields.io/badge/Aiogram-3.x-green.svg)](https://docs.aiogram.dev/)
[![License](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![CI](https://github.com/EDeev/y.calendarkin/actions/workflows/ci.yml/badge.svg)](https://github.com/EDeev/y.calendarkin/actions/workflows/ci.yml)
[![Docker](https://github.com/EDeev/y.calendarkin/actions/workflows/docker.yml/badge.svg)](https://github.com/EDeev/y.calendarkin/actions/workflows/docker.yml)
[![License](https://img.shields.io/github/license/EDeev/y.calendarkin)](LICENSE)
## 📖 Описание
Telegram-бот, который присылает уведомления о событиях из Яндекс.Календаря: утреннюю сводку на день,
два напоминания перед событием и сообщение в момент начала. Достаточно прислать ссылку экспорта календаря.
Y.Calendarkin — некоммерческий проект, предназначенный для автоматизации уведомлений о календарных событиях из Яндекс.Календаря через Telegram. Бот позволяет настроить гибкую систему напоминаний и получать актуальную информацию о предстоящих событиях.
**Статус:** личный проект, поддерживается · бот [@calendarkin_ybot](https://t.me/calendarkin_ybot)
## ✨ Основные возможности
**Стек:** Python 3.12 · aiogram 3 · aiohttp · icalendar · python-dateutil · pytz · SQLite · Docker
- 📅 **Импорт календаря** — поддержка ICal ссылок из Яндекс.Календаря
- ⏰ **Гибкие уведомления** — настройка времени напоминаний (до 60 минут)
- 🌅 **Ежедневные отчеты** — утренние сводки событий на день
- 🎯 **Моментальные уведомления** — оповещения в момент начала события
- 🌍 **Поддержка часовых поясов** — автоматическое определение из календаря
- 📋 **Просмотр событий** — список мероприятий на текущий день
## Возможности
## 🛠️ Технологический стек
- Подписка по ссылке экспорта iCal; часовой пояс берётся из ссылки
- Повторяющиеся события (`RRULE`) разворачиваются на сегодняшний день
- Сводка на день в 8:00 по часовому поясу календаря (`/daily`)
- Два напоминания перед событием: по умолчанию за 15 и 5 минут, меняются в `/edit_alarm`
- Сообщение в момент начала события (`/moment`)
- Календарь обновляется раз в 13 минут; если обновление не удалось, остаётся прошлая версия
- **Backend**: Python 3.8+
- **Bot Framework**: Aiogram 3.x
- **Database**: SQLite
- **Calendar Processing**: icalendar, dateutil
- **HTTP Client**: wget
- **Timezone Support**: pytz
Пример уведомления:
## 🚀 Установка и запуск
```
Напоминаю!
Через 15 минут будет событие:
### Предварительные требования
Созвон по проекту
Организатор: boss@example.com
10:00 — 11:00
```
## Команды
| Команда | Что делает |
|---|---|
| `/link` | как получить ссылку на календарь |
| `/list` | события на сегодня |
| `/daily` | включить или выключить утреннюю сводку |
| `/moment` | включить или выключить сообщение в момент начала |
| `/get_alarm`, `/edit_alarm` | посмотреть и изменить время напоминаний |
| `/stop_alarm` | включить или выключить второе напоминание |
| `/notif` | приостановить или возобновить все уведомления |
| `/unsubscribe` | удалить подписку на календарь |
## Запуск
```bash
pip install -r requirements.txt
git clone https://github.com/EDeev/y.calendarkin.git && cd y.calendarkin
cp .env.example .env # BOT_TOKEN от @BotFather
docker compose up -d
```
### Настройка
Готовый образ: `docker pull ghcr.io/edeev/y.calendarkin` или `docker pull dcr.deev.su/edeev/y.calendarkin`.
Базы SQLite создаются при первом запуске.
1. Создайте нового бота через [@BotFather](https://t.me/BotFather)
2. Получите токен и добавьте его в `config.py`:
Без Docker: Python 3.12, `pip install -r requirements.txt`, затем `cd code && BOT_TOKEN=… python bot.py`.
```python
TOKEN = "your_bot_token_here"
## Как устроено
```
code/bot.py запуск и два фоновых цикла: проверка событий раз в минуту, обновление календарей
code/handlers.py команды и приём ссылки
code/script.py проверка ссылки, скачивание, разбор iCal и повторений, текст уведомления
code/sql.py пользователи, подписки и настройки напоминаний (SQLite)
```
3. Создайте структуру каталогов:
Принимаются только ссылки `https://calendar.yandex.*`: бот не скачивает файлы по произвольным адресам.
Ошибка у одного пользователя — битый календарь или заблокированный бот — не останавливает рассылку
остальным.
## Разработка
```bash
mkdir -p data/icals db
pip install -r requirements-dev.txt
ruff check --select E9,F code tests && pytest
```
### Запуск
Тесты проверяют разбор календаря: события на сегодня, повторения, `UNTIL` в разных форматах, проверку
ссылки и текст уведомления. Docker-образ собирается по тегу `v*` и публикуется в GitHub Packages и
`dcr.deev.su`.
```bash
cd code
python bot.py
```
## Лицензия
## 📝 Использование
MIT — см. [LICENSE](LICENSE).
### Начало работы
## Автор
1. Запустите бота командой `/start`
2. Отправьте ICal ссылку вашего Яндекс.Календаря
3. Настройте уведомления по своему усмотрению
### Основные команды
| Команда | Описание |
|---------|----------|
| `/help` | Справочная информация |
| `/list` | События на сегодня |
| `/notif` | Включить/выключить уведомления |
| `/daily` | Ежедневные утренние сводки |
| `/moment` | Уведомления в момент события |
| `/get_alarm` | Информация о текущих настройках |
| `/edit_alarm` | Изменение времени уведомлений |
| `/stop_alarm` | Отключение второго уведомления |
### Получение ICal ссылки
1. Откройте Яндекс.Календарь
2. Перейдите в настройки календаря
3. Найдите раздел "Экспорт"
4. Скопируйте ссылку в формате ICal
5. Отправьте ссылку боту
## 🏗️ Архитектура проекта
```
y.calendarkin/
├── code/
│ ├── bot.py # Основная логика бота
│ ├── config.py # Конфигурация
│ ├── script.py # Обработка календарных файлов
│ └── sql.py # Работа с базой данных
├── data/
│ ├── icals/ # Загруженные календари
│ └── photo_edit_alarm.jpg
├── db/ # База данных SQLite
└── requirements.txt
```
## 📊 База данных
Проект использует две основные таблицы:
- **user** — информация о пользователях и их календарях
- **alarm** — настройки уведомлений для каждого пользователя
## 🔧 Настройка уведомлений
- **Первое уведомление**: за 15 минут до события (по умолчанию)
- **Второе уведомление**: за 5 минут до события (по умолчанию)
- **Диапазон настройки**: от 1 до 59 минут
- **Ежедневные сводки**: в 8:00 по часовому поясу пользователя
## 📄 Лицензия
Этот проект является некоммерческим и распространяется под лицензией MIT.
## 👨‍💻 Автор
**Деев Егор Викторович** - Backend Developer
- GitHub: [@EDeev](https://github.com/EDeev)
- Email: egor@deev.space
- Telegram: [@Egor_Deev](https://t.me/Egor_Deev)
**Деев Егор Викторович** — [GitHub](https://github.com/EDeev) · [Telegram](https://t.me/DeevEgor) · [egor@deev.space](mailto:egor@deev.space)
---
<div align="center">
<sub>Этот бот предназначен для личного использования и не является коммерческим продуктом.</sub>
<p><sub>Создано с ❤️ от вашего дорогого - deev.space ©</sub></p>
<sub>⭐ Если проект оказался полезным, поставьте звёздочку на GitHub!</sub>
<p><sub>Сделано с ❤️ — <a href="https://deev.space">deev.space</a></sub></p>
</div>

View file

@ -1,321 +1,113 @@
import logging, config, sql, asyncio, 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 минут!")
# ПРОВЕРКА НА СОБЫТИЕ
async def alarm(wait_for):
while True:
await asyncio.sleep(wait_for)
users_id = du.all_users()
for user_id in users_id:
user_id = user_id[0]
if dc.clock_exists(user_id):
if du.get_status(user_id):
events = sorted(text_ical(user_id, du.get_tz(user_id)))
tg_id = str(du.get_first_user_id(user_id))
start = dc.get_start(user_id)
daily = dc.get_daily(user_id)
alarm = dc.get_alarm(user_id)
alarm2_status = dc.get_status2(user_id)
today = dt_now(du.get_tz(user_id)).date()
time_check = dt_now(du.get_tz(user_id)).time()
# ДЛЯ DAILY
counter = 0
txt = "<b>События сегодня</b>\n\n"
delta_daily1 = time(hour=8, minute=0)
delta_daily2 = time(hour=8, minute=1)
# --------------------------------
for event in events:
if event[0] == today:
d_event = datetime.combine(today, event[1])
if daily and delta_daily1 <= time_check < delta_daily2:
counter += 1
txt += message_form(counter, event[3])
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,
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,
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,
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)
else:
await bot.send_message(chat_id=tg_id, parse_mode=ParseMode.HTML,
text=f"Сегодня событий <b>нет</b>")
async def update(wait_for):
while True:
await asyncio.sleep(wait_for)
users_id = du.all_users()
for user_id in users_id:
user_id = user_id[0]
if du.url_exists(user_id):
if du.get_status(user_id):
wget.download(du.get_url(user_id), f'../data/icals/{str(user_id)}_new.ics')
os.remove(f'../data/icals/{str(user_id)}.ics')
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)
import asyncio
import logging
from datetime import datetime, time
from init import bot, dp, du, dc
from script import delta_time, download_ical, dt_now, message_form, text_ical
from handlers import router
async def send(tg_id, text):
await bot.send_message(chat_id=tg_id, disable_web_page_preview=True, text=text)
async def check_user(user_id):
events = sorted(text_ical(user_id, du.get_tz(user_id)), key=lambda e: (e[0], e[1], e[2]))
tg_id = str(du.get_first_user_id(user_id))
start = dc.get_start(user_id)
daily = dc.get_daily(user_id)
alarm = dc.get_alarm(user_id)
alarm2_status = dc.get_status2(user_id)
today = dt_now(du.get_tz(user_id)).date()
time_check = dt_now(du.get_tz(user_id)).time()
# ДЛЯ DAILY
counter = 0
txt = "<b>События сегодня</b>\n\n"
delta_daily1 = time(hour=8, minute=0)
delta_daily2 = time(hour=8, minute=1)
# --------------------------------
for event in events:
if event[0] == today:
d_event = datetime.combine(today, event[1])
if daily and delta_daily1 <= time_check < delta_daily2:
counter += 1
txt += message_form(counter, event[3])
# для событий на весь день напоминания по времени не шлём
if event[3]["all_day"]:
continue
delta_start = delta_time(d_event, 1, 0)
if start and delta_start[0] < time_check <= delta_start[1]:
await send(tg_id, "<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 send(tg_id, 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 send(tg_id, 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 send(tg_id, txt)
else:
await send(tg_id, "Сегодня событий <b>нет</b>")
# ПРОВЕРКА НА СОБЫТИЕ
async def alarm(wait_for):
while True:
await asyncio.sleep(wait_for)
for (user_id,) in du.all_users():
# ошибка у одного пользователя (заблокировал бота, битый календарь) не должна
# останавливать рассылку остальным и весь бот
try:
if dc.clock_exists(user_id) and du.get_status(user_id):
await check_user(user_id)
except Exception:
logging.exception("Ошибка проверки событий пользователя %s", user_id)
async def update(wait_for):
while True:
await asyncio.sleep(wait_for)
for (user_id,) in du.all_users():
try:
if du.url_exists(user_id) and du.get_status(user_id):
await download_ical(du.get_url(user_id), user_id)
except Exception as err:
# при сбое остаётся прошлая версия календаря
logging.warning("Не удалось обновить календарь пользователя %s: %s", user_id, err)
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())
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

View file

@ -1 +1,3 @@
TOKEN = "**********:***************************"
import os
botToken = os.getenv("BOT_TOKEN", "XXXXXXXXXXXXXXXXXXXXXXXX") # @calendarkin_ybot

238
code/handlers.py Normal file
View file

@ -0,0 +1,238 @@
import logging
import os
from aiogram import F, Router
from aiogram.types import Message, CallbackQuery, ContentType, InlineKeyboardMarkup, InlineKeyboardButton, FSInputFile
from aiogram.filters import Command
from init import du, dc
from script import ICAL_DIR, download_ical, dt_now, message_form, parse_ical_url, text_ical
router = Router()
NO_LINK = "Для взаимодействия с событиями вы должны прислать ical-ссылку на календарь! Как её получить — /link"
EDIT_ALARM_TEXT = ("Для изменения времени вам надо в ответ на это сообщение прислать два числа через пробел: "
"разница времени первого и второго таймера по ходу времени соответственно")
def user_id_of(msg: Message) -> int:
"""Короткий id пользователя; заводит запись, если человек пишет боту, не нажав /start"""
tg_id = int(msg.chat.id)
if not du.user_exists(tg_id):
du.add_user(tg_id)
return du.get_user_id(tg_id)
# ПРИВЕТСТВЕННОЕ СООБЩЕНИЕ
@router.message(Command('start', 'help'))
async def helps(msg: Message) -> None:
user_id_of(msg)
buttons = [[InlineKeyboardButton(text="КОМАНДЫ", callback_data="com"),
InlineKeyboardButton(text="АВТОР", callback_data="auth")]]
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
await msg.answer(text="<b>Я.Календаркин</b> - бот для оповещения о событиях из <b>Яндекс.Календаря</b>. "
"Для начала работы вам нужно всего лишь прислать в чат ссылку экспорта календаря в "
"<b>формате ICal</b> (как её получить — /link). После получения ссылки, бот начнёт "
"оповещать о всех новых событиях и появится возможность настройки оповещений. О том, "
"какие команды есть для настройки, вы можете ознакомиться по кнопке <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/EDeev">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>/link</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> - команда, отключающая второе оповещение о событии\n\n'
'<b>/unsubscribe</b> - удалить подписку на календарь')
# КОМАНДЫ
@router.message(Command('link'))
async def link_help(msg: Message) -> None:
await msg.answer("<b>Как получить ссылку</b>\n\n"
"1. Откройте Яндекс.Календарь в браузере и наведите на нужный календарь в списке слева\n"
"2. Нажмите на шестерёнку → <b>Экспорт</b>\n"
"3. Выберите формат <b>iCal</b> и скопируйте ссылку\n"
"4. Пришлите её сюда — в любой момент, новая ссылка заменит старую")
@router.message(Command('unsubscribe'))
async def unsubscribe(msg: Message) -> None:
user_id = user_id_of(msg)
if du.url_exists(user_id):
du.delete_url(user_id)
dc.delete_clock(user_id)
try: os.remove(f"{ICAL_DIR}/{user_id}.ics")
except FileNotFoundError: pass
await msg.answer("Подписка на календарь удалена, уведомлений больше не будет. "
"Чтобы подписаться снова, пришлите ссылку.")
else: await msg.answer("Подписки на календарь нет.")
@router.message(Command('list'))
async def check_list(msg: Message) -> None:
user_id = user_id_of(msg)
if du.url_exists(user_id):
txt = "<b>Имеющиеся события на сегодня</b>\n\n"
lst_events = sorted(text_ical(user_id, du.get_tz(user_id)), key=lambda e: (e[0], e[1], e[2]))
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(NO_LINK)
@router.message(Command('notif'))
async def notif_up(msg: Message) -> None:
user_id = user_id_of(msg)
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(NO_LINK)
@router.message(Command('daily'))
async def daily_up(msg: Message) -> None:
user_id = user_id_of(msg)
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(NO_LINK)
@router.message(Command('moment'))
async def moment_up(msg: Message) -> None:
user_id = user_id_of(msg)
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(NO_LINK)
@router.message(Command('get_alarm'))
async def alarm_get(msg: Message) -> None:
user_id = user_id_of(msg)
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(NO_LINK)
@router.message(Command('edit_alarm'))
async def alarm_edit(msg: Message) -> None:
user_id = user_id_of(msg)
if dc.clock_exists(user_id):
await msg.answer_photo(photo=FSInputFile(path="../data/photo_edit_alarm.jpg"), caption=EDIT_ALARM_TEXT)
else:
await msg.answer(NO_LINK)
@router.message(Command('stop_alarm'))
async def alarm_stop(msg: Message) -> None:
user_id = user_id_of(msg)
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(NO_LINK)
# ЗАГРУЗКА ССЫЛКИ И ИЗМЕНЕНИЕ ВРЕМЕНИ ОПОВЕЩЕНИЙ
@router.message(F.content_type == ContentType.TEXT)
async def downloading_file_ics(msg: Message) -> None:
user_id = user_id_of(msg)
if msg.text.startswith("http"):
url = msg.text.strip()
time_zone = parse_ical_url(url)
if time_zone is None:
await msg.answer("Это не ссылка экспорта Яндекс.Календаря. Как её получить — /link")
return
try:
await download_ical(url, user_id)
except Exception as err:
logging.warning("Не удалось скачать календарь пользователя %s: %s", user_id, err)
await msg.answer("Ошибка скачивания! Проверьте правильность ссылки и пришлите ещё раз")
return
if not du.url_exists(user_id):
du.add_url(user_id, url, time_zone)
dc.add_clock(user_id)
else:
du.update_url(user_id, url, time_zone)
await msg.answer("Ссылка успешно добавлена! Уведомления уже включены!")
return
if msg.reply_to_message is not None and msg.reply_to_message.caption == EDIT_ALARM_TEXT \
and dc.clock_exists(user_id):
alarm_new = msg.text.split()
if len(alarm_new) != 2 or not all(x.isdigit() for x in alarm_new):
await msg.answer("Пришлите два числа через пробел, например: <b>15 5</b>")
elif all(0 < int(x) < 60 for x in alarm_new):
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("Время отправки уведомлений должно быть от 1 до 59 минут!")
return
await msg.answer("Не понял сообщение. Пришлите ссылку на календарь (/link) или посмотрите команды — /help")

16
code/init.py Normal file
View file

@ -0,0 +1,16 @@
import os
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
from sql import Users, Clock
bot = Bot(token=botToken, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())
os.makedirs("../db", exist_ok=True)
du = Users('../db/users.db')
dc = Clock('../db/clock.db')

View file

@ -1,112 +1,178 @@
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)
import os
import re
from html import escape
from datetime import datetime, time
from urllib.parse import parse_qs, urlparse
import aiohttp
import icalendar
import pytz
from dateutil.parser import isoparse, parse
from dateutil.rrule import rrulestr
ICAL_DIR = "../data/icals"
# только экспорт Яндекс.Календаря: произвольная ссылка позволила бы заставить бота ходить по любым
# адресам, в том числе внутренним адресам сервера
YANDEX_HOST = re.compile(r"^calendar\.yandex\.(ru|com|by|kz|uz|com\.tr)$")
def parse_ical_url(url):
"""Проверяет ссылку экспорта и возвращает часовой пояс из неё или None, если ссылка не подходит"""
parts = urlparse(url.strip())
if parts.scheme != "https" or not YANDEX_HOST.match(parts.hostname or ""):
return None
tz = parse_qs(parts.query).get("tz_id", ["Europe/Moscow"])[0]
return tz if tz in pytz.all_timezones_set else "Europe/Moscow"
async def download_ical(url, user_id):
"""Скачивает календарь, проверяет, что это iCal, и только потом заменяет старый файл"""
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url, allow_redirects=False) as response:
response.raise_for_status()
data = await response.read()
icalendar.Calendar.from_ical(data) # бросит исключение, если пришёл не календарь
os.makedirs(ICAL_DIR, exist_ok=True)
path_new = f"{ICAL_DIR}/{user_id}_new.ics"
with open(path_new, "wb") as f:
f.write(data)
os.replace(path_new, f"{ICAL_DIR}/{user_id}.ics")
def text_ical(user_id, tz):
date = dt_now(tz)
path = f"{ICAL_DIR}/{user_id}.ics"
if not os.path.exists(path):
return []
with open(path, "rb") as e:
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")
all_day = len(dt) == 1
if all_day:
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_utc(until_old, tz), 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": str(desc).strip() if desc else "",
"org": str(org).removeprefix("mailto:") if org else "", "all_day": all_day,
"datetime": [dt_start, dt_end]}
events.append([dt_start.date(), dt_start.time(), i, event])
return events
def until_utc(value, tz):
"""UNTIL в UTC: dateutil требует его, когда начало события задано с часовым поясом,
а Яндекс иногда отдаёт UNTIL местным временем или просто датой"""
if value.endswith("Z"):
return value
until = isoparse(value)
if "T" not in value: # только дата — повторения включительно по этот день
until = until.replace(hour=23, minute=59, second=59)
if until.tzinfo is None:
until = tz.localize(until)
return until.astimezone(pytz.utc).strftime("%Y%m%dT%H%M%SZ")
def message_form(k, event):
txt = ""
if k:
txt += "-------\n"
txt += f"<b>{k}. {escape(str(event['name']))}</b>\n"
else:
txt += f"<b>{escape(str(event['name']))}</b>\n"
# пустые описание и организатора не показываем
if event["desc"]:
txt += f"<b>Описание:</b> {escape(event['desc'])}\n"
if event["org"]:
txt += f"<b>Организатор:</b> {escape(event['org'])}\n"
# события всегда сегодняшние, поэтому дата не нужна — только время
start, end = event["datetime"]
if event["all_day"]:
txt += "\n<b>Весь день</b>\n"
elif end.date() != start.date():
txt += f"\n<b>{start.strftime('%H:%M')} — {end.strftime('%d.%m %H:%M')}</b>\n"
else:
txt += f"\n<b>{start.strftime('%H:%M')} — {end.strftime('%H:%M')}</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)

View file

@ -8,50 +8,56 @@ class Users:
self.connection = sqlite3.connect(database)
self.cursor = self.connection.cursor()
# на новой установке (например, в Docker с пустым томом) таблиц ещё нет
with self.connection:
self.cursor.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL)")
self.cursor.execute("CREATE TABLE IF NOT EXISTS url (user_id INTEGER NOT NULL, "
"status BOOLEAN NOT NULL DEFAULT (True), url_ical STRING, time_zone STRING)")
# КОМАНДЫ USER
def user_exists(self, user_id):
"""Проверяем, есть ли уже пользователь в базе"""
with self.connection:
result = self.cursor.execute(f'SELECT * FROM `user` WHERE `user_id` = ?', (user_id,)).fetchall()
result = self.cursor.execute('SELECT * FROM `user` WHERE `user_id` = ?', (user_id,)).fetchall()
return bool(len(result))
def all_users(self):
"""Список айди"""
with self.connection:
return self.cursor.execute(f'SELECT `id` FROM `user`').fetchall()
return self.cursor.execute('SELECT `id` FROM `user`').fetchall()
def add_user(self, user_id):
"""Добавляем нового пользователя"""
with self.connection:
return self.cursor.execute(f"INSERT INTO `user` (`user_id`) VALUES(?)", (user_id,))
return self.cursor.execute("INSERT INTO `user` (`user_id`) VALUES(?)", (user_id,))
def get_user_id(self, user_id):
"""Получаем короткое айди юзера"""
with self.connection:
return self.cursor.execute(f'SELECT `id` FROM `user` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `id` FROM `user` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
def get_first_user_id(self, user_id):
"""Получаем длинное айди юзера"""
with self.connection:
return self.cursor.execute(f'SELECT `user_id` FROM `user` WHERE `id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `user_id` FROM `user` WHERE `id` = ?', (user_id,)).fetchone()[0]
# КОМАНДЫ URL
def url_exists(self, user_id):
"""Проверяем, есть ли данные уже в базе"""
with self.connection:
result = self.cursor.execute(f'SELECT * FROM `url` WHERE `user_id` = ?', (user_id,)).fetchall()
result = self.cursor.execute('SELECT * FROM `url` WHERE `user_id` = ?', (user_id,)).fetchall()
return bool(len(result))
def add_url(self, user_id, url_ical, time_zone):
"""Добавляем ссылку на календарь"""
with self.connection:
return self.cursor.execute(f"INSERT INTO `url` (`user_id`, `url_ical`, `time_zone`) VALUES(?, ?, ?)",
return self.cursor.execute("INSERT INTO `url` (`user_id`, `url_ical`, `time_zone`) VALUES(?, ?, ?)",
(user_id, url_ical, time_zone))
def update_status(self, user_id):
"""Обновляем статус рассылки уведомлений"""
with self.connection:
status = self.cursor.execute(f'SELECT `status` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
status = self.cursor.execute('SELECT `status` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute("UPDATE `url` SET `status` = ? WHERE `user_id` = ?", (not status, user_id))
def update_url(self, user_id, url_ical, time_zone):
@ -63,17 +69,22 @@ class Users:
def get_status(self, user_id):
"""Получаем статус работы"""
with self.connection:
return self.cursor.execute(f'SELECT `status` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `status` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
def get_url(self, user_id):
"""Получаем ссылку на календарь"""
with self.connection:
return self.cursor.execute(f'SELECT `url_ical` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `url_ical` FROM `url` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
def delete_url(self, user_id):
"""Удаляем подписку на календарь"""
with self.connection:
return self.cursor.execute("DELETE FROM `url` WHERE `user_id` = ?", (user_id,))
def get_tz(self, user_id):
"""Получаем указанный в календаре часовой пояс"""
with self.connection:
return timezone(self.cursor.execute(f'SELECT `time_zone` FROM `url` WHERE `user_id` = ?',
return timezone(self.cursor.execute('SELECT `time_zone` FROM `url` WHERE `user_id` = ?',
(user_id,)).fetchone()[0])
# ЗАКРЫТИЕ ВЫЗОВА
@ -88,52 +99,62 @@ class Clock:
self.connection = sqlite3.connect(database)
self.cursor = self.connection.cursor()
with self.connection:
self.cursor.execute("CREATE TABLE IF NOT EXISTS alarm (user_id INTEGER NOT NULL, "
"daily BOOLEAN NOT NULL DEFAULT (False), start BOOLEAN NOT NULL DEFAULT (True), "
"alarm_1 INTEGER, alarm_2 INTEGER, status_2 BOOLEAN NOT NULL DEFAULT (True))")
# КОМАНДЫ ALARM
def clock_exists(self, user_id):
"""Проверяем, есть ли данные уже в базе"""
with self.connection:
result = self.cursor.execute(f'SELECT * FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchall()
result = self.cursor.execute('SELECT * FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchall()
return bool(len(result))
def add_clock(self, user_id):
"""Добавляем параметры уведомления"""
with self.connection:
return self.cursor.execute(f"INSERT INTO `alarm` (`user_id`, `alarm_1`, `alarm_2`) VALUES(?, ?, ?)",
return self.cursor.execute("INSERT INTO `alarm` (`user_id`, `alarm_1`, `alarm_2`) VALUES(?, ?, ?)",
(user_id, 15, 5))
def delete_clock(self, user_id):
"""Удаляем настройки оповещений"""
with self.connection:
return self.cursor.execute("DELETE FROM `alarm` WHERE `user_id` = ?", (user_id,))
# КОМАНДЫ ПОЛУЧЕНИЯ ССЫЛОК
def get_alarm(self, user_id):
"""Получаем задержки таймера"""
with self.connection:
return self.cursor.execute(f'SELECT `alarm_1`, `alarm_2` FROM `alarm` WHERE `user_id` = ?',
return self.cursor.execute('SELECT `alarm_1`, `alarm_2` FROM `alarm` WHERE `user_id` = ?',
(user_id,)).fetchone()
def get_start(self, user_id):
"""Получаем статус уведомления в момент исполнения"""
with self.connection:
return self.cursor.execute(f'SELECT `start` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `start` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
def get_daily(self, user_id):
"""Получаем статус ежедневного уведомления"""
with self.connection:
return self.cursor.execute(f'SELECT `daily` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `daily` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
def get_status2(self, user_id):
"""Получаем статус отправки второго уведомления"""
with self.connection:
return self.cursor.execute(f'SELECT `status_2` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute('SELECT `status_2` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
# КОМАНДЫ ОБНОВЛЕНИЯ БУЛЕВЫХ СТАТУСОВ
def update_daily(self, user_id):
"""Обновляем статус ежедневного уведомления"""
with self.connection:
daily = self.cursor.execute(f'SELECT `daily` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
daily = self.cursor.execute('SELECT `daily` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute("UPDATE `alarm` SET `daily` = ? WHERE `user_id` = ?", (not daily, user_id))
def update_start(self, user_id):
"""Обновляем статус уведомления в момент исполнения"""
with self.connection:
start = self.cursor.execute(f'SELECT `start` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
start = self.cursor.execute('SELECT `start` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute("UPDATE `alarm` SET `start` = ? WHERE `user_id` = ?", (not start, user_id))
# КОМАНДЫ ОБНОВЛЕНИЯ ВРЕМЕННОГО ДИАПАЗОНА
@ -150,7 +171,7 @@ class Clock:
def update_status2(self, user_id):
"""Обновляем статус отправки второго уведомления"""
with self.connection:
status_2 = self.cursor.execute(f'SELECT `status_2` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
status_2 = self.cursor.execute('SELECT `status_2` FROM `alarm` WHERE `user_id` = ?', (user_id,)).fetchone()[0]
return self.cursor.execute("UPDATE `alarm` SET `status_2` = ? WHERE `user_id` = ?", (not status_2, user_id))
# ЗАКРЫТИЕ ВЫЗОВА

15
compose.yaml Normal file
View file

@ -0,0 +1,15 @@
services:
bot:
build: .
image: ghcr.io/edeev/y.calendarkin:latest
env_file: .env
environment:
TZ: Europe/Moscow
volumes:
- icals:/app/data/icals
- db:/app/db
restart: unless-stopped
volumes:
icals:
db:

0
data/icals/.gitkeep Normal file
View file

0
db/.gitkeep Normal file
View file

Binary file not shown.

Binary file not shown.

3
requirements-dev.txt Normal file
View file

@ -0,0 +1,3 @@
-r requirements.txt
pytest==8.4.2
ruff==0.14.0

View file

@ -1,18 +1,5 @@
# Core bot framework
aiogram>=3.0.0
# Calendar processing
icalendar>=5.0.0
python-dateutil>=2.8.0
# Timezone support
pytz>=2023.3
# HTTP client for downloading calendar files
wget>=3.2
# Database (built-in with Python)
# sqlite3 - included in Python standard library
# Additional dependencies for date/time handling
typing-extensions>=4.0.0
aiogram==3.23.0
aiohttp==3.13.2
icalendar==6.3.2
python-dateutil==2.9.0.post0
pytz==2025.2

4
tests/conftest.py Normal file
View file

@ -0,0 +1,4 @@
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "code"))

84
tests/test_script.py Normal file
View file

@ -0,0 +1,84 @@
from datetime import datetime, timedelta
import pytz
import script
TZ = pytz.timezone("Europe/Moscow")
def make_ics(events):
body = "\r\n".join(events)
return ("BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//test//EN\r\n" + body + "\r\nEND:VCALENDAR\r\n").encode()
def vevent(uid, start, end, summary, extra=""):
return (f"BEGIN:VEVENT\r\nUID:{uid}\r\nDTSTART;TZID=Europe/Moscow:{start:%Y%m%dT%H%M%S}\r\n"
f"DTEND;TZID=Europe/Moscow:{end:%Y%m%dT%H%M%S}\r\nSUMMARY:{summary}\r\n{extra}END:VEVENT")
def test_parse_ical_url_accepts_only_yandex_export():
ok = "https://calendar.yandex.ru/export/ics.xml?private_token=abc&tz_id=Asia/Krasnoyarsk"
assert script.parse_ical_url(ok) == "Asia/Krasnoyarsk"
assert script.parse_ical_url("https://calendar.yandex.ru/export/ics.xml?private_token=abc") == "Europe/Moscow"
assert script.parse_ical_url("https://calendar.yandex.ru/x?tz_id=Not/AZone") == "Europe/Moscow"
for bad in ("http://calendar.yandex.ru/export/ics.xml", "https://127.0.0.1/ics", "file:///etc/passwd",
"https://calendar.yandex.ru.evil.com/ics", "https://example.com/?tz_id=Europe/Moscow"):
assert script.parse_ical_url(bad) is None, bad
def test_text_ical_today_events_and_recurrence(tmp_path, monkeypatch):
monkeypatch.setattr(script, "ICAL_DIR", str(tmp_path))
now = datetime.now(TZ).replace(tzinfo=None, second=0, microsecond=0)
today = now.replace(hour=10, minute=0)
events = [
vevent("once", today, today + timedelta(hours=1), "Разовое <важное> & срочное",
"DESCRIPTION:Описание\r\nORGANIZER:mailto:boss@example.com\r\n"),
vevent("weekly", today.replace(hour=12) - timedelta(weeks=3),
today.replace(hour=13) - timedelta(weeks=3), "Еженедельное", "RRULE:FREQ=WEEKLY\r\n"),
vevent("tomorrow", today + timedelta(days=1), today + timedelta(days=1, hours=1), "Завтра"),
]
(tmp_path / "1.ics").write_bytes(make_ics(events))
found = {e[3]["name"]: e for e in script.text_ical(1, TZ)}
assert set(found) == {"Разовое <важное> & срочное", "Еженедельное"}
assert found["Еженедельное"][1].hour == 12
text = script.message_form(1, found["Разовое <важное> & срочное"][3])
assert "&lt;важное&gt; &amp; срочное" in text
assert "boss@example.com" in text and "mailto:" not in text
assert "10:00 — 11:00" in text and str(now.year) not in text
def test_message_form_hides_empty_fields():
event = {"name": "Встреча", "desc": "", "org": "", "all_day": False,
"datetime": [datetime(2026, 1, 1, 9, 0), datetime(2026, 1, 1, 9, 30)]}
text = script.message_form(0, event)
assert "Описание" not in text and "Организатор" not in text
assert "09:00 — 09:30" in text
def test_missing_calendar_file_gives_no_events(tmp_path, monkeypatch):
monkeypatch.setattr(script, "ICAL_DIR", str(tmp_path))
assert script.text_ical(42, TZ) == []
def test_until_not_in_utc(tmp_path, monkeypatch):
# повторяющиеся события с UNTIL местным временем или датой раньше роняли разбор календаря
monkeypatch.setattr(script, "ICAL_DIR", str(tmp_path))
now = datetime.now(TZ).replace(tzinfo=None, second=0, microsecond=0)
first = now.replace(hour=9, minute=0) - timedelta(weeks=5)
events = [
vevent("ended-local", first, first + timedelta(hours=1), "Закончилось",
f"RRULE:FREQ=WEEKLY;UNTIL={first + timedelta(weeks=2):%Y%m%dT%H%M%S}\r\n"),
vevent("ended-date", first, first + timedelta(hours=1), "Закончилось датой",
f"RRULE:FREQ=WEEKLY;UNTIL={first + timedelta(weeks=2):%Y%m%d}\r\n"),
vevent("running", first, first + timedelta(hours=1), "Идёт",
f"RRULE:FREQ=WEEKLY;UNTIL={now + timedelta(weeks=4):%Y%m%dT%H%M%S}\r\n"),
]
(tmp_path / "2.ics").write_bytes(make_ics(events))
names = {e[3]["name"] for e in script.text_ical(2, TZ)}
assert names == {"Идёт"}
assert script.until_utc("20260101", TZ) == "20260101T205959Z"