755 lines
29 KiB
Python
755 lines
29 KiB
Python
import requests
|
||
import json
|
||
import asyncio
|
||
from bs4 import BeautifulSoup
|
||
from datetime import datetime, timedelta
|
||
from aiogram import Bot, Dispatcher, types
|
||
from aiogram.filters import Command
|
||
from aiogram.types import Message, KeyboardButton
|
||
from aiogram.utils.keyboard import ReplyKeyboardBuilder
|
||
import logging
|
||
import os
|
||
from dotenv import load_dotenv
|
||
from typing import Optional, List, Dict
|
||
|
||
# Загрузка переменных окружения
|
||
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: Optional[List[Dict]] = 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 с графиком отключений"""
|
||
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
|
||
|
||
left = hour_block.select_one(".half.left")
|
||
right = hour_block.select_one(".half.right")
|
||
|
||
def parse_half(half, force_off: bool = False) -> Dict:
|
||
if not half:
|
||
return {"status": "unknown", "queue": None, "confirmed": None}
|
||
|
||
status = "off" if force_off else "unknown"
|
||
if not force_off:
|
||
classes = half.get("class", [])
|
||
if "has_disconnection" in classes:
|
||
status = "off"
|
||
elif "no_disconnection" in classes:
|
||
status = "on"
|
||
|
||
queue = None
|
||
confirmed = None
|
||
disconnection_div = half.select_one(".disconnection")
|
||
|
||
if disconnection_div and disconnection_div.has_attr("title"):
|
||
title = disconnection_div["title"]
|
||
if "Номер черги" in title:
|
||
try:
|
||
queue = title.split(":")[-1].strip()
|
||
except:
|
||
pass
|
||
|
||
disconnection_classes = disconnection_div.get("class", [])
|
||
if "disconnection_confirm_1" in disconnection_classes:
|
||
confirmed = True
|
||
elif "disconnection_confirm_0" in disconnection_classes:
|
||
confirmed = False
|
||
|
||
if force_off and confirmed is None:
|
||
if 'confirm_1' in cell_classes:
|
||
confirmed = True
|
||
elif 'confirm_0' in cell_classes:
|
||
confirmed = False
|
||
|
||
return {"status": status, "queue": queue, "confirmed": confirmed}
|
||
|
||
left_data = parse_half(left, force_off=full_hour_off)
|
||
right_data = parse_half(right, force_off=full_hour_off)
|
||
|
||
schedule.append({
|
||
"hour": current_hour,
|
||
"day": current_day,
|
||
"first_half": left_data,
|
||
"second_half": right_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(f"└ 🟢 Решта часу світло")
|
||
|
||
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: Optional[List[Dict]], new_schedule: Optional[List[Dict]]) -> 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):
|
||
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.1 (Оптимізована)\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())
|
||
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.1 (ОПТИМІЗОВАНА ВЕРСІЯ)")
|
||
logger.info("=" * 50)
|
||
|
||
if not TELEGRAM_TOKEN or TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "YOUR_TOKEN_HERE")
|
||
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}")
|
||
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("⏹ Завершено") |