From ea2ef321ee8381f16e78a75ef94182aa2602447b Mon Sep 17 00:00:00 2001 From: Yuval Adam <_@yuv.al> Date: Sun, 16 May 2021 13:40:38 +0300 Subject: Split into classes --- Pipfile | 2 +- alarmpy/alarm.py | 146 ++++++++++++++++++++++++++++++++++++++++++++++ alarmpy/alarmpy.py | 168 ----------------------------------------------------- alarmpy/cli.py | 27 +++++++++ alarmpy/display.py | 3 + setup.py | 2 +- 6 files changed, 178 insertions(+), 170 deletions(-) create mode 100644 alarmpy/alarm.py delete mode 100644 alarmpy/alarmpy.py create mode 100644 alarmpy/cli.py create mode 100644 alarmpy/display.py diff --git a/Pipfile b/Pipfile index 6a9c049..fdb2f33 100644 --- a/Pipfile +++ b/Pipfile @@ -16,7 +16,7 @@ twine = "*" build = "*" [scripts] -alarmpy = "python alarmpy/alarmpy.py" +alarmpy = "python alarmpy/cli.py" format = "black alarmpy" formatcheck = "black --check alarmpy" lint = "pylint -j 0 -d R,C alarmpy" diff --git a/alarmpy/alarm.py b/alarmpy/alarm.py new file mode 100644 index 0000000..45fe98a --- /dev/null +++ b/alarmpy/alarm.py @@ -0,0 +1,146 @@ +import click +import json +import requests + +from collections import defaultdict +from datetime import datetime +from pathlib import Path +from time import sleep, time + + +class Alarm: + + URL = "https://www.oref.org.il/WarningMessages/alert/alerts.json" + + HEADERS = { + "Referer": "https://www.oref.org.il/11226-he/pakar.aspx", + "X-Requested-With": "XMLHttpRequest", + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36", + } + + def __init__( + self, + language="he", + delay=1, + routine_delay=60 * 5, + alarm_id=False, + repeat_alarms=False, + quiet=False, + ): + self.language = language + self.delay = delay + self.last_routine_delay = routine_delay + self.alarm_id = alarm_id + self.repeat_alarms = repeat_alarms + self.quiet = quiet + + self.current_alarms = [] + self.last_routine_output = 0 + + self.session = self.init_session() + self.labels = self.load_labels() + + def init_session(self): + return requests.Session() + + def load_labels(self): + DATA_DIR = Path(__file__).parent / "data" + with open(DATA_DIR / "labels.json", "r") as f: + return json.load(f) + + def start(self): + while True: + try: + cities, alarm_id = self.fetch() + self.update(cities, alarm_id) + except Exception as e: # pylint: disable=broad-except + self.output_error(f"Exception: {e}") + finally: + sleep(self.delay) + + def fetch(self): + try: + res = self.session.get(self.URL, headers=self.HEADERS, timeout=1) + except requests.Timeout as e: + raise Exception("HTTP request timed out") from e + + if not res.content: + # empty content means no alarms + return [], None + + try: + data = res.json() + alarm_id = data["id"] + cities = data["data"] + return cities, alarm_id + except ValueError as ve: + raise Exception(f"Error parsing JSON: {res.content}") from ve + except KeyError as ke: + raise Exception(f"Missing keys in JSON data: {data}") from ke + + def update(self, cities, alarm_id): + if cities: + self.update_alarm(cities, alarm_id) + else: + self.update_routine() + + def update_alarm(self, cities, alarm_id): + if self.repeat_alarms or set(cities) != set(self.current_alarms): + self.output_alarms(cities, alarm_id) + self.current_alarms = cities + + def update_routine(self): + now = time() + if ( + self.current_alarms + or now - self.last_routine_output > self.last_routine_delay + ): + self.output_routine() + self.last_routine_output = now + self.current_alarms = [] + + def output_leading_timestamp(self, nl=False): + now = datetime.now() + ts = now.strftime("%Y-%m-%d %H:%M:%S") + click.secho(f"{ts} ", nl=nl) + + def output_error(self, err): + if not self.quiet: + self.output_leading_timestamp() + click.secho(err, fg="yellow") + + def output_routine(self): + if not self.quiet: + self.output_leading_timestamp() + click.secho("No active alarms", fg="green") + + def output_alarms(self, cities, alarm_id): + areas = self.group_areas_and_localize(cities) + multiple_areas = len(areas) > 1 + self.output_leading_timestamp(nl=multiple_areas) + for area, cities in areas.items(): + cities_str = ", ".join(cities) + leading_tab = "\t" if multiple_areas else "" + click.secho(f"{leading_tab}{area} ", fg="red", bold=True, nl=False) + click.secho(f"\t{cities_str} ", fg="red") + if self.alarm_id: + click.secho(f"({alarm_id})") + + def group_areas_and_localize(self, cities): + res = defaultdict(list) + for city in cities: + try: + area = self.labels[city][f"areaname_{self.language}"] + label = self.labels[city][f"label_{self.language}"] + except KeyError: + area = "Unknown" + label = city + res[area].append(label) + return res + + def localize_cities(self, cities): + localized_cities = [ + self.labels.get(city, {}).get(f"label_{self.language}", city) + for city in cities + ] + return ", ".join(localized_cities) diff --git a/alarmpy/alarmpy.py b/alarmpy/alarmpy.py deleted file mode 100644 index f230cd4..0000000 --- a/alarmpy/alarmpy.py +++ /dev/null @@ -1,168 +0,0 @@ -import click -import json -import requests - -from collections import defaultdict -from datetime import datetime -from pathlib import Path -from time import sleep, time - - -class Alarm: - - URL = "https://www.oref.org.il/WarningMessages/alert/alerts.json" - - HEADERS = { - "Referer": "https://www.oref.org.il/11226-he/pakar.aspx", - "X-Requested-With": "XMLHttpRequest", - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36", - } - - def __init__( - self, - language="he", - delay=1, - routine_delay=60 * 5, - alarm_id=False, - repeat_alarms=False, - quiet=False, - ): - self.language = language - self.delay = delay - self.last_routine_delay = routine_delay - self.alarm_id = alarm_id - self.repeat_alarms = repeat_alarms - self.quiet = quiet - - self.current_alarms = [] - self.last_routine_output = 0 - - self.session = self.init_session() - self.labels = self.load_labels() - - def init_session(self): - return requests.Session() - - def load_labels(self): - DATA_DIR = Path(__file__).parent / "data" - with open(DATA_DIR / "labels.json", "r") as f: - return json.load(f) - - def start(self): - while True: - try: - cities, alarm_id = self.fetch() - self.update(cities, alarm_id) - except Exception as e: # pylint: disable=broad-except - self.output_error(f"Exception: {e}") - finally: - sleep(self.delay) - - def fetch(self): - try: - res = self.session.get(self.URL, headers=self.HEADERS, timeout=1) - except requests.Timeout as e: - raise Exception("HTTP request timed out") from e - - if not res.content: - # empty content means no alarms - return [], None - - try: - data = res.json() - alarm_id = data["id"] - cities = data["data"] - return cities, alarm_id - except ValueError as ve: - raise Exception(f"Error parsing JSON: {res.content}") from ve - except KeyError as ke: - raise Exception(f"Missing keys in JSON data: {data}") from ke - - def update(self, cities, alarm_id): - if cities: - self.update_alarm(cities, alarm_id) - else: - self.update_routine() - - def update_alarm(self, cities, alarm_id): - if self.repeat_alarms or set(cities) != set(self.current_alarms): - self.output_alarms(cities, alarm_id) - self.current_alarms = cities - - def update_routine(self): - now = time() - if ( - self.current_alarms - or now - self.last_routine_output > self.last_routine_delay - ): - self.output_routine() - self.last_routine_output = now - self.current_alarms = [] - - def output_leading_timestamp(self, nl=False): - now = datetime.now() - ts = now.strftime("%Y-%m-%d %H:%M:%S") - click.secho(f"{ts} ", nl=nl) - - def output_error(self, err): - if not self.quiet: - self.output_leading_timestamp() - click.secho(err, fg="yellow") - - def output_routine(self): - if not self.quiet: - self.output_leading_timestamp() - click.secho("No active alarms", fg="green") - - def output_alarms(self, cities, alarm_id): - areas = self.group_areas_and_localize(cities) - multiple_areas = len(areas) > 1 - self.output_leading_timestamp(nl=multiple_areas) - for area, cities in areas.items(): - cities_str = ", ".join(cities) - leading_tab = "\t" if multiple_areas else "" - click.secho(f"{leading_tab}{area} ", fg="red", bold=True, nl=False) - click.secho(f"\t{cities_str} ", fg="red") - if self.alarm_id: - click.secho(f"({alarm_id})") - - def group_areas_and_localize(self, cities): - res = defaultdict(list) - for city in cities: - try: - area = self.labels[city][f"areaname_{self.language}"] - label = self.labels[city][f"label_{self.language}"] - except KeyError: - area = "Unknown" - label = city - res[area].append(label) - return res - - def localize_cities(self, cities): - localized_cities = [ - self.labels.get(city, {}).get(f"label_{self.language}", city) - for city in cities - ] - return ", ".join(localized_cities) - - -@click.command() -@click.option( - "--language", - default="he", - type=click.Choice(["en", "he", "ar", "ru"]), - help="Alert language ", -) -@click.option("--delay", default=1, help="Polling delay in seconds") -@click.option( - "--routine-delay", default=60 * 5, help="Routine message delay in seconds" -) -@click.option("--alarm-id", is_flag=True, help="Print alarm IDs") -@click.option("--repeat-alarms", is_flag=True, help="Do not suppress ongoing alarms") -@click.option("--quiet", is_flag=True, help="Print only active alarms") -def cli(**kwargs): - Alarm(**kwargs).start() - - -if __name__ == "__main__": - cli() diff --git a/alarmpy/cli.py b/alarmpy/cli.py new file mode 100644 index 0000000..790fcd2 --- /dev/null +++ b/alarmpy/cli.py @@ -0,0 +1,27 @@ +import click + +from alarm import Alarm +from display import Display + + +@click.command() +@click.option( + "--language", + default="he", + type=click.Choice(["en", "he", "ar", "ru"]), + help="Alert language ", +) +@click.option("--delay", default=1, help="Polling delay in seconds") +@click.option( + "--routine-delay", default=60 * 5, help="Routine message delay in seconds" +) +@click.option("--alarm-id", is_flag=True, help="Print alarm IDs") +@click.option("--repeat-alarms", is_flag=True, help="Do not suppress ongoing alarms") +@click.option("--quiet", is_flag=True, help="Print only active alarms") +def cli(**kwargs): + _d = Display(**kwargs) + Alarm(**kwargs).start() + + +if __name__ == "__main__": + cli() diff --git a/alarmpy/display.py b/alarmpy/display.py new file mode 100644 index 0000000..d1ecb09 --- /dev/null +++ b/alarmpy/display.py @@ -0,0 +1,3 @@ +class Display: + def __init__(self, **_kwargs): + pass diff --git a/setup.py b/setup.py index 0f3aa40..b9afa4b 100644 --- a/setup.py +++ b/setup.py @@ -27,5 +27,5 @@ setuptools.setup( package_data={"alarmpy": ["data/*.json"]}, python_requires=">=3.6", install_requires=["requests", "click", "blessed"], - entry_points={"console_scripts": ["alarmpy = alarmpy:cli"]}, + entry_points={"console_scripts": ["alarmpy = cli:cli"]}, ) -- cgit v1.3.1