Initial commit

This commit is contained in:
Anry Das 2025-11-01 10:25:24 +02:00
commit 2f970d8799
5 changed files with 419 additions and 0 deletions

75
config_file.py Normal file
View File

@ -0,0 +1,75 @@
import json
import os
import yaml
def read_config(name):
if not os.path.isfile(name):
raise Exception(f"File {name} doesn't exists")
filename, ext = os.path.splitext(name)
if 'json' in ext:
return read_json(name)
elif 'properties' in ext:
return read_prop(name)
elif 'yaml' in ext or 'yml' in ext:
return read_yaml(name)
else:
raise Exception("Wrong file type")
def read_json(name):
with open(name, 'r', encoding='utf-8') as f:
j_conf = json.load(f)
conf = {}
for key, value in j_conf.items():
conf[key] = value
return conf
def read_prop(filepath, sep='=', comment_char='#'):
"""
Read the file passed as parameter as a properties file.
"""
conf = {}
with open(filepath, "rt") as f:
for line in f:
l = line.strip()
if l and not l.startswith(comment_char):
key_value = l.split(sep)
key = key_value[0].strip()
value = sep.join(key_value[1:]).strip().strip('"')
conf[key] = value
return conf
def read_yaml(name, secrets_file='secrets.yaml'):
# Load secrets first
secrets = {}
secrets_path = os.path.join(os.path.dirname(name), secrets_file)
if os.path.exists(secrets_path):
with open(secrets_path, 'r', encoding='utf-8') as f:
secrets = yaml.safe_load(f) or {}
# Define a custom constructor for !secret tag
def secret_constructor(loader, node):
secret_key = loader.construct_scalar(node)
if secret_key not in secrets:
raise ValueError(f"Secret '{secret_key}' not found in {secrets_file}")
return secrets[secret_key]
# Register the custom constructor
yaml.add_constructor('!secret', secret_constructor, Loader=yaml.SafeLoader)
# Load the main configuration
conf = {}
with open(name, 'r', encoding='utf-8') as f:
y_conf = yaml.safe_load(f)
print(f'{[y_conf]}')
if y_conf:
for key, value in y_conf.items():
conf[key] = value
return conf
def main():
pass
if __name__ == "__main__":
main()

319
get_eo.py Normal file
View File

@ -0,0 +1,319 @@
#!/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
MENTIONED_USERS = '@Kirden0'
def read_config():
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']
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)
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 <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 = [int(hours_off[i].split(':')[0]), int(hours_on[i].split(':')[0])]
msg = get_digit_message(arr, True)
message += '\n' + msg
message += (f'\n{MENTIONED_USERS}')
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'{MENTIONED_USERS}')
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 = "<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/Пр.&quot;ЗБВIК&quot;/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)
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()

9
get_eo_config.yaml Normal file
View File

@ -0,0 +1,9 @@
token: !secret token
chat_id: !secret chat_id
account: !secret account
debug: "true"
settlement: !secret settlement
street: !secret street
house: "9"
building_part_number: ""
apartment: ""

5
requirements.txt Normal file
View File

@ -0,0 +1,5 @@
requests~=2.32.3
urllib3~=2.5.0
numpy~=2.2.0
lxml~=5.3.1
PyYAML~=6.0.3

11
start.sh Executable file
View File

@ -0,0 +1,11 @@
#!/bin/bash
cd /home/das/app/EO_info_bot
. ./.venv/bin/activate
python ./get_eo.py
deactivate
exit 0