heimdall/core
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

[issue-3] [mod] store timestamp of last notification in state file

Browse source
This commit is contained in:
Christian Fraß 2023-03-22 11:24:13 +01:00
parent 386e1d5891
commit 3ae65c08db
1 changed files with 47 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
source/logic/main.py
View file

@ -3,6 +3,7 @@ def state_encode(state):
"timestamp": state["timestamp"],
"condition": condition_encode(state["condition"]),
"count": state["count"],
"last_notification_timestamp": state["last_notification_timestamp"],
}
@ -11,6 +12,7 @@ def state_decode(state_encoded):
"timestamp": state_encoded["timestamp"],
"condition": condition_decode(state_encoded["condition"]),
"count": state_encoded["count"],
"last_notification_timestamp": state_encoded["last_notification_timestamp"],
}
@ -234,10 +236,7 @@ def main():
"error": str(error),
},
}
new_item_state = {
"timestamp": timestamp,
"condition": result["condition"],
"count": (
count = (
1
if (
(old_item_state is None)
@ -253,33 +252,47 @@ def main():
) else
None
)
),
}
state_data[check_data["name"]] = state_encode(new_item_state)
file_write(state_path, _json.dumps(state_data, indent = "\t"))
### send notifications
if (
)
shall_send_notification = (
(
(
(new_item_state["count"] is not None)
(count is not None)
and
(new_item_state["count"] == check_data["threshold"])
(count == check_data["threshold"])
)
or
(
(new_item_state["count"] is None)
(count is None)
and
check_data["annoy"]
)
)
and
(
(new_item_state["condition"] != enum_condition.ok)
(result["condition"] != enum_condition.ok)
or
args.send_ok_notifications
)
):
)
new_item_state = {
"timestamp": timestamp,
"condition": result["condition"],
"count": count,
"last_notification_timestamp": (
timestamp
if shall_send_notification else
(
None
if (old_item_state is None) else
old_item_state["last_notification_timestamp"]
)
),
}
state_data[check_data["name"]] = state_encode(new_item_state)
file_write(state_path, _json.dumps(state_data, indent = "\t"))
### send notifications
if shall_send_notification:
for notification in check_data["notifications"]:
notification_channel_implementations[notification["kind"]].notify(
notification["parameters"],