This commit is contained in:
269
src/cli.py
Normal file
269
src/cli.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user