| """ |
| 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() |
|
|