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
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ classifiers = [

dependencies = [
"pytz>=2023.3",
"colorama>=0.4.6",
]

[project.optional-dependencies]
Expand Down
139 changes: 95 additions & 44 deletions src/cronpal/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,13 @@
import sys
from datetime import datetime

from cronpal.color_utils import (
ColorConfig,
format_error_message,
format_success_message,
get_color_config,
set_color_config,
)
from cronpal.error_handler import ErrorHandler, suggest_fix
from cronpal.exceptions import CronPalError
from cronpal.field_parser import FieldParser
Expand All @@ -28,19 +35,24 @@ def main(args=None):
parser = create_parser()
parsed_args = parser.parse_args(args)

# Initialize color configuration
use_colors = not getattr(parsed_args, 'no_color', False)
color_config = ColorConfig(use_colors=use_colors)
set_color_config(color_config)

# Handle version flag
if parsed_args.version:
from cronpal import __version__
print(f"cronpal {__version__}")
print(color_config.info(f"cronpal {__version__}"))
return 0

# Handle list timezones flag
if parsed_args.list_timezones:
print("Available timezones:")
print(color_config.header("Available timezones:"))
timezones = list_common_timezones()
for tz in timezones:
print(f" {tz}")
print(f"\nTotal: {len(timezones)} timezones")
print(f" {color_config.value(tz)}")
print(f"\n{color_config.info(f'Total: {len(timezones)} timezones')}")
return 0

# Handle cron expression
Expand All @@ -57,7 +69,8 @@ def main(args=None):
current_time = get_current_time(timezone)
tz_offset = get_timezone_offset(timezone, current_time)
tz_abbrev = get_timezone_abbreviation(timezone, current_time)
print(f"Using timezone: {parsed_args.timezone} ({tz_abbrev} {tz_offset})")
tz_info = f"{parsed_args.timezone} ({tz_abbrev} {tz_offset})"
print(color_config.info(f"Using timezone: {tz_info}"))
except ValueError as e:
raise CronPalError(f"Invalid timezone: {e}")

Expand All @@ -73,26 +86,33 @@ def main(args=None):
if parsed_args.pretty:
# Pretty print mode
if cron_expr.raw_expression.lower() == "@reboot":
print("\n✔ Valid cron expression: @reboot")
print("\nThis expression runs at system startup/reboot only.")
print(format_success_message(
f"Valid cron expression: {cron_expr.raw_expression}",
"This expression runs at system startup/reboot only."
))
else:
printer = PrettyPrinter(cron_expr)
printer = PrettyPrinter(cron_expr, use_colors=use_colors)
print()
print(printer.print_table())
print()
print(f"Summary: {printer.get_summary()}")
print(color_config.header("Summary: ") +
color_config.info(printer.get_summary()))

if parsed_args.verbose:
print()
print(printer.print_detailed())
else:
# Normal output
print(f"✔ Valid cron expression: {cron_expr}")
print(format_success_message(
f"Valid cron expression: {cron_expr.raw_expression}"
))

if parsed_args.verbose:
print(f" Special string: {cron_expr.raw_expression}")
print(f" {color_config.field('Special string')}: "
f"{color_config.value(cron_expr.raw_expression)}")
description = special_parser.get_description(cron_expr.raw_expression)
print(f" Description: {description}")
print(f" {color_config.field('Description')}: "
f"{color_config.info(description)}")

# For @reboot, we don't have fields to show
if cron_expr.raw_expression.lower() != "@reboot":
Expand All @@ -114,22 +134,27 @@ def main(args=None):

if parsed_args.pretty:
# Pretty print mode
printer = PrettyPrinter(cron_expr)
printer = PrettyPrinter(cron_expr, use_colors=use_colors)
print()
print(printer.print_table())
print()
print(f"Summary: {printer.get_summary()}")
print(color_config.header("Summary: ") +
color_config.info(printer.get_summary()))

if parsed_args.verbose:
print()
print(printer.print_detailed())
else:
# Normal output
print(f"✔ Valid cron expression: {cron_expr}")
print(format_success_message(
f"Valid cron expression: {cron_expr.raw_expression}"
))

if parsed_args.verbose:
print(f" Raw expression: {cron_expr.raw_expression}")
print(f" Validation: PASSED")
print(f" {color_config.field('Raw expression')}: "
f"{color_config.value(cron_expr.raw_expression)}")
print(f" {color_config.field('Validation')}: "
f"{color_config.success('PASSED')}")
_print_verbose_fields(cron_expr)

# Show next run times if requested
Expand All @@ -144,18 +169,20 @@ def main(args=None):

except CronPalError as e:
# Use error handler for formatting
error_handler.print_error(e, parsed_args.expression)
print(format_error_message(str(e)), file=sys.stderr)

# Suggest a fix if possible
suggestion = suggest_fix(e, parsed_args.expression)
if suggestion:
print(f" 💡 Suggestion: {suggestion}", file=sys.stderr)
color_config = get_color_config()
print(f" {color_config.warning('💡 Suggestion:')} {suggestion}",
file=sys.stderr)

return 1

except Exception as e:
# Handle unexpected errors
error_handler.print_error(e, parsed_args.expression)
print(format_error_message(f"Unexpected error: {e}"), file=sys.stderr)
return 2

# If no arguments provided, show help
Expand All @@ -170,29 +197,36 @@ def _print_verbose_fields(cron_expr: CronExpression):
Args:
cron_expr: The CronExpression to print fields for.
"""
config = get_color_config()

if cron_expr.minute:
print(f" Minute field: {cron_expr.minute.raw_value}")
print(f" {config.field('Minute field')}: "
f"{config.value(cron_expr.minute.raw_value)}")
if cron_expr.minute.parsed_values:
_print_field_values(" ", cron_expr.minute.parsed_values)

if cron_expr.hour:
print(f" Hour field: {cron_expr.hour.raw_value}")
print(f" {config.field('Hour field')}: "
f"{config.value(cron_expr.hour.raw_value)}")
if cron_expr.hour.parsed_values:
_print_field_values(" ", cron_expr.hour.parsed_values)

if cron_expr.day_of_month:
print(f" Day of month field: {cron_expr.day_of_month.raw_value}")
print(f" {config.field('Day of month field')}: "
f"{config.value(cron_expr.day_of_month.raw_value)}")
if cron_expr.day_of_month.parsed_values:
_print_field_values(" ", cron_expr.day_of_month.parsed_values)

if cron_expr.month:
print(f" Month field: {cron_expr.month.raw_value}")
print(f" {config.field('Month field')}: "
f"{config.value(cron_expr.month.raw_value)}")
if cron_expr.month.parsed_values:
_print_field_values(" ", cron_expr.month.parsed_values)
_print_month_names(" ", cron_expr.month.parsed_values)

if cron_expr.day_of_week:
print(f" Day of week field: {cron_expr.day_of_week.raw_value}")
print(f" {config.field('Day of week field')}: "
f"{config.value(cron_expr.day_of_week.raw_value)}")
if cron_expr.day_of_week.parsed_values:
_print_field_values(" ", cron_expr.day_of_week.parsed_values)
_print_day_names(" ", cron_expr.day_of_week.parsed_values)
Expand All @@ -206,12 +240,16 @@ def _print_field_values(prefix: str, values: set):
prefix: Prefix for each line.
values: Set of values to print.
"""
config = get_color_config()
sorted_values = sorted(values)
if len(sorted_values) <= 10:
print(f"{prefix}Values: {sorted_values}")
values_str = str(sorted_values)
print(f"{prefix}{config.field('Values')}: {config.value(values_str)}")
else:
print(f"{prefix}Values: {sorted_values[:5]} ... {sorted_values[-5:]}")
print(f"{prefix}Total: {len(sorted_values)} values")
truncated = f"{sorted_values[:5]} ... {sorted_values[-5:]}"
print(f"{prefix}{config.field('Values')}: {config.value(truncated)}")
print(f"{prefix}{config.field('Total')}: "
f"{config.highlight(f'{len(sorted_values)} values')}")


def _print_month_names(prefix: str, values: set):
Expand All @@ -222,14 +260,16 @@ def _print_month_names(prefix: str, values: set):
prefix: Prefix for each line.
values: Set of month numbers to convert to names.
"""
config = get_color_config()
month_names = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

sorted_values = sorted(values)
names = [month_names[v - 1] for v in sorted_values if 1 <= v <= 12]

if len(names) <= 10:
print(f"{prefix}Months: {', '.join(names)}")
print(f"{prefix}{config.field('Months')}: "
f"{config.info(', '.join(names))}")


def _print_day_names(prefix: str, values: set):
Expand All @@ -240,13 +280,15 @@ def _print_day_names(prefix: str, values: set):
prefix: Prefix for each line.
values: Set of day numbers to convert to names.
"""
config = get_color_config()
day_names = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]

sorted_values = sorted(values)
names = [day_names[v] for v in sorted_values if 0 <= v <= 6]

if len(names) <= 7:
print(f"{prefix}Days: {', '.join(names)}")
print(f"{prefix}{config.field('Days')}: "
f"{config.info(', '.join(names))}")


def _print_next_runs(cron_expr: CronExpression, count: int, timezone=None):
Expand All @@ -258,21 +300,23 @@ def _print_next_runs(cron_expr: CronExpression, count: int, timezone=None):
count: Number of next runs to show.
timezone: Optional timezone for calculations.
"""
config = get_color_config()

# Don't show next runs for @reboot
if cron_expr.raw_expression.lower() == "@reboot":
print("\nNext runs: @reboot only runs at system startup")
print(f"\n{config.warning('Next 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")
print(f"\n{config.error('Next runs: Cannot calculate - incomplete expression')}")
return

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

print(f"\nNext {count} run{'s' if count != 1 else ''}:")
print(f"\n{config.header(f'Next {count} run')}{'s' if count != 1 else ''}:")
for i, run_time in enumerate(next_runs, 1):
# Format the datetime with timezone info
if timezone:
Expand Down Expand Up @@ -300,14 +344,16 @@ def _print_next_runs(cron_expr: CronExpression, count: int, timezone=None):
relative = ""

if relative:
print(f" {i}. {formatted} ({relative})")
print(f" {config.value(f'{i}.')} "
f"{config.highlight(formatted)} "
f"{config.separator('(')}({config.info(relative)}){config.separator(')')}")
else:
print(f" {i}. {formatted}")
print(f" {config.value(f'{i}.')} {config.highlight(formatted)}")
else:
print(f" {i}. {formatted}")
print(f" {config.value(f'{i}.')} {config.highlight(formatted)}")

except Exception as e:
print(f"\nNext runs: Error calculating - {e}")
print(f"\n{config.error(f'Next runs: Error calculating - {e}')}")


def _print_previous_runs(cron_expr: CronExpression, count: int, timezone=None):
Expand All @@ -319,21 +365,24 @@ def _print_previous_runs(cron_expr: CronExpression, count: int, timezone=None):
count: Number of previous runs to show.
timezone: Optional timezone for calculations.
"""
config = get_color_config()

# Don't show previous runs for @reboot
if cron_expr.raw_expression.lower() == "@reboot":
print("\nPrevious runs: @reboot only runs at system startup")
print(f"\n{config.warning('Previous 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")
print(f"\n{config.error('Previous runs: Cannot calculate - incomplete expression')}")
return

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

print(f"\nPrevious {count} run{'s' if count != 1 else ''} (most recent first):")
print(f"\n{config.header(f'Previous {count} run')}{'s' if count != 1 else ''} "
f"{config.info('(most recent first)')}:")
for i, run_time in enumerate(previous_runs, 1):
# Format the datetime with timezone info
if timezone:
Expand Down Expand Up @@ -361,14 +410,16 @@ def _print_previous_runs(cron_expr: CronExpression, count: int, timezone=None):
relative = ""

if relative:
print(f" {i}. {formatted} ({relative})")
print(f" {config.value(f'{i}.')} "
f"{config.info(formatted)} "
f"{config.separator('(')}({config.warning(relative)}){config.separator(')')}")
else:
print(f" {i}. {formatted}")
print(f" {config.value(f'{i}.')} {config.info(formatted)}")
else:
print(f" {i}. {formatted}")
print(f" {config.value(f'{i}.')} {config.info(formatted)}")

except Exception as e:
print(f"\nPrevious runs: Error calculating - {e}")
print(f"\n{config.error(f'Previous runs: Error calculating - {e}')}")


if __name__ == "__main__":
Expand Down
Loading
Loading