|
2 | 2 | import logging
|
3 | 3 | import re
|
4 | 4 | from datetime import datetime, timedelta
|
5 |
| -from typing import List, Union |
| 5 | +from typing import cast, List, Union |
6 | 6 | from threading import Thread
|
7 | 7 | from time import sleep
|
8 | 8 |
|
@@ -50,7 +50,9 @@ def __init__(self, config: dict, notify_manager: NotifyManager):
|
50 | 50 | f"Summary notifications will be sent out every {self._frequency_hours} "
|
51 | 51 | f"hours starting from {self._notify_time['hour']:02d}:{self._notify_time['minute']:02d}"
|
52 | 52 | )
|
53 |
| - self._datetime_next_summary = datetime.now().replace(hour=self._notify_time['hour'], minute=self._notify_time['minute'], second=0, microsecond=0) |
| 53 | + self._datetime_next_summary = datetime.now().replace( |
| 54 | + hour=self._notify_time["hour"], minute=self._notify_time["minute"], second=0, microsecond=0 |
| 55 | + ) |
54 | 56 | while datetime.now() > self._datetime_next_summary:
|
55 | 57 | self._datetime_next_summary += timedelta(hours=self._frequency_hours)
|
56 | 58 |
|
@@ -103,12 +105,13 @@ def _run_loop(self):
|
103 | 105 | def stop(self):
|
104 | 106 | self._is_running = False
|
105 | 107 |
|
106 |
| - def _parse_notify_time(self, value: Union[str, int], default: dict = {'hour': 21, 'minute': 0}) -> dict: |
| 108 | + def _parse_notify_time(self, value: Union[str, int], default: dict = {"hour": 21, "minute": 0}) -> dict: |
107 | 109 | if type(value) == int:
|
108 |
| - return {'hour': value, 'minute': 0} |
| 110 | + return {"hour": value, "minute": 0} |
109 | 111 |
|
110 |
| - match = re.match('(?:[01]\d|2[0-3]):(?:[0-5]\d)', value) |
| 112 | + value = cast(str, value) |
| 113 | + match = re.match(r"(?:[01]\d|2[0-3]):(?:[0-5]\d)", value) |
111 | 114 | if match:
|
112 |
| - return {'hour': int(value[:2]), 'minute': int(value[-2:])} |
| 115 | + return {"hour": int(value[:2]), "minute": int(value[-2:])} |
113 | 116 |
|
114 | 117 | return default
|
0 commit comments