From 2f970d8799e7480e0610882996cbd5617b73a730 Mon Sep 17 00:00:00 2001 From: Anry Das Date: Sat, 1 Nov 2025 10:25:24 +0200 Subject: [PATCH] Initial commit --- config_file.py | 75 +++++++++++ get_eo.py | 319 +++++++++++++++++++++++++++++++++++++++++++++ get_eo_config.yaml | 9 ++ requirements.txt | 5 + start.sh | 11 ++ 5 files changed, 419 insertions(+) create mode 100644 config_file.py create mode 100644 get_eo.py create mode 100644 get_eo_config.yaml create mode 100644 requirements.txt create mode 100755 start.sh diff --git a/config_file.py b/config_file.py new file mode 100644 index 0000000..40f4caa --- /dev/null +++ b/config_file.py @@ -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() \ No newline at end of file diff --git a/get_eo.py b/get_eo.py new file mode 100644 index 0000000..92877d5 --- /dev/null +++ b/get_eo.py @@ -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 ГПВ на {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{MENTIONED_USERS}') + 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'{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 = "
Район
Населений пункт
Вулиця
№ буд
" \ + "
Корпус
Диспетчерська назва об'єкту
Вид робіт
Причина робіт
Дата розміщення
" \ + "
Дата і час вимкнення
Дата і час включення
Наступна дата робіт
Піднянський район
" \ + "
Гірна
Тевського
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() \ No newline at end of file diff --git a/get_eo_config.yaml b/get_eo_config.yaml new file mode 100644 index 0000000..b5003d0 --- /dev/null +++ b/get_eo_config.yaml @@ -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: "" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..18280be --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +requests~=2.32.3 +urllib3~=2.5.0 +numpy~=2.2.0 +lxml~=5.3.1 +PyYAML~=6.0.3 \ No newline at end of file diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..045922e --- /dev/null +++ b/start.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +cd /home/das/app/EO_info_bot + +. ./.venv/bin/activate + +python ./get_eo.py + +deactivate + +exit 0