feat: add functions to check and update 4:20 timezone cache
Build and Deploy Docker Container / test (push) Successful in 14s
Build and Deploy Docker Container / build-and-deploy (push) Failing after 30s

This commit is contained in:
2026-05-10 13:06:31 +02:00
parent 565c4078bb
commit f88f60a019
+41 -32
View File
@@ -71,6 +71,45 @@ def get_state_snapshot() -> dict:
return dict(STATE)
def _check_420(tz_list: list[str] | None = None) -> str:
cache = get_tzdb_cache()
timezones = load_timezones() if cache is None else []
countries = load_countries() if cache is None else []
if tz_list is None:
if cache is None:
tz_list = where_is_it_420(timezones, countries)
else:
tz_list = where_is_it_420(
timezones,
countries,
tz_names=cache.get("tz_names"),
tz_to_country_code=cache.get("tz_to_country_code"),
country_code_to_name=cache.get("country_code_to_name"),
)
if tz_list:
tz_str = "\n".join(tz_list)
return f"\nIt's 4:20 in:\n{tz_str}"
return ""
def _update_420_cache() -> None:
try:
cache = get_tzdb_cache()
if cache is None:
tz_list = where_is_it_420(load_timezones(), load_countries())
else:
tz_list = where_is_it_420(
[],
[],
tz_names=cache.get("tz_names"),
tz_to_country_code=cache.get("tz_to_country_code"),
country_code_to_name=cache.get("country_code_to_name"),
)
_update_state(last_locations=tz_list or [])
except Exception as e:
_update_state(last_locations=[], last_error=str(e))
def get_next_scheduled_event(now: datetime | None = None) -> dict:
"""Return the next scheduled notification time/type based on known minute marks."""
if now is None:
@@ -98,9 +137,6 @@ def create_embed(type: str, tz_list: list[str] | None = None) -> dict:
"""
templates_path = os.getenv("TEMPLATES_PATH", "templates.json")
messages = load_templates(templates_path)
cache = get_tzdb_cache()
timezones = load_timezones() if cache is None else []
countries = load_countries() if cache is None else []
if type in messages:
msg = messages[type]
image_url = msg.get("image_url")
@@ -108,20 +144,7 @@ def create_embed(type: str, tz_list: list[str] | None = None) -> dict:
msg["image"] = {"url": image_url}
if type == "420":
# Check where it's 4:20
if tz_list is None:
if cache is None:
tz_list = where_is_it_420(timezones, countries)
else:
tz_list = where_is_it_420(
timezones,
countries,
tz_names=cache.get("tz_names"),
tz_to_country_code=cache.get("tz_to_country_code"),
country_code_to_name=cache.get("country_code_to_name"),
)
if tz_list:
tz_str = "\n".join(tz_list)
msg["text"] += f"\nIt's 4:20 in:\n{tz_str}"
msg["text"] += _check_420(tz_list)
else:
msg = {"text": "Unknown notification type", "color": 0xFF0000}
embed = {
@@ -159,21 +182,7 @@ def send_notification(message: str) -> None:
tz_list: list[str] | None = None
if message == "420":
try:
cache = get_tzdb_cache()
if cache is None:
tz_list = where_is_it_420(load_timezones(), load_countries())
else:
tz_list = where_is_it_420(
[],
[],
tz_names=cache.get("tz_names"),
tz_to_country_code=cache.get("tz_to_country_code"),
country_code_to_name=cache.get("country_code_to_name"),
)
_update_state(last_locations=tz_list or [])
except Exception as e:
_update_state(last_locations=[], last_error=str(e))
_update_420_cache()
embed = create_embed(message, tz_list=tz_list)
data = {"embeds": [embed]}