Files
tracker-cli/tracker/cli/commands.py
Daniel Arroyo b36b60353d Implement complete CLI commands for MVP-1 Personal Tracker
- Refactored CLI commands from nested Typer subapps to direct command functions
- Fixed main.py to use app.command() instead of app.add_typer_command()
- Fixed project_service.py to properly load projects from YAML
- Fixed file_storage.py to save session JSON files alongside markdown
- Added missing methods: write_file, read_file, extract_autogen_section, get_recent_sessions
- Fixed root_path and repo_path to use strings instead of Path objects
2026-03-23 09:02:21 -03:00

674 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""CLI commands implementation."""
import uuid
from datetime import datetime
from pathlib import Path
from typing import Optional
import typer
from tracker.models import Project, Session, NoteType, Change
from tracker.services import (
create_project,
get_project,
update_project,
list_projects,
ensure_project_structure,
get_active_session,
set_active_session,
clear_active_session,
validate_no_other_active_session,
add_note,
consolidate_notes,
suggest_next_steps,
generate_summary,
)
from tracker.storage import FileStorage, MarkdownReader, MarkdownWriter
# Initialize storage and markdown utilities
storage = FileStorage(projects_root=Path("projects"))
markdown_reader = MarkdownReader()
markdown_writer = MarkdownWriter()
# =============================================================================
# init-project command
# =============================================================================
def init_project(
name: str = typer.Argument(..., help="Project name"),
type: str = typer.Option("misc", help="Project type (code, homelab, automation, agent, research, misc)"),
tags: str = typer.Option("", help="Comma-separated tags"),
repo_path: Optional[str] = typer.Option(None, help="Path to git repository"),
description: str = typer.Option("", help="Project description"),
) -> None:
"""Create a new project with standard directory structure and files."""
from tracker.utils.slug import generate_slug
slug = generate_slug(name)
tag_list = [t.strip() for t in tags.split(",") if t.strip()]
# Create project using service
project = create_project(
name=name,
slug=slug,
description=description,
type=type,
tags=tag_list,
repo_path=Path(repo_path) if repo_path else None,
)
# Ensure directory structure
ensure_project_structure(slug)
# Generate files from templates
readme_content = _generate_readme(project)
log_content = "# Log\n\n_Project activity log_\n\n"
changelog_content = "# Changelog\n\n_Project changes_\n\n"
tasks_content = _generate_tasks_template()
meta_content = _generate_meta(project)
# Write files
storage.write_file(slug, "README.md", readme_content)
storage.write_file(slug, "LOG.md", log_content)
storage.write_file(slug, "CHANGELOG.md", changelog_content)
storage.write_file(slug, "TASKS.md", tasks_content)
storage.write_file(slug, "meta/project.yaml", meta_content)
typer.echo(f"Created project '{name}' with slug '{slug}'")
typer.echo(f"Location: {storage._project_path(slug)}")
# =============================================================================
# list command
# =============================================================================
def list_projects_cmd() -> None:
"""Show all projects with their status, last session, and next steps."""
projects = list_projects()
if not projects:
typer.echo("No projects found. Create one with: tracker init-project <name>")
return
typer.echo(f"\n{'Name':<25} {'Slug':<20} {'Status':<10} {'Last Session':<20} {'Next Step'}")
typer.echo("-" * 100)
for project in projects:
last_session = "Never"
next_step = ""
# Try to get last session info from LOG.md
log_content = storage.read_log(project.slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
if parsed.get("date_range"):
last_session = parsed["date_range"].split("")[0][:19]
if parsed.get("next_steps"):
next_step = parsed["next_steps"][0][:40]
typer.echo(
f"{project.name:<25} {project.slug:<20} {project.status:<10} "
f"{last_session:<20} {next_step}"
)
typer.echo(f"\nTotal: {len(projects)} project(s)")
# =============================================================================
# show command
# =============================================================================
def show_project(slug: str = typer.Argument(..., help="Project slug")) -> None:
"""Show detailed project information including status, context, last summary, blockers, and next steps."""
# Load project
project_dict = storage.read_project_meta(slug)
if not project_dict:
typer.echo(f"Error: Project '{slug}' not found.", err=True)
raise typer.Exit(code=1)
project = Project(**project_dict)
typer.echo(f"\n{'='*60}")
typer.echo(f"Project: {project.name}")
typer.echo(f"{'='*60}")
# Basic info
typer.echo(f"\nSlug: {project.slug}")
typer.echo(f"Type: {project.type}")
typer.echo(f"Status: {project.status}")
typer.echo(f"Description: {project.description or 'N/A'}")
if project.tags:
typer.echo(f"Tags: {', '.join(project.tags)}")
# Read LOG.md for context
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
# Last summary
if parsed.get("summary"):
typer.echo(f"\n--- Last Summary ---")
typer.echo(parsed["summary"][:300] + ("..." if len(parsed.get("summary", "")) > 300 else ""))
# Blockers
if parsed.get("blockers"):
typer.echo(f"\n--- Blockers ---")
for blocker in parsed["blockers"]:
typer.echo(f" - {blocker}")
# Next steps
if parsed.get("next_steps"):
typer.echo(f"\n--- Next Steps ---")
for step in parsed["next_steps"]:
typer.echo(f" - {step}")
# Last activity
if parsed.get("date_range"):
typer.echo(f"\n--- Last Activity ---")
typer.echo(f" {parsed['date_range']}")
typer.echo(f"\nLocation: {storage._project_path(slug)}")
# =============================================================================
# start command
# =============================================================================
def start_session(
slug: str = typer.Argument(..., help="Project slug"),
objective: Optional[str] = typer.Option(None, help="Session objective"),
) -> None:
"""Start a new work session for the project.
Validates that no other session is active, creates an active session,
and shows recent context from the project.
"""
# Check project exists
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
# Validate no other active session
if not validate_no_other_active_session(slug):
active = get_active_session()
typer.echo(
f"Error: There is already an active session for project '{active.project_slug}'.",
err=True,
)
typer.echo(f"Stop that session first with: tracker stop {active.project_slug}")
raise typer.Exit(code=1)
# Create session
session = Session(
id=str(uuid.uuid4()),
project_slug=slug,
started_at=datetime.now(),
objective=objective or "",
)
# Save active session
set_active_session(session)
typer.echo(f"Started session for project '{slug}'")
if objective:
typer.echo(f"Objective: {objective}")
# Show recent context from LOG.md
typer.echo("\n--- Recent Context ---")
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
if parsed.get("work_done"):
typer.echo("Recent work:")
for item in parsed["work_done"][:3]:
typer.echo(f" - {item[:60]}")
if parsed.get("next_steps"):
typer.echo("Next steps:")
for item in parsed["next_steps"][:3]:
typer.echo(f" - {item}")
else:
typer.echo("No previous sessions found.")
# =============================================================================
# note command
# =============================================================================
def add_note_cmd(
text: str = typer.Argument(..., help="Note text"),
type: str = typer.Option("work", help="Note type (work, change, blocker, decision, idea, reference)"),
) -> None:
"""Add a note to the currently active session.
Note types:
- work: Work performed
- change: A change made
- blocker: Something blocking progress
- decision: A decision made
- idea: An idea
- reference: A reference or link
"""
# Get active session
session_data = storage.read_active_session()
if not session_data:
typer.echo("Error: No active session. Start one with: tracker start <slug>", err=True)
raise typer.Exit(code=1)
# Reconstruct session
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
# Add note
try:
note = add_note(session, type, text)
# Update active session file directly
storage.write_active_session(session.model_dump(mode="json"))
typer.echo(f"Added [{note['type']}] note: {text[:50]}{'...' if len(text) > 50 else ''}")
except ValueError as e:
typer.echo(f"Error: {e}", err=True)
raise typer.Exit(code=1)
# =============================================================================
# stop command
# =============================================================================
def stop_session(
slug: str = typer.Argument(..., help="Project slug"),
add_to_changelog: bool = typer.Option(False, "--changelog", help="Add session summary to CHANGELOG.md"),
) -> None:
"""Stop the current session and generate summary.
Calculates duration, consolidates notes, generates summary,
suggests next steps, creates session file, updates LOG.md,
and optionally updates CHANGELOG.md.
"""
# Get active session
session_data = storage.read_active_session()
if not session_data:
typer.echo("Error: No active session to stop.", err=True)
raise typer.Exit(code=1)
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
# Verify it matches the slug
if session.project_slug != slug:
typer.echo(
f"Error: Active session is for project '{session.project_slug}', not '{slug}'.",
err=True,
)
raise typer.Exit(code=1)
# End session
session.ended_at = datetime.now()
session.duration_minutes = int((session.ended_at - session.started_at).total_seconds() / 60)
# Consolidate notes
consolidated = consolidate_notes(session.raw_notes)
session.work_done = consolidated["work_done"]
session.changes = consolidated["changes"]
session.decisions = consolidated["decisions"]
session.blockers = consolidated["blockers"]
session.references = consolidated["references"]
# Generate summary
session.summary = generate_summary(session)
# Get project for heuristics
project_dict = storage.read_project_meta(slug)
project = Project(**project_dict) if project_dict else None
# Suggest next steps
if project:
session.next_steps = suggest_next_steps(session, project)
else:
session.next_steps = suggest_next_steps(session, session)
# Save session file
storage.write_session_file(session)
# Format and append to LOG.md
log_entry = markdown_writer.format_log_entry(session, session.summary)
storage.append_to_log(slug, log_entry)
# Update README AUTOGEN sections
_update_readme_autogen(slug, session)
# Optionally add to changelog
if add_to_changelog and session.changes:
change = Change(
date=datetime.now().date(),
type="code",
title=f"Session: {session.objective or 'Work session'}",
impact=f"{session.duration_minutes} min - {len(session.work_done)} tasks completed",
)
changelog_entry = f"\n- **{change.date}** [{change.type}] {change.title}: {change.impact}"
storage.append_to_changelog(slug, changelog_entry)
# Clear active session
storage.delete_active_session()
# Print summary
typer.echo(f"\nSession completed for '{slug}'")
typer.echo(f"Duration: {session.duration_minutes} minutes")
typer.echo(f"\nSummary:\n{session.summary}")
if session.next_steps:
typer.echo("\nSuggested next steps:")
for step in session.next_steps:
typer.echo(f" - {step}")
# =============================================================================
# change command
# =============================================================================
def add_change(
slug: str = typer.Argument(..., help="Project slug"),
type: str = typer.Option("code", help="Change type (code, infra, config, docs, automation, decision)"),
title: str = typer.Option(..., help="Change title"),
impact: str = typer.Option("", help="Impact description"),
) -> None:
"""Add a manual entry to the project's CHANGELOG.md."""
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
change = Change(
date=datetime.now().date(),
type=type,
title=title,
impact=impact,
)
entry = f"- **{change.date}** [{change.type}] {change.title}: {change.impact}"
storage.append_to_changelog(slug, entry)
typer.echo(f"Added change to CHANGELOG.md:")
typer.echo(f" {entry}")
# =============================================================================
# next command
# =============================================================================
def suggest_next(slug: str = typer.Argument(..., help="Project slug")) -> None:
"""Suggest next steps based on project history and heuristics.
Uses simple rules:
- If there are open blockers, suggest unblocking
- If there are unvalidated changes, suggest validation
- If work is incomplete, suggest closing open threads
- If no progress, suggest redefining objective
"""
if not storage._project_path(slug).exists():
typer.echo(f"Error: Project '{slug}' does not exist.", err=True)
raise typer.Exit(code=1)
# Get active session if exists
session_data = storage.read_active_session()
if session_data:
session = Session(**session_data)
if isinstance(session.started_at, str):
session.started_at = datetime.fromisoformat(session.started_at)
else:
# Create a minimal session from project context
session = Session(
id="",
project_slug=slug,
started_at=datetime.now(),
)
# Load recent work from LOG
log_content = storage.read_log(slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
session.work_done = parsed.get("work_done", [])
session.changes = parsed.get("changes", [])
session.blockers = parsed.get("blockers", [])
session.decisions = parsed.get("decisions", [])
# Get project
project_dict = storage.read_project_meta(slug)
project = Project(**project_dict) if project_dict else None
# Get suggestions
suggestions = suggest_next_steps(session, project or Project(
id="",
name="",
slug=slug,
description="",
type="misc",
status="active",
tags=[],
root_path="",
created_at=datetime.now(),
updated_at=datetime.now(),
))
if suggestions:
typer.echo(f"\nSuggested next steps for '{slug}':")
for i, suggestion in enumerate(suggestions, 1):
typer.echo(f" {i}. {suggestion}")
else:
typer.echo(f"\nNo specific suggestions for '{slug}'.")
typer.echo("Consider defining new objectives or reviewing the project status.")
# =============================================================================
# review command
# =============================================================================
def review() -> None:
"""Show an overview of all projects.
Displays:
- Active projects
- Recent sessions
- Open blockers
- Projects without recent activity
"""
projects = list_projects()
if not projects:
typer.echo("No projects to review.")
return
typer.echo("\n" + "=" * 60)
typer.echo("PROJECT REVIEW")
typer.echo("=" * 60)
# Categorize projects
active_projects = [p for p in projects if p.status == "active"]
blocked_projects = [p for p in projects if p.status == "blocked"]
other_projects = [p for p in projects if p.status not in ("active", "blocked")]
# Active projects
if active_projects:
typer.echo(f"\n--- Active Projects ({len(active_projects)}) ---")
for p in active_projects:
typer.echo(f" * {p.name} ({p.slug})")
else:
typer.echo("\n--- No Active Projects ---")
# Blocked projects
if blocked_projects:
typer.echo(f"\n--- Blocked Projects ({len(blocked_projects)}) ---")
for p in blocked_projects:
typer.echo(f" ! {p.name} ({p.slug})")
# Check for blockers in LOG
log_content = storage.read_log(p.slug)
if log_content:
parsed = markdown_reader.parse_log_entry(log_content)
for blocker in parsed.get("blockers", []):
typer.echo(f" - {blocker}")
# Recent sessions
typer.echo(f"\n--- Recent Sessions ---")
has_recent = False
for p in projects:
sessions = storage.get_recent_sessions(p.slug, limit=1)
if sessions:
recent = sessions[0]
typer.echo(f" {p.name}: {recent.started_at.strftime('%Y-%m-%d %H:%M')} ({recent.duration_minutes} min)")
has_recent = True
if not has_recent:
typer.echo(" No recent sessions")
# Check for stale projects (no activity in 7 days)
typer.echo(f"\n--- Projects Without Recent Activity ---")
stale_threshold = datetime.now().timestamp() - (7 * 24 * 60 * 60)
stale_found = False
for p in projects:
if p.last_session_at:
if p.last_session_at.timestamp() < stale_threshold:
days = int((datetime.now() - p.last_session_at).days)
typer.echo(f" {p.name}: {days} days since last session")
stale_found = True
else:
# Check LOG for any sessions
log_content = storage.read_log(p.slug)
if not log_content:
typer.echo(f" {p.name}: Never had a session")
stale_found = True
if not stale_found:
typer.echo(" All projects have recent activity")
typer.echo("\n" + "=" * 60)
# =============================================================================
# Helper functions
# =============================================================================
def _generate_readme(project: Project) -> str:
"""Generate README.md content for a new project."""
return f"""# {project.name}
{project.description or '_No description_'}
## Objective
_TODO: Define objective_
## Status
**Current Status:** {project.status}
<!-- AUTOGEN:STATUS_START -->
Status: {project.status}
<!-- AUTOGEN:STATUS_END -->
## Context
_Current context and background_
## Stack / Tools
- _Tool 1_
- _Tool 2_
## Architecture
_Brief architecture description_
## Technical Decisions
_No decisions recorded yet_
## Risks / Blockers
_No blockers_
<!-- AUTOGEN:NEXT_STEPS_START -->
<!-- AUTOGEN:NEXT_STEPS_END -->
## Recent Sessions
<!-- AUTOGEN:SESSIONS_START -->
<!-- AUTOGEN:SESSIONS_END -->
_Last updated: {datetime.now().strftime('%Y-%m-%d')}_
"""
def _generate_tasks_template() -> str:
"""Generate TASKS.md content for a new project."""
return """# Tasks
## Inbox
-
## Next
-
## In Progress
-
## Blocked
-
## Waiting
-
## Done
-
"""
def _generate_meta(project: Project) -> str:
"""Generate meta/project.yaml content."""
import yaml
data = {
"id": project.id,
"name": project.name,
"slug": project.slug,
"description": project.description,
"type": project.type,
"status": project.status,
"tags": project.tags,
"root_path": str(project.root_path),
"repo_path": str(project.repo_path) if project.repo_path else None,
"created_at": project.created_at.isoformat(),
"updated_at": project.updated_at.isoformat(),
"last_session_at": None,
}
return yaml.dump(data, default_flow_style=False, sort_keys=False)
def _update_readme_autogen(slug: str, session: Session) -> None:
"""Update README.md AUTOGEN sections with session info."""
# Update sessions section
session_line = f"- {session.started_at.strftime('%Y-%m-%d %H:%M')} ({session.duration_minutes} min): {session.summary[:50]}..."
sessions_content = storage.extract_autogen_section(slug, "SESSIONS") or ""
sessions_content = sessions_content + f"\n{session_line}" if sessions_content else session_line
storage.update_readme_autogen(slug, "SESSIONS", sessions_content.strip())
# Update next steps section
if session.next_steps:
next_steps_content = "\n".join([f"- {step}" for step in session.next_steps])
existing_next = storage.extract_autogen_section(slug, "NEXT_STEPS") or ""
if existing_next:
# Keep existing and add new
next_steps_content = existing_next + "\n" + next_steps_content
storage.update_readme_autogen(slug, "NEXT_STEPS", next_steps_content.strip())
# Register all commands at module level for direct access
__all__ = [
"init_project",
"list_projects_cmd",
"show_project",
"start_session",
"add_note_cmd",
"stop_session",
"add_change",
"suggest_next",
"review",
]