forust
fb43306571
Dockerfile fixes: - edu_master/phpsessid-bot: add --no-install-recommends, pin pip versions - edu_master/webinar-checker: pin pip versions with --no-cache-dir - userbot: add SHELL with pipefail for pipe operations Ruff fixes (173 → 0): - W293/W291/W292: whitespace clean via ruff format - N806: camelCase → snake_case (anilist, safone, hearts, flux, etc.) - ARG001/ARG002: prefix unused params with _ - SIM115: use context managers for file I/O - SIM117: combine nested with statements - S608: noqa on SQL f-strings (module name is validated) - E402/N812/N817: import fixes - B023: pass loop variable as argument - I001: auto-sorted imports - syntax: fixed = vs == in dtek_notif/main.py
749 lines
30 KiB
Python
749 lines
30 KiB
Python
import asyncio
|
||
import contextlib
|
||
import logging
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
|
||
import requests
|
||
from aiogram import Bot, Dispatcher
|
||
from aiogram.filters import Command
|
||
from aiogram.types import KeyboardButton, Message
|
||
from aiogram.utils.keyboard import ReplyKeyboardBuilder
|
||
from bs4 import BeautifulSoup
|
||
from dotenv import load_dotenv
|
||
|
||
# Загрузка переменных окружения
|
||
load_dotenv()
|
||
|
||
# Настройки
|
||
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN', 'YOUR_TOKEN_HERE')
|
||
ALLOWED_CHAT_IDS = list(map(int, os.getenv('ALLOWED_CHAT_IDS', '').split(','))) if os.getenv('ALLOWED_CHAT_IDS') else []
|
||
CHECK_INTERVAL = int(os.getenv('CHECK_INTERVAL', '120'))
|
||
|
||
# Параметры для запроса
|
||
VOE_CITY_ID = int(os.getenv('VOE_CITY_ID', 'VOE_CITY_ID'))
|
||
VOE_STREET_ID = int(os.getenv('VOE_STREET_ID', 'VOE_STREET_ID'))
|
||
VOE_HOUSE_ID = int(os.getenv('VOE_HOUSE_ID', 'VOE_HOUSE_ID'))
|
||
|
||
# Настройка логирования
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Глобальные переменные
|
||
bot = Bot(token=TELEGRAM_TOKEN)
|
||
dp = Dispatcher()
|
||
last_schedule: list[dict] | None = None
|
||
last_notification_time: dict[str, datetime] = {}
|
||
|
||
|
||
# ============================================================================
|
||
# УТИЛИТЫ
|
||
# ============================================================================
|
||
|
||
|
||
def format_time_duration(minutes: int) -> str:
|
||
"""Форматирует время из минут в часы и минуты"""
|
||
hours = minutes // 60
|
||
mins = minutes % 60
|
||
|
||
if hours == 0:
|
||
return f'{mins}м'
|
||
elif mins == 0:
|
||
return f'{hours}ч'
|
||
return f'{hours}ч {mins}м'
|
||
|
||
|
||
def get_day_statistics(day_blocks: list[dict]) -> dict[str, int]:
|
||
"""Получает статистику по дню"""
|
||
total_minutes = 0
|
||
confirmed_minutes = 0
|
||
possible_minutes = 0
|
||
|
||
for block in day_blocks:
|
||
for half in [block['first_half'], block['second_half']]:
|
||
if half['status'] == 'off':
|
||
total_minutes += 30
|
||
if half['confirmed']:
|
||
confirmed_minutes += 30
|
||
else:
|
||
possible_minutes += 30
|
||
|
||
return {'total': total_minutes, 'confirmed': confirmed_minutes, 'possible': possible_minutes}
|
||
|
||
|
||
# ============================================================================
|
||
# ПАРСИНГ ДАННЫХ
|
||
# ============================================================================
|
||
|
||
|
||
def parse_html(html: str) -> list[dict]:
|
||
"""Парсит HTML с графиком отключений (логика от 15.11.2024)"""
|
||
soup = BeautifulSoup(html, 'html.parser')
|
||
cells = soup.select('.disconnection-detailed-table-cell.cell')
|
||
|
||
schedule = []
|
||
current_hour = 0
|
||
current_day = 0
|
||
|
||
for cell in cells:
|
||
if 'legend' in cell.get('class', []) or 'head' in cell.get('class', []):
|
||
continue
|
||
|
||
cell_classes = cell.get('class', [])
|
||
|
||
# ПРоверка статуса отключения на весь час
|
||
full_hour_off = 'has_disconnection' in cell_classes and 'full_hour' in cell_classes
|
||
|
||
hour_block = cell.select_one('.hour_block')
|
||
if not hour_block:
|
||
continue
|
||
|
||
# Проверка подтверждённости отключения для всего часа
|
||
cell_confirmed = None
|
||
if 'confirm_1' in cell_classes:
|
||
cell_confirmed = True
|
||
elif 'confirm_0' in cell_classes:
|
||
cell_confirmed = False
|
||
|
||
# Проверка половин часа
|
||
left = hour_block.select_one('.half.left')
|
||
right = hour_block.select_one('.half.right')
|
||
|
||
def parse_half(half, is_full_hour_off: bool, cell_confirmed: bool | None = None) -> dict:
|
||
"""Парсит половину часа"""
|
||
if not half:
|
||
return {'status': 'on', 'queue': None, 'confirmed': None}
|
||
|
||
half_classes = half.get('class', [])
|
||
|
||
# Если вся ячейка full_hour - используем статус ячейки
|
||
if is_full_hour_off:
|
||
return {'status': 'off', 'queue': None, 'confirmed': cell_confirmed}
|
||
|
||
# Определяем статус половины
|
||
if 'has_disconnection' in half_classes:
|
||
status = 'off'
|
||
elif 'no_disconnection' in half_classes:
|
||
status = 'on'
|
||
else:
|
||
status = 'on' # По умолчанию считаем включенным
|
||
|
||
# Если выключено - ищем подробности
|
||
queue = None
|
||
confirmed = None
|
||
|
||
if status == 'off':
|
||
disconnection_div = half.select_one('.disconnection')
|
||
if disconnection_div:
|
||
# Ищем номер черги в title
|
||
if disconnection_div.has_attr('title'):
|
||
title = disconnection_div['title']
|
||
if 'Номер черги' in title or 'Номер черги:' in title:
|
||
with contextlib.suppress(BaseException):
|
||
queue = title.split(':')[-1].strip()
|
||
|
||
# Определяем подтверждение
|
||
disc_classes = disconnection_div.get('class', [])
|
||
if 'disconnection_confirm_1' in disc_classes:
|
||
confirmed = True
|
||
elif 'disconnection_confirm_0' in disc_classes:
|
||
confirmed = False
|
||
|
||
return {'status': status, 'queue': queue, 'confirmed': confirmed}
|
||
|
||
first_half_data = parse_half(left, full_hour_off, cell_confirmed)
|
||
second_half_data = parse_half(right, full_hour_off, cell_confirmed)
|
||
|
||
schedule.append(
|
||
{
|
||
'hour': current_hour,
|
||
'day': current_day,
|
||
'first_half': first_half_data,
|
||
'second_half': second_half_data,
|
||
}
|
||
)
|
||
|
||
current_hour += 1
|
||
if current_hour >= 24:
|
||
current_hour = 0
|
||
current_day += 1
|
||
|
||
return schedule
|
||
|
||
|
||
def get_voe_html(city_id: int, street_id: int, house_id: int) -> str:
|
||
"""Получает HTML с сайта VOE"""
|
||
url = 'https://www.voe.com.ua/disconnection/detailed?ajax_form=1&_wrapper_format=drupal_ajax'
|
||
headers = {
|
||
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
|
||
'X-Requested-With': 'XMLHttpRequest',
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
||
}
|
||
data = {
|
||
'search_type': 0,
|
||
'city_id': city_id,
|
||
'street_id': street_id,
|
||
'house_id': house_id,
|
||
'form_build_id': 'form-Irv5aHw1R2FT_Ik2apyHOZ47hTH5xPNH_LQnBrmpSTc',
|
||
'form_id': 'disconnection_detailed_search_form',
|
||
'_triggering_element_name': 'search',
|
||
'_triggering_element_value': 'Показати',
|
||
'_drupal_ajax': 1,
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, headers=headers, data=data, timeout=10)
|
||
response.raise_for_status()
|
||
resp_json = response.json()
|
||
|
||
insert_html = next((item['data'] for item in resp_json if item.get('command') == 'insert'), None)
|
||
|
||
if not insert_html:
|
||
raise ValueError('HTML не найден в ответе')
|
||
|
||
return insert_html
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f'Ошибка запроса VOE: {e}')
|
||
raise
|
||
|
||
|
||
# ============================================================================
|
||
# ФОРМАТИРОВАНИЕ СООБЩЕНИЙ
|
||
# ============================================================================
|
||
|
||
|
||
def get_main_keyboard():
|
||
"""Создает главную клавиатуру"""
|
||
builder = ReplyKeyboardBuilder()
|
||
builder.row(KeyboardButton(text='📊 Графік'), KeyboardButton(text='🔄 Оновити'))
|
||
builder.row(KeyboardButton(text='📅 Сьогодні'), KeyboardButton(text='📅 Завтра'))
|
||
builder.row(KeyboardButton(text='ℹ️ Про бота'))
|
||
return builder.as_markup(resize_keyboard=True)
|
||
|
||
|
||
def format_schedule_message(schedule: list[dict], days_to_show: int = 2) -> str:
|
||
"""Форматирует полный график на несколько дней"""
|
||
lines = [
|
||
'⚡️ <b>Графік відключень світла</b>',
|
||
f'🕐 Оновлено: {datetime.now().strftime("%d.%m.%Y %H:%M:%S")}',
|
||
'─' * 30,
|
||
'',
|
||
]
|
||
|
||
start_date = datetime.now()
|
||
|
||
for day in range(min(days_to_show, 2)):
|
||
day_blocks = [b for b in schedule if b['day'] == day]
|
||
if not day_blocks:
|
||
continue
|
||
|
||
date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y')
|
||
day_name = '🌅 <b>Сьогодні</b>' if day == 0 else '🌄 <b>Завтра</b>'
|
||
|
||
lines.append(f'{day_name} ({date_str})')
|
||
|
||
# Статистика
|
||
stats = get_day_statistics(day_blocks)
|
||
if stats['total'] > 0:
|
||
lines.append(f'⏱ Всього: <code>{format_time_duration(stats["total"])}</code>')
|
||
if stats['confirmed'] > 0:
|
||
lines.append(f'🔴 Підтверджено: <code>{format_time_duration(stats["confirmed"])}</code>')
|
||
if stats['possible'] > 0:
|
||
lines.append(f'🟠 Можливо: <code>{format_time_duration(stats["possible"])}</code>')
|
||
else:
|
||
lines.append('🟢 <b>Відключень немає!</b>')
|
||
|
||
lines.append('')
|
||
|
||
# Детальный список отключений
|
||
disconnections = []
|
||
current_status = None
|
||
start_time = None
|
||
current_confirmed = None
|
||
current_queue = None
|
||
|
||
for block in day_blocks:
|
||
hour = block['hour']
|
||
|
||
for half_idx, half in enumerate([block['first_half'], block['second_half']]):
|
||
time_str = f'{hour:02d}:00' if half_idx == 0 else f'{hour:02d}:30'
|
||
|
||
if half['status'] == 'off':
|
||
if current_status != 'off':
|
||
start_time = time_str
|
||
current_confirmed = half['confirmed']
|
||
current_queue = half['queue']
|
||
current_status = 'off'
|
||
else:
|
||
if current_status == 'off':
|
||
icon = '🔴' if current_confirmed else '🟠'
|
||
queue_text = f' (Ч{current_queue})' if current_queue else ''
|
||
disconnections.append(f'{icon} <code>{start_time} - {time_str}</code>{queue_text}')
|
||
current_status = half['status']
|
||
|
||
# Если день закончился на отключении
|
||
if current_status == 'off':
|
||
icon = '🔴' if current_confirmed else '🟠'
|
||
queue_text = f' (Ч{current_queue})' if current_queue else ''
|
||
next_hour = (day_blocks[-1]['hour'] + 1) % 24
|
||
end_time = f'{next_hour:02d}:00'
|
||
disconnections.append(f'{icon} <code>{start_time} - {end_time}</code>{queue_text}')
|
||
|
||
if disconnections:
|
||
for idx, disc in enumerate(disconnections, 1):
|
||
lines.append(f'{idx}. {disc}')
|
||
|
||
lines.append('')
|
||
|
||
lines.append('<i>🔴 = підтверджено • 🟠 = можливо • 🟢 = світло</i>')
|
||
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def format_single_day_schedule(schedule: list[dict], day: int) -> str:
|
||
"""Форматирует график на один день"""
|
||
day_blocks = [b for b in schedule if b['day'] == day]
|
||
if not day_blocks:
|
||
return '❌ Немає даних для цього дня'
|
||
|
||
start_date = datetime.now()
|
||
date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y')
|
||
day_name = '🟠 <b>Сьогодні</b>' if day == 0 else '🔶 <b>Завтра</b>'
|
||
|
||
lines = [f'{day_name} • {date_str}', '']
|
||
|
||
# Статистика
|
||
lines.append('<b>📊 Статистика</b>')
|
||
stats = get_day_statistics(day_blocks)
|
||
|
||
if stats['total'] == 0:
|
||
lines.append('└ 🟢 <b>Відключень немає!</b>')
|
||
else:
|
||
total_time = format_time_duration(stats['total'])
|
||
lines.append(f'├ ⏱ Всього: <code>{total_time}</code>')
|
||
|
||
if stats['confirmed'] > 0:
|
||
confirmed_time = format_time_duration(stats['confirmed'])
|
||
lines.append(f'├ 🔴 Підтверджено: <code>{confirmed_time}</code>')
|
||
|
||
if stats['possible'] > 0:
|
||
possible_time = format_time_duration(stats['possible'])
|
||
lines.append(f'└ 🟠 Можливо: <code>{possible_time}</code>')
|
||
else:
|
||
lines.append('└ 🟢 Решта часу світло')
|
||
|
||
lines.append('')
|
||
|
||
# Детальный список отключений
|
||
disconnections = []
|
||
current_status = None
|
||
start_time = None
|
||
current_confirmed = None
|
||
current_queue = None
|
||
|
||
for block in day_blocks:
|
||
hour = block['hour']
|
||
|
||
for half_idx, half in enumerate([block['first_half'], block['second_half']]):
|
||
time_str = f'{hour:02d}:00' if half_idx == 0 else f'{hour:02d}:30'
|
||
|
||
if half['status'] == 'off':
|
||
if current_status != 'off':
|
||
start_time = time_str
|
||
current_confirmed = half['confirmed']
|
||
current_queue = half['queue']
|
||
current_status = 'off'
|
||
else:
|
||
if current_status == 'off':
|
||
icon = '🔴' if current_confirmed else '🟠'
|
||
queue_text = f' (Ч.{current_queue})' if current_queue else ''
|
||
disconnections.append(f'{icon} <code>{start_time} - {time_str}</code>{queue_text}')
|
||
current_status = half['status']
|
||
|
||
# Если день закончился на отключении
|
||
if current_status == 'off':
|
||
icon = '🔴' if current_confirmed else '🟠'
|
||
queue_text = f' (Ч.{current_queue})' if current_queue else ''
|
||
next_hour = (day_blocks[-1]['hour'] + 1) % 24
|
||
end_time = f'{next_hour:02d}:00'
|
||
disconnections.append(f'{icon} <code>{start_time} - {end_time}</code>{queue_text}')
|
||
|
||
if disconnections:
|
||
lines.append('<b>⚡️ Розклад відключень</b>')
|
||
for idx, disc in enumerate(disconnections, 1):
|
||
lines.append(f'{idx}. {disc}')
|
||
|
||
lines.append('')
|
||
lines.append('<i>🔴 підтверджено • 🟠 можливо • 🟢 світло</i>')
|
||
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def schedules_differ(old_schedule: list[dict] | None, new_schedule: list[dict] | None) -> bool:
|
||
"""Проверяет отличия между графиками"""
|
||
if old_schedule is None or new_schedule is None:
|
||
return True
|
||
|
||
if len(old_schedule) != len(new_schedule):
|
||
return True
|
||
|
||
for old, new in zip(old_schedule, new_schedule, strict=False):
|
||
if old['day'] >= 2:
|
||
break
|
||
|
||
if old['first_half'] != new['first_half'] or old['second_half'] != new['second_half']:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
# ============================================================================
|
||
# УВЕДОМЛЕНИЯ
|
||
# ============================================================================
|
||
|
||
|
||
async def send_to_all_users(message_text: str, parse_mode: str = 'HTML'):
|
||
"""Отправляет сообщение всем пользователям"""
|
||
if not ALLOWED_CHAT_IDS:
|
||
logger.warning('Нет допущенных ID чатов для отправки уведомлений')
|
||
return
|
||
|
||
for chat_id in ALLOWED_CHAT_IDS:
|
||
try:
|
||
await bot.send_message(chat_id, message_text, parse_mode=parse_mode)
|
||
logger.info(f'✅ Сообщение отправлено пользователю {chat_id}')
|
||
except Exception as e:
|
||
logger.error(f'❌ Ошибка отправки пользователю {chat_id}: {e}')
|
||
await asyncio.sleep(0.5)
|
||
|
||
|
||
async def check_schedule():
|
||
"""Проверяет график и отправляет уведомления"""
|
||
global last_schedule
|
||
|
||
try:
|
||
logger.info('🔍 Проверка графика...')
|
||
html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID)
|
||
new_schedule = parse_html(html)
|
||
|
||
if schedules_differ(last_schedule, new_schedule):
|
||
logger.info('✨ Обнаружены изменения!')
|
||
message = format_schedule_message(new_schedule, days_to_show=2)
|
||
|
||
if last_schedule is not None:
|
||
await send_to_all_users(f'🔄 <b>Графік оновлено!</b>\n\n{message}')
|
||
|
||
last_schedule = new_schedule
|
||
else:
|
||
logger.info('✓ Графік без змін')
|
||
|
||
except Exception as e:
|
||
logger.error(f'❌ Ошибка при проверке графика: {e}')
|
||
|
||
|
||
async def check_upcoming_disconnections():
|
||
"""Проверяет предстоящие события и отправляет предупреждения за 5 минут"""
|
||
global last_notification_time
|
||
|
||
if last_schedule is None:
|
||
return
|
||
|
||
now = datetime.now()
|
||
today_blocks = [b for b in last_schedule if b['day'] == 0]
|
||
|
||
# Создаем список всех переходов (off -> on или on -> off)
|
||
transitions = []
|
||
prev_status = None
|
||
|
||
for block in today_blocks:
|
||
hour = block['hour']
|
||
|
||
for half_idx, half in enumerate([block['first_half'], block['second_half']]):
|
||
minute = 0 if half_idx == 0 else 30
|
||
time_str = f'{hour:02d}:{minute:02d}'
|
||
|
||
current_status = half['status']
|
||
|
||
# Если статус изменился - это переход
|
||
if prev_status is not None and prev_status != current_status:
|
||
transitions.append(
|
||
{
|
||
'hour': hour,
|
||
'minute': minute,
|
||
'time_str': time_str,
|
||
'from_status': prev_status,
|
||
'to_status': current_status,
|
||
'confirmed': half.get('confirmed'),
|
||
'queue': half.get('queue'),
|
||
}
|
||
)
|
||
|
||
prev_status = current_status
|
||
|
||
# Проверяем переходы
|
||
for transition in transitions:
|
||
event_time = now.replace(hour=transition['hour'], minute=transition['minute'], second=0, microsecond=0)
|
||
|
||
time_until = (event_time - now).total_seconds() / 60
|
||
notification_key = f'{transition["hour"]}:{transition["minute"]}_{transition["to_status"]}'
|
||
|
||
# Если за 5 минут до события (±1 минута) и еще не отправляли
|
||
if 4 <= time_until <= 6:
|
||
# Проверяем, не отправляли ли уже уведомление сегодня
|
||
if notification_key in last_notification_time:
|
||
last_notif_time = last_notification_time[notification_key]
|
||
if last_notif_time.date() == now.date():
|
||
continue # Уже отправляли сегодня
|
||
|
||
# Переход на ОТКЛЮЧЕНИЕ (on -> off)
|
||
if transition['from_status'] == 'on' and transition['to_status'] == 'off':
|
||
icon = '🔴' if transition['confirmed'] else '🟠'
|
||
status = 'підтверджено' if transition['confirmed'] else 'можливе'
|
||
queue_info = f' (Черга {transition["queue"]})' if transition['queue'] else ''
|
||
|
||
warning = (
|
||
f'⚠️ <b>УВАГА! ВІДКЛЮЧЕННЯ</b>\n\n'
|
||
f'Через ~5 хвилин\n'
|
||
f'Час: <code>{transition["time_str"]}</code>\n'
|
||
f'Статус: {icon} {status}{queue_info}'
|
||
)
|
||
|
||
await send_to_all_users(warning)
|
||
last_notification_time[notification_key] = now
|
||
logger.info(f'📢 Відправлено попередження про ВІДКЛЮЧЕННЯ в {transition["time_str"]}')
|
||
|
||
# Переход на ВКЛЮЧЕНИЕ (off -> on)
|
||
elif transition['from_status'] == 'off' and transition['to_status'] == 'on':
|
||
warning = (
|
||
f'✅ <b>УВАГА! ВКЛЮЧЕННЯ</b>\n\n'
|
||
f'Через ~5 хвилин буде світло\n'
|
||
f'Час: <code>{transition["time_str"]}</code>'
|
||
)
|
||
|
||
await send_to_all_users(warning)
|
||
last_notification_time[notification_key] = now
|
||
logger.info(f'📢 Відправлено попередження про ВКЛЮЧЕННЯ в {transition["time_str"]}')
|
||
|
||
|
||
async def monitoring_loop():
|
||
"""Основной цикл мониторинга"""
|
||
await check_schedule()
|
||
|
||
while True:
|
||
try:
|
||
await asyncio.sleep(CHECK_INTERVAL)
|
||
await check_schedule()
|
||
await check_upcoming_disconnections()
|
||
except Exception as e:
|
||
logger.error(f'Ошибка в цикле мониторинга: {e}')
|
||
await asyncio.sleep(5)
|
||
|
||
|
||
# ============================================================================
|
||
# ОБРАБОТЧИКИ КОМАНД
|
||
# ============================================================================
|
||
|
||
|
||
@dp.message(Command('start'))
|
||
async def cmd_start(message: Message):
|
||
"""Обработчик /start"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
await message.answer('❌ У вас немає доступу до цього бота.')
|
||
return
|
||
|
||
await message.answer(
|
||
'👋 <b>Ласкаво просимо!</b>\n\n'
|
||
'🤖 <b>Бот для моніторингу графіку відключень світла</b>\n\n'
|
||
'✨ <b>Можливості:</b>\n'
|
||
'• 📊 Перегляд графіку на сьогодні і завтра\n'
|
||
'• 🔔 Автоматичні сповіщення за 5 хвилин до подій\n'
|
||
'• 🔄 Моніторинг змін графіку\n\n'
|
||
'Використовуйте кнопки нижче 👇',
|
||
parse_mode='HTML',
|
||
reply_markup=get_main_keyboard(),
|
||
)
|
||
|
||
|
||
@dp.message(lambda msg: msg.text == 'ℹ️ Про бота')
|
||
async def cmd_info(message: Message):
|
||
"""Показывает информацию о боте"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
return
|
||
|
||
await message.answer(
|
||
'<b>ℹ️ Про бота</b>\n\n'
|
||
'🚀 <b>Версія:</b> 2.2 (Стабільна)\n\n'
|
||
'📝 <b>Реліз-ноути:</b>\n'
|
||
'├ 15.11.2024: Адаптація під оновлену логіку сайту VOE\n'
|
||
'├ Виправлено парсинг half.left та half.right\n'
|
||
'├ Покращено визначення підтвердження відключень\n'
|
||
'└ Оптимізовано обробку статусу для всієї години\n\n'
|
||
'⚡ <b>Функціональність:</b>\n'
|
||
'├ Моніторинг графіку 24/7\n'
|
||
'├ Сповіщення за 5 хвилин\n'
|
||
'├ Детальна статистика дня\n'
|
||
'└ Красива візуалізація\n\n'
|
||
'🔐 <b>Безпека:</b> Використовуються .env файли\n'
|
||
'💾 <b>Джерело:</b> voe.com.ua',
|
||
parse_mode='HTML',
|
||
reply_markup=get_main_keyboard(),
|
||
)
|
||
|
||
|
||
@dp.message(Command('schedule'))
|
||
async def cmd_schedule(message: Message):
|
||
"""Показывает полный график"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
await message.answer('❌ У вас немає доступу.')
|
||
return
|
||
|
||
try:
|
||
await message.answer('⏳ Завантаження графіку...')
|
||
html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID)
|
||
schedule = parse_html(html)
|
||
text = format_schedule_message(schedule, days_to_show=2)
|
||
await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
except Exception as e:
|
||
await message.answer(f'❌ <b>Помилка:</b> {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
|
||
|
||
@dp.message(Command('today'))
|
||
async def cmd_today(message: Message):
|
||
"""Показывает график на сегодня"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
await message.answer('❌ У вас немає доступу.')
|
||
return
|
||
|
||
try:
|
||
await message.answer('⏳ Завантаження графіку сьогодні...')
|
||
html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID)
|
||
schedule = parse_html(html)
|
||
text = format_single_day_schedule(schedule, 0)
|
||
await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
except Exception as e:
|
||
await message.answer(f'❌ <b>Помилка:</b> {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
|
||
|
||
@dp.message(Command('tomorrow'))
|
||
async def cmd_tomorrow(message: Message):
|
||
"""Показывает график на завтра"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
await message.answer('❌ У вас немає доступу.')
|
||
return
|
||
|
||
try:
|
||
await message.answer('⏳ Завантаження графіку завтра...')
|
||
html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID)
|
||
schedule = parse_html(html)
|
||
text = format_single_day_schedule(schedule, 1)
|
||
await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
# await message.answer("❌ Функція тимчасово недоступна. Чекаємо на оновлення сайту", parse_mode="HTML", reply_markup=get_main_keyboard())
|
||
except Exception as e:
|
||
await message.answer(f'❌ <b>Помилка:</b> {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
|
||
|
||
@dp.message(Command('check'))
|
||
async def cmd_check(message: Message):
|
||
"""Принудительная проверка графика"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
await message.answer('❌ У вас немає доступу.')
|
||
return
|
||
|
||
try:
|
||
await message.answer('🔄 <b>Перевіряю графік...</b>', parse_mode='HTML')
|
||
html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID)
|
||
new_schedule = parse_html(html)
|
||
|
||
prefix = (
|
||
'✅ <b>Знайдено зміни!</b>\n\n'
|
||
if schedules_differ(last_schedule, new_schedule)
|
||
else '✓ <b>Графік без змін</b>\n\n'
|
||
)
|
||
result = prefix + format_schedule_message(new_schedule, days_to_show=2)
|
||
|
||
await message.answer(result, parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
except Exception as e:
|
||
await message.answer(f'❌ <b>Помилка:</b> {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard())
|
||
|
||
|
||
@dp.message()
|
||
async def handle_text(message: Message):
|
||
"""Обработчик текстовых сообщений и кнопок"""
|
||
if message.chat.id not in ALLOWED_CHAT_IDS:
|
||
return
|
||
|
||
text = message.text
|
||
|
||
# Кнопка "Графік"
|
||
if text == '📊 Графік':
|
||
await cmd_schedule(message)
|
||
|
||
# Кнопка "Сьогодні"
|
||
elif text == '📅 Сьогодні':
|
||
await cmd_today(message)
|
||
|
||
# Кнопка "Завтра"
|
||
elif text == '📅 Завтра':
|
||
await cmd_tomorrow(message)
|
||
|
||
# Кнопка "Оновити"
|
||
elif text == '🔄 Оновити':
|
||
await cmd_check(message)
|
||
|
||
# Кнопка "Про бота"
|
||
elif text == 'ℹ️ Про бота':
|
||
await cmd_info(message)
|
||
|
||
# Неизвестная команда
|
||
else:
|
||
await message.answer(
|
||
'❓ <b>Команда не розпізнана</b>\n\n'
|
||
'Використовуйте кнопки на клавіатурі або команди:\n'
|
||
'/start • /today • /tomorrow • /schedule • /check',
|
||
parse_mode='HTML',
|
||
reply_markup=get_main_keyboard(),
|
||
)
|
||
|
||
|
||
# ============================================================================
|
||
# ГЛАВНАЯ ФУНКЦИЯ
|
||
# ============================================================================
|
||
|
||
|
||
async def main():
|
||
"""Главная функция"""
|
||
logger.info('=' * 50)
|
||
logger.info('ЗАПУСК БОТА V2.2 (stable 2.2, 15.11.2025)')
|
||
logger.info('=' * 50)
|
||
|
||
if not TELEGRAM_TOKEN or os.getenv('TELEGRAM_TOKEN', 'YOUR_TOKEN_HERE') == TELEGRAM_TOKEN:
|
||
logger.error('❌ TELEGRAM_TOKEN не конфігурований! Напишіть токен в .env файл')
|
||
return
|
||
|
||
if not ALLOWED_CHAT_IDS:
|
||
logger.error('❌ ALLOWED_CHAT_IDS не конфігуровані! Напишіть ID в .env файл')
|
||
return
|
||
|
||
logger.info(f'📌 Allowed chat ids: {ALLOWED_CHAT_IDS}')
|
||
logger.info(f'⏱ Інтервал перевірки: {CHECK_INTERVAL} сек')
|
||
logger.info('=' * 50)
|
||
|
||
# Запускаем мониторинг
|
||
monitoring_task = asyncio.create_task(monitoring_loop())
|
||
|
||
try:
|
||
await dp.start_polling(bot)
|
||
except KeyboardInterrupt:
|
||
logger.info('⏹ Бот зупинений користувачем')
|
||
finally:
|
||
monitoring_task.cancel()
|
||
await bot.session.close()
|
||
logger.info('✓ Підключення закрито')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
asyncio.run(main())
|
||
except KeyboardInterrupt:
|
||
logger.info('⏹ Завершено')
|