Add NLP, scheduler, generator, and describer modules
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
This commit is contained in:
108
src/cronparse/scheduler.py
Normal file
108
src/cronparse/scheduler.py
Normal file
@@ -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"
|
||||||
Reference in New Issue
Block a user