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"" 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")