Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/cronpal/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
from cronpal.field_parser import FieldParser
from cronpal.models import CronExpression, CronField, FieldType
from cronpal.parser import create_parser
from cronpal.scheduler import CronScheduler
from cronpal.special_parser import SpecialStringParser
from cronpal.validators import validate_expression

Expand All @@ -25,6 +26,7 @@
"FieldType",
"FieldParser",
"SpecialStringParser",
"CronScheduler",
"validate_expression",
"CronPalError",
"InvalidCronExpression",
Expand Down
63 changes: 63 additions & 0 deletions src/cronpal/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,14 @@
"""Main CLI entry point for CronPal."""

import sys
from datetime import datetime

from cronpal.error_handler import ErrorHandler, suggest_fix
from cronpal.exceptions import CronPalError
from cronpal.field_parser import FieldParser
from cronpal.models import CronExpression
from cronpal.parser import create_parser
from cronpal.scheduler import CronScheduler
from cronpal.special_parser import SpecialStringParser
from cronpal.validators import validate_expression, validate_expression_format

Expand Down Expand Up @@ -70,6 +72,10 @@ def main(args=None):
print(f" Validation: PASSED")
_print_verbose_fields(cron_expr)

# Show next run times if requested
if parsed_args.next is not None:
_print_next_runs(cron_expr, parsed_args.next)

return 0

except CronPalError as e:
Expand Down Expand Up @@ -179,5 +185,62 @@ def _print_day_names(prefix: str, values: set):
print(f"{prefix}Days: {', '.join(names)}")


def _print_next_runs(cron_expr: CronExpression, count: int):
"""
Print the next run times for a cron expression.

Args:
cron_expr: The CronExpression to calculate runs for.
count: Number of next runs to show.
"""
# Don't show next runs for @reboot
if cron_expr.raw_expression.lower() == "@reboot":
print("\nNext runs: @reboot only runs at system startup")
return

# Make sure we have parsed fields
if not cron_expr.is_valid():
print("\nNext runs: Cannot calculate - incomplete expression")
return

try:
scheduler = CronScheduler(cron_expr)
next_runs = scheduler.get_next_runs(count)

print(f"\nNext {count} run{'s' if count != 1 else ''}:")
for i, run_time in enumerate(next_runs, 1):
# Format the datetime nicely
formatted = run_time.strftime("%Y-%m-%d %H:%M:%S %A")

# Add relative time for first few entries
if i <= 3:
now = datetime.now()
delta = run_time - now

if delta.days == 0:
if delta.seconds < 3600:
minutes = delta.seconds // 60
relative = f"in {minutes} minute{'s' if minutes != 1 else ''}"
else:
hours = delta.seconds // 3600
relative = f"in {hours} hour{'s' if hours != 1 else ''}"
elif delta.days == 1:
relative = "tomorrow"
elif delta.days < 7:
relative = f"in {delta.days} days"
else:
relative = ""

if relative:
print(f" {i}. {formatted} ({relative})")
else:
print(f" {i}. {formatted}")
else:
print(f" {i}. {formatted}")

except Exception as e:
print(f"\nNext runs: Error calculating - {e}")


if __name__ == "__main__":
sys.exit(main())
84 changes: 54 additions & 30 deletions src/cronpal/scheduler.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Scheduler for calculating cron expression run times."""

from datetime import datetime, timedelta
from typing import Optional
from typing import List, Optional

from cronpal.exceptions import CronPalError
from cronpal.models import CronExpression
Expand Down Expand Up @@ -67,6 +67,38 @@ def get_next_run(self, after: Optional[datetime] = None) -> datetime:

raise CronPalError("Could not find next run time within reasonable limits")

def get_next_runs(self, count: int, after: Optional[datetime] = None) -> List[datetime]:
"""
Calculate multiple next run times for the cron expression.

Args:
count: Number of next run times to calculate.
after: The datetime to start searching from.
Defaults to current time if not provided.

Returns:
List of next run times.

Raises:
ValueError: If count is less than 1.
"""
if count < 1:
raise ValueError("Count must be at least 1")

if after is None:
after = datetime.now()

runs = []
current = after

for _ in range(count):
next_run = self.get_next_run(current)
runs.append(next_run)
# Start next search 1 minute after the found time
current = next_run + timedelta(minutes=1)

return runs

def _matches_time(self, dt: datetime) -> bool:
"""
Check if a datetime matches the cron expression.
Expand Down Expand Up @@ -200,7 +232,7 @@ def _get_next_day(self, dt: datetime) -> Optional[datetime]:
break

test_dt = dt.replace(day=day, hour=first_hour, minute=first_minute,
second=0, microsecond=0)
second=0, microsecond=0)

# Check if this day matches day constraints
day_of_month_match = day in self.cron_expr.day_of_month.parsed_values
Expand Down Expand Up @@ -234,15 +266,29 @@ def _get_next_month(self, dt: datetime) -> datetime:
first_hour = valid_hours[0] if valid_hours else 0
first_minute = valid_minutes[0] if valid_minutes else 0

# Try remaining months this year
for month in valid_months:
if month > dt.month:
# Start searching from current year
current_year = dt.year
current_month = dt.month

# Search for up to 10 years to handle February 29th cases
for year_offset in range(10):
search_year = current_year + year_offset

# Determine which months to check this year
if year_offset == 0:
# For current year, only check months after current month
months_to_check = [m for m in valid_months if m > current_month]
else:
# For future years, check all valid months
months_to_check = valid_months

for month in months_to_check:
# Find first valid day in this month
for day in range(1, 32):
if not is_valid_day_in_month(dt.year, month, day):
break
if not is_valid_day_in_month(search_year, month, day):
continue

test_dt = datetime(dt.year, month, day, first_hour, first_minute, 0, 0)
test_dt = datetime(search_year, month, day, first_hour, first_minute, 0, 0)

# Check day constraints
day_of_month_match = day in self.cron_expr.day_of_month.parsed_values
Expand All @@ -255,27 +301,5 @@ def _get_next_month(self, dt: datetime) -> datetime:
if day_of_month_match and day_of_week_match:
return test_dt

# No valid month found this year, try next year
next_year = dt.year + 1
first_month = valid_months[0]

# Find first valid day in the first month of next year
for day in range(1, 32):
if not is_valid_day_in_month(next_year, first_month, day):
break

test_dt = datetime(next_year, first_month, day, first_hour, first_minute, 0, 0)

# Check day constraints
day_of_month_match = day in self.cron_expr.day_of_month.parsed_values
day_of_week_match = get_weekday(test_dt) in self.cron_expr.day_of_week.parsed_values

if not self.cron_expr.day_of_month.is_wildcard() and not self.cron_expr.day_of_week.is_wildcard():
if day_of_month_match or day_of_week_match:
return test_dt
else:
if day_of_month_match and day_of_week_match:
return test_dt

# This should rarely happen unless the cron expression is very restrictive
raise CronPalError("Could not find valid next month")
Loading
Loading