| | """ |
| | Interactive workflow editor for reference checking configuration. |
| | |
| | Provides a terminal-based UI using rich for customizing the order |
| | and enabled state of fetchers in the verification workflow. |
| | """ |
| | from typing import Optional |
| | from pathlib import Path |
| |
|
| | from rich.console import Console |
| | from rich.panel import Panel |
| | from rich.table import Table |
| | from rich.prompt import Prompt, Confirm |
| | from rich.text import Text |
| |
|
| | from ..config.workflow import WorkflowConfig, get_default_workflow |
| |
|
| |
|
| | class WorkflowEditor: |
| | """Interactive terminal editor for workflow configuration.""" |
| | |
| | def __init__(self, config: Optional[WorkflowConfig] = None): |
| | self.console = Console() |
| | self.config = config or get_default_workflow() |
| | self.selected_index = 0 |
| | self.modified = False |
| | |
| | def display_workflow(self): |
| | """Display current workflow configuration as a table.""" |
| | self.console.clear() |
| | |
| | |
| | self.console.print(Panel( |
| | "[bold blue]📋 Reference Check Workflow Editor[/bold blue]\n" |
| | "[dim]Customize the order and sources for metadata verification[/dim]", |
| | border_style="blue" |
| | )) |
| | |
| | |
| | self.console.print() |
| | self.console.print("[dim]Commands: [cyan]u[/cyan]=move up, [cyan]d[/cyan]=move down, " |
| | "[cyan]t[/cyan]=toggle, [cyan]s[/cyan]=save, [cyan]r[/cyan]=reset, [cyan]q[/cyan]=quit[/dim]") |
| | self.console.print() |
| | |
| | |
| | table = Table(show_header=True, header_style="bold magenta", box=None) |
| | table.add_column("#", style="dim", width=3) |
| | table.add_column("Status", width=8) |
| | table.add_column("Source", width=25) |
| | table.add_column("Description", style="dim") |
| | |
| | for i, step in enumerate(self.config.steps): |
| | |
| | row_style = "reverse" if i == self.selected_index else "" |
| | |
| | |
| | if step.enabled: |
| | status = "[green]✓ ON[/green]" |
| | else: |
| | status = "[red]✗ OFF[/red]" |
| | |
| | |
| | priority = f"{i + 1}" |
| | |
| | table.add_row( |
| | priority, |
| | status, |
| | step.display_name, |
| | step.description, |
| | style=row_style |
| | ) |
| | |
| | self.console.print(table) |
| | self.console.print() |
| | |
| | |
| | if 0 <= self.selected_index < len(self.config.steps): |
| | step = self.config.steps[self.selected_index] |
| | info = Text() |
| | info.append("Selected: ", style="dim") |
| | info.append(step.display_name, style="cyan bold") |
| | info.append(f" (search type: {step.search_type})", style="dim") |
| | self.console.print(info) |
| | |
| | if self.modified: |
| | self.console.print("[yellow]* Unsaved changes[/yellow]") |
| | |
| | def run(self) -> WorkflowConfig: |
| | """Run the interactive editor loop.""" |
| | while True: |
| | self.display_workflow() |
| | |
| | |
| | try: |
| | cmd = Prompt.ask( |
| | "\n[bold]Enter command[/bold]", |
| | choices=["u", "d", "t", "s", "r", "q", "1", "2", "3", "4", "5", "6", "7", "8"], |
| | default="q", |
| | show_choices=False |
| | ) |
| | except KeyboardInterrupt: |
| | cmd = "q" |
| | |
| | if cmd == "q": |
| | if self.modified: |
| | if Confirm.ask("Discard unsaved changes?", default=False): |
| | break |
| | else: |
| | break |
| | elif cmd == "u": |
| | if self.config.move_step_up(self.selected_index): |
| | self.selected_index -= 1 |
| | self.modified = True |
| | elif cmd == "d": |
| | if self.config.move_step_down(self.selected_index): |
| | self.selected_index += 1 |
| | self.modified = True |
| | elif cmd == "t": |
| | self.config.toggle_step(self.selected_index) |
| | self.modified = True |
| | elif cmd == "s": |
| | self._save_workflow() |
| | elif cmd == "r": |
| | if Confirm.ask("Reset to default workflow?", default=False): |
| | self.config = get_default_workflow() |
| | self.selected_index = 0 |
| | self.modified = True |
| | elif cmd.isdigit(): |
| | num = int(cmd) |
| | if 1 <= num <= len(self.config.steps): |
| | self.selected_index = num - 1 |
| | |
| | return self.config |
| | |
| | def _save_workflow(self): |
| | """Save workflow configuration to file.""" |
| | default_path = Path.home() / ".bibguard" / "workflow.json" |
| | |
| | path_str = Prompt.ask( |
| | "Save to", |
| | default=str(default_path) |
| | ) |
| | |
| | try: |
| | self.config.save(path_str) |
| | self.console.print(f"[green]✓ Saved to {path_str}[/green]") |
| | self.modified = False |
| | except Exception as e: |
| | self.console.print(f"[red]✗ Failed to save: {e}[/red]") |
| | |
| | Prompt.ask("Press Enter to continue") |
| |
|
| |
|
| | def launch_workflow_editor(config_path: Optional[str] = None) -> WorkflowConfig: |
| | """Launch the workflow editor and return the resulting configuration.""" |
| | config = None |
| | if config_path: |
| | try: |
| | config = WorkflowConfig.load(config_path) |
| | except FileNotFoundError: |
| | pass |
| | |
| | editor = WorkflowEditor(config) |
| | return editor.run() |
| |
|