Add DTEK notification bot with Docker configuration and requirements

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Установка зависимостей
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Копирование кода
+COPY main.py .
+
+# Запуск бота
+CMD ["python", "-u", "main.py"]

dtek-docker-compose.yaml

@@ -0,0 +1,14 @@
+services:
+  dtek_notif:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    restart: unless-stopped
+    environment:
+      - TZ=Europe/Kyiv
+    
+    dns:
+      - 1.1.1.1
+      - 8.8.8.8
+    networks:
+      - default

main.py

@@ -0,0 +1,319 @@
+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
+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 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"📅 <b>{day_name} ({date_str})</b>\n"
+        message += "━━━━━━━━━━━━━━━━━━━━\n"
+        
+        for block in day_blocks:
+            first = block["first_half"]
+            second = block["second_half"]
+            
+            def get_icon(half_data):
+                if half_data["status"] == "off":
+                    if half_data["confirmed"] is True:
+                        return "🔴"  # Точно відключено
+                    elif half_data["confirmed"] is False:
+                        return "🟠"  # Можливо відключено
+                    return "🔴"
+                elif half_data["status"] == "on":
+                    return "🟢"
+                return "⚪️"
+            
+            first_icon = get_icon(first)
+            second_icon = get_icon(second)
+            
+            # Форматируем время
+            hour_start = f"{block['hour']:02d}:00"
+            hour_mid = f"{block['hour']:02d}:30"
+            hour_end = f"{block['hour']+1:02d}:00"
+            
+            # Информация о черге
+            queue_info = ""
+            if first["queue"] or second["queue"]:
+                queue = first["queue"] or second["queue"]
+                queue_info = f" | Черга {queue}"
+            
+            message += f"{hour_start}-{hour_mid} {first_icon}  {hour_mid}-{hour_end} {second_icon}{queue_info}\n"
+        
+        message += "\n"
+    
+    message += "🟢 - Світло є\n"
+    message += "🔴 - Відключення підтверджено\n"
+    message += "🟠 - Можливе відключення\n"
+    message += "⚪️ - Інформація відсутня"
+    
+    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:
+                # Первый запуск
+                await send_to_all_users(f"✨ <b>Бот запущено!</b>\n\n{message}")
+            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 monitoring_loop():
+    """Основной цикл мониторинга"""
+    # Первая проверка сразу при запуске
+    await check_schedule()
+    
+    # Затем проверяем каждые 2 минуты
+    while True:
+        await asyncio.sleep(CHECK_INTERVAL)
+        await check_schedule()
+
+
+@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"
+        "Я буду відстежувати зміни в графіку відключень світла.\n\n"
+        "Доступні команди:\n"
+        "/schedule - Показати поточний графік\n"
+        "/check - Перевірити графік зараз",
+        parse_mode="HTML"
+    )
+
+
+@dp.message(Command("schedule"))
+async def cmd_schedule(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_schedule_message(schedule, days_to_show=2)
+        await message.answer(text, parse_mode="HTML")
+    except Exception as e:
+        await message.answer(f"❌ Помилка: {str(e)}")
+
+
+@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("🔄 Перевіряю графік...")
+    await check_schedule()
+
+
+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())

pyproject.toml

@@ -0,0 +1,7 @@
+[project]
+name = "dtek-notif"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = []

requirements.txt

@@ -0,0 +1,3 @@
+bs4
+requests
+aiogram