405 lines
16 KiB
Python
405 lines
16 KiB
Python
#!/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
|
||
STATUS_ERROR_SEND_TO_MX = 40
|
||
|
||
SCRIPT_PATH = os.path.dirname(__file__)
|
||
CONFIG_FILE_NAME = SCRIPT_PATH + '/get_eo_config.yaml'
|
||
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 = ""
|
||
USE_MATRIX = False
|
||
MX_SERVER = ""
|
||
MX_ROOM = ""
|
||
MX_TOKEN = ""
|
||
|
||
|
||
def read_config():
|
||
global MENTIONS, USE_MATRIX, MX_SERVER, MX_ROOM, MX_TOKEN
|
||
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']
|
||
USE_MATRIX = j['use_matrix'].lower() == 'true'
|
||
MX_SERVER = j['mx_server']
|
||
MX_ROOM = j['mx_room_id']
|
||
MX_TOKEN = j['mx_access_token']
|
||
# i = 1
|
||
# for m in j['mentions']:
|
||
# MENTIONS += f'<a href="{m}">user {i}</a>' + '\n'
|
||
# i += 1
|
||
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 send_matrix_notification(message: str) -> bool:
|
||
"""Send notification to Matrix"""
|
||
if not USE_MATRIX:
|
||
return False
|
||
|
||
try:
|
||
url = f"{MX_SERVER}/_matrix/client/r0/rooms/{MX_ROOM}/send/m.room.message"
|
||
|
||
headers = {
|
||
'Authorization': f'Bearer {MX_TOKEN}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
payload = {
|
||
'msgtype': 'm.text',
|
||
'body': message,
|
||
'format': 'org.matrix.custom.html',
|
||
'formatted_body': message.replace('\n', '<br/>')
|
||
}
|
||
|
||
response = requests.post(url, headers=headers, json=payload)
|
||
return response.status_code == 200
|
||
except Exception as e:
|
||
# logger.error(f"Error sending Matrix notification: {e}")
|
||
return False
|
||
|
||
|
||
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 = number.split(':')[0]
|
||
m = number.split(':')[1]
|
||
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:
|
||
first = arr[0]
|
||
|
||
last = 0
|
||
if len(arr) > 1:
|
||
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 <b>ГПВ на {event_date} для {queue} черги</b>'
|
||
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 = [hours_off[i], hours_on[i]]
|
||
msg = get_digit_message(arr, True)
|
||
message += '\n' + msg
|
||
|
||
# message += (f'\n{MENTIONS}')
|
||
else:
|
||
message += '\nВимкнень немає \U0001F44C \U0001F483'
|
||
message += (f'\n<b><i>{approved_from}</i></b>')
|
||
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 "<div class='shutdowns'>" 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 <b>Увага!</b>\nНа {start_date} за адресою {address} заплановано <b>{PO_type}</b> відключення (заявка від {placement_date}).\n'
|
||
f'Причина відключення: {reason}\n'
|
||
f'Початок вимкнення <b>{start_date} об {start_time}</b>\n'
|
||
f'Кінець вимкнення <b>{stop_date} об {stop_time}</b>\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)
|
||
|
||
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 = "<html><body><div id='shutdowns-table'><div class='table'><div class='flex table-header'><div class='region'>Район</div><div class='city'>Населений пункт</div><div class='street'>Вулиця</div><div class='house_number'>№ буд</div>" \
|
||
"<div class='building_part_number'>Корпус</div><div class='dispatch_name'>Диспетчерська назва об'єкту</div><div class='type-shutdown'>Вид робіт</div><div class='reason-shutdown'>Причина робіт</div><div class='placement_date'>Дата розміщення</div>" \
|
||
"<div class='shutdown_date'>Дата і час вимкнення</div><div class='turn_on_date'>Дата і час включення</div><div class='next_date'>Наступна дата робіт</div></div></div><div class='shutdowns'><div><div><div class='table-row flex'><div class='region'>Піднянський район</div>" \
|
||
"<div class='city'>Гірна</div><div class='street'>Тевського</div><div class='house_number'>9</div><div class='building_part_number'/><div class='dispatch_name'>ДСП 110/Пр."ЗБВIК"/30-413/Л-6</div><div class='type-shutdown'>Планове</div>" \
|
||
"<div class='reason-shutdown reason-text-container'><div class='reason-text-hide'>Поточний ремонт</div></div><div class='placement_date'><div class='date'>2025.03.24</div><div class='time'>11:50</div></div><div class='shutdown_date'><div class='date'>2025.04.01</div>" \
|
||
"<div class='time'>09:00</div></div><div class='turn_on_date'><div class='date'>2025.04.01</div><div class='time'>18:00</div></div><div class='next_date'><div class='reason-text-hide'/></div></div></div></div></div></div></body></html>"
|
||
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() |