#!/usr/bin/python3 # Get Energy Outages script import os import sys from io import StringIO from types import SimpleNamespace import requests import json import urllib3 import numpy as np from numpy.ma.core import append from config_file import read_config as read_cfg from datetime import datetime as dt from lxml import etree STATUS_ERROR_SEND_TO_TG = 20 STATUS_ERROR_GPV_RESPONSE = 10 STATUS_ERROR_PO_RESPONSE = 30 STATUS_EMPTY_CONFIG = 5 SCRIPT_PATH = os.path.dirname(__file__) #CONFIG_FILE_NAME = SCRIPT_PATH + '/get_eo_config.yaml' CONFIG_FILE_NAME = SCRIPT_PATH + '/eo_config.json' LAST_DATAS_FILE_NAME_GPV = SCRIPT_PATH + '/get_eo.last_info' LAST_DATAS_FILE_NAME_PO = SCRIPT_PATH + '/get_po.last_info' LAST_DATAS_FORMAT_GPV = '{array} {date} {day}' LAST_DATAS_FORMAT_PO = '{placed} {star_dt} {start_tm} {stop_dt} {stop_tm} {current_date}' IS_DEBUG = False MENTIONS = "" def read_config(): global MENTIONS j = read_cfg(CONFIG_FILE_NAME) tg_token = j['token'] tg_chat = j['chat_id'] oe_account_number = j['account'] is_debug = j['debug'].lower() == 'true' settlement = j['settlement'] street = j['street'] house = j['house'] building_part_number = j['building_part_number'] apartment = j['apartment'] for m in j['mentions']: MENTIONS += m + '\n' return tg_token, tg_chat, oe_account_number, is_debug, settlement, street, house, building_part_number, apartment def send_message_to_tg(msg, token, chat): url = f"https://api.telegram.org/bot{token}/sendMessage?chat_id={chat}&parse_mode=html&text={msg}" return requests.get(url) # return requests.post( # url='https://api.telegram.org/bot{0}/{1}'.format(token, 'sendMessage'), # data={'chat_id': {chat}, 'text': {msg}, 'parse_mode': 'html'} # ) def get_GPV_response_from_oe(account): url = 'https://be-svitlo.oe.if.ua/schedule-by-search' headers = { "Host": "be-svitlo.oe.if.ua", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:144.0) Gecko/20100101 Firefox/144.0", "Accept": "application/json, text/plain, */*", "Accept-Language": "en-US,en;q=0.5", "Content-Type": "application/x-www-form-urlencoded;charset=utf-8", "Referer": "https://svitlo.oe.if.ua/", "Origin": "https://svitlo.oe.if.ua", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site", "Connection": "keep-alive", "Priority": "u=0", } data = { "accountNumber": account, "userSearchChoice": "pob", "address": "" } urllib3.disable_warnings() return requests.post(url, headers=headers, data=data, verify=False) def get_PO_response(settlement, street, house, building_part_number='', apartment=''): # Power Outage utf8 = '✓' url = 'https://oe.if.ua/uk/shutdowns_table' params = { 'utf8': utf8, 'settlement': settlement, 'street': street, 'house_number': house, 'building_part_number': building_part_number, 'apartment' : apartment, 'commit': 'Пошук' } headers = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36' } urllib3.disable_warnings() return requests.get(url=url, headers=headers, params=params) def has_numbers(input): return any(char.isdigit() for char in str(input)) def get_utf_chars(number, is_time=False): digit_chars = ['\U00000030\U000020E3', '\U00000031\U000020E3', '\U00000032\U000020E3', '\U00000033\U000020E3', '\U00000034\U000020E3', '\U00000035\U000020E3', '\U00000036\U000020E3', '\U00000037\U000020E3', '\U00000038\U000020E3', '\U00000039\U000020E3'] res = '' if is_time: h = int(number) m = int((number * 60) % 60) if m != 0: value = str(h) + ':' + str(m) else: value = str(h) + ':00' else: value = str(number) for c in value: if c.isdigit and c not in ['.', ':']: res += digit_chars[int(c)] else: res += c return res def get_digit_message(arr, is_time=False): first = 0 if len(arr)> 0 and arr[0] > 0: first = arr[0] last = 0 if len(arr)> 1 and arr[0] > 0: last = arr[1] msg = 'з {num0} до {num1}' return msg.format(num0=get_utf_chars(first, is_time), num1=get_utf_chars(last, is_time)) def save_last_datas_GPV(hours_on, hours_off, date, day): with open(LAST_DATAS_FILE_NAME_GPV, 'w') as f: f.write(LAST_DATAS_FORMAT_GPV.format(array=str(hours_on)+str(hours_off), date=date, day=day)) def save_last_datas_PO(placed='', star_dt='', start_tm='', stop_dt='', stop_tm=''): date = dt.today().strftime('%Y-%m-%d') with open(LAST_DATAS_FILE_NAME_PO, 'w') as f: f.write(LAST_DATAS_FORMAT_PO.format(placed=str(placed), star_dt=str(star_dt), start_tm=str(start_tm), stop_dt=str(stop_dt), stop_tm=str(stop_tm), current_date=date)) def load_last_datas(file_name): if os.path.isfile(file_name): with open(file_name, 'r') as f: return f.read() else: return '' def is_old_and_new_datas_equals_GPV(hours_on, hours_off, date, day): loaded = load_last_datas(LAST_DATAS_FILE_NAME_GPV) return LAST_DATAS_FORMAT_GPV.format(array=str(hours_on)+str(hours_off), date=date, day=day) == loaded def is_old_and_new_datas_equals_PO(placed='', star_dt='', start_tm='', stop_dt='', stop_tm=''): date = dt.today().strftime('%Y-%m-%d') loaded = load_last_datas(LAST_DATAS_FILE_NAME_PO) return LAST_DATAS_FORMAT_PO.format(placed=str(placed), star_dt=str(star_dt), start_tm=str(start_tm), stop_dt=str(stop_dt), stop_tm=str(stop_tm), current_date=date) == loaded def get_GPV_message(approved_from, event_date, hours_off, hours_on, outage_queue): queue = get_utf_chars(outage_queue) message = f'\U000026A1 ГПВ на {event_date} для {queue} черги' need_to_send = not is_old_and_new_datas_equals_GPV(hours_off, hours_on, approved_from, event_date) if need_to_send: save_last_datas_GPV(hours_off, hours_on, approved_from, event_date) if len(hours_off) > 0: message += '\nВимкнення:' for i in range(len(hours_off)): arr = [int(hours_off[i].split(':')[0]), int(hours_on[i].split(':')[0])] msg = get_digit_message(arr, True) message += '\n' + msg message += (f'\n{MENTIONS}') else: message += '\nВимкнень немає \U0001F44C \U0001F483' message += (f'\n{approved_from}') return message, need_to_send def get_address(city, street, house): return f'{city}, {street}, {house}' def is_outage_date_before_or_same(stop_date): parsed_date = dt.strptime(stop_date, "%Y.%m.%d").date() current_date = dt.now().date() return current_date <= parsed_date def get_message_with_PO(root, html): #city, street, house, placement_date, PO_type, reason, start_date, stop_date, start_time, stop_time if "
" in html: city = root.xpath("//div[@class='table-row flex']/div[@class='city']/text()")[0] street = root.xpath("//div[@class='table-row flex']/div[@class='street']/text()")[0] house = root.xpath("//div[@class='table-row flex']/div[@class='house_number']/text()")[0] PO_type = root.xpath("//div[@class='table-row flex']/div[@class='type-shutdown']/text()")[0] reason = root.xpath("//div[@class='table-row flex']/div[@class='reason-shutdown reason-text-container']/div[@class='reason-text-hide']/text()")[0] placement_date = root.xpath("//div[@class='table-row flex']/div[@class='placement_date']/div[@class='date']/text()")[0] start_date = root.xpath("//div[@class='table-row flex']/div[@class='shutdown_date']/div[@class='date']/text()")[0] stop_date = root.xpath("//div[@class='table-row flex']/div[@class='turn_on_date']/div[@class='date']/text()")[0] start_time = root.xpath("//div[@class='table-row flex']/div[@class='shutdown_date']/div[@class='time']/text()")[0] stop_time = root.xpath("//div[@class='table-row flex']/div[@class='turn_on_date']/div[@class='time']/text()")[0] address = get_address(city, street, house) message = (f'\U000026A0 Увага!\nНа {start_date} за адресою {address} заплановано {PO_type} відключення (заявка від {placement_date}).\n' f'Причина відключення: {reason}\n' f'Початок вимкнення {start_date} об {start_time}\n' f'Кінець вимкнення {stop_date} об {stop_time}\n' f'{MENTIONS}') need_to_send = (not is_old_and_new_datas_equals_PO(placed=placement_date, star_dt=start_date, start_tm=start_time, stop_dt=stop_date, stop_tm=stop_time)) if is_outage_date_before_or_same(stop_date): need_to_send = need_to_send and True else: message = '\nПланових вимкнень немає \U0001F44C \U0001F483' need_to_send = not is_old_and_new_datas_equals_PO(placed=placement_date, star_dt=start_date, start_tm=start_time, stop_dt=stop_date, stop_tm=stop_time) if need_to_send: save_last_datas_PO(placed=placement_date, star_dt=start_date, start_tm=start_time, stop_dt=stop_date, stop_tm=stop_time) else: message = '\nПланових вимкнень немає \U0001F44C \U0001F483' need_to_send = not is_old_and_new_datas_equals_PO() if need_to_send: save_last_datas_PO() return message, need_to_send def process_GPV(oe_account_number, tg_chat, tg_token): global IS_DEBUG if not IS_DEBUG: response = get_GPV_response_from_oe(oe_account_number) if response.status_code != 200: print("code=" + str(response.status_code)) return STATUS_ERROR_GPV_RESPONSE x = response.json() else: x = json.loads( '{"current": {"hasQueue": "yes", "note": "ZZZ xxx YYY", "queue": 5, "subQueue": 1}, "schedule": [{"createdAt": "30.10.2025 20:59", "eventDate": "31.10.2025", "queues": {"5.1": [{"from": "22:00", "shutdownHours": "22:00-00:00", "status": 1, "to": "00:00"}]}, "scheduleApprovedSince": "31.10.2025 10:52"}]}') hours_off = [] hours_on = [] outage_queue = str(x["current"]["queue"]) if x["current"]["subQueue"]: outage_queue += '.' + str(x["current"]["subQueue"]) datas = '' if 'queues' in str(x["schedule"]): datas = x["schedule"][0] if len(datas) > 0: approved_from = 'Запроваджено ' + datas['scheduleApprovedSince'] event_date = datas['eventDate'] hours_list = datas['queues'][outage_queue] else: hours_list = [] approved_from = 'Hемає даних про дату запровадження' event_date = dt.today().strftime('%Y-%m-%d') if IS_DEBUG: print(f'hours_list{hours_list}, approved_from={approved_from}, event_date={event_date}') for h in hours_list: if h["status"] == 1: hours_off.append(h["from"]) hours_on.append(h["to"]) if IS_DEBUG: print(f'hours_off = {hours_off}') print(f'hours_on = {hours_on}') message, need_to_send = get_GPV_message(approved_from, event_date, hours_off, hours_on, outage_queue) if IS_DEBUG: print(f'{message}\nneed_to_send={need_to_send}') if need_to_send and not IS_DEBUG: tg_response = send_message_to_tg(message, tg_token, tg_chat) if tg_response.status_code != 200: sys.exit(STATUS_ERROR_SEND_TO_TG) def process_PO(apart, house, part_number, settlement, street, tg_chat, tg_token): global IS_DEBUG if not IS_DEBUG: response = get_PO_response(settlement, street, house, part_number, apart) if response.status_code != 200: print("code=" + str(response.status_code)) return STATUS_ERROR_PO_RESPONSE html = response.text parser = etree.HTMLParser() root = etree.parse(StringIO(html), parser) else: html = "
Район
Населений пункт
Вулиця
№ буд
" \ "
Корпус
Диспетчерська назва об'єкту
Вид робіт
Причина робіт
Дата розміщення
" \ "
Дата і час вимкнення
Дата і час включення
Наступна дата робіт
Піднянський район
" \ "
Гірна
Тевського
9
ДСП 110/Пр."ЗБВIК"/30-413/Л-6
Планове
" \ "
Поточний ремонт
2025.03.24
11:50
2025.04.01
" \ "
09:00
2025.04.01
18:00
" root = etree.fromstring(html) message, need_to_send = get_message_with_PO(root=root, html=html) if IS_DEBUG: print(f'{message}\nneed_to_send={need_to_send}') if need_to_send and not IS_DEBUG: tg_response = send_message_to_tg(message, tg_token, tg_chat) if tg_response.status_code != 200: sys.exit(STATUS_ERROR_SEND_TO_TG) def main(): global IS_DEBUG tg_token, tg_chat, oe_account_number, IS_DEBUG, settlement, street, house, part_number, apart = read_config() if IS_DEBUG: print('-=: Debug Mode :=-') if (not tg_token or not tg_chat or not oe_account_number or not settlement or not street or not house): sys.exit(STATUS_EMPTY_CONFIG) process_GPV(oe_account_number, tg_chat, tg_token) process_PO(apart, house, part_number, settlement, street, tg_chat, tg_token) if __name__ == "__main__": main()