122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""Output utilities with Rich formatting."""
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.text import Text
|
|
from rich.theme import Theme
|
|
|
|
|
|
class OutputFormatter:
|
|
"""Provides colorized output using Rich."""
|
|
|
|
DEFAULT_THEME = Theme({
|
|
"info": "cyan",
|
|
"success": "green",
|
|
"warning": "yellow",
|
|
"error": "red",
|
|
"title": "bold magenta",
|
|
"property": "blue",
|
|
"type": "yellow",
|
|
})
|
|
|
|
def __init__(self, use_color: bool = True):
|
|
self.console = Console(force_terminal=use_color, theme=self.DEFAULT_THEME if use_color else None)
|
|
self.use_color = use_color
|
|
|
|
def print_title(self, text: str) -> None:
|
|
"""Print a title."""
|
|
self.console.print(f"[title]{text}[/]")
|
|
|
|
def print_success(self, text: str) -> None:
|
|
"""Print a success message."""
|
|
self.console.print(f"[success]{text}[/]")
|
|
|
|
def print_error(self, text: str) -> None:
|
|
"""Print an error message."""
|
|
self.console.print(f"[error]{text}[/]")
|
|
|
|
def print_warning(self, text: str) -> None:
|
|
"""Print a warning message."""
|
|
self.console.print(f"[warning]{text}[/]")
|
|
|
|
def print_info(self, text: str) -> None:
|
|
"""Print an info message."""
|
|
self.console.print(f"[info]{text}[/]")
|
|
|
|
def print_schema_table(self, schema: Dict[str, Any]) -> None:
|
|
"""Print schema as a formatted table."""
|
|
table = Table(title="Inferred Schema")
|
|
table.add_column("Property", style="property")
|
|
table.add_column("Type", style="type")
|
|
table.add_column("Required", style="type")
|
|
table.add_column("Description", style="dim")
|
|
|
|
properties = schema.get("properties", [])
|
|
for prop in properties:
|
|
table.add_row(
|
|
prop.get("name", ""),
|
|
prop.get("type", ""),
|
|
"Yes" if prop.get("required", True) else "No",
|
|
prop.get("description", ""),
|
|
)
|
|
|
|
self.console.print(table)
|
|
|
|
def print_schema_tree(self, schema: Dict[str, Any], indent: int = 0) -> None:
|
|
"""Print schema as a tree structure."""
|
|
prefix = " " * indent
|
|
root_type = schema.get("root_type", "unknown")
|
|
|
|
if root_type == "object":
|
|
self.console.print(f"{prefix}└─ [property]{{ }}[/] [type]object[/]")
|
|
for i, prop in enumerate(schema.get("properties", [])):
|
|
is_last = i == len(schema.get("properties", [])) - 1
|
|
connector = "└─ " if is_last else "├─ "
|
|
self._print_prop_tree(prop, indent, connector)
|
|
elif root_type == "array":
|
|
self.console.print(f"{prefix}└─ [type]array[/]")
|
|
items = schema.get("items")
|
|
if items:
|
|
self._print_prop_tree(items, indent, "└─ ")
|
|
else:
|
|
self.console.print(f"{prefix}└─ [type]{root_type}[/]")
|
|
|
|
def _print_prop_tree(self, prop: Dict[str, Any], indent: int = 0, connector: str = "└─ ") -> None:
|
|
"""Print a single property in tree format."""
|
|
prefix = " " * indent
|
|
prop_name = prop.get("name", "unknown")
|
|
prop_type = prop.get("type", "unknown")
|
|
|
|
self.console.print(f"{prefix}{connector}[property]{prop_name}[/]: [type]{prop_type}[/]")
|
|
|
|
new_indent = indent + 1
|
|
|
|
if prop_type == "object" and prop.get("properties"):
|
|
for i, child in enumerate(prop["properties"]):
|
|
is_last = i == len(prop["properties"]) - 1
|
|
child_connector = "└─ " if is_last else "├─ "
|
|
self._print_prop_tree(child, new_indent, child_connector)
|
|
elif prop_type == "array" and prop.get("items"):
|
|
self._print_prop_tree(prop["items"], new_indent, "└─ ")
|
|
|
|
def print_conversion_result(
|
|
self, source_format: str, target_format: str, success: bool, details: Optional[str] = None
|
|
) -> None:
|
|
"""Print conversion result."""
|
|
status = "[success]✓ Success[/]" if success else "[error]✗ Failed[/]"
|
|
self.console.print(f" {status} {source_format} → {target_format}")
|
|
if details:
|
|
self.console.print(f" {details}")
|
|
|
|
def print_batch_progress(
|
|
self, total: int, current: int, success: int, failed: int
|
|
) -> None:
|
|
"""Print batch processing progress."""
|
|
self.console.print(
|
|
f" Progress: {current}/{total} | "
|
|
f"[success]✓ {success}[/] | "
|
|
f"[error]✗ {failed}[/]"
|
|
)
|