122 lines
3.4 KiB
Python
122 lines
3.4 KiB
Python
"""Main CLI entry point for Term Flow."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from . import __version__
|
|
from .commands.record import record_cmd
|
|
from .commands.visualize import visualize_cmd
|
|
from .commands.compare import compare_cmd
|
|
from .commands.export import export_cmd
|
|
from .utils.config import load_config, get_default_db_path
|
|
|
|
console = Console()
|
|
|
|
|
|
@click.group()
|
|
@click.version_option(version=__version__)
|
|
@click.option(
|
|
"--db",
|
|
default=None,
|
|
type=click.Path(path_type=Path),
|
|
help="Path to database file",
|
|
)
|
|
@click.option(
|
|
"--style",
|
|
default="detailed",
|
|
type=click.Choice(["compact", "detailed", "minimal"]),
|
|
help="Visualization style",
|
|
)
|
|
@click.pass_context
|
|
def cli(ctx: click.Context, db: Path, style: str) -> None:
|
|
"""Term Flow - Record terminal sessions and generate flowcharts."""
|
|
config = load_config()
|
|
if db is None:
|
|
db = Path(config.get("database", {}).get("path", get_default_db_path()))
|
|
ctx.ensure_object(dict)
|
|
ctx.obj["db"] = db
|
|
ctx.obj["style"] = style
|
|
ctx.obj["config"] = config
|
|
|
|
|
|
@cli.command(name="list")
|
|
@click.pass_context
|
|
def list_sessions(ctx: click.Context) -> None:
|
|
"""List all recorded sessions."""
|
|
from .core.database import SessionDatabase
|
|
db_path = ctx.obj["db"]
|
|
db = SessionDatabase(db_path)
|
|
sessions = db.get_all_sessions()
|
|
|
|
if not sessions:
|
|
console.print("[yellow]No sessions recorded yet.[/yellow]")
|
|
return
|
|
|
|
table = Table(title="Recorded Sessions")
|
|
table.add_column("ID", justify="right")
|
|
table.add_column("Name")
|
|
table.add_column("Commands", justify="right")
|
|
table.add_column("Duration")
|
|
table.add_column("Created")
|
|
|
|
for session in sessions:
|
|
duration = session.get("end_time") - session.get("start_time") if session.get("end_time") else "Active"
|
|
created_at = session.get("created_at")
|
|
if isinstance(created_at, str):
|
|
created_at = created_at[:16]
|
|
else:
|
|
created_at = created_at.strftime("%Y-%m-%d %H:%M") if created_at else "N/A"
|
|
table.add_row(
|
|
str(session["id"]),
|
|
session["name"],
|
|
str(session["command_count"]),
|
|
str(duration)[:20],
|
|
created_at,
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
|
|
@cli.command(name="show")
|
|
@click.argument("session_id", type=int)
|
|
@click.pass_context
|
|
def show_session(ctx: click.Context, session_id: int) -> None:
|
|
"""Show details of a specific session."""
|
|
from .core.database import SessionDatabase
|
|
db_path = ctx.obj["db"]
|
|
db = SessionDatabase(db_path)
|
|
session = db.get_session(session_id)
|
|
|
|
if not session:
|
|
console.print(f"[red]Session {session_id} not found.[/red]")
|
|
return
|
|
|
|
console.print(f"[bold]Session: {session['name']}[/bold]")
|
|
console.print(f"ID: {session['id']}")
|
|
console.print(f"Created: {session.get('created_at', 'N/A')}")
|
|
console.print(f"Commands: {session['command_count']}")
|
|
|
|
commands = db.get_session_commands(session_id)
|
|
if commands:
|
|
console.print("\n[bold]Commands:[/bold]")
|
|
for cmd in commands:
|
|
console.print(f" {cmd['timestamp']}: [cyan]{cmd['command']}[/cyan]")
|
|
|
|
|
|
cli.add_command(record_cmd, "record")
|
|
cli.add_command(visualize_cmd, "visualize")
|
|
cli.add_command(compare_cmd, "compare")
|
|
cli.add_command(export_cmd, "export")
|
|
|
|
|
|
def main() -> None:
|
|
cli(obj={})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|