" in html:
for el in PARSED_PO_DATAS:
if message:
message += '\n'
address = get_address(el.city, el.street, el.house)
message = message + (
f'\U000026A0
Увага!\nНа {el.start_date} за адресою {address} заплановано
{el.po_type} відключення (заявка від {el.placement_date}).\n'
f'Причина відключення: {el.reason}\n'
f'Початок вимкнення
{el.start_date} об {el.start_time}\n'
f'Кінець вимкнення
{el.stop_date} об {el.stop_time}\n'
f'{MENTIONS}')
# need_to_send = el.need_to_send or False
need_to_send = not is_old_and_new_datas_equals_PO(PARSED_PO_DATAS)
else:
message = '\nПланових вимкнень немає \U0001F44C \U0001F483'
need_to_send = not is_old_and_new_datas_equals_PO()
if need_to_send:
save_last_datas_PO(PARSED_PO_DATAS)
return message, need_to_send
def process_GPV(oe_account_number, tg_chat, tg_token):
global IS_DEBUG
if IS_DEBUG:
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"}]}'
"""{
"current":
{
"gav":
{
"message": "Черга спеціальних аварійних відключень (СГАВ):",
"queue": null
},
"hasQueue": "yes",
"note": "Станом на 21:05 14.01.2026 за вказаним особовим рахунком '30014180' споживач підпадає під чергу '5.1' Графіку погодинного відключення(ГПВ)",
"queue": 5,
"sgav":
{
"message": "Черга спеціальних аварійних відключень (СГАВ):",
"queue": null
},
"subQueue": 1
},
"schedule":
[
{
"createdAt": "14.01.2026 19:11",
"eventDate": "15.01.2026",
"queues":
{
"5.1":
[
{
"from": "07:30",
"shutdownHours": "07:30-11:00",
"status": 1,
"to": "11:00"
},
{
"from": "14:30",
"shutdownHours": "14:30-20:00",
"status": 1,
"to": "20:00"
},
{
"from": "22:00",
"shutdownHours": "22:00-00:00",
"status": 1,
"to": "00:00"
}
]
},
"scheduleApprovedSince": "14.01.2026 19:11"
},
{
"createdAt": "13.01.2026 20:28",
"eventDate": "14.01.2026",
"queues":
{
"5.1":
[
{
"from": "03:00",
"shutdownHours": "03:00-07:00",
"status": 1,
"to": "07:00"
},
{
"from": "11:00",
"shutdownHours": "11:00-17:30",
"status": 1,
"to": "17:30"
}
]
},
"scheduleApprovedSince": "14.01.2026 09:56"
}
]
}
"""
)
else:
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()
outage_queue = str(x["current"]["queue"])
if x["current"]["subQueue"]:
outage_queue += '.' + str(x["current"]["subQueue"])
datas = ''
if 'queues' in str(x["schedule"]):
datas = sorted(x["schedule"], key=lambda item: item['eventDate'])
need_to_send = not is_old_and_new_datas_equals_GPV(datas)
if need_to_send:
save_last_datas_GPV(datas)
for entry in datas:
hours_off = []
hours_on = []
if len(entry) > 0:
approved_from = 'Запроваджено ' + entry['scheduleApprovedSince']
event_date = entry['eventDate']
hours_list = entry['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 = 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)
if USE_MATRIX:
res = send_matrix_notification(message)
if not res:
sys.exit(STATUS_ERROR_SEND_TO_MX)
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
Планове
" \
"
"
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)
if USE_MATRIX:
res = send_matrix_notification(message)
if not res:
sys.exit(STATUS_ERROR_SEND_TO_MX)
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()