#!/usr/bin/env python # -*- coding: utf-8 -*- import threading import paho.mqtt.client as mqtt import urllib3 import os import requests import re from loguru import logger import time import codecs import apprise import json from whatsapp_api_client_python import API os.environ['PYTHONIOENCODING'] = 'utf-8' os.environ['LANG'] = 'C.UTF-8' #mqtt connection Params server = os.getenv('MQTT_HOST') #Default port is 1883 port = int(os.getenv('MQTT_PORT', '1883')) user = os.getenv('MQTT_USER') passw = os.getenv('MQTT_PASS') debug = os.getenv('DEBUG_MODE') region = os.getenv('REGION') NOTIFIERS = os.getenv("NOTIFIERS") MQTT_TOPIC = os.environ.get("MQTT_TOPIC", "/redalert") INCLUDE_TEST_ALERTS = os.getenv("INCLUDE_TEST_ALERTS") GREEN_API_INSTANCE = os.getenv("GREEN_API_INSTANCE") GREEN_API_TOKEN = os.getenv("GREEN_API_TOKEN") WHATSAPP_NUMBER = os.getenv("WHATSAPP_NUMBER") logger.info("Monitoring alerts for :" + region) #Setting Request Headers http = urllib3.PoolManager() _headers = {'Referer':'https://www.oref.org.il/heb','Accept':'application/json, text/plain, */*','User-Agent':"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.97 Safari/537.36",'X-Requested-With':'XMLHttpRequest'} url= 'https://www.oref.org.il/warningMessages/alert/Alerts.json' if debug == 'True': url = 'http://localhost/alerts.json' #Check Connection Status def on_connect(client, userdata, flags, rc): if rc==0: client.connected_flag=True #set flag logger.info("connected OK Returned code=" + str(rc)) else: if rc==1: logger.error("Connection refused – incorrect protocol version") if rc==2: logger.error("Connection refused – invalid client identifier") if rc==3: logger.error("Connection refused – server unavailable") if rc==4: logger.error("Connection refused – bad username or password") if rc==5: logger.error("Connection refused – not authorised") def on_disconnect(client, userdata, rc): logger.info("disconnecting reason " +str(rc)) client.connected_flag=False client.disconnect_flag=True client.connect(server) alerts = [0] # Setting apprise Job Manager apobj = apprise.Apprise() #Setting up MqttClient (optional - only when MQTT_HOST is set) client = None if server: client = mqtt.Client("redalert") client.username_pw_set(user,passw) client.on_connect=on_connect client.on_disconnect=on_disconnect client.loop_start() logger.info("Connecting to broker") mqtt.Client.connected_flag=False#create flag in class client.connect(server, port=port, keepalive=3600) while not client.connected_flag: #wait in loop logger.info("In wait loop") time.sleep(1) logger.info("in Main Loop") client.loop_stop() #Stop loop else: logger.info("MQTT_HOST not set, MQTT publishing disabled") def mqtt_publish(topic, payload): if client is not None: client.publish(topic, payload, qos=0, retain=False) if len(NOTIFIERS)!=0: logger.info("Setting Apprise Alert") jobs=NOTIFIERS.split() for job in jobs: logger.info("Adding: " + job) apobj.add(job) def alarm_on(data): mqtt_publish(MQTT_TOPIC + "/data",str(data["data"])) mqtt_publish(MQTT_TOPIC,'on') title=str(data["title"]) wapp_title = f"*{title}*" categorized_places = categorize_places(lamas, data["data"]) places = format_output(categorized_places) body='באזורים הבאים: \r\n ' + places if len(NOTIFIERS)!=0: logger.info("Alerting using Notifires") apobj.notify( body=body, title=title, ) try: if GREEN_API_INSTANCE and GREEN_API_TOKEN: green_api = API.GreenAPI(GREEN_API_INSTANCE, GREEN_API_TOKEN) green_api.sending.sendMessage(WHATSAPP_NUMBER,wapp_title + "\r\n" + body) except Exception as e: logger.error(f"Error sending whatsapp message. \n {str(e)}") def alarm_off(): mqtt_publish(MQTT_TOPIC + "/alarm",'off') mqtt_publish(MQTT_TOPIC,"No active alerts") def is_test_alert(alert): # if includes, all alerts are treated as not test return INCLUDE_TEST_ALERTS == 'False' and ('בדיקה' in alert['data'] or 'בדיקה מחזורית' in alert['data']) def standardize_name(name): return re.sub(r'[\(\)\'\"]+', '', name).strip() def load_lamas_data(): file_path = 'lamas.json' github_url = "https://raw.githubusercontent.com/t0mer/Redalert/refs/heads/master/lamas.json" try: with open(file_path, 'r', encoding='utf-8') as file: lamas_data = json.load(file) logger.info("Lamas data loaded from local file") except FileNotFoundError: logger.error(f"Error: Lamas data file not found at {file_path}, attempting to download from GitHub") lamas_data = download_lamas_data(github_url, file_path) except json.JSONDecodeError: logger.error("Error: Invalid JSON in Lamas data file, attempting to download from GitHub") lamas_data = download_lamas_data(github_url, file_path) except Exception as e: logger.error(f"Error loading Lamas data from local file: {e}") lamas_data = download_lamas_data(github_url, file_path) if 'areas' not in lamas_data: logger.error("Unexpected JSON structure in Lamas data") return None for area, cities in lamas_data['areas'].items(): if isinstance(cities, dict): standardized_cities = set([ standardize_name(city) for city in cities.keys() ]) else: logger.error(f"Unexpected cities structure in area {area}: {cities}") standardized_cities = set() lamas_data['areas'][area] = standardized_cities return lamas_data def download_lamas_data(url, file_path): try: response = requests.get(url) response.raise_for_status() lamas_data = response.json() with open(file_path, 'w', encoding='utf-8') as file: json.dump(lamas_data, file, ensure_ascii=False, indent=2) logger.info(f"Lamas data downloaded from GitHub and saved to {file_path}") return lamas_data except requests.exceptions.RequestException as e: logger.error(f"Error downloading Lamas data from GitHub: {e}") return None except json.JSONDecodeError as e: logger.error(f"Error decoding Lamas data from GitHub: {e}") return None except Exception as e: logger.error(f"Unexpected error: {e}") return None def categorize_places(data, places): categorized_places = {} areas_dict = {area: set(area_places) for area, area_places in data['areas'].items()} for place in places: found = False for area, area_places in areas_dict.items(): if place in area_places: categorized_places.setdefault(area, []).append(place) found = True break if not found: categorized_places.setdefault("כללי", []).append(place) return categorized_places def format_output(categorized_places): result = [] for area in sorted(categorized_places.keys()): # Sort places a-z in area sorted_places = sorted(categorized_places[area]) # Join the cities/places with commas places_str = "\r\n".join(sorted_places) result.append(f"*ישובי {area}*: {places_str}") return "\r\n".join(result) def monitor(): #start the timer threading.Timer(1, monitor).start() #Check if data contains alert data try: r = http.request('GET',url,headers=_headers) r.encoding = 'utf-8' alert_data = r.data.decode('utf-8-sig') alert_data = alert_data.replace('\x00', '').strip() if alert_data and alert_data != '' and not alert_data.isspace(): alert = json.loads(alert_data) logger.info("Alert data successfully parsed.") if region in alert["data"] or region=="*": if alert["id"] not in alerts and not is_test_alert(alert): alerts.append(alert["id"]) alarm_on(alert) logger.info(str(alert)) else: alarm_off() except Exception as ex: logger.error(str(ex)) finally: r.release_conn() if __name__ == '__main__': lamas = load_lamas_data() monitor()