diff --git a/src/cronparse/scheduler.py b/src/cronparse/scheduler.py new file mode 100644 index 0000000..3ea1232 --- /dev/null +++ b/src/cronparse/scheduler.py @@ -0,0 +1,108 @@ +"""Cron execution scheduling module.""" + +from datetime import datetime, timedelta +from typing import List +from croniter import croniter + + +def get_next_executions(expression: str, count: int = 5) -> List[datetime]: + """Get the next execution times for a cron expression. + + Args: + expression: The cron expression. + count: Number of executions to return. + + Returns: + List of datetime objects for the next executions. + """ + now = datetime.now() + cron = croniter(expression, now) + executions = [] + for _ in range(count): + next_ts = cron.get_next() + exec_datetime = datetime.fromtimestamp(next_ts) + executions.append(exec_datetime) + return executions + + +def format_timeline(executions: List[datetime]) -> str: + """Format execution times as an ASCII timeline. + + Args: + executions: List of datetime objects. + + Returns: + ASCII timeline string. + """ + if not executions: + return "No executions scheduled." + + now = datetime.now() + lines = [] + lines.append("Timeline:") + + max_label_len = max(len(f"#{i}") for i in range(1, len(executions) + 1)) + max_date_len = max(len(e.strftime("%Y-%m-%d %H:%M")) for e in executions) + max_rel_len = 15 + + for i, exec_time in enumerate(executions, 1): + label = f"#{i}" + date_str = exec_time.strftime("%Y-%m-%d %H:%M") + day_name = exec_time.strftime("%A") + + delta = exec_time - now + if delta.days > 0: + relative = f"in {delta.days} day{'s' if delta.days > 1 else ''}" + elif delta.seconds > 3600: + hours = delta.seconds // 3600 + relative = f"in {hours} hour{'s' if hours > 1 else ''}" + elif delta.seconds > 60: + minutes = delta.seconds // 60 + relative = f"in {minutes} minute{'s' if minutes > 1 else ''}" + elif delta.seconds >= 0: + relative = "soon" + else: + relative = "past" + + padding_date = " " * (max_date_len - len(date_str)) + padding_rel = " " * (max_rel_len - len(relative)) + + lines.append(f" {label:>{max_label_len}} | {date_str} {padding_date} | {day_name:10} | {relative}{padding_rel}") + + lines.append("") + lines.append("Legend: # = execution number, date = execution date/time, day = day of week, relative = time from now") + + return "\n".join(lines) + + +def format_relative_time(exec_time: datetime) -> str: + """Format a datetime as relative time from now. + + Args: + exec_time: The execution datetime. + + Returns: + Human-readable relative time string. + """ + now = datetime.now() + delta = exec_time - now + + if delta.days < -365: + years = abs(delta.days) // 365 + return f"{years} year{'s' if years > 1 else ''} ago" + elif delta.days < -30: + months = abs(delta.days) // 30 + return f"{months} month{'s' if months > 1 else ''} ago" + elif delta.days < 0: + days = abs(delta.days) + return f"{days} day{'s' if days > 1 else ''} ago" + elif delta.seconds > 3600: + hours = delta.seconds // 3600 + return f"{hours} hour{'s' if hours > 1 else ''} from now" + elif delta.seconds > 60: + minutes = delta.seconds // 60 + return f"{minutes} minute{'s' if minutes > 1 else ''} from now" + elif delta.seconds >= 0: + return "in a few seconds" + else: + return "a moment ago"