diff --git a/src/timmyctl/cli.py b/src/timmyctl/cli.py index 18649160..1f230dfb 100644 --- a/src/timmyctl/cli.py +++ b/src/timmyctl/cli.py @@ -95,6 +95,106 @@ def _get_config_dir() -> Path: return DEFAULT_CONFIG_DIR +def _load_daily_run_config() -> dict[str, Any]: + """Load and validate the daily run configuration.""" + config_path = _get_config_dir() / "daily_run.json" + config = _load_json_config(config_path) + + if not config: + console.print("[yellow]No daily run configuration found.[/yellow]") + raise typer.Exit(1) + + return config + + +def _display_schedules_table(schedules: dict[str, Any]) -> None: + """Display the daily run schedules in a table.""" + table = Table(title="Daily Run Schedules") + table.add_column("Schedule", style="cyan") + table.add_column("Description", style="green") + table.add_column("Automations", style="yellow") + + for schedule_name, schedule_data in schedules.items(): + automations = schedule_data.get("automations", []) + table.add_row( + schedule_name, + schedule_data.get("description", ""), + ", ".join(automations) if automations else "—", + ) + + console.print(table) + console.print() + + +def _display_triggers_table(triggers: dict[str, Any]) -> None: + """Display the triggers in a table.""" + trigger_table = Table(title="Triggers") + trigger_table.add_column("Trigger", style="cyan") + trigger_table.add_column("Description", style="green") + trigger_table.add_column("Automations", style="yellow") + + for trigger_name, trigger_data in triggers.items(): + automations = trigger_data.get("automations", []) + trigger_table.add_row( + trigger_name, + trigger_data.get("description", ""), + ", ".join(automations) if automations else "—", + ) + + console.print(trigger_table) + console.print() + + +def _execute_automation(auto: dict[str, Any], verbose: bool) -> None: + """Execute a single automation and display results.""" + cmd = auto.get("command") + name = auto.get("name", auto.get("id", "unnamed")) + if not cmd: + console.print(f"[yellow]Skipping {name} — no command defined.[/yellow]") + return + + console.print(f"[cyan]▶ Running: {name}[/cyan]") + if verbose: + console.print(f"[dim] $ {cmd}[/dim]") + + try: + result = subprocess.run( # noqa: S602 + cmd, + shell=True, + capture_output=True, + text=True, + timeout=120, + ) + if result.stdout.strip(): + console.print(result.stdout.strip()) + if result.returncode != 0: + console.print(f"[red] ✗ {name} exited with code {result.returncode}[/red]") + if result.stderr.strip(): + console.print(f"[red]{result.stderr.strip()}[/red]") + else: + console.print(f"[green] ✓ {name} completed successfully[/green]") + except subprocess.TimeoutExpired: + console.print(f"[red] ✗ {name} timed out after 120s[/red]") + except Exception as exc: + console.print(f"[red] ✗ {name} failed: {exc}[/red]") + + +def _execute_all_automations(verbose: bool) -> None: + """Execute all enabled automations.""" + console.print("[green]Executing daily run automations...[/green]") + auto_config_path = _get_config_dir() / "automations.json" + auto_config = _load_json_config(auto_config_path) + all_automations = auto_config.get("automations", []) + enabled = [a for a in all_automations if a.get("enabled", False)] + + if not enabled: + console.print("[yellow]No enabled automations found.[/yellow]") + return + + for auto in enabled: + _execute_automation(auto, verbose) + + @app.command() def daily_run( dry_run: bool = typer.Option( @@ -113,93 +213,22 @@ def daily_run( console.print("[bold green]Timmy Daily Run[/bold green]") console.print() - config_path = _get_config_dir() / "daily_run.json" - config = _load_json_config(config_path) - - if not config: - console.print("[yellow]No daily run configuration found.[/yellow]") - raise typer.Exit(1) - + config = _load_daily_run_config() schedules = config.get("schedules", {}) triggers = config.get("triggers", {}) if verbose: + config_path = _get_config_dir() / "daily_run.json" console.print(f"[dim]Config loaded from: {config_path}[/dim]") console.print() - # Show the daily run schedule - table = Table(title="Daily Run Schedules") - table.add_column("Schedule", style="cyan") - table.add_column("Description", style="green") - table.add_column("Automations", style="yellow") - - for schedule_name, schedule_data in schedules.items(): - automations = schedule_data.get("automations", []) - table.add_row( - schedule_name, - schedule_data.get("description", ""), - ", ".join(automations) if automations else "—", - ) - - console.print(table) - console.print() - - # Show triggers - trigger_table = Table(title="Triggers") - trigger_table.add_column("Trigger", style="cyan") - trigger_table.add_column("Description", style="green") - trigger_table.add_column("Automations", style="yellow") - - for trigger_name, trigger_data in triggers.items(): - automations = trigger_data.get("automations", []) - trigger_table.add_row( - trigger_name, - trigger_data.get("description", ""), - ", ".join(automations) if automations else "—", - ) - - console.print(trigger_table) - console.print() + _display_schedules_table(schedules) + _display_triggers_table(triggers) if dry_run: console.print("[yellow]Dry run mode — no actions executed.[/yellow]") else: - console.print("[green]Executing daily run automations...[/green]") - auto_config_path = _get_config_dir() / "automations.json" - auto_config = _load_json_config(auto_config_path) - all_automations = auto_config.get("automations", []) - enabled = [a for a in all_automations if a.get("enabled", False)] - if not enabled: - console.print("[yellow]No enabled automations found.[/yellow]") - for auto in enabled: - cmd = auto.get("command") - name = auto.get("name", auto.get("id", "unnamed")) - if not cmd: - console.print(f"[yellow]Skipping {name} — no command defined.[/yellow]") - continue - console.print(f"[cyan]▶ Running: {name}[/cyan]") - if verbose: - console.print(f"[dim] $ {cmd}[/dim]") - try: - result = subprocess.run( # noqa: S602 - cmd, - shell=True, - capture_output=True, - text=True, - timeout=120, - ) - if result.stdout.strip(): - console.print(result.stdout.strip()) - if result.returncode != 0: - console.print(f"[red] ✗ {name} exited with code {result.returncode}[/red]") - if result.stderr.strip(): - console.print(f"[red]{result.stderr.strip()}[/red]") - else: - console.print(f"[green] ✓ {name} completed successfully[/green]") - except subprocess.TimeoutExpired: - console.print(f"[red] ✗ {name} timed out after 120s[/red]") - except Exception as exc: - console.print(f"[red] ✗ {name} failed: {exc}[/red]") + _execute_all_automations(verbose) @app.command()