diff --git a/src/cli.py b/src/cli.py new file mode 100644 index 0000000..277cec4 --- /dev/null +++ b/src/cli.py @@ -0,0 +1,269 @@ +"""Command-line interface for DevTrace.""" + +import os +import sys +import json +from datetime import datetime +from pathlib import Path +from typing import Optional, List + +import click +from rich.console import Console +from rich.table import Table +from rich.panel import Panel + +from . import __version__ +from .storage import Database, Session +from .ui import TimelineView, DisplayManager + + +console = Console() + + +def get_db_path() -> Path: + """Get the database path from environment or default.""" + db_path = os.environ.get("DEVTRACE_DB", "") + if db_path: + return Path(db_path) + devtrace_dir = Path(os.environ.get("DEVTRACE_DIR", Path.home() / ".devtrace")) + devtrace_dir.mkdir(parents=True, exist_ok=True) + return devtrace_dir / "devtrace.db" + + +def get_watch_patterns() -> List[str]: + """Get watch patterns from environment or default.""" + patterns = os.environ.get( + "DEVTRACE_WATCH_PATTERNS", + "*.py,*.js,*.ts,*.html,*.css,*.md,*.json,*.yaml,*.yml" + ) + return [p.strip() for p in patterns.split(",")] + + +@click.group() +@click.version_option(version=__version__) +@click.option("--db", "db_path", help="Path to database file") +@click.pass_context +def main(ctx: click.Context, db_path: Optional[str]) -> None: + """DevTrace - Local Development Workflow Tracker.""" + ctx.ensure_object(dict) + if db_path: + ctx.obj["db_path"] = Path(db_path) + else: + ctx.obj["db_path"] = get_db_path() + + +@main.command() +def init(db_path: click.Context) -> None: + """Initialize DevTrace database and directories.""" + db_path = get_db_path() + db = Database(db_path) + db.initialize() + console.print(Panel( + f"[bold green]DevTrace initialized successfully![/]\n" + f"Database: {db_path}\n" + f"Watch patterns: {', '.join(get_watch_patterns())}", + title="DevTrace" + )) + + +@main.command() +@click.option("--project", "-p", help="Project name for this session") +@click.option("--directory", "-d", help="Directory to monitor", default=".") +@click.pass_context +def start(ctx: click.Context, project: Optional[str], directory: str) -> None: + """Start monitoring a development session.""" + db_path = get_db_path() + db = Database(db_path) + db.initialize() + + watch_dir = Path(directory).absolute() + + if not watch_dir.exists(): + console.print(f"[bold red]Error: Directory {watch_dir} does not exist[/]") + sys.exit(1) + + session = db.create_session( + name=project or watch_dir.name, + directory=str(watch_dir) + ) + + console.print(Panel( + f"[bold green]Session started![/]\n" + f"Session ID: {session.id}\n" + f"Project: {session.name}\n" + f"Directory: {watch_dir}", + title="DevTrace" + )) + + +@main.command() +@click.argument("session_id", type=int) +@click.pass_context +def stop(ctx: click.Context, session_id: int) -> None: + """Stop a monitoring session.""" + db_path = get_db_path() + db = Database(db_path) + session = db.get_session(session_id) + + if session: + db.end_session(session_id) + console.print(f"[bold green]Session {session_id} stopped[/]") + else: + console.print(f"[bold red]Session {session_id} not found[/]") + + +@main.command() +@click.argument("session_id", type=int) +@click.option("--limit", "-l", default=50, help="Maximum events to show") +@click.pass_context +def timeline(ctx: click.Context, session_id: int, limit: int) -> None: + """View timeline of events for a session.""" + db_path = get_db_path() + db = Database(db_path) + + session = db.get_session(session_id) + if not session: + console.print(f"[bold red]Session {session_id} not found[/]") + sys.exit(1) + + events = db.get_session_events(session_id, limit=limit) + view = TimelineView(events) + view.display() + + +@main.command() +@click.argument("session_id", type=int) +@click.option("--format", "output_format", type=click.Choice(["json", "text"]), default="json") +@click.option("--limit", "-l", default=100, help="Maximum events to export") +@click.pass_context +def export(ctx: click.Context, session_id: int, output_format: str, limit: int) -> None: + """Export session data to JSON.""" + db_path = get_db_path() + db = Database(db_path) + + session = db.get_session(session_id) + if not session: + console.print(f"[bold red]Session {session_id} not found[/]") + sys.exit(1) + + events = db.get_session_events(session_id, limit=limit) + + if output_format == "json": + output = { + "session": { + "id": session.id, + "name": session.name, + "start_time": session.start_time.isoformat() if session.start_time else None, + "end_time": session.end_time.isoformat() if session.end_time else None, + }, + "events": [ + { + "id": e.id, + "event_type": e.event_type, + "timestamp": e.timestamp.isoformat() if e.timestamp else None, + "details": e.details + } + for e in events + ] + } + console.print_json(json.dumps(output, default=str)) + else: + for event in events: + console.print(f"[{event.timestamp}] {event.event_type}: {event.details}") + + +@main.command() +@click.argument("query", nargs=-1, type=str) +@click.option("--session", "-s", type=int, help="Filter by session ID") +@click.option("--limit", "-l", default=10, help="Maximum results") +@click.pass_context +def search(ctx: click.Context, query: tuple, session: Optional[int], limit: int) -> None: + """Search workflow history.""" + query_text = " ".join(query) + + db_path = get_db_path() + db = Database(db_path) + + table = Table(title="Search Results") + table.add_column("ID", justify="right") + table.add_column("Type") + table.add_column("Details") + + events = db.get_session_events(session or 0, limit=limit * 2) + + for event in events: + if query_text.lower() in str(event.details).lower() or query_text.lower() in event.event_type.lower(): + table.add_row( + str(event.id), + event.event_type, + str(event.details)[:100] + ) + + console.print(table) + + +@main.command() +@click.argument("session_id", type=int) +@click.pass_context +def sessions(ctx: click.Context) -> None: + """List all sessions.""" + db_path = get_db_path() + db = Database(db_path) + + sessions = db.get_all_sessions() + + table = Table(title="Sessions") + table.add_column("ID", justify="right") + table.add_column("Name") + table.add_column("Start Time") + table.add_column("End Time") + table.add_column("Status") + + for session in sessions: + status = "Active" if session.end_time is None else "Completed" + table.add_row( + str(session.id), + session.name, + session.start_time.isoformat() if session.start_time else "N/A", + session.end_time.isoformat() if session.end_time else "N/A", + status + ) + + console.print(table) + + +@main.command() +@click.argument("event_type", type=click.Choice(["file", "command", "git"])) +@click.argument("session_id", type=int) +@click.pass_context +def events(ctx: click.Context, event_type: str, session_id: int) -> None: + """Show events of a specific type for a session.""" + db_path = get_db_path() + db = Database(db_path) + + if event_type == "file": + events = db.get_file_events(session_id) + elif event_type == "command": + events = db.get_command_events(session_id) + else: + events = db.get_git_events(session_id) + + for event in events: + console.print(f"[{event.timestamp}] {event.event_type}: {event.details}") + + +@main.command() +@click.pass_context +def status(ctx: click.Context) -> None: + """Show DevTrace status and statistics.""" + db_path = get_db_path() + + if not db_path.exists(): + console.print("[bold yellow]DevTrace not initialized. Run 'devtrace init' first.[/]") + return + + db = Database(db_path) + stats = db.get_stats() + + display = DisplayManager() + display.show_status(stats)