570 lines
23 KiB
Python
570 lines
23 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, ReplyKeyboardMarkup, KeyboardButton
|
|
from aiogram.utils.keyboard import ReplyKeyboardBuilder
|
|
import logging
|
|
|
|
# Настройки
|
|
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 = 120 # 2 минуты
|
|
|
|
# Настройка логирования
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Глобальные переменные
|
|
bot = Bot(token=TELEGRAM_TOKEN)
|
|
dp = Dispatcher()
|
|
last_schedule = None
|
|
|
|
|
|
def parse_html(html: str):
|
|
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=False):
|
|
if not half:
|
|
return {"status": "unknown", "queue": None, "confirmed": None}
|
|
|
|
if force_off:
|
|
status = "off"
|
|
else:
|
|
classes = half.get("class", [])
|
|
if "has_disconnection" in classes:
|
|
status = "off"
|
|
elif "no_disconnection" in classes:
|
|
status = "on"
|
|
else:
|
|
status = "unknown"
|
|
|
|
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:
|
|
queue = title.split(":")[-1].strip()
|
|
|
|
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 not confirmed and not queue:
|
|
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, street_id, house_id):
|
|
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 (compatible; PowerBot/1.0)"
|
|
}
|
|
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,
|
|
}
|
|
r = requests.post(url, headers=headers, data=data)
|
|
r.raise_for_status()
|
|
resp_json = json.loads(r.text)
|
|
insert_html = next((i["data"] for i in resp_json if i.get("command") == "insert"), None)
|
|
if not insert_html:
|
|
raise ValueError("HTML не найден в ответе")
|
|
return insert_html
|
|
|
|
|
|
def get_main_keyboard():
|
|
"""Создает главную клавиатуру с кнопками"""
|
|
builder = ReplyKeyboardBuilder()
|
|
builder.row(
|
|
KeyboardButton(text="📊 Графік"),
|
|
KeyboardButton(text="🔄 Перевірити")
|
|
)
|
|
builder.row(
|
|
KeyboardButton(text="📅 Сьогодні"),
|
|
KeyboardButton(text="📅 Завтра")
|
|
)
|
|
return builder.as_markup(resize_keyboard=True)
|
|
|
|
|
|
def format_schedule_message(schedule, days_to_show=2):
|
|
"""Форматирует расписание для отправки в Telegram"""
|
|
message = f"⚡️ <b>Графік відключень світла</b>\n"
|
|
message += f"🕐 Оновлено: {datetime.now().strftime('%d.%m.%Y %H:%M')}\n\n"
|
|
|
|
start_date = datetime.now()
|
|
|
|
for day in range(days_to_show):
|
|
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')
|
|
day_name = "🌅 Сьогодні" if day == 0 else "🌄 Завтра"
|
|
|
|
message += f"{'─' * 32}\n"
|
|
message += f"📅 <b>{day_name} ({date_str})</b>\n"
|
|
message += f"{'─' * 32}\n\n"
|
|
|
|
# Группируем последовательные отключения
|
|
current_time = None
|
|
current_status = None
|
|
current_confirmed = None
|
|
current_queue = None
|
|
start_time = None
|
|
|
|
disconnection_periods = []
|
|
|
|
for i, block in enumerate(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 is True else "🟠"
|
|
queue_text = f" • Черга {current_queue}" if current_queue else ""
|
|
disconnection_periods.append(f"{icon} <code>{start_time} - {time_str}</code>{queue_text}")
|
|
current_status = half["status"]
|
|
|
|
# Если день закончился на отключении
|
|
if current_status == "off":
|
|
icon = "🔴" if current_confirmed is True 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"
|
|
disconnection_periods.append(f"{icon} <code>{start_time} - {end_time}</code>{queue_text}")
|
|
|
|
# Выводим результат
|
|
if disconnection_periods:
|
|
for period in disconnection_periods:
|
|
message += period + "\n"
|
|
else:
|
|
message += "🟢 <b>Відключень немає!</b>\n"
|
|
|
|
message += "\n"
|
|
|
|
message += f"{'─' * 32}\n"
|
|
message += "💡 <i>Легенда:</i>\n"
|
|
message += "🔴 — Підтверджено\n"
|
|
message += "🟠 — Можливо\n"
|
|
message += "🟢 — Світло є\n"
|
|
|
|
return message
|
|
|
|
|
|
def format_single_day_schedule(schedule, day_num):
|
|
"""Форматирует расписание для одного дня"""
|
|
start_date = datetime.now()
|
|
date_str = (start_date + timedelta(days=day_num)).strftime('%d.%m')
|
|
day_name = "🌅 Сьогодні" if day_num == 0 else "🌄 Завтра"
|
|
|
|
message = f"{'─' * 32}\n"
|
|
message += f"📅 <b>{day_name} ({date_str})</b>\n"
|
|
message += f"{'─' * 32}\n\n"
|
|
|
|
day_blocks = [b for b in schedule if b["day"] == day_num]
|
|
if not day_blocks:
|
|
return message + "❓ <i>Немає даних</i>"
|
|
|
|
# Создаем визуальную шкалу времени
|
|
message += "⏰ <b>Часова шкала:</b>\n\n"
|
|
|
|
# Выводим по 4 часа в строке для читаемости
|
|
for start_hour in range(0, 24, 4):
|
|
line = ""
|
|
for h in range(start_hour, min(start_hour + 4, 24)):
|
|
block = day_blocks[h] if h < len(day_blocks) else None
|
|
if not block:
|
|
line += f"{h:02d}|❓❓ "
|
|
continue
|
|
|
|
first = block["first_half"]
|
|
second = block["second_half"]
|
|
|
|
def get_icon(half_data):
|
|
if half_data["status"] == "off":
|
|
return "🔴" if half_data["confirmed"] is True else "🟠"
|
|
elif half_data["status"] == "on":
|
|
return "🟢"
|
|
return "❓"
|
|
|
|
first_icon = "●" if first["status"] == "off" else "·"
|
|
second_icon = "●" if second["status"] == "off" else "·"
|
|
|
|
line += f"<code>{h:02d}|{first_icon}{second_icon}</code> "
|
|
|
|
message += line + "\n"
|
|
|
|
message += "\n<i>● = відключено, · = світло</i>\n\n"
|
|
|
|
# Детальний список відключень
|
|
message += "📋 <b>Детально:</b>\n\n"
|
|
|
|
has_disconnections = False
|
|
for block in day_blocks:
|
|
hour = block["hour"]
|
|
first = block["first_half"]
|
|
second = block["second_half"]
|
|
|
|
if first["status"] == "off":
|
|
has_disconnections = True
|
|
icon = "🔴" if first["confirmed"] is True else "🟠"
|
|
status = "підтв." if first["confirmed"] is True else "можл."
|
|
queue = f"Ч{first['queue']}" if first['queue'] else "—"
|
|
message += f"{icon} <code>{hour:02d}:00-{hour:02d}:30</code> • {status} • {queue}\n"
|
|
|
|
if second["status"] == "off":
|
|
has_disconnections = True
|
|
icon = "🔴" if second["confirmed"] is True else "🟠"
|
|
status = "підтв." if second["confirmed"] is True else "можл."
|
|
queue = f"Ч{second['queue']}" if second['queue'] else "—"
|
|
message += f"{icon} <code>{hour:02d}:30-{hour+1:02d}:00</code> • {status} • {queue}\n"
|
|
|
|
if not has_disconnections:
|
|
message += "🟢 <b>Відключень немає!</b>\n"
|
|
|
|
message += f"\n{'─' * 32}\n"
|
|
message += "<i>🔴 підтв. • 🟠 можл. • 🟢 світло</i>"
|
|
|
|
return message
|
|
|
|
|
|
def schedules_differ(old_schedule, new_schedule):
|
|
"""Проверяет, отличаются ли графики"""
|
|
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):
|
|
# Сравниваем только первые 2 дня
|
|
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):
|
|
"""Отправляет сообщение всем пользователям"""
|
|
for chat_id in ALLOWED_CHAT_IDS:
|
|
try:
|
|
await bot.send_message(chat_id, message_text, parse_mode="HTML")
|
|
logger.info(f"Сообщение отправлено пользователю {chat_id}")
|
|
except Exception as e:
|
|
logger.error(f"Ошибка отправки пользователю {chat_id}: {e}")
|
|
|
|
|
|
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 None:
|
|
pass # Ничего не делать, если изменений нет
|
|
else:
|
|
# Изменения в графике
|
|
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}")
|
|
await send_to_all_users(f"❌ <b>Помилка!</b>\n\nНе вдалося отримати графік: {str(e)}")
|
|
|
|
|
|
async def check_upcoming_disconnections():
|
|
"""Проверяет предстоящие отключения и включения, отправляет предупреждения"""
|
|
global last_schedule
|
|
|
|
if last_schedule is None:
|
|
return
|
|
|
|
now = datetime.now()
|
|
current_time = now.time()
|
|
|
|
# Проверяем только текущий день
|
|
today_blocks = [b for b in last_schedule if b["day"] == 0]
|
|
|
|
for block in today_blocks:
|
|
hour = block["hour"]
|
|
|
|
# Проверяем первую половину часа (XX:00 - XX:30)
|
|
first_half_start = now.replace(hour=hour, minute=0, second=0, microsecond=0)
|
|
time_until_first = (first_half_start - now).total_seconds() / 60
|
|
|
|
# Если до события осталось 5 минут (с погрешностью ±1 минута)
|
|
if 4 <= time_until_first <= 6:
|
|
first = block["first_half"]
|
|
|
|
# ОТКЛЮЧЕНИЕ
|
|
if first["status"] == "off":
|
|
icon = "🔴" if first["confirmed"] is True else "🟠"
|
|
status_text = "підтверджено" if first["confirmed"] is True else "можливе"
|
|
queue_info = f" (Черга {first['queue']})" if first["queue"] else ""
|
|
|
|
warning = (
|
|
f"⚠️ <b>УВАГА! ВІДКЛЮЧЕННЯ</b>\n\n"
|
|
f"Через 5 хвилин відключення {icon}\n"
|
|
f"Час: {hour:02d}:00 - {hour:02d}:30\n"
|
|
f"Статус: {status_text}{queue_info}"
|
|
)
|
|
await send_to_all_users(warning)
|
|
logger.info(f"Отправлено предупреждение о отключении в {hour:02d}:00")
|
|
|
|
# ВКЛЮЧЕНИЕ
|
|
elif first["status"] == "on":
|
|
# Проверяем, было ли отключение до этого
|
|
prev_block = today_blocks[today_blocks.index(block) - 1] if today_blocks.index(block) > 0 else None
|
|
was_off = False
|
|
|
|
if prev_block:
|
|
was_off = prev_block["second_half"]["status"] == "off"
|
|
|
|
if was_off:
|
|
warning = (
|
|
f"✅ <b>УВАГА! ВКЛЮЧЕННЯ</b>\n\n"
|
|
f"Через 5 хвилин світло має з'явитися 🟢\n"
|
|
f"Час: {hour:02d}:00"
|
|
)
|
|
await send_to_all_users(warning)
|
|
logger.info(f"Отправлено предупреждение о включении в {hour:02d}:00")
|
|
|
|
# Проверяем вторую половину часа (XX:30 - XX+1:00)
|
|
second_half_start = now.replace(hour=hour, minute=30, second=0, microsecond=0)
|
|
time_until_second = (second_half_start - now).total_seconds() / 60
|
|
|
|
if 4 <= time_until_second <= 6:
|
|
second = block["second_half"]
|
|
|
|
# ОТКЛЮЧЕНИЕ
|
|
if second["status"] == "off":
|
|
icon = "🔴" if second["confirmed"] is True else "🟠"
|
|
status_text = "підтверджено" if second["confirmed"] is True else "можливе"
|
|
queue_info = f" (Черга {second['queue']})" if second["queue"] else ""
|
|
|
|
warning = (
|
|
f"⚠️ <b>УВАГА! ВІДКЛЮЧЕННЯ</b>\n\n"
|
|
f"Через 5 хвилин відключення {icon}\n"
|
|
f"Час: {hour:02d}:30 - {hour+1:02d}:00\n"
|
|
f"Статус: {status_text}{queue_info}"
|
|
)
|
|
await send_to_all_users(warning)
|
|
logger.info(f"Отправлено предупреждение о отключении в {hour:02d}:30")
|
|
|
|
# ВКЛЮЧЕНИЕ
|
|
elif second["status"] == "on":
|
|
# Проверяем, было ли отключение в первой половине
|
|
was_off = block["first_half"]["status"] == "off"
|
|
|
|
if was_off:
|
|
warning = (
|
|
f"✅ <b>УВАГА! ВКЛЮЧЕННЯ</b>\n\n"
|
|
f"Через 5 хвилин світло має з'явитися 🟢\n"
|
|
f"Час: {hour:02d}:30"
|
|
)
|
|
await send_to_all_users(warning)
|
|
logger.info(f"Отправлено предупреждение о включении в {hour:02d}:30")
|
|
|
|
|
|
async def monitoring_loop():
|
|
"""Основной цикл мониторинга"""
|
|
# Первая проверка сразу при запуске
|
|
await check_schedule()
|
|
|
|
# Затем проверяем каждые 2 минуты
|
|
while True:
|
|
await asyncio.sleep(CHECK_INTERVAL)
|
|
await check_schedule()
|
|
await check_upcoming_disconnections()
|
|
|
|
|
|
@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"
|
|
"Я відстежую зміни в графіку відключень світла та попереджаю за 5 хвилин до відключення.\n\n"
|
|
"Використовуйте кнопки нижче 👇",
|
|
parse_mode="HTML",
|
|
reply_markup=get_main_keyboard()
|
|
)
|
|
|
|
|
|
@dp.message(lambda message: message.text == "📊 Графік")
|
|
@dp.message(Command("schedule"))
|
|
async def cmd_schedule(message: Message):
|
|
"""Показывает текущий график (2 дня)"""
|
|
if message.chat.id not in ALLOWED_CHAT_IDS:
|
|
await message.answer("❌ У вас немає доступу до цього бота.")
|
|
return
|
|
|
|
try:
|
|
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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
|
|
|
|
|
|
@dp.message(lambda message: message.text == "📅 Сьогодні")
|
|
async def cmd_today(message: Message):
|
|
"""Показывает график на сегодня"""
|
|
if message.chat.id not in ALLOWED_CHAT_IDS:
|
|
await message.answer("❌ У вас немає доступу до цього бота.")
|
|
return
|
|
|
|
try:
|
|
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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
|
|
|
|
|
|
@dp.message(lambda message: message.text == "📅 Завтра")
|
|
async def cmd_tomorrow(message: Message):
|
|
"""Показывает график на завтра"""
|
|
if message.chat.id not in ALLOWED_CHAT_IDS:
|
|
await message.answer("❌ У вас немає доступу до цього бота.")
|
|
return
|
|
|
|
try:
|
|
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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
|
|
|
|
|
|
@dp.message(lambda message: message.text == "🔄 Перевірити")
|
|
@dp.message(Command("check"))
|
|
async def cmd_check(message: Message):
|
|
"""Принудительная проверка графика"""
|
|
if message.chat.id not in ALLOWED_CHAT_IDS:
|
|
await message.answer("❌ У вас немає доступу до цього бота.")
|
|
return
|
|
|
|
await message.answer("🔄 Перевіряю графік...", reply_markup=get_main_keyboard())
|
|
|
|
# Выполняем проверку и отправляем результат
|
|
try:
|
|
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):
|
|
result_message = "✅ <b>Знайдено зміни в графіку!</b>\n\n"
|
|
result_message += format_schedule_message(new_schedule, days_to_show=2)
|
|
await message.answer(result_message, parse_mode="HTML", reply_markup=get_main_keyboard())
|
|
else:
|
|
result_message = "✅ <b>Графік без змін</b>\n\n"
|
|
result_message += format_schedule_message(new_schedule, days_to_show=2)
|
|
await message.answer(result_message, parse_mode="HTML", reply_markup=get_main_keyboard())
|
|
except Exception as e:
|
|
await message.answer(f"❌ Помилка при перевірці: {str(e)}", reply_markup=get_main_keyboard())
|
|
|
|
|
|
async def main():
|
|
"""Главная функция"""
|
|
logger.info("Запуск бота...")
|
|
|
|
# Запускаем мониторинг в фоне
|
|
monitoring_task = asyncio.create_task(monitoring_loop())
|
|
|
|
# Запускаем бота
|
|
try:
|
|
await dp.start_polling(bot)
|
|
finally:
|
|
monitoring_task.cancel()
|
|
await bot.session.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |