Download diffweave/cli/commands/log_cmd.py from shak3008/DiffWeave: direct link, hf CLI and curl.
- Browser
- Download file 2.22 kB
-
https://huggingface.co/spaces/shak3008/DiffWeave/resolve/main/diffweave/cli/commands/log_cmd.py
- Command line
-
hf download hf://spaces/shak3008/DiffWeave/diffweave/cli/commands/log_cmd.py
-
curl -L -o log_cmd.py https://huggingface.co/spaces/shak3008/DiffWeave/resolve/main/diffweave/cli/commands/log_cmd.py
2.22 kB
| """ | |
| `dw log` command. | |
| Chronological knowledge audit history and telemetry timeline. | |
| """ | |
| from __future__ import annotations | |
| from typing import Optional | |
| import typer | |
| from rich.table import Table | |
| from diffweave.cli.config import get_active_workspace | |
| from diffweave.cli.output import print_error, print_json, console | |
| from diffweave.mcp.client import DiffWeaveMCPClient | |
| def log_command( | |
| limit: int = typer.Option(20, "--limit", "-n", help="Number of history events to display"), | |
| workspace_id: Optional[str] = typer.Option(None, "--workspace", "-w", help="Workspace UUID override"), | |
| as_json: bool = typer.Option(False, "--json", help="Output machine-readable JSON"), | |
| ): | |
| """ | |
| View chronological knowledge audit log (commits, reviews, workflows). | |
| """ | |
| try: | |
| active_ws_id = get_active_workspace(workspace_id) | |
| except Exception as e: | |
| print_error(str(e)) | |
| raise typer.Exit(code=1) | |
| client = DiffWeaveMCPClient() | |
| try: | |
| feed = client.call_tool_sync("get_activity_feed", {"workspace_id": active_ws_id, "limit": limit}) | |
| except Exception as e: | |
| print_error(f"Failed to fetch activity log: {e}") | |
| raise typer.Exit(code=1) | |
| if as_json: | |
| print_json(feed) | |
| return | |
| events = feed.get("events", []) | |
| if not events: | |
| console.print("[dim]No activity events recorded in this workspace yet.[/dim]") | |
| return | |
| table = Table(title=f"Knowledge Audit History (Last {len(events)} events)", header_style="bold blue") | |
| table.add_column("Timestamp", style="dim") | |
| table.add_column("Event Type", style="cyan") | |
| table.add_column("Details", style="white") | |
| for ev in events: | |
| ev_type = ev.get("type", "event") | |
| if "approved" in ev_type or "completed" in ev_type: | |
| badge = f"[green]{ev_type}[/green]" | |
| elif "rejected" in ev_type: | |
| badge = f"[red]{ev_type}[/red]" | |
| elif "archived" in ev_type: | |
| badge = f"[yellow]{ev_type}[/yellow]" | |
| else: | |
| badge = f"[cyan]{ev_type}[/cyan]" | |
| table.add_row( | |
| str(ev.get("timestamp", ""))[:19], | |
| badge, | |
| ev.get("message", ""), | |
| ) | |
| console.print(table) | |