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
65 changes: 63 additions & 2 deletions src/cronpal/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ def main(args=None):
# Parse as special string
cron_expr = special_parser.parse(parsed_args.expression)

print(f" Valid cron expression: {cron_expr}")
print(f" Valid cron expression: {cron_expr}")

if parsed_args.verbose:
print(f" Special string: {cron_expr.raw_expression}")
Expand All @@ -65,7 +65,7 @@ def main(args=None):
cron_expr.month = field_parser.parse_month(fields[3])
cron_expr.day_of_week = field_parser.parse_day_of_week(fields[4])

print(f" Valid cron expression: {cron_expr}")
print(f" Valid cron expression: {cron_expr}")

if parsed_args.verbose:
print(f" Raw expression: {cron_expr.raw_expression}")
Expand All @@ -76,6 +76,10 @@ def main(args=None):
if parsed_args.next is not None:
_print_next_runs(cron_expr, parsed_args.next)

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

return 0

except CronPalError as e:
Expand Down Expand Up @@ -242,5 +246,62 @@ def _print_next_runs(cron_expr: CronExpression, count: int):
print(f"\nNext runs: Error calculating - {e}")


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

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

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

try:
scheduler = CronScheduler(cron_expr)
previous_runs = scheduler.get_previous_runs(count)

print(f"\nPrevious {count} run{'s' if count != 1 else ''} (most recent first):")
for i, run_time in enumerate(previous_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 = now - run_time

if delta.days == 0:
if delta.seconds < 3600:
minutes = delta.seconds // 60
relative = f"{minutes} minute{'s' if minutes != 1 else ''} ago"
else:
hours = delta.seconds // 3600
relative = f"{hours} hour{'s' if hours != 1 else ''} ago"
elif delta.days == 1:
relative = "yesterday"
elif delta.days < 7:
relative = f"{delta.days} days ago"
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"\nPrevious runs: Error calculating - {e}")


if __name__ == "__main__":
sys.exit(main())
7 changes: 7 additions & 0 deletions src/cronpal/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -56,4 +56,11 @@ def create_parser():
help="Show next N execution times (default: 5)"
)

parser.add_argument(
"-p", "--previous",
type=int,
metavar="N",
help="Show previous N execution times"
)

return parser
235 changes: 234 additions & 1 deletion src/cronpal/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,14 @@
get_next_hour,
get_next_minute,
get_next_month,
get_previous_day,
get_previous_hour,
get_previous_minute,
get_previous_month,
get_weekday,
is_valid_day_in_month,
round_to_next_minute,
round_to_previous_minute,
)


Expand Down Expand Up @@ -99,6 +104,71 @@ def get_next_runs(self, count: int, after: Optional[datetime] = None) -> List[da

return runs

def get_previous_run(self, before: Optional[datetime] = None) -> datetime:
"""
Calculate the previous run time for the cron expression.

Args:
before: The datetime to start searching from.
Defaults to current time if not provided.

Returns:
The previous datetime when the cron expression ran.
"""
if before is None:
before = datetime.now()

# Round down to previous minute if needed
current = round_to_previous_minute(before)

# Maximum iterations to prevent infinite loops
max_iterations = 10000
iterations = 0

while iterations < max_iterations:
iterations += 1

# Check if current time matches the cron expression
if self._matches_time(current):
return current

# Move to previous possible time
current = self._retreat_to_previous_possible(current)

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

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

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

Returns:
List of previous run times (most recent first).

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

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

runs = []
current = before

for _ in range(count):
previous_run = self.get_previous_run(current)
runs.append(previous_run)
# Start next search 1 minute before the found time
current = previous_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 @@ -166,6 +236,34 @@ def _advance_to_next_possible(self, dt: datetime) -> datetime:
# If no valid day this month, try next month
return self._get_next_month(dt)

def _retreat_to_previous_possible(self, dt: datetime) -> datetime:
"""
Retreat datetime to the previous possible matching time.

Args:
dt: The current datetime.

Returns:
The previous datetime that could potentially match.
"""
# Try to retreat minute first
prev_minute = self._get_previous_minute(dt)
if prev_minute is not None:
return prev_minute

# If no valid minute in this hour, try previous hour
prev_hour = self._get_previous_hour(dt)
if prev_hour is not None:
return prev_hour

# If no valid hour today, try previous day
prev_day = self._get_previous_day(dt)
if prev_day is not None:
return prev_day

# If no valid day this month, try previous month
return self._get_previous_month(dt)

def _get_next_minute(self, dt: datetime) -> Optional[datetime]:
"""
Get the next valid minute after the current time.
Expand All @@ -185,6 +283,25 @@ def _get_next_minute(self, dt: datetime) -> Optional[datetime]:

return None

def _get_previous_minute(self, dt: datetime) -> Optional[datetime]:
"""
Get the previous valid minute before the current time.

Args:
dt: The current datetime.

Returns:
Previous valid minute in the same hour, or None if no valid minute.
"""
current_minute = dt.minute
valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True)

for minute in valid_minutes:
if minute < current_minute:
return dt.replace(minute=minute, second=0, microsecond=0)

return None

def _get_next_hour(self, dt: datetime) -> Optional[datetime]:
"""
Get the next valid hour after the current time.
Expand All @@ -208,6 +325,29 @@ def _get_next_hour(self, dt: datetime) -> Optional[datetime]:

return None

def _get_previous_hour(self, dt: datetime) -> Optional[datetime]:
"""
Get the previous valid hour before the current time.

Args:
dt: The current datetime.

Returns:
Previous valid hour in the same day, or None if no valid hour.
"""
current_hour = dt.hour
valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True)
valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True)

# Last minute of the previous valid hour
last_minute = valid_minutes[0] if valid_minutes else 59

for hour in valid_hours:
if hour < current_hour:
return dt.replace(hour=hour, minute=last_minute, second=0, microsecond=0)

return None

def _get_next_day(self, dt: datetime) -> Optional[datetime]:
"""
Get the next valid day after the current time.
Expand Down Expand Up @@ -248,6 +388,43 @@ def _get_next_day(self, dt: datetime) -> Optional[datetime]:

return None

def _get_previous_day(self, dt: datetime) -> Optional[datetime]:
"""
Get the previous valid day before the current time.

Args:
dt: The current datetime.

Returns:
Previous valid day in the same month, or None if no valid day.
"""
current_day = dt.day
valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True)
valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True)

# Last time of the day
last_hour = valid_hours[0] if valid_hours else 23
last_minute = valid_minutes[0] if valid_minutes else 59

# Check previous days in the month
for day in range(current_day - 1, 0, -1):
test_dt = dt.replace(day=day, hour=last_hour, minute=last_minute,
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
day_of_week_match = get_weekday(test_dt) in self.cron_expr.day_of_week.parsed_values

# Apply OR logic for day fields if both are restricted
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

return None

def _get_next_month(self, dt: datetime) -> datetime:
"""
Get the first valid time in the next valid month.
Expand Down Expand Up @@ -302,4 +479,60 @@ def _get_next_month(self, dt: datetime) -> datetime:
return test_dt

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

def _get_previous_month(self, dt: datetime) -> datetime:
"""
Get the last valid time in the previous valid month.

Args:
dt: The current datetime.

Returns:
Last valid time in the previous valid month.
"""
valid_months = sorted(self.cron_expr.month.parsed_values, reverse=True)
valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True)
valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True)

# Last time of any day
last_hour = valid_hours[0] if valid_hours else 23
last_minute = valid_minutes[0] if valid_minutes else 59

# Start searching from current year
current_year = dt.year
current_month = dt.month

# Search for up to 10 years back
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 before current month
months_to_check = [m for m in valid_months if m < current_month]
else:
# For past years, check all valid months
months_to_check = valid_months

for month in months_to_check:
# Find last valid day in this month
for day in range(31, 0, -1):
if not is_valid_day_in_month(search_year, month, day):
continue

test_dt = datetime(search_year, month, day, last_hour, last_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 previous month")
Loading
Loading