import json import asyncio import os import qrcode # Библиотека для отрисовки QR from telethon import TelegramClient, events from google import genai from telethon.errors import SessionPasswordNeededError, ApiIdInvalidError from datetime import datetime, timedelta # --- НАСТРОЙКИ --- API_ID = 33837936 API_HASH = "7ad6e2e2a28f8f6efd228c66c293c195" CHANNEL = "pat_cherkasyoblenergo" GEMINI_KEY = "AIzaSyDTql5-rthOqXcKjlFff9fs7kRSINtb7TY" # <--- ПРОВЕРЬ КЛЮЧ WEB_SERVER_PORT = 2101 # Порт WEB-сервера BASE_DIR = os.path.dirname(os.path.abspath(__file__)) OUTPUT_FILE = os.path.join(BASE_DIR, "light.json") ALERTS_FILE = os.path.join(BASE_DIR, "alerts.json") # Настройка AI ai_client = genai.Client(api_key=GEMINI_KEY) # Настройка Телеграм client = TelegramClient('session_name', API_ID, API_HASH) # Файл для хранения ID последнего обработанного сообщения (чтобы не тратить лимиты при перезапуске) LAST_MSG_ID_FILE = os.path.join(BASE_DIR, "last_msg_id.txt") def get_last_msg_id(): if os.path.exists(LAST_MSG_ID_FILE): with open(LAST_MSG_ID_FILE, 'r') as f: return f.read().strip() return None def save_last_msg_id(msg_id): with open(LAST_MSG_ID_FILE, 'w') as f: f.write(str(msg_id)) async def ask_gemini(text): print("Отправляю запрос в Gemini...") now = datetime.now() now_str = now.strftime("%Y-%m-%d %H:%M") today_str = now.strftime("%Y-%m-%d") tomorrow_str = (now + timedelta(days=1)).strftime("%Y-%m-%d") prompt = f""" РОЛЬ: Ты парсер графиков отключения электроэнергии. ВВОДНЫЕ: 1. Текст сообщения: "{text}" 2. Текущая дата и время: {now_str} 3. Сегодня: {today_str} 4. Завтра: {tomorrow_str} ЗАДАЧА: 1. Определи, содержит ли сообщение график отключений электроэнергии. 2. Если НЕ содержит (например, просто новость или поздравление) — верни {{"is_schedule": false}}. 3. Если содержит — извлеки периоды отключения для ВСЕХ очередей (1.1, 1.2, 2.1, ..., 6.2), которые упомянуты. 4. Раздели периоды на "today" (сегодня) и "tomorrow" (завтра) по датам в сообщении. ФОРМАТ JSON (без markdown, без ```): {{ "is_schedule": true, "last_updated": "{now_str}", "queues": {{ "1.1": {{ "today": {{ "date": "{today_str}", "periods": [ {{"start": "HH:MM", "end": "HH:MM"}}, ... ] }}, "tomorrow": {{ "date": "{tomorrow_str}", "periods": [ ... ] }} или null }}, "1.2": {{ ... }}, ... "6.2": {{ ... }} }} }} ПРИМЕЧАНИЕ: - ОБЯЗАТЕЛЬНО указывай "date" для today и tomorrow. - Если периодов нет, "periods": []. - Если завтрашнего графика нет вообще, "tomorrow": null. """ retries = 3 for attempt in range(retries): try: # Reverting to gemini-2.0-flash as 1.5-flash returned 404 response = ai_client.models.generate_content( model='gemini-2.5-flash', contents=prompt ) clean_json = response.text.replace('```json', '').replace('```', '').strip() print(f"Ответ Gemini: {clean_json}") return json.loads(clean_json) except Exception as e: if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e): print(f"Превышен лимит Gemini (429). Ждем 60 сек... (Попытка {attempt+1}/{retries})") await asyncio.sleep(60) else: print(f"Ошибка AI: {e}") return None print("Не удалось получить ответ от AI после нескольких попыток.") return None async def save_to_file(data): # Не сохраняем если это не график if not data or not data.get("is_schedule"): print("Сообщение не содержит график. Файл не изменён.") return False # Пытаемся прочитать существующий файл для слияния existing_data = {} if os.path.exists(OUTPUT_FILE): try: with open(OUTPUT_FILE, 'r', encoding='utf-8') as f: existing_data = json.load(f) except Exception as e: print(f"Ошибка чтения старого файла: {e}") # Логика слияния new_queues = data.get("queues", {}) old_queues = existing_data.get("queues", {}) for q_id, q_data in new_queues.items(): # Если очереди нет в старом файле, просто берем новую if q_id not in old_queues: continue old_q = old_queues[q_id] # --- СЛИЯНИЕ TODAY --- # Если в новом "today" пусто (0 периодов), а в старом было что-то и даты совпадают -> оставляем старое new_today = q_data.get("today", {}) old_today = old_q.get("today", {}) # Поддержка старого формата (если старый файл был массивом, то мы его перезапишем новым форматом) # Но если новый формат {date, periods}, то работаем. if isinstance(new_today, dict) and isinstance(old_today, dict): new_periods = new_today.get("periods", []) old_periods = old_today.get("periods", []) # Если новый список пуст, а старый нет, и даты совпадают if not new_periods and old_periods: if new_today.get("date") == old_today.get("date"): print(f"[{q_id}] Сохраняем старый график на сегодня ({old_today.get('date')})") q_data["today"] = old_today # --- СЛИЯНИЕ TOMORROW --- new_tomorrow = q_data.get("tomorrow") old_tomorrow = old_q.get("tomorrow") if isinstance(new_tomorrow, dict) and isinstance(old_tomorrow, dict): new_periods = new_tomorrow.get("periods", []) old_periods = old_tomorrow.get("periods", []) if not new_periods and old_periods: if new_tomorrow.get("date") == old_tomorrow.get("date"): print(f"[{q_id}] Сохраняем старый график на завтра") q_data["tomorrow"] = old_tomorrow os.makedirs(os.path.dirname(OUTPUT_FILE), exist_ok=True) with open(OUTPUT_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"Файл {OUTPUT_FILE} успешно обновлен!") return True @client.on(events.NewMessage(chats=CHANNEL)) async def handler(event): print(f"\n--- Новое сообщение в канале! ---") text = event.message.message msg_id = event.message.id if not text: return # Можно добавить проверку дублей тут, но обычно handler ловит только новые data = await ask_gemini(text) saved = await save_to_file(data) if saved: save_last_msg_id(msg_id) import threading from http.server import SimpleHTTPRequestHandler from socketserver import TCPServer import aiohttp async def fetch_and_save_alerts(): """Фоновая задача для загрузки воздушных тревог и сохранения в alerts.json""" url = "https://ubilling.net.ua/aerialalerts/" while True: try: timestamp = int(datetime.now().timestamp() * 1000) async with aiohttp.ClientSession() as session: async with session.get(f"{url}?t={timestamp}", timeout=10) as response: if response.status == 200: data = await response.json(content_type=None) os.makedirs(os.path.dirname(ALERTS_FILE), exist_ok=True) with open(ALERTS_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False) # print(f"[{datetime.now().strftime('%H:%M:%S')}] Обновлены воздушные тревоги") except Exception as e: print(f"Ошибка обновления тревог: {e}") await asyncio.sleep(60) # Обновление каждую минуту # ... imports ... async def main(): print("Запуск...") # --- ЗАПУСК WEB-СЕРВЕРА --- # Чтобы ESP8266 мог скачать light.json (а также Flutter Web без ошибки CORS) def run_server(): PORT = WEB_SERVER_PORT web_dir = os.path.dirname(OUTPUT_FILE) if os.path.exists(web_dir): os.chdir(web_dir) class CORSRequestHandler(SimpleHTTPRequestHandler): def end_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET') self.send_header('Access-Control-Allow-Headers', 'Origin, Content-Type') self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate') super().end_headers() def log_message(self, format, *args): pass # Отключаем логи сервера в консоль with TCPServer(("", PORT), CORSRequestHandler) as httpd: print(f"WEB-сервер запущен: http://0.0.0.0:{PORT}/light.json") print(f"WEB-сервер запущен: http://0.0.0.0:{PORT}/alerts.json") httpd.serve_forever() # Запускаем сервер в отдельном потоке (daemon=True, чтобы закрылся вместе со скриптом) server_thread = threading.Thread(target=run_server, daemon=True) server_thread.start() # -------------------------- # Запуск фонового обновления тревог asyncio.create_task(fetch_and_save_alerts()) # 1. Подключаемся к серверам Telegram await client.connect() # 2. Проверяем, авторизованы ли мы уже if not await client.is_user_authorized(): # Force UTF-8 encoding (полезно для вывода QR-кода в логах) import sys sys.stdout.reconfigure(encoding='utf-8') print("Нужна авторизация через QR-код...") try: while True: # Получаем объект логина qr_login = await client.qr_login() # Рисуем QR-код в консоли print("---------------------------------------") print("Отсканируй этот код в Telegram (Настройки -> Устройства):") qr = qrcode.QRCode() qr.add_data(qr_login.url) qr.make(fit=True) # invert=True делает код белым на черном фоне (для консоли Windows) qr.print_ascii(invert=True) print("---------------------------------------") # Ждем, пока юзер отсканирует try: user = await qr_login.wait() print(f"Успешный вход! Пользователь: {user.username}") break except SessionPasswordNeededError: # Если стоит облачный пароль (2FA) password = input("Введите ваш облачный пароль (2FA): ") await client.sign_in(password=password) break except Exception as e: # Check for AuthTokenExpiredError by name since we haven't imported it yet # or we can just retry on specific errors. # Best practice: import it. But to avoid modifying imports at top, I'll check repr if "AuthTokenExpiredError" in repr(e): print("QR-код истек, генерирую новый...") continue else: raise e except ApiIdInvalidError: print("\n!!! ОШИБКА: API_ID или API_HASH неверны! !!!") print("Перейдите на https://my.telegram.org, создайте приложение и скопируйте новые данные в код.") return else: print("Вы уже авторизованы (файл сессии найден).") # 3. Чтение истории (только если это новое сообщение) print(f"Проверяю последнее сообщение в {CHANNEL}...") try: last_processed = get_last_msg_id() async for message in client.iter_messages(CHANNEL, limit=1): if message.message: print(f"Последнее сообщение ID: {message.id}") if str(message.id) == str(last_processed): print(">>> Это сообщение уже обработано ранее. Пропускаем запрос к AI (экономим лимиты).") else: print(f"Обрабатываем: {message.message[:50]}...") data = await ask_gemini(message.message) saved = await save_to_file(data) if saved: save_last_msg_id(message.id) except Exception as e: print(f"Ошибка чтения истории (возможно канал приватный или неверный юзернейм): {e}") print("Слушаю новые сообщения...") await client.run_until_disconnected() if __name__ == '__main__': asyncio.run(main())