119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime
|
|
import logging
|
|
import os
|
|
from typing import TYPE_CHECKING
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import dateparser
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from apscheduler.triggers.date import DateTrigger
|
|
from apscheduler.triggers.interval import IntervalTrigger
|
|
|
|
if TYPE_CHECKING:
|
|
from apscheduler.job import Job
|
|
|
|
logger: logging.Logger = logging.getLogger("discord_reminder_bot")
|
|
|
|
|
|
def parse_time(date_to_parse: str | None, timezone: str | None = os.getenv("TIMEZONE")) -> datetime.datetime | None:
|
|
"""Parse a date string into a datetime object.
|
|
|
|
Args:
|
|
date_to_parse(str): The date string to parse.
|
|
timezone(str, optional): The timezone to use. Defaults to the TIMEZONE environment variable.
|
|
|
|
Returns:
|
|
datetime.datetime: The parsed datetime object.
|
|
"""
|
|
if not date_to_parse:
|
|
logger.error("No date provided to parse.")
|
|
return None
|
|
|
|
if not timezone:
|
|
logger.error("No timezone provided to parse date.")
|
|
return None
|
|
|
|
logger.info("Parsing date: '%s' with timezone: '%s'", date_to_parse, timezone)
|
|
|
|
try:
|
|
parsed_date: datetime.datetime | None = dateparser.parse(
|
|
date_string=date_to_parse,
|
|
settings={
|
|
"PREFER_DATES_FROM": "future",
|
|
"TIMEZONE": f"{timezone}",
|
|
"RETURN_AS_TIMEZONE_AWARE": True,
|
|
"RELATIVE_BASE": datetime.datetime.now(tz=ZoneInfo(str(timezone))),
|
|
},
|
|
)
|
|
except (ValueError, TypeError):
|
|
logger.exception("Failed to parse date: '%s'", date_to_parse)
|
|
return None
|
|
|
|
logger.debug("Parsed date: %s with timezone: %s", parsed_date, timezone)
|
|
|
|
return parsed_date
|
|
|
|
|
|
def calculate(job: Job) -> str:
|
|
"""Calculate the time left for a job.
|
|
|
|
Args:
|
|
job: The job to calculate the time for.
|
|
|
|
Returns:
|
|
str: The time left for the job or "Paused" if the job is paused or has no next run time.
|
|
"""
|
|
trigger_time = None
|
|
if isinstance(job.trigger, DateTrigger | IntervalTrigger):
|
|
if not hasattr(job, "next_run_time"):
|
|
logger.debug("No next run time found for '%s'", job.id)
|
|
logger.debug("%s", job.__getstate__())
|
|
return "Paused"
|
|
trigger_time = job.next_run_time or None
|
|
|
|
elif isinstance(job.trigger, CronTrigger):
|
|
if not job.next_run_time:
|
|
logger.debug("No next run time found for '%s', probably paused?", job.id)
|
|
logger.debug("%s", job.__getstate__())
|
|
return "Paused"
|
|
|
|
trigger_time = job.trigger.get_next_fire_time(None, datetime.datetime.now(tz=job._scheduler.timezone)) # noqa: SLF001
|
|
|
|
logger.debug("Trigger type: %s, Trigger time: %s", type(job.trigger), trigger_time)
|
|
|
|
if not trigger_time:
|
|
logger.debug("No trigger time found")
|
|
return "Paused"
|
|
|
|
return f"<t:{int(trigger_time.timestamp())}:R>"
|
|
|
|
|
|
def get_human_readable_time(job: Job) -> str:
|
|
"""Get the human-readable time for a job.
|
|
|
|
Args:
|
|
job: The job to get the time for.
|
|
|
|
Returns:
|
|
str: The human-readable time.
|
|
"""
|
|
trigger_time = None
|
|
if isinstance(job.trigger, DateTrigger | IntervalTrigger):
|
|
trigger_time = job.next_run_time or None
|
|
|
|
elif isinstance(job.trigger, CronTrigger):
|
|
if not job.next_run_time:
|
|
logger.debug("No next run time found for '%s', probably paused?", job.id)
|
|
logger.debug("%s", job.__getstate__())
|
|
return "Paused"
|
|
|
|
trigger_time = job.trigger.get_next_fire_time(None, datetime.datetime.now(tz=job._scheduler.timezone)) # noqa: SLF001
|
|
|
|
if not trigger_time:
|
|
logger.debug("No trigger time found")
|
|
return "Paused"
|
|
|
|
return trigger_time.strftime("%Y-%m-%d %H:%M:%S")
|