Daniel Arroyo
bd48122db2
Implement task add and task move subcommands for managing tasks in TASKS.md. task add <slug> <title> [--section <section>]: - Adds a task to the specified section (default: Inbox) - Valid sections: Inbox, Proximo, En curso, Bloqueado, En espera, Hecho - Creates section if it doesn't exist task move <slug> <task_title> <to_section>: - Searches for task by title across all sections - Moves task to destination section - Marks task as completed ([x]) when moved to Hecho - Updates checkbox state based on destination Includes parsing and building functions for TASKS.md with: - Section normalization (English to Spanish names) - Merge support for duplicate normalized sections - Standard section ordering Uses app.add_typer to register task subcommand group.
899 lines
29 KiB
Python
899 lines
29 KiB
Python
"""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())
|
||
|
||
|
||
# =============================================================================
|
||
# task add command
|
||
# =============================================================================
|
||
def task_add(
|
||
slug: str = typer.Argument(..., help="Project slug"),
|
||
title: str = typer.Argument(..., help="Task title"),
|
||
section: str = typer.Option("Inbox", help="Section name (Inbox, Próximo, En curso, Bloqueado, En espera, Hecho)"),
|
||
) -> None:
|
||
"""Add a task to the project's TASKS.md.
|
||
|
||
Valid sections: Inbox, Próximo, En curso, Bloqueado, En espera, Hecho.
|
||
If the section doesn't exist, it will be created.
|
||
"""
|
||
# Validate 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)
|
||
|
||
# Valid sections (Spanish names as per spec)
|
||
valid_sections = {"Inbox", "Próximo", "En curso", "Bloqueado", "En espera", "Hecho"}
|
||
if section not in valid_sections:
|
||
typer.echo(
|
||
f"Error: Invalid section '{section}'. Valid sections are: {', '.join(sorted(valid_sections))}",
|
||
err=True,
|
||
)
|
||
raise typer.Exit(code=1)
|
||
|
||
# Read current TASKS.md
|
||
tasks_content = storage.read_tasks(slug)
|
||
|
||
# Parse sections and tasks
|
||
sections = _parse_tasks_sections(tasks_content)
|
||
|
||
# Add task to the specified section
|
||
new_task = f"- [ ] {title}"
|
||
if section in sections:
|
||
sections[section].append(new_task)
|
||
else:
|
||
# Create new section with the task
|
||
sections[section] = [new_task]
|
||
|
||
# Rebuild TASKS.md content
|
||
new_content = _build_tasks_content(sections)
|
||
storage.write_tasks(slug, new_content)
|
||
|
||
typer.echo(f"Added task to '{section}' section: {title}")
|
||
|
||
|
||
# =============================================================================
|
||
# task move command
|
||
# =============================================================================
|
||
def task_move(
|
||
slug: str = typer.Argument(..., help="Project slug"),
|
||
task_title: str = typer.Argument(..., help="Task title to search for"),
|
||
to_section: str = typer.Argument(..., help="Destination section"),
|
||
) -> None:
|
||
"""Move a task to a different section in TASKS.md.
|
||
|
||
Searches for the task by title in any section and moves it to the destination.
|
||
Tasks moved to 'Hecho' will be marked as completed ([x]).
|
||
"""
|
||
# Validate 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)
|
||
|
||
# Valid sections
|
||
valid_sections = {"Inbox", "Próximo", "En curso", "Bloqueado", "En espera", "Hecho"}
|
||
if to_section not in valid_sections:
|
||
typer.echo(
|
||
f"Error: Invalid section '{to_section}'. Valid sections are: {', '.join(sorted(valid_sections))}",
|
||
err=True,
|
||
)
|
||
raise typer.Exit(code=1)
|
||
|
||
# Read current TASKS.md
|
||
tasks_content = storage.read_tasks(slug)
|
||
|
||
# Parse sections and tasks
|
||
sections = _parse_tasks_sections(tasks_content)
|
||
|
||
# Find the task
|
||
found_task = None
|
||
found_in_section = None
|
||
task_pattern = f"- [ ] {task_title}"
|
||
task_pattern_done = f"- [x] {task_title}"
|
||
|
||
for section_name, tasks in sections.items():
|
||
for task in tasks:
|
||
if task == task_pattern or task == task_pattern_done:
|
||
found_task = task
|
||
found_in_section = section_name
|
||
break
|
||
if found_task:
|
||
break
|
||
|
||
if not found_task:
|
||
typer.echo(f"Error: Task '{task_title}' not found in any section.", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
if found_in_section == to_section:
|
||
typer.echo(f"Task '{task_title}' is already in '{to_section}' section.")
|
||
return
|
||
|
||
# Remove from original section
|
||
sections[found_in_section].remove(found_task)
|
||
if not sections[found_in_section]:
|
||
del sections[found_in_section]
|
||
|
||
# Determine checkbox state based on destination section
|
||
if to_section == "Hecho":
|
||
new_task = f"- [x] {task_title}"
|
||
else:
|
||
new_task = f"- [ ] {task_title}"
|
||
|
||
# Add to destination section
|
||
if to_section in sections:
|
||
sections[to_section].append(new_task)
|
||
else:
|
||
sections[to_section] = [new_task]
|
||
|
||
# Rebuild TASKS.md content
|
||
new_content = _build_tasks_content(sections)
|
||
storage.write_tasks(slug, new_content)
|
||
|
||
checkbox = "[x]" if to_section == "Hecho" else "[ ]"
|
||
typer.echo(f"Moved task '{task_title}' from '{found_in_section}' to '{to_section}' ({checkbox})")
|
||
|
||
|
||
# =============================================================================
|
||
# Helper functions for TASKS.md parsing
|
||
# =============================================================================
|
||
def _parse_tasks_sections(content: str) -> dict:
|
||
"""Parse TASKS.md content into sections and tasks.
|
||
|
||
Returns a dict mapping section names to lists of task strings.
|
||
Normalizes English section names to Spanish and merges duplicate sections.
|
||
"""
|
||
import re
|
||
|
||
# Mapping from English to Spanish section names
|
||
section_mapping = {
|
||
"Next": "Próximo",
|
||
"In Progress": "En curso",
|
||
"Blocked": "Bloqueado",
|
||
"Waiting": "En espera",
|
||
"Done": "Hecho",
|
||
"Inbox": "Inbox", # Same in both
|
||
}
|
||
|
||
sections = {}
|
||
current_section = None
|
||
current_tasks = []
|
||
|
||
# Match section headers (## Section Name)
|
||
section_pattern = re.compile(r"^##\s+(.+)$")
|
||
# Match task items (- [ ] task or - [x] task)
|
||
task_pattern = re.compile(r"^(- \[[ x]\]) (.+)$")
|
||
|
||
def save_current_section():
|
||
"""Save current section tasks, merging if normalized name already exists."""
|
||
if current_section is not None and current_tasks:
|
||
if current_section in sections:
|
||
sections[current_section].extend(current_tasks)
|
||
else:
|
||
sections[current_section] = current_tasks
|
||
|
||
for line in content.split("\n"):
|
||
section_match = section_pattern.match(line)
|
||
if section_match:
|
||
# Save previous section if exists
|
||
save_current_section()
|
||
raw_section = section_match.group(1)
|
||
# Normalize section name
|
||
current_section = section_mapping.get(raw_section, raw_section)
|
||
current_tasks = []
|
||
else:
|
||
task_match = task_pattern.match(line)
|
||
if task_match and current_section is not None:
|
||
checkbox = task_match.group(1)
|
||
title = task_match.group(2)
|
||
current_tasks.append(f"{checkbox} {title}")
|
||
|
||
# Save last section
|
||
save_current_section()
|
||
|
||
return sections
|
||
|
||
|
||
def _build_tasks_content(sections: dict) -> str:
|
||
"""Build TASKS.md content from sections dict.
|
||
|
||
Maintains the order of sections as specified.
|
||
"""
|
||
section_order = ["Inbox", "Próximo", "En curso", "Bloqueado", "En espera", "Hecho"]
|
||
|
||
lines = ["# Tasks", ""]
|
||
|
||
for section_name in section_order:
|
||
lines.append(f"## {section_name}")
|
||
tasks = sections.get(section_name, [])
|
||
if tasks:
|
||
for task in tasks:
|
||
lines.append(task)
|
||
else:
|
||
lines.append("-")
|
||
lines.append("")
|
||
|
||
# Add any sections not in the standard order
|
||
for section_name in sections:
|
||
if section_name not in section_order:
|
||
lines.append(f"## {section_name}")
|
||
tasks = sections[section_name]
|
||
if tasks:
|
||
for task in tasks:
|
||
lines.append(task)
|
||
else:
|
||
lines.append("-")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
# 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",
|
||
"task_add",
|
||
"task_move",
|
||
]
|