diff --git a/dtek_notif/Dockerfile b/dtek_notif/Dockerfile
index 2656973..52d4464 100644
--- a/dtek_notif/Dockerfile
+++ b/dtek_notif/Dockerfile
@@ -8,6 +8,6 @@ RUN pip install --no-cache-dir -r requirements.txt
# Копирование кода
COPY main.py .
-
+COPY .env .
# Запуск бота
CMD ["python", "-u", "main.py"]
\ No newline at end of file
diff --git a/dtek_notif/main.py b/dtek_notif/main.py
index a1919ea..67c3735 100644
--- a/dtek_notif/main.py
+++ b/dtek_notif/main.py
@@ -5,26 +5,84 @@ 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.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 = 120 # 2 минуты
+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)
+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 = None
+last_schedule: Optional[List[Dict]] = None
+last_notification_time: Dict[str, datetime] = {}
-def parse_html(html: str):
+# ============================================================================
+# УТИЛИТЫ
+# ============================================================================
+
+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")
@@ -46,28 +104,29 @@ def parse_html(html: str):
left = hour_block.select_one(".half.left")
right = hour_block.select_one(".half.right")
- def parse_half(half, force_off=False):
+ def parse_half(half, force_off: bool = False) -> Dict:
if not half:
return {"status": "unknown", "queue": None, "confirmed": None}
- if force_off:
- status = "off"
- else:
+ 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"
- 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()
+ try:
+ queue = title.split(":")[-1].strip()
+ except:
+ pass
disconnection_classes = disconnection_div.get("class", [])
if "disconnection_confirm_1" in disconnection_classes:
@@ -75,7 +134,7 @@ def parse_html(html: str):
elif "disconnection_confirm_0" in disconnection_classes:
confirmed = False
- if force_off and not confirmed and not queue:
+ if force_off and confirmed is None:
if 'confirm_1' in cell_classes:
confirmed = True
elif 'confirm_0' in cell_classes:
@@ -101,12 +160,13 @@ def parse_html(html: str):
return schedule
-def get_voe_html(city_id, street_id, house_id):
+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 (compatible; PowerBot/1.0)"
+ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
data = {
"search_type": 0,
@@ -119,184 +179,208 @@ def get_voe_html(city_id, street_id, house_id):
"_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
+
+ 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="🔄 Перевірити")
+ KeyboardButton(text="🔄 Оновити")
)
builder.row(
KeyboardButton(text="📅 Сьогодні"),
KeyboardButton(text="📅 Завтра")
)
+ builder.row(KeyboardButton(text="ℹ️ Про бота"))
return builder.as_markup(resize_keyboard=True)
-def format_schedule_message(schedule, days_to_show=2):
- """Форматирует расписание для отправки в Telegram"""
- message = f"⚡️ Графік відключень світла\n"
- message += f"🕐 Оновлено: {datetime.now().strftime('%d.%m.%Y %H:%M')}\n\n"
+def format_schedule_message(schedule: List[Dict], days_to_show: int = 2) -> str:
+ """Форматирует полный график на несколько дней"""
+ lines = [
+ "⚡️ Графік відключень світла",
+ f"🕐 Оновлено: {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}",
+ "─" * 30,
+ ""
+ ]
start_date = datetime.now()
- for day in range(days_to_show):
+ 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')
- day_name = "🌅 Сьогодні" if day == 0 else "🌄 Завтра"
+ date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y')
+ day_name = "🌅 Сьогодні" if day == 0 else "🌄 Завтра"
- message += f"{'─' * 32}\n"
- message += f"📅 {day_name} ({date_str})\n"
- message += f"{'─' * 32}\n\n"
+ lines.append(f"{day_name} ({date_str})")
- # Группируем последовательные отключения
- current_time = None
+ # Статистика
+ stats = get_day_statistics(day_blocks)
+ if stats["total"] > 0:
+ lines.append(f"⏱ Всього: {format_time_duration(stats['total'])}")
+ if stats["confirmed"] > 0:
+ lines.append(f"🔴 Підтверджено: {format_time_duration(stats['confirmed'])}")
+ if stats["possible"] > 0:
+ lines.append(f"🟠 Можливо: {format_time_duration(stats['possible'])}")
+ else:
+ lines.append("🟢 Відключень немає!")
+
+ lines.append("")
+
+ # Детальный список отключений
+ disconnections = []
current_status = None
+ start_time = None
current_confirmed = None
current_queue = None
- start_time = None
- disconnection_periods = []
-
- for i, block in enumerate(day_blocks):
+ 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 is True else "🟠"
- queue_text = f" • Черга {current_queue}" if current_queue else ""
- disconnection_periods.append(f"{icon} {start_time} - {time_str}{queue_text}")
+ icon = "🔴" if current_confirmed else "🟠"
+ queue_text = f" (Ч{current_queue})" if current_queue else ""
+ disconnections.append(f"{icon} {start_time} - {time_str}{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 ""
+ 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"
- disconnection_periods.append(f"{icon} {start_time} - {end_time}{queue_text}")
+ disconnections.append(f"{icon} {start_time} - {end_time}{queue_text}")
- # Выводим результат
- if disconnection_periods:
- for period in disconnection_periods:
- message += period + "\n"
- else:
- message += "🟢 Відключень немає!\n"
+ if disconnections:
+ for idx, disc in enumerate(disconnections, 1):
+ lines.append(f"{idx}. {disc}")
- message += "\n"
+ lines.append("")
- message += f"{'─' * 32}\n"
- message += "💡 Легенда:\n"
- message += "🔴 — Підтверджено\n"
- message += "🟠 — Можливо\n"
- message += "🟢 — Світло є\n"
+ lines.append("🔴 = підтверджено • 🟠 = можливо • 🟢 = світло")
- return message
+ return "\n".join(lines)
-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"📅 {day_name} ({date_str})\n"
- message += f"{'─' * 32}\n\n"
-
- day_blocks = [b for b in schedule if b["day"] == day_num]
+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 message + "❓ Немає даних"
+ return "❌ Немає даних для цього дня"
- # Создаем визуальную шкалу времени
- message += "⏰ Часова шкала:\n\n"
+ start_date = datetime.now()
+ date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y')
+ day_name = "🟠 Сьогодні" if day == 0 else "🔶 Завтра"
- # Выводим по 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"{h:02d}|{first_icon}{second_icon} "
+ lines = [
+ f"{day_name} • {date_str}",
+ ""
+ ]
+
+ # Статистика
+ lines.append("📊 Статистика")
+ stats = get_day_statistics(day_blocks)
+
+ if stats["total"] == 0:
+ lines.append("└ 🟢 Відключень немає!")
+ else:
+ total_time = format_time_duration(stats["total"])
+ lines.append(f"├ ⏱ Всього: {total_time}")
- message += line + "\n"
+ if stats["confirmed"] > 0:
+ confirmed_time = format_time_duration(stats["confirmed"])
+ lines.append(f"├ 🔴 Підтверджено: {confirmed_time}")
+
+ if stats["possible"] > 0:
+ possible_time = format_time_duration(stats["possible"])
+ lines.append(f"└ 🟠 Можливо: {possible_time}")
+ else:
+ lines.append(f"└ 🟢 Решта часу світло")
- message += "\n● = відключено, · = світло\n\n"
+ lines.append("")
- # Детальний список відключень
- message += "📋 Детально:\n\n"
+ # Детальный список отключений
+ disconnections = []
+ current_status = None
+ start_time = None
+ current_confirmed = None
+ current_queue = None
- 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} {hour:02d}:00-{hour:02d}:30 • {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} {hour:02d}:30-{hour+1:02d}:00 • {status} • {queue}\n"
+ 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} {start_time} - {time_str}{queue_text}")
+ current_status = half["status"]
- if not has_disconnections:
- message += "🟢 Відключень немає!\n"
+ # Если день закончился на отключении
+ 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} {start_time} - {end_time}{queue_text}")
- message += f"\n{'─' * 32}\n"
- message += "🔴 підтв. • 🟠 можл. • 🟢 світло"
+ if disconnections:
+ lines.append("⚡️ Розклад відключень")
+ for idx, disc in enumerate(disconnections, 1):
+ lines.append(f"{idx}. {disc}")
- return message
+ lines.append("")
+ lines.append("🔴 підтверджено • 🟠 можливо • 🟢 світло")
+
+ return "\n".join(lines)
-def schedules_differ(old_schedule, new_schedule):
- """Проверяет, отличаются ли графики"""
+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
@@ -304,7 +388,6 @@ def schedules_differ(old_schedule, new_schedule):
return True
for old, new in zip(old_schedule, new_schedule):
- # Сравниваем только первые 2 дня
if old["day"] >= 2:
break
@@ -315,256 +398,358 @@ def schedules_differ(old_schedule, new_schedule):
return False
-async def send_to_all_users(message_text: str):
+# ============================================================================
+# УВЕДОМЛЕНИЯ
+# ============================================================================
+
+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="HTML")
- logger.info(f"Сообщение отправлено пользователю {chat_id}")
+ 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}")
+ logger.error(f"❌ Ошибка отправки пользователю {chat_id}: {e}")
+ await asyncio.sleep(0.5)
async def check_schedule():
- """Проверяет график и отправляет уведомления при изменениях"""
+ """Проверяет график и отправляет уведомления"""
global last_schedule
try:
- logger.info("Проверка графика...")
+ 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("Обнаружены изменения в графике!")
+ logger.info("✨ Обнаружены изменения!")
message = format_schedule_message(new_schedule, days_to_show=2)
- if last_schedule is None:
- pass # Ничего не делать, если изменений нет
- else:
- # Изменения в графике
+ if last_schedule is not None:
await send_to_all_users(f"🔄 Графік оновлено!\n\n{message}")
last_schedule = new_schedule
else:
- logger.info("Изменений нет")
+ logger.info("✓ Графік без змін")
except Exception as e:
- logger.error(f"Ошибка при проверке графика: {e}")
- await send_to_all_users(f"❌ Помилка!\n\nНе вдалося отримати графік: {str(e)}")
+ logger.error(f"❌ Ошибка при проверке графика: {e}")
async def check_upcoming_disconnections():
- """Проверяет предстоящие отключения и включения, отправляет предупреждения"""
- global last_schedule
+ """Проверяет предстоящие события и отправляет предупреждения за 5 минут"""
+ global last_notification_time
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]
+ # Создаем список всех переходов (off -> on или on -> off)
+ transitions = []
+ prev_status = None
+
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"]
+ 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}"
- # ОТКЛЮЧЕНИЕ
- 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 ""
+ 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"⚠️ УВАГА! ВІДКЛЮЧЕННЯ\n\n"
- f"Через 5 хвилин відключення {icon}\n"
- f"Час: {hour:02d}:00 - {hour:02d}:30\n"
- f"Статус: {status_text}{queue_info}"
+ f"Через ~5 хвилин\n"
+ f"Час: {transition['time_str']}\n"
+ f"Статус: {icon} {status}{queue_info}"
)
+
await send_to_all_users(warning)
- logger.info(f"Отправлено предупреждение о отключении в {hour:02d}:00")
+ last_notification_time[notification_key] = now
+ logger.info(f"📢 Відправлено попередження про ВІДКЛЮЧЕННЯ в {transition['time_str']}")
- # ВКЛЮЧЕНИЕ
- 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"✅ УВАГА! ВКЛЮЧЕННЯ\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 ""
-
+ # Переход на ВКЛЮЧЕНИЕ (off -> on)
+ elif transition["from_status"] == "off" and transition["to_status"] == "on":
warning = (
- f"⚠️ УВАГА! ВІДКЛЮЧЕННЯ\n\n"
- f"Через 5 хвилин відключення {icon}\n"
- f"Час: {hour:02d}:30 - {hour+1:02d}:00\n"
- f"Статус: {status_text}{queue_info}"
+ f"✅ УВАГА! ВКЛЮЧЕННЯ\n\n"
+ f"Через ~5 хвилин буде світло\n"
+ f"Час: {transition['time_str']}"
)
- 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"✅ УВАГА! ВКЛЮЧЕННЯ\n\n"
- f"Через 5 хвилин світло має з'явитися 🟢\n"
- f"Час: {hour:02d}:30"
- )
- await send_to_all_users(warning)
- logger.info(f"Отправлено предупреждение о включении в {hour:02d}:30")
+ 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()
- # Затем проверяем каждые 2 минуты
while True:
- await asyncio.sleep(CHECK_INTERVAL)
- await check_schedule()
- await check_upcoming_disconnections()
+ 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"""
+ """Обработчик /start"""
if message.chat.id not in ALLOWED_CHAT_IDS:
await message.answer("❌ У вас немає доступу до цього бота.")
return
await message.answer(
- "👋 Вітаю!\n\n"
- "Я відстежую зміни в графіку відключень світла та попереджаю за 5 хвилин до відключення.\n\n"
+ "👋 Ласкаво просимо!\n\n"
+ "🤖 Бот для моніторингу графіку відключень світла\n\n"
+ "✨ Можливості:\n"
+ "• 📊 Перегляд графіку на сьогодні і завтра\n"
+ "• 🔔 Автоматичні сповіщення за 5 хвилин до подій\n"
+ "• 🔄 Моніторинг змін графіку\n\n"
"Використовуйте кнопки нижче 👇",
parse_mode="HTML",
reply_markup=get_main_keyboard()
)
-@dp.message(lambda message: message.text == "📊 Графік")
+@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(
+ "ℹ️ Про бота\n\n"
+ "🚀 Версія: 2.1 (Оптимізована)\n"
+ "⚡ Функціональність:\n"
+ "├ Моніторинг графіку 24/7\n"
+ "├ Сповіщення за 5 хвилин\n"
+ "├ Детальна статистика дня\n"
+ "└ Красива візуалізація\n\n"
+ "🔐 Безпека: Використовуються .env файли\n"
+ "💾 Джерело: voe.com.ua",
+ parse_mode="HTML",
+ reply_markup=get_main_keyboard()
+ )
+
+
@dp.message(Command("schedule"))
async def cmd_schedule(message: Message):
- """Показывает текущий график (2 дня)"""
+ """Показывает полный график"""
if message.chat.id not in ALLOWED_CHAT_IDS:
- await message.answer("❌ У вас немає доступу до цього бота.")
+ 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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
+ await message.answer(
+ f"❌ Помилка: {str(e)}",
+ parse_mode="HTML",
+ reply_markup=get_main_keyboard()
+ )
-@dp.message(lambda message: message.text == "📅 Сьогодні")
+@dp.message(Command("today"))
async def cmd_today(message: Message):
"""Показывает график на сегодня"""
if message.chat.id not in ALLOWED_CHAT_IDS:
- await message.answer("❌ У вас немає доступу до цього бота.")
+ 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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
+ await message.answer(
+ f"❌ Помилка: {str(e)}",
+ parse_mode="HTML",
+ reply_markup=get_main_keyboard()
+ )
-@dp.message(lambda message: message.text == "📅 Завтра")
+@dp.message(Command("tomorrow"))
async def cmd_tomorrow(message: Message):
"""Показывает график на завтра"""
if message.chat.id not in ALLOWED_CHAT_IDS:
- await message.answer("❌ У вас немає доступу до цього бота.")
+ 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"❌ Помилка: {str(e)}", reply_markup=get_main_keyboard())
+ await message.answer(
+ f"❌ Помилка: {str(e)}",
+ parse_mode="HTML",
+ 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("❌ У вас немає доступу до цього бота.")
+ await message.answer("❌ У вас немає доступу.")
return
- await message.answer("🔄 Перевіряю графік...", reply_markup=get_main_keyboard())
-
- # Выполняем проверку и отправляем результат
try:
+ await message.answer("🔄 Перевіряю графік...", parse_mode="HTML")
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 = "✅ Знайдено зміни в графіку!\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 = "✅ Графік без змін\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())
+ prefix = "✅ Знайдено зміни!\n\n" if schedules_differ(last_schedule, new_schedule) else "✓ Графік без змін\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"❌ Помилка при перевірці: {str(e)}", reply_markup=get_main_keyboard())
+ await message.answer(
+ f"❌ Помилка: {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(
+ "❓ Команда не розпізнана\n\n"
+ "Використовуйте кнопки на клавіатурі або команди:\n"
+ "/start • /today • /tomorrow • /schedule • /check",
+ parse_mode="HTML",
+ reply_markup=get_main_keyboard()
+ )
+
+
+# ============================================================================
+# ГЛАВНАЯ ФУНКЦИЯ
+# ============================================================================
+
async def main():
"""Главная функция"""
- logger.info("Запуск бота...")
+ 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__":
- asyncio.run(main())
\ No newline at end of file
+ try:
+ asyncio.run(main())
+ except KeyboardInterrupt:
+ logger.info("⏹ Завершено")
\ No newline at end of file
diff --git a/dtek_notif/requirements.txt b/dtek_notif/requirements.txt
index 7bf17e3..72e2a35 100644
--- a/dtek_notif/requirements.txt
+++ b/dtek_notif/requirements.txt
@@ -1,3 +1,5 @@
-bs4
-requests
-aiogram
\ No newline at end of file
+requests>=2.31.0
+beautifulsoup4>=4.12.0
+aiogram>=3.3.0
+python-dotenv>=1.0.0
+aiohttp>=3.9.0