1
0
Fork 0
mirror of https://github.com/EDeev/npr_meteobot.git synced 2026-10-07 20:49:33 +03:00

Compare commits

...

5 commits

Author SHA1 Message Date
69830f156f README на русском и английском 2026-10-02 23:36:49 +00:00
646943c21e requirements.txt и CI (ruff, создание базы) 2026-10-02 23:36:49 +00:00
28a5170be0 Мелкие исправления по ruff: дублирующееся имя обработчика /weather, неиспользуемые переменные 2026-10-02 23:36:49 +00:00
f223570bab Токен из окружения, база создаётся сама
- токен бота — из BOT_TOKEN, а не из config.py;
- sub.db (с идентификатором подписчика) убран из git: SQLighter создаёт таблицы и
  служебные строки из schema.sql, если базы нет.
2026-10-02 23:36:49 +00:00
be4411e269 Переводы строк приведены к LF 2026-10-02 23:33:14 +00:00
12 changed files with 480 additions and 475 deletions

2
.gitattributes vendored Normal file
View file

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

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

@ -0,0 +1,21 @@
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.10"
- run: pip install ruff==0.14.0
- name: Ruff (синтаксис и ошибки)
run: ruff check --select E9,F .
- name: База создаётся из schema.sql
run: |
python -c "import sql; db = sql.SQLighter('/tmp/t.db'); db.add_user('1'); assert db.user_exists('1')"

5
.gitignore vendored Normal file
View file

@ -0,0 +1,5 @@
__pycache__/
*.py[cod]
.venv/
.env
*.db

41
README.en.md Normal file
View file

@ -0,0 +1,41 @@
# NPR MeteoBot
[Русский](README.md) · **English**
[![CI](https://github.com/EDeev/npr_meteobot/actions/workflows/ci.yml/badge.svg)](https://github.com/EDeev/npr_meteobot/actions/workflows/ci.yml)
A Telegram bot for schoolchildren in the Norilsk industrial area: it sends class cancellations due to frost
and wind ("aktirovka") for the user's shift and district and shows the weather forecast.
**Status:** school project (2022), completed, archived
**Stack:** Python 3.10 · aiogram 2 · SQLite · Selenium + BeautifulSoup (city administration site, Gismeteo) · pymorphy2
- Subscription to cancellations by shift (1 or 2) and district: Norilsk, Talnakh, Oganer, Kayerkan
- Regular checks of the administration site and broadcasts to subscribers
- Weather forecast for the day and up to 10 days ahead (Gismeteo data)
## Running
```bash
pip install -r requirements.txt
BOT_TOKEN=token_from_BotFather python bot.py
```
The `sub.db` database is created on first run from `schema.sql`. Cancellations need Chrome: the
administration site is read with Selenium.
## License
School project (2022). The code is open for study; there is no separate 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>

138
README.md
View file

@ -1,139 +1,41 @@
# NPR MeteoBot
Автоматизированная система мониторинга метеорологических условий и уведомлений об актировках учебных занятий для Норильского промышленного района (НПР).
**Русский** · [English](README.en.md)
## 📋 Описание проекта
[![CI](https://github.com/EDeev/npr_meteobot/actions/workflows/ci.yml/badge.svg)](https://github.com/EDeev/npr_meteobot/actions/workflows/ci.yml)
NPR MeteoBot представляет собой интеллектуальную систему на базе Telegram-бота, предназначенную для:
- Мониторинга погодных условий в четырех районах НПР
- Автоматического отслеживания актировок учебных занятий
- Персонализированных уведомлений пользователей
- Предоставления детализированных метеорологических данных
Telegram-бот для школьников Норильского промышленного района: присылает актировки (отмену занятий
из-за мороза и ветра) для своей смены и района и показывает прогноз погоды.
### Основные возможности:
- **Персонализация**: Настройка по смене обучения и району проживания
- **Автоматизация**: Регулярный мониторинг официальных источников
- **Уведомления**: Система подписок с гибкими настройками
- **Актуальность**: Интеграция с Gismeteo и официальным сайтом города
**Статус:** школьный проект (2022), завершён, в архиве
## 🏗️ Архитектура системы
**Стек:** Python 3.10 · aiogram 2 · SQLite · Selenium + BeautifulSoup (сайт администрации, Gismeteo) · pymorphy2
### Компоненты:
- `bot.py` - Основной модуль Telegram-бота с обработкой команд
- `sql.py` - Модуль работы с базой данных SQLite
- `info.py` - Парсинг данных с внешних источников
- `config.py` - Конфигурационные параметры системы
- Подписка на актировки по смене (1 или 2) и району: Норильск, Талнах, Оганер, Кайеркан
- Регулярная проверка сайта администрации и рассылка подписчикам
- Прогноз погоды на день и на 10 дней вперёд (данные Gismeteo)
### Структура базы данных:
- **users**: Профили пользователей с настройками
- **weather**: Кэш метеорологических данных
- **parts**: Данные об актировках по сменам
- **data**: Системная информация
## Запуск
## 🛠️ Технологический стек
### Backend:
- **Python 3.7+** - Основной язык разработки
- **aiogram 2.x** - Фреймворк для Telegram Bot API
- **SQLite** - Локальная база данных
- **asyncio** - Асинхронное программирование
### Парсинг данных:
- **Selenium WebDriver** - Автоматизация браузера
- **BeautifulSoup4** - HTML-парсинг
- **requests** - HTTP-запросы
### Обработка данных:
- **pymorphy2** - Морфологический анализ русского языка
## 🚀 Установка и настройка
### Предварительные требования:
```python
# Установка ChromeDriver для Selenium
# Ubuntu/Debian:
sudo apt-get install chromium-chromedriver
# Windows: загрузить с https://chromedriver.chromium.org/
```
### Установка зависимостей:
```python
```bash
pip install -r requirements.txt
BOT_TOKEN=токен_от_BotFather python bot.py
```
### Конфигурация:
1. Получите токен бота в [@BotFather](https://t.me/BotFather)
2. Обновите `API_TOKEN` в `config.py`
3. Создайте структуру базы данных SQLite
База `sub.db` создаётся при первом запуске из `schema.sql`. Для актировок нужен Chrome: сайт
администрации читается через Selenium.
### Запуск:
```python
python bot.py
```
## Лицензия
## 📊 Функциональные возможности
Школьный проект (2022). Код открыт для изучения, отдельной лицензии нет.
### Команды бота:
- `/start` - Инициализация и настройка профиля
- `/town` - Изменение города и смены обучения
- `/acta` - Просмотр текущих актировок
- `/weather` - Детализированный прогноз погоды
- `/storm` - Штормовые предупреждения
- `/letter` - Управление подпиской на уведомления
- `/notice` - Настройка типа уведомлений
## Автор
### Автоматические процессы:
- **timer()** - Мониторинг актировок (каждую минуту)
- **wind()** - Обновление погодных данных (каждые 20 минут)
- **clear()** - Очистка устаревших данных (каждые 5 минут)
## 🎯 Целевые районы
Система обслуживает четыре района НПР:
- **Центральный район** (Норильск)
- **Район Талнах**
- **Район Оганер**
- **Район Кайеркан**
## 🔧 Конфигурационные параметры
### Временные интервалы мониторинга:
- Прогноз погоды: 1, 4, 7, 10, 13, 16, 19, 22 часа
- Проверка актировок: ежеминутно
- Обновление погоды: каждые 20 минут
### Источники данных:
- **Актировки**: [norilsk-city.ru/meteo](https://www.norilsk-city.ru/meteo/)
- **Погода**: [gismeteo.ru](https://www.gismeteo.ru/)
## 📈 Масштабируемость
Система спроектирована для:
- Обработки 17,000+ активных пользователей
- Асинхронной обработки запросов
- Отказоустойчивой работы с внешними API
## ⚠️ Ограничения
- Зависимость от доступности внешних источников данных
- Требует настройки ChromeDriver для корректного парсинга
- Локальная SQLite база данных (не подходит для кластерного развертывания)
## 📝 Лицензия
Проект разработан для образовательных и некоммерческих целей.
## 👥 Автор
**Деев Егор Викторович** - 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>

524
bot.py
View file

@ -1,262 +1,262 @@
import config, asyncio, info
import logging, pymorphy2, time
from aiogram import Bot, Dispatcher, executor, types
from sql import SQLighter
from datetime import datetime as dt
# задаем уровень логов
logging.basicConfig(level=logging.INFO)
# инициализируем бота
bot = Bot(token=config.API_TOKEN)
dp = Dispatcher(bot)
keyboard = types.InlineKeyboardMarkup()
morph = pymorphy2.MorphAnalyzer()
# инициализируем соединение с БД
db = SQLighter('sub.db')
# СТАРТОВЫЕ КОМАНДЫ
@dp.message_handler(commands=['start'])
async def start(message: types.Message):
if not db.user_exists(message.from_user.id):
# если юзера нет в базе, добавляем его
db.add_user(message.from_user.id)
buttons = [types.InlineKeyboardButton(text="1-я смена", callback_data="prt1"),
types.InlineKeyboardButton(text="2-я смена", callback_data="prt2")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await message.answer(f"Привет, {message.from_user.first_name}!\nЯ - погодный бот, созданный для выдачи информации о"
f" вероятных актировках. Прошу вас выбрать вашу смену обучения!\n\n<i>Для последующей "
f"смены: города или смены обучения воспользуйтесь командой <b>/town</b></i>",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.message_handler(content_types=['sticker'])
async def stick(message: types.Message):
print(message.sticker.file_id)
@dp.message_handler(commands=['town'])
async def town(message: types.Message):
buttons = [types.InlineKeyboardButton(text="1-я смена", callback_data="prt1"),
types.InlineKeyboardButton(text="2-я смена", callback_data="prt2")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await message.answer("<b>Прошу вас выбрать вашу смену обучения!</b>", parse_mode=types.ParseMode.HTML,
reply_markup=keyboard)
# ИЛАЙН КЛАВИАТУРА
@dp.callback_query_handler(text="prt1")
async def part_one(call: types.CallbackQuery):
db.update_user(call.message.chat.id, 3, 1)
buttons = [types.InlineKeyboardButton(text="Норильск", callback_data="nor"),
types.InlineKeyboardButton(text="Талнах", callback_data="tal"),
types.InlineKeyboardButton(text="Оганер", callback_data="oga"),
types.InlineKeyboardButton(text="Каеркан", callback_data="kae")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await call.message.answer(text="<b><i>Отлично!</i></b> Теперь вам надо выбрать район в котором вы проживаете "
"на территории НПР!",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.callback_query_handler(text="prt2")
async def part_two(call: types.CallbackQuery):
db.update_user(call.message.chat.id, 3, 2)
buttons = [types.InlineKeyboardButton(text="Норильск", callback_data="nor"),
types.InlineKeyboardButton(text="Талнах", callback_data="tal"),
types.InlineKeyboardButton(text="Оганер", callback_data="oga"),
types.InlineKeyboardButton(text="Каеркан", callback_data="kae")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await call.message.answer(text="<b><i>Конечно я вам не завидую, но куда деваться)</i></b> Теперь вам надо выбрать "
"район в котором вы проживаете на территории НПР!",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.callback_query_handler(text=["nor", "tal", "oga", "kae"])
async def function(call: types.CallbackQuery):
txt = call.data
if txt == "nor":
db.update_user(call.message.chat.id, 4, 1)
elif txt == "tal":
db.update_user(call.message.chat.id, 4, 2)
elif txt == "oga":
db.update_user(call.message.chat.id, 4, 3)
elif txt == "kae":
db.update_user(call.message.chat.id, 4, 4)
await call.message.answer(text='<b>Прекрасно!</b> Вы заполнили данные, теперь если вы хотите подписаться на '
'утреннюю рассылку об актировках воспользуйтесь командой <b>/letter</b>.\n\n'
'<i><b>Для полного ознакомления можете воспользоваться командой /help</b></i>',
parse_mode=types.ParseMode.HTML)
# КОМАНДЫ АКТИВАЦИИ ПОДПИСОК
@dp.message_handler(commands=['notice'])
async def notification(message: types.Message):
if str(message.chat.id) not in db.get_send():
db.update_user(message.chat.id, 2, 1)
await message.answer("Увидомление приходит каждый день не зависимо от наличия актировки!")
else:
db.update_user(message.chat.id, 2, 0)
await message.answer("Уведомление приходит, только если есть актировка!")
@dp.message_handler(commands=['letter'])
async def letter(message: types.Message):
if str(message.chat.id) not in db.get_subs():
db.update_user(message.chat.id, 1, 1)
await message.answer("Вы подписались на рассылку!")
else:
db.update_user(message.chat.id, 1, 0)
await message.answer("Вы отписались от расслыки!")
@dp.message_handler(commands=['storm'])
async def storm(message: types.Message):
await message.answer(info.storm(), parse_mode=types.ParseMode.MARKDOWN)
@dp.message_handler(commands=['acta'])
async def act(message: types.Message):
mail = f"<b>Актировка за {'.'.join(reversed(str(db.get_data()[0]).split('-')))}</b>\n"
for part in [1, 2]:
mail = mail + f"\n<b>Для {part}-ой смены:</b>\n"
if db.check_acta(part):
cities = config.NAMES
city = db.get_acta(part)
for i, town in enumerate(city):
if str(town) != "нет":
mail = mail + f"<b>*</b> В {morph.parse(cities[i])[0].inflect({'loct'}).word.title()} актировка " \
f"с 1 по {town} классы!\n"
else:
mail = mail + f"<b>*</b> В {morph.parse(cities[i])[0].inflect({'loct'}).word.title()} актировки нет!\n"
else:
mail = mail + "На данный момент информации на сайте нету!\n\n"
await message.answer(mail, parse_mode=types.ParseMode.HTML)
@dp.message_handler(commands=['weather'])
async def storm(message: types.Message):
text = message.text.split()
try:
day = int(text[1])
if 0 <= day <= 9:
weather = db.get_weather(day)
txt = "<b>Погода:</b>\n\n"
for hour in weather:
dat = weather[hour]
if len(dat[1]) == 2:
par = f"{dat[1][0]}-{dat[1][1]}"
else:
par = str(dat[1][0])
txt = txt + f"<b>В {hour}:</b>\n Температура: {dat[0]}\n Скорость и порывы ветра: {par}\n Направление ветра: {dat[2]}\n Осадки: {dat[3]}\n\n"
await message.answer(txt, parse_mode=types.ParseMode.HTML)
else:
raise NameError('Число не вподходящем диапозоне!')
except Exception:
await message.answer("Введи число дня по счёту от сегодня, погода которого тебе нужна!\n\nТо есть:\nСегодня - {/waether 0}\nЗавтра - {/weather 1}",
parse_mode=types.ParseMode.MARKDOWN)
@dp.message_handler(commands=['help'])
async def main(message: types.Message):
await message.answer('*Инструкция использования:*\n\n'
'*/town* - с помощью этой команды вы можете изменить данные о городе и смене обучения\n'
'*/acta* - команда, выдающая данные об актировке на все города на данный момент\n'
'*/storm* - команда, выдающая данные о штормовом предупреждении\n'
'*/letter* - команда, позваляющая подписаться на утреннюю рассылку о наличии актировки\n'
'*/notice* - команда, дающая выбор - получать уведомления в любом случае или только при '
'наличии актировки\n', types.ParseMode.MARKDOWN)
# УВЕДОМЛЕНИЕ
async def timer(wait_for):
while True:
await asyncio.sleep(wait_for)
city, part = info.act()
if city:
date_old, part_old = db.get_data()
date_now, part_num = str(dt.now().date()), int(part[0])
if (part_num == 1 and date_old != date_now) or (part_num == 2 and date_old == date_now):
if part_old != part_num:
db.update_data(dt.now().date(), part_num)
db.save_acta(city, part_num)
for twn in city:
town = morph.parse(twn[0][1])[0].inflect({'datv'}).word
if isinstance(twn[1], int):
for s in db.get_users(part_num, twn[0][0]):
try:
await bot.send_sticker(chat_id=s, sticker=config.ID_STICKERS + config.STICKERS[1])
await bot.send_message(chat_id=s, text=f"*Для {part} по {town.title()} объявлена актировка c 1 по {twn[1]} классы!*",
parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
print(repr(e))
db.update_status(s)
else:
for s in db.get_sends(part_num, twn[0][0]):
try:
await bot.send_sticker(chat_id=s, sticker=config.ID_STICKERS + config.STICKERS[6])
await bot.send_message(chat_id=s, text=f"*Для {part} по {town.title()} актировка отсутствует!",
parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
print(repr(e))
db.update_status(s)
print("---------------------------------")
# ОБНОВЛЕНИЕ ДАННЫХ О ПОГОДЫ
async def wind(wait_for):
while True:
await asyncio.sleep(wait_for)
for moment in db.get_moments():
weather = info.weather(moment)
if type(weather) == dict:
for tm in weather:
db.update_weather(weather[tm], moment, tm)
else:
print(weather)
time.sleep(10)
# ОЧИСТКА ДАННЫХ ОБ АКТИРОВКАХ
async def clear(wait_for):
while True:
await asyncio.sleep(wait_for)
if (db.get_data()[0] != str(dt.now().date())) and (db.check_acta(1) or db.check_acta(2)):
db.del_acta()
# ЗАПУСК ЛОНГ ПОЛЛИНГА
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.create_task(timer(60)) # ПРОВЕРКА МЕТЕО НОРИЛЬСК КАЖДУЮ МИНУТУ
loop.create_task(clear(300)) # ПРОВЕРКА НА НОВЫЙ ДЕНЬ КАЖДЫЕ 10 МИНУТ
loop.create_task(wind(1200)) # ПРОВЕРКА НА ОБНОВЛЕНИЕ ПОГОДЫ КАЖДЫЕ 20 МИНУТ
executor.start_polling(dp, skip_updates=True)
import config, asyncio, info
import logging, pymorphy2, time
from aiogram import Bot, Dispatcher, executor, types
from sql import SQLighter
from datetime import datetime as dt
# задаем уровень логов
logging.basicConfig(level=logging.INFO)
# инициализируем бота
bot = Bot(token=config.API_TOKEN)
dp = Dispatcher(bot)
keyboard = types.InlineKeyboardMarkup()
morph = pymorphy2.MorphAnalyzer()
# инициализируем соединение с БД
db = SQLighter('sub.db')
# СТАРТОВЫЕ КОМАНДЫ
@dp.message_handler(commands=['start'])
async def start(message: types.Message):
if not db.user_exists(message.from_user.id):
# если юзера нет в базе, добавляем его
db.add_user(message.from_user.id)
buttons = [types.InlineKeyboardButton(text="1-я смена", callback_data="prt1"),
types.InlineKeyboardButton(text="2-я смена", callback_data="prt2")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await message.answer(f"Привет, {message.from_user.first_name}!\nЯ - погодный бот, созданный для выдачи информации о"
f" вероятных актировках. Прошу вас выбрать вашу смену обучения!\n\n<i>Для последующей "
f"смены: города или смены обучения воспользуйтесь командой <b>/town</b></i>",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.message_handler(content_types=['sticker'])
async def stick(message: types.Message):
print(message.sticker.file_id)
@dp.message_handler(commands=['town'])
async def town(message: types.Message):
buttons = [types.InlineKeyboardButton(text="1-я смена", callback_data="prt1"),
types.InlineKeyboardButton(text="2-я смена", callback_data="prt2")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await message.answer("<b>Прошу вас выбрать вашу смену обучения!</b>", parse_mode=types.ParseMode.HTML,
reply_markup=keyboard)
# ИЛАЙН КЛАВИАТУРА
@dp.callback_query_handler(text="prt1")
async def part_one(call: types.CallbackQuery):
db.update_user(call.message.chat.id, 3, 1)
buttons = [types.InlineKeyboardButton(text="Норильск", callback_data="nor"),
types.InlineKeyboardButton(text="Талнах", callback_data="tal"),
types.InlineKeyboardButton(text="Оганер", callback_data="oga"),
types.InlineKeyboardButton(text="Каеркан", callback_data="kae")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await call.message.answer(text="<b><i>Отлично!</i></b> Теперь вам надо выбрать район в котором вы проживаете "
"на территории НПР!",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.callback_query_handler(text="prt2")
async def part_two(call: types.CallbackQuery):
db.update_user(call.message.chat.id, 3, 2)
buttons = [types.InlineKeyboardButton(text="Норильск", callback_data="nor"),
types.InlineKeyboardButton(text="Талнах", callback_data="tal"),
types.InlineKeyboardButton(text="Оганер", callback_data="oga"),
types.InlineKeyboardButton(text="Каеркан", callback_data="kae")]
keyboard = types.InlineKeyboardMarkup(row_width=2)
keyboard.add(*buttons)
await call.message.answer(text="<b><i>Конечно я вам не завидую, но куда деваться)</i></b> Теперь вам надо выбрать "
"район в котором вы проживаете на территории НПР!",
parse_mode=types.ParseMode.HTML, reply_markup=keyboard)
@dp.callback_query_handler(text=["nor", "tal", "oga", "kae"])
async def function(call: types.CallbackQuery):
txt = call.data
if txt == "nor":
db.update_user(call.message.chat.id, 4, 1)
elif txt == "tal":
db.update_user(call.message.chat.id, 4, 2)
elif txt == "oga":
db.update_user(call.message.chat.id, 4, 3)
elif txt == "kae":
db.update_user(call.message.chat.id, 4, 4)
await call.message.answer(text='<b>Прекрасно!</b> Вы заполнили данные, теперь если вы хотите подписаться на '
'утреннюю рассылку об актировках воспользуйтесь командой <b>/letter</b>.\n\n'
'<i><b>Для полного ознакомления можете воспользоваться командой /help</b></i>',
parse_mode=types.ParseMode.HTML)
# КОМАНДЫ АКТИВАЦИИ ПОДПИСОК
@dp.message_handler(commands=['notice'])
async def notification(message: types.Message):
if str(message.chat.id) not in db.get_send():
db.update_user(message.chat.id, 2, 1)
await message.answer("Увидомление приходит каждый день не зависимо от наличия актировки!")
else:
db.update_user(message.chat.id, 2, 0)
await message.answer("Уведомление приходит, только если есть актировка!")
@dp.message_handler(commands=['letter'])
async def letter(message: types.Message):
if str(message.chat.id) not in db.get_subs():
db.update_user(message.chat.id, 1, 1)
await message.answer("Вы подписались на рассылку!")
else:
db.update_user(message.chat.id, 1, 0)
await message.answer("Вы отписались от расслыки!")
@dp.message_handler(commands=['storm'])
async def storm(message: types.Message):
await message.answer(info.storm(), parse_mode=types.ParseMode.MARKDOWN)
@dp.message_handler(commands=['acta'])
async def act(message: types.Message):
mail = f"<b>Актировка за {'.'.join(reversed(str(db.get_data()[0]).split('-')))}</b>\n"
for part in [1, 2]:
mail = mail + f"\n<b>Для {part}-ой смены:</b>\n"
if db.check_acta(part):
cities = config.NAMES
city = db.get_acta(part)
for i, town in enumerate(city):
if str(town) != "нет":
mail = mail + f"<b>*</b> В {morph.parse(cities[i])[0].inflect({'loct'}).word.title()} актировка " \
f"с 1 по {town} классы!\n"
else:
mail = mail + f"<b>*</b> В {morph.parse(cities[i])[0].inflect({'loct'}).word.title()} актировки нет!\n"
else:
mail = mail + "На данный момент информации на сайте нету!\n\n"
await message.answer(mail, parse_mode=types.ParseMode.HTML)
@dp.message_handler(commands=['weather'])
async def weather_command(message: types.Message):
text = message.text.split()
try:
day = int(text[1])
if 0 <= day <= 9:
weather = db.get_weather(day)
txt = "<b>Погода:</b>\n\n"
for hour in weather:
dat = weather[hour]
if len(dat[1]) == 2:
par = f"{dat[1][0]}-{dat[1][1]}"
else:
par = str(dat[1][0])
txt = txt + f"<b>В {hour}:</b>\n Температура: {dat[0]}\n Скорость и порывы ветра: {par}\n Направление ветра: {dat[2]}\n Осадки: {dat[3]}\n\n"
await message.answer(txt, parse_mode=types.ParseMode.HTML)
else:
raise NameError('Число не вподходящем диапозоне!')
except Exception:
await message.answer("Введи число дня по счёту от сегодня, погода которого тебе нужна!\n\nТо есть:\nСегодня - {/waether 0}\nЗавтра - {/weather 1}",
parse_mode=types.ParseMode.MARKDOWN)
@dp.message_handler(commands=['help'])
async def main(message: types.Message):
await message.answer('*Инструкция использования:*\n\n'
'*/town* - с помощью этой команды вы можете изменить данные о городе и смене обучения\n'
'*/acta* - команда, выдающая данные об актировке на все города на данный момент\n'
'*/storm* - команда, выдающая данные о штормовом предупреждении\n'
'*/letter* - команда, позваляющая подписаться на утреннюю рассылку о наличии актировки\n'
'*/notice* - команда, дающая выбор - получать уведомления в любом случае или только при '
'наличии актировки\n', types.ParseMode.MARKDOWN)
# УВЕДОМЛЕНИЕ
async def timer(wait_for):
while True:
await asyncio.sleep(wait_for)
city, part = info.act()
if city:
date_old, part_old = db.get_data()
date_now, part_num = str(dt.now().date()), int(part[0])
if (part_num == 1 and date_old != date_now) or (part_num == 2 and date_old == date_now):
if part_old != part_num:
db.update_data(dt.now().date(), part_num)
db.save_acta(city, part_num)
for twn in city:
town = morph.parse(twn[0][1])[0].inflect({'datv'}).word
if isinstance(twn[1], int):
for s in db.get_users(part_num, twn[0][0]):
try:
await bot.send_sticker(chat_id=s, sticker=config.ID_STICKERS + config.STICKERS[1])
await bot.send_message(chat_id=s, text=f"*Для {part} по {town.title()} объявлена актировка c 1 по {twn[1]} классы!*",
parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
print(repr(e))
db.update_status(s)
else:
for s in db.get_sends(part_num, twn[0][0]):
try:
await bot.send_sticker(chat_id=s, sticker=config.ID_STICKERS + config.STICKERS[6])
await bot.send_message(chat_id=s, text=f"*Для {part} по {town.title()} актировка отсутствует!",
parse_mode=types.ParseMode.MARKDOWN)
except Exception as e:
print(repr(e))
db.update_status(s)
print("---------------------------------")
# ОБНОВЛЕНИЕ ДАННЫХ О ПОГОДЫ
async def wind(wait_for):
while True:
await asyncio.sleep(wait_for)
for moment in db.get_moments():
weather = info.weather(moment)
if type(weather) == dict:
for tm in weather:
db.update_weather(weather[tm], moment, tm)
else:
print(weather)
time.sleep(10)
# ОЧИСТКА ДАННЫХ ОБ АКТИРОВКАХ
async def clear(wait_for):
while True:
await asyncio.sleep(wait_for)
if (db.get_data()[0] != str(dt.now().date())) and (db.check_acta(1) or db.check_acta(2)):
db.del_acta()
# ЗАПУСК ЛОНГ ПОЛЛИНГА
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.create_task(timer(60)) # ПРОВЕРКА МЕТЕО НОРИЛЬСК КАЖДУЮ МИНУТУ
loop.create_task(clear(300)) # ПРОВЕРКА НА НОВЫЙ ДЕНЬ КАЖДЫЕ 10 МИНУТ
loop.create_task(wind(1200)) # ПРОВЕРКА НА ОБНОВЛЕНИЕ ПОГОДЫ КАЖДЫЕ 20 МИНУТ
executor.start_polling(dp, skip_updates=True)

View file

@ -1,19 +1,21 @@
API_TOKEN = 'token from BotFather' # ОСНОВНОЙ
# API_TOKEN = 'token from BotFather' # ТЕСТЫ
CITY = {'Центральный район': [1, 'Норильск'], 'район Талнах': [2, 'Талнах'],
'ж/о Оганер': [3, 'Оганер'], 'район Кайеркан': [4, 'Кайеркан']}
NAMES = ['Норильск', 'Талнах', 'Оганер', 'Кайеркан']
TIME = [1, 4, 7, 10, 13, 16, 19, 22]
ID_STICKERS = r"CAACAgIAAxkBAAID"
STICKERS = {"пасмурно": r"2mIKSBBtFNR1uHLhdztkhVwqTxRRAAKHEgACljNZSAgtQYHfmE4WIwQ", # облако
"поземок": r"22IKSBIdUSxApfTAEk5mMjldDexZAAIPFwAClIBRSCSyRnG71n6CIwQ", # солнце и облако
1: r"3GIKSBRW4sCgYX8AAaq01U6DaNF1IwACXhMAAlzDUUirRDtYEkGJ3iME", # ветер
4: r"3WIKSBec5Ch8vL-czNkXikARo-JmAALmFAACGHZRSHmiNc0n7VZPIwQ", # гроза
"снег": r"3mIKSBpHLfkJOnI4HCQny_a_XSVbAAI8FAACSixRSN7lm05cAAEj0yME", # снегопад
6: r"32IKSByANsmxbok8VePu7k6mTzexAALMFgACVopQSHXJa0spfMe7IwQ"} # ясно
# await message.answer_sticker(ID_STICKERS + STICKERS[1])
import os
# Токен бота от @BotFather — из переменной окружения BOT_TOKEN
API_TOKEN = os.environ.get('BOT_TOKEN', '')
CITY = {'Центральный район': [1, 'Норильск'], 'район Талнах': [2, 'Талнах'],
'ж/о Оганер': [3, 'Оганер'], 'район Кайеркан': [4, 'Кайеркан']}
NAMES = ['Норильск', 'Талнах', 'Оганер', 'Кайеркан']
TIME = [1, 4, 7, 10, 13, 16, 19, 22]
ID_STICKERS = r"CAACAgIAAxkBAAID"
STICKERS = {"пасмурно": r"2mIKSBBtFNR1uHLhdztkhVwqTxRRAAKHEgACljNZSAgtQYHfmE4WIwQ", # облако
"поземок": r"22IKSBIdUSxApfTAEk5mMjldDexZAAIPFwAClIBRSCSyRnG71n6CIwQ", # солнце и облако
1: r"3GIKSBRW4sCgYX8AAaq01U6DaNF1IwACXhMAAlzDUUirRDtYEkGJ3iME", # ветер
4: r"3WIKSBec5Ch8vL-czNkXikARo-JmAALmFAACGHZRSHmiNc0n7VZPIwQ", # гроза
"снег": r"3mIKSBpHLfkJOnI4HCQny_a_XSVbAAI8FAACSixRSN7lm05cAAEj0yME", # снегопад
6: r"32IKSByANsmxbok8VePu7k6mTzexAALMFgACVopQSHXJa0spfMe7IwQ"} # ясно
# await message.answer_sticker(ID_STICKERS + STICKERS[1])

152
info.py
View file

@ -1,77 +1,77 @@
# - * - coding: utf-8 - * -
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
import requests, config
def act():
try:
soup = BeautifulSoup(requests.get('https://www.norilsk-city.ru/meteo/').text, 'html.parser')
acta = [list(map(lambda x: x.get_text(), _.find_all("p"))) for _ in soup.find_all("tbody")[0] if _ != "\n"]
count = '1-ой смены' if "1" in str(soup.find_all("div",
class_="textContent",
limit=1)[0].find_all("p",
limit=1)[0].get_text()) else '2-ой смены'
for lst in acta:
if "нет" in lst[1]:
acta[acta.index(lst)] = [config.CITY[lst[0]], "нет"]
else:
acta[acta.index(lst)] = [config.CITY[lst[0]], int(lst[1].split("-")[1].split()[0])]
return sorted(acta), count
except Exception as e:
return [], ""
def storm():
soup = BeautifulSoup(requests.get('https://www.norilsk-city.ru/meteo/').text, 'html.parser')
message = str(str(soup.find_all("p")[13])[35:-40])
stm = "*" + str(str(soup.find_all("h2")[1])[4:-5]) + \
"*\n\n" + message[:-67] + " " + message[96:-36] + message[118:-24] + message[129:-1]
return stm
def weather(moment):
try:
# путь к драйверу chrome
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument('--no-sandbox')
chromedriver = 'chromedriver'
browser = webdriver.Chrome(executable_path=chromedriver, chrome_options=chrome_options)
# После успешного входа в систему переходим на страницу «Gismeteo»
browser.get(f'https://www.gismeteo.ru/weather-norilsk-3957/{moment}')
# Получение HTML-содержимого
requiredHtml = browser.page_source
soup = BeautifulSoup(requiredHtml, 'html.parser')
# weather = soup.find_all()
temp = soup.find_all("span", class_="unit unit_temperature_c")[6:]
wind = soup.find_all("span", class_=["wind-unit unit unit_wind_m_s", "wind-unit unit unit_wind_m_s warning"])[:8]
dire = soup.find_all("div", class_="direction")
prec = soup.find_all("div", class_="item-unit")
k, s, weather = 0, 0, {}
for i in range(8):
par = wind[i].get_text().split()[0].split('-')
if len(par) == 1:
par = [int(par[0])]
else:
par = [int(par[0]), int(par[1])]
pre = prec[i].get_text()
# pre = prec[i].get_text().split('\n')
# precti = pre[1].split(' ')[-1] if len(pre) > 1 else pre[0]
weather[config.TIME[i]] = [temp[i].get_text(), par, dire[i].get_text(), pre]
return weather
except Exception as e:
# - * - coding: utf-8 - * -
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
import requests, config
def act():
try:
soup = BeautifulSoup(requests.get('https://www.norilsk-city.ru/meteo/').text, 'html.parser')
acta = [list(map(lambda x: x.get_text(), _.find_all("p"))) for _ in soup.find_all("tbody")[0] if _ != "\n"]
count = '1-ой смены' if "1" in str(soup.find_all("div",
class_="textContent",
limit=1)[0].find_all("p",
limit=1)[0].get_text()) else '2-ой смены'
for lst in acta:
if "нет" in lst[1]:
acta[acta.index(lst)] = [config.CITY[lst[0]], "нет"]
else:
acta[acta.index(lst)] = [config.CITY[lst[0]], int(lst[1].split("-")[1].split()[0])]
return sorted(acta), count
except Exception:
return [], ""
def storm():
soup = BeautifulSoup(requests.get('https://www.norilsk-city.ru/meteo/').text, 'html.parser')
message = str(str(soup.find_all("p")[13])[35:-40])
stm = "*" + str(str(soup.find_all("h2")[1])[4:-5]) + \
"*\n\n" + message[:-67] + " " + message[96:-36] + message[118:-24] + message[129:-1]
return stm
def weather(moment):
try:
# путь к драйверу chrome
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument('--no-sandbox')
chromedriver = 'chromedriver'
browser = webdriver.Chrome(executable_path=chromedriver, chrome_options=chrome_options)
# После успешного входа в систему переходим на страницу «Gismeteo»
browser.get(f'https://www.gismeteo.ru/weather-norilsk-3957/{moment}')
# Получение HTML-содержимого
requiredHtml = browser.page_source
soup = BeautifulSoup(requiredHtml, 'html.parser')
# weather = soup.find_all()
temp = soup.find_all("span", class_="unit unit_temperature_c")[6:]
wind = soup.find_all("span", class_=["wind-unit unit unit_wind_m_s", "wind-unit unit unit_wind_m_s warning"])[:8]
dire = soup.find_all("div", class_="direction")
prec = soup.find_all("div", class_="item-unit")
weather = {}
for i in range(8):
par = wind[i].get_text().split()[0].split('-')
if len(par) == 1:
par = [int(par[0])]
else:
par = [int(par[0]), int(par[1])]
pre = prec[i].get_text()
# pre = prec[i].get_text().split('\n')
# precti = pre[1].split(' ')[-1] if len(pre) > 1 else pre[0]
weather[config.TIME[i]] = [temp[i].get_text(), par, dire[i].get_text(), pre]
return weather
except Exception as e:
return [repr(e)]

6
requirements.txt Normal file
View file

@ -0,0 +1,6 @@
aiogram==2.25.2
pymorphy2==0.9.1
pymorphy2-dicts-ru==2.4.417127.4579844
selenium==4.15.2
beautifulsoup4==4.12.3
requests==2.31.0

15
schema.sql Normal file
View file

@ -0,0 +1,15 @@
-- Схема базы бота и служебные строки, которые он ожидает (подписчиков нет)
CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, id_user VARCHAR (255) NOT NULL,
rating INTEGER (4) NOT NULL DEFAULT (0), status BOOLEAN NOT NULL DEFAULT (False),
send BOOLEAN NOT NULL DEFAULT (True), part INTEGER (1) NOT NULL, city INTEGER (1) NOT NULL);
CREATE TABLE IF NOT EXISTS parts (part INTEGER PRIMARY KEY AUTOINCREMENT, nor VARCHAR (3), tal VARCHAR (3),
oga VARCHAR (3), kae VARCHAR (3));
CREATE TABLE IF NOT EXISTS weather (id STRING, "1" VARCHAR, "4" VARCHAR, "7" VARCHAR, "10" VARCHAR,
"13" VARCHAR, "16" VARCHAR, "19" VARCHAR, "22" VARCHAR);
CREATE TABLE IF NOT EXISTS data (id INTEGER PRIMARY KEY AUTOINCREMENT, date DATE NOT NULL, part INTEGER (2) NOT NULL,
weath STRING);
INSERT INTO parts (part, nor, tal, oga, kae) VALUES (1, '11', '11', '11', '11'), (2, '11', '11', '11', '11');
INSERT INTO weather (id) VALUES (' '), ('tomorrow'), ('3-day'), ('4-day'), ('5-day'), ('6-day'), ('7-day'),
('8-day'), ('9-day'), ('10-day');
INSERT INTO data (id, date, part) VALUES (1, '2022-02-24', 1);

11
sql.py
View file

@ -1,11 +1,22 @@
import os
import sqlite3
SCHEMA = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'schema.sql')
class SQLighter:
def __init__(self, database):
"""Подключаемся к БД и сохраняем курсор соединения"""
self.connection = sqlite3.connect(database)
self.cursor = self.connection.cursor()
self._create_if_empty()
def _create_if_empty(self):
"""Новая база: создаём таблицы и служебные строки из schema.sql"""
exists = self.cursor.execute("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users'").fetchone()
if not exists:
with open(SCHEMA, encoding='utf-8') as f:
self.connection.executescript(f.read())
def user_exists(self, id_user):
"""Проверяем, есть ли уже пользователь в базе"""

BIN
sub.db

Binary file not shown.