Added ability to send notifications in Matrix

get_eo.py

@@ -19,10 +19,10 @@ 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'
+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}'
@@ -30,9 +30,14 @@ LAST_DATAS_FORMAT_PO = '{placed} {star_dt} {start_tm} {stop_dt} {stop_tm} {curre
 
 IS_DEBUG = False
 MENTIONS = ""
+USE_MATRIX = False
+MX_SERVER = ""
+MX_ROOM = ""
+MX_TOKEN = ""
+
 
 def read_config():
-    global MENTIONS
+    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']
@@ -43,17 +48,53 @@ def read_config():
     house = j['house']
     building_part_number = j['building_part_number']
     apartment = j['apartment']
-    for m in j['mentions']:
+    USE_MATRIX = j['use_matrix'].lower() == 'true'
-        MENTIONS += m + '\n'
+    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'}
+# 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):
@@ -80,6 +121,7 @@ def get_GPV_response_from_oe(account):
     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'
@@ -89,7 +131,7 @@ def get_PO_response(settlement, street, house, building_part_number='', apartmen
         'street': street,
         'house_number': house,
         'building_part_number': building_part_number,
-        'apartment' : apartment,
+        'apartment': apartment,
         'commit': 'Пошук'
     }
     headers = {
@@ -99,17 +141,19 @@ def get_PO_response(settlement, street, house, building_part_number='', apartmen
     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)
+        h = number.split(':')[0]
-        m = int((number * 60) % 60)
+        m = number.split(':')[1]
         if m != 0:
             value = str(h) + ':' + str(m)
         else:
@@ -123,28 +167,33 @@ def get_utf_chars(number, is_time=False):
             res += c
     return res
 
+
 def get_digit_message(arr, is_time=False):
     first = 0
-    if len(arr)> 0 and arr[0] > 0:
+    if len(arr) > 0:
         first = arr[0]
 
     last = 0
-    if len(arr)> 1 and arr[0] > 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))
+        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),
+        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:
@@ -152,16 +201,20 @@ def load_last_datas(file_name):
     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
+    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),
+    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>'
@@ -171,39 +224,49 @@ def get_GPV_message(approved_from, event_date, hours_off, hours_on, outage_queue
     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])]
+            arr = [hours_off[i], hours_on[i]]
             msg = get_digit_message(arr, True)
             message += '\n' + msg
 
-        message += (f'\n{MENTIONS}')
+        # 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
+
+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]
+        reason = root.xpath(
-        placement_date = root.xpath("//div[@class='table-row flex']/div[@class='placement_date']/div[@class='date']/text()")[0]
+            "//div[@class='table-row flex']/div[@class='reason-shutdown reason-text-container']/div[@class='reason-text-hide']/text()")[
-        start_date = root.xpath("//div[@class='table-row flex']/div[@class='shutdown_date']/div[@class='date']/text()")[0]
+            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]
+        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'
+        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'
@@ -219,7 +282,8 @@ def get_message_with_PO(root, html): #city, street, house, placement_date, PO_ty
                                                               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)
+            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()
@@ -228,6 +292,7 @@ def get_message_with_PO(root, html): #city, street, house, placement_date, PO_ty
 
     return message, need_to_send
 
+
 def process_GPV(oe_account_number, tg_chat, tg_token):
     global IS_DEBUG
     if not IS_DEBUG:
@@ -282,6 +347,12 @@ def process_GPV(oe_account_number, tg_chat, tg_token):
         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:
@@ -310,6 +381,12 @@ def process_PO(apart, house, part_number, settlement, street, tg_chat, tg_token)
         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()
@@ -323,5 +400,6 @@ def main():
 
     process_PO(apart, house, part_number, settlement, street, tg_chat, tg_token)
 
+
 if __name__ == "__main__":
     main()

get_eo_config.yaml

@@ -1,10 +1,14 @@
 token: !secret token
 chat_id: !secret chat_id
 account: !secret account
-debug: "true"
+debug: "false"
 settlement: !secret settlement
 street: !secret street
 house: "9"
 building_part_number: ""
 apartment: ""
 mentions: !secret mentions
+use_matrix: "true"
+mx_server: !secret mx_server
+mx_room_id: !secret mx_room_id
+mx_access_token: !secret mx_access_token

requirements.txt

@@ -1,4 +1,4 @@
-requests~=2.32.3
+requests~=2.32.4
 urllib3~=2.5.0
 numpy~=2.2.0
 lxml~=5.3.1