Implement storage layer for MVP-1 Personal Tracker CLI
Add storage layer with FileStorage, MarkdownReader, and MarkdownWriter classes. Add data models (Project, Session, Note, Change).
This commit is contained in:
5
tracker/storage/__init__.py
Normal file
5
tracker/storage/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
from tracker.storage.file_storage import FileStorage
|
||||
from tracker.storage.markdown_reader import MarkdownReader
|
||||
from tracker.storage.markdown_writer import MarkdownWriter
|
||||
|
||||
__all__ = ["FileStorage", "MarkdownReader", "MarkdownWriter"]
|
||||
158
tracker/storage/file_storage.py
Normal file
158
tracker/storage/file_storage.py
Normal file
@@ -0,0 +1,158 @@
|
||||
"""Storage layer for file-based persistence."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import yaml
|
||||
|
||||
from tracker.models.session import Session
|
||||
|
||||
|
||||
class FileStorage:
|
||||
"""Maneja lectura/escritura de archivos del proyecto."""
|
||||
|
||||
def __init__(self, projects_root: Path):
|
||||
self.projects_root = projects_root
|
||||
|
||||
def _project_path(self, slug: str) -> Path:
|
||||
return self.projects_root / slug
|
||||
|
||||
def _meta_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "meta" / "project.yaml"
|
||||
|
||||
def _log_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "LOG.md"
|
||||
|
||||
def _changelog_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "CHANGELOG.md"
|
||||
|
||||
def _tasks_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "TASKS.md"
|
||||
|
||||
def _readme_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "README.md"
|
||||
|
||||
def _sessions_path(self, slug: str) -> Path:
|
||||
return self._project_path(slug) / "sessions"
|
||||
|
||||
def read_project_meta(self, slug: str) -> dict:
|
||||
"""Lee projects/<slug>/meta/project.yaml"""
|
||||
meta_path = self._meta_path(slug)
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
def write_project_meta(self, slug: str, data: dict) -> None:
|
||||
"""Escribe projects/<slug>/meta/project.yaml"""
|
||||
meta_path = self._meta_path(slug)
|
||||
meta_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(meta_path, "w", encoding="utf-8") as f:
|
||||
yaml.safe_dump(data, f, default_flow_style=False, allow_unicode=True)
|
||||
|
||||
def read_log(self, slug: str) -> str:
|
||||
"""Lee projects/<slug>/LOG.md"""
|
||||
log_path = self._log_path(slug)
|
||||
if not log_path.exists():
|
||||
return ""
|
||||
with open(log_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def append_to_log(self, slug: str, entry: str) -> None:
|
||||
"""Append a LOG.md entry."""
|
||||
log_path = self._log_path(slug)
|
||||
with open(log_path, "a", encoding="utf-8") as f:
|
||||
f.write(entry)
|
||||
|
||||
def read_changelog(self, slug: str) -> str:
|
||||
"""Lee projects/<slug>/CHANGELOG.md"""
|
||||
changelog_path = self._changelog_path(slug)
|
||||
if not changelog_path.exists():
|
||||
return ""
|
||||
with open(changelog_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def append_to_changelog(self, slug: str, change: str) -> None:
|
||||
"""Append a CHANGELOG.md entry."""
|
||||
changelog_path = self._changelog_path(slug)
|
||||
with open(changelog_path, "a", encoding="utf-8") as f:
|
||||
f.write(change)
|
||||
|
||||
def read_tasks(self, slug: str) -> str:
|
||||
"""Lee projects/<slug>/TASKS.md"""
|
||||
tasks_path = self._tasks_path(slug)
|
||||
if not tasks_path.exists():
|
||||
return ""
|
||||
with open(tasks_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def write_tasks(self, slug: str, tasks_content: str) -> None:
|
||||
"""Escribe projects/<slug>/TASKS.md"""
|
||||
tasks_path = self._tasks_path(slug)
|
||||
with open(tasks_path, "w", encoding="utf-8") as f:
|
||||
f.write(tasks_content)
|
||||
|
||||
def read_readme(self, slug: str) -> str:
|
||||
"""Lee projects/<slug>/README.md"""
|
||||
readme_path = self._readme_path(slug)
|
||||
if not readme_path.exists():
|
||||
return ""
|
||||
with open(readme_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
def update_readme_autogen(self, slug: str, section: str, content: str) -> None:
|
||||
"""Actualiza una seccion autogenerada en README.md.
|
||||
|
||||
Busca <!-- AUTOGEN:{section}_START --> ... <!-- AUTOGEN:{section}_END -->
|
||||
y reemplaza el contenido entre esos marcadores.
|
||||
"""
|
||||
from tracker.storage.markdown_writer import MarkdownWriter
|
||||
|
||||
readme_path = self._readme_path(slug)
|
||||
current_content = self.read_readme(slug)
|
||||
|
||||
writer = MarkdownWriter()
|
||||
new_content = writer.format_autogen_section(current_content, section, content)
|
||||
|
||||
with open(readme_path, "w", encoding="utf-8") as f:
|
||||
f.write(new_content)
|
||||
|
||||
def write_session_file(self, session: Session) -> None:
|
||||
"""Crea projects/<slug>/sessions/YYYY-MM-DD_HHMM.md"""
|
||||
from tracker.storage.markdown_writer import MarkdownWriter
|
||||
|
||||
sessions_path = self._sessions_path(session.project_slug)
|
||||
sessions_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
started = session.started_at
|
||||
filename = started.strftime("%Y-%m-%d_%H%M.md")
|
||||
session_path = sessions_path / filename
|
||||
|
||||
writer = MarkdownWriter()
|
||||
content = writer.format_session_file(session)
|
||||
|
||||
with open(session_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
|
||||
def active_session_path(self) -> Path:
|
||||
"""Returns Path to projects/.active_session.json"""
|
||||
return self.projects_root / ".active_session.json"
|
||||
|
||||
def read_active_session(self) -> Optional[dict]:
|
||||
"""Lee la sesion activa desde .active_session.json"""
|
||||
path = self.active_session_path()
|
||||
if not path.exists():
|
||||
return None
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
def write_active_session(self, session_data: dict) -> None:
|
||||
"""Escribe la sesion activa a .active_session.json"""
|
||||
path = self.active_session_path()
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(session_data, f, indent=2, default=str)
|
||||
|
||||
def delete_active_session(self) -> None:
|
||||
"""Elimina .active_session.json"""
|
||||
path = self.active_session_path()
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
138
tracker/storage/markdown_reader.py
Normal file
138
tracker/storage/markdown_reader.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""Markdown reader utility."""
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class MarkdownReader:
|
||||
"""Lectura de archivos Markdown del proyecto."""
|
||||
|
||||
def parse_log_entry(self, content: str) -> dict:
|
||||
"""Parse una entrada de LOG.md.
|
||||
|
||||
Formato esperado:
|
||||
## 2026-03-23 10:00–11:20
|
||||
**Objetivo**
|
||||
...
|
||||
**Trabajo realizado**
|
||||
- ...
|
||||
**Cambios relevantes**
|
||||
- ...
|
||||
**Bloqueos**
|
||||
- ...
|
||||
**Decisiones**
|
||||
- ...
|
||||
**Próximos pasos**
|
||||
- ...
|
||||
**Resumen**
|
||||
...
|
||||
|
||||
Returns dict con:
|
||||
- date_range: str
|
||||
- objective: str
|
||||
- work_done: list[str]
|
||||
- changes: list[str]
|
||||
- blockers: list[str]
|
||||
- decisions: list[str]
|
||||
- next_steps: list[str]
|
||||
- summary: str
|
||||
"""
|
||||
result = {
|
||||
"date_range": "",
|
||||
"objective": "",
|
||||
"work_done": [],
|
||||
"changes": [],
|
||||
"blockers": [],
|
||||
"decisions": [],
|
||||
"next_steps": [],
|
||||
"summary": "",
|
||||
}
|
||||
|
||||
# Extraer fecha/rango
|
||||
date_match = re.search(r"##\s+(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}[–-]\d{2}:\d{2})", content)
|
||||
if date_match:
|
||||
result["date_range"] = date_match.group(1)
|
||||
|
||||
# Extraer secciones
|
||||
sections = {
|
||||
"objective": r"\*\*Objetivo\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"work_done": r"\*\*Trabajo realizado\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"changes": r"\*\*Cambios relevantes\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"blockers": r"\*\*Bloqueos\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"decisions": r"\*\*Decisiones\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"next_steps": r"\*\*Próximos pasos\*\*\s*\n(.*?)(?=\n\*\*|\n##|\Z)",
|
||||
"summary": r"\*\*Resumen\*\*\s*\n(.*?)(?=\n##|\Z)",
|
||||
}
|
||||
|
||||
for key, pattern in sections.items():
|
||||
match = re.search(pattern, content, re.DOTALL)
|
||||
if match:
|
||||
text = match.group(1).strip()
|
||||
if key in ("work_done", "changes", "blockers", "decisions", "next_steps"):
|
||||
# Extraer listas con bullet points
|
||||
items = re.findall(r"^\s*-\s+(.+)$", text, re.MULTILINE)
|
||||
result[key] = items
|
||||
else:
|
||||
result[key] = text
|
||||
|
||||
return result
|
||||
|
||||
def extract_autogen_section(self, content: str, section: str) -> str:
|
||||
"""Extrae contenido de una seccion AUTOGEN.
|
||||
|
||||
Busca <!-- AUTOGEN:{section}_START --> ... <!-- AUTOGEN:{section}_END -->
|
||||
Returns el contenido entre esos marcadores, o string vacio si no existe.
|
||||
"""
|
||||
pattern = rf"<!--\s*AUTOGEN:{section}_START\s*-->(.*?)<!--\s*AUTOGEN:{section}_END\s*-->"
|
||||
match = re.search(pattern, content, re.DOTALL)
|
||||
if match:
|
||||
return match.group(1).strip()
|
||||
return ""
|
||||
|
||||
def parse_tasks(self, content: str) -> dict:
|
||||
"""Parse TASKS.md por secciones.
|
||||
|
||||
Secciones esperadas:
|
||||
- Inbox
|
||||
- Próximo
|
||||
- En curso
|
||||
- Bloqueado
|
||||
- En espera
|
||||
- Hecho
|
||||
|
||||
Returns dict con nombre de seccion -> lista de tareas
|
||||
"""
|
||||
result = {}
|
||||
current_section = None
|
||||
current_tasks = []
|
||||
|
||||
lines = content.split("\n")
|
||||
|
||||
for line in lines:
|
||||
# Detectar headers de seccion (## )
|
||||
section_match = re.match(r"^##\s+(.+)$", line)
|
||||
if section_match:
|
||||
# Guardar seccion anterior
|
||||
if current_section is not None:
|
||||
result[current_section] = current_tasks
|
||||
|
||||
current_section = section_match.group(1).strip()
|
||||
current_tasks = []
|
||||
elif current_section is not None:
|
||||
# Parsear bullet points
|
||||
task_match = re.match(r"^\s*-\s+\[([ x])\]\s*(.+)$", line)
|
||||
if task_match:
|
||||
checked = task_match.group(1) == "x"
|
||||
task_text = task_match.group(2).strip()
|
||||
current_tasks.append({"text": task_text, "done": checked})
|
||||
elif line.strip():
|
||||
# Lineas que no son bullet ni header, agregar a la ultima tarea
|
||||
if current_tasks:
|
||||
current_tasks[-1]["text"] += " " + line.strip()
|
||||
|
||||
# Guardar ultima seccion
|
||||
if current_section is not None:
|
||||
result[current_section] = current_tasks
|
||||
|
||||
return result
|
||||
241
tracker/storage/markdown_writer.py
Normal file
241
tracker/storage/markdown_writer.py
Normal file
@@ -0,0 +1,241 @@
|
||||
"""Markdown writer utility."""
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from tracker.models.session import Session
|
||||
|
||||
|
||||
class MarkdownWriter:
|
||||
"""Escritura de archivos Markdown del proyecto."""
|
||||
|
||||
def format_log_entry(self, session: Session, summary: str) -> str:
|
||||
"""Formatea una entrada para LOG.md.
|
||||
|
||||
Formato:
|
||||
## 2026-03-23 10:00–11:20
|
||||
**Objetivo**
|
||||
...
|
||||
|
||||
**Trabajo realizado**
|
||||
- ...
|
||||
|
||||
**Cambios relevantes**
|
||||
- ...
|
||||
|
||||
**Bloqueos**
|
||||
- ...
|
||||
|
||||
**Decisiones**
|
||||
- ...
|
||||
|
||||
**Próximos pasos**
|
||||
- ...
|
||||
|
||||
**Resumen**
|
||||
...
|
||||
|
||||
Returns string formateado.
|
||||
"""
|
||||
started = session.started_at.strftime("%Y-%m-%d %H:%M")
|
||||
ended = session.ended_at.strftime("%H:%M") if session.ended_at else "En Curso"
|
||||
date_range = f"{started}–{ended}"
|
||||
|
||||
lines = [
|
||||
f"## {date_range}",
|
||||
"",
|
||||
"**Objetivo**",
|
||||
f"{session.objective or 'No especificado'}",
|
||||
"",
|
||||
"**Trabajo realizado**",
|
||||
]
|
||||
|
||||
if session.work_done:
|
||||
for item in session.work_done:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin trabajo registrado")
|
||||
|
||||
lines.extend(["", "**Cambios relevantes**"])
|
||||
if session.changes:
|
||||
for item in session.changes:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin cambios")
|
||||
|
||||
lines.extend(["", "**Bloqueos**"])
|
||||
if session.blockers:
|
||||
for item in session.blockers:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin bloqueos")
|
||||
|
||||
lines.extend(["", "**Decisiones**"])
|
||||
if session.decisions:
|
||||
for item in session.decisions:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin decisiones")
|
||||
|
||||
lines.extend(["", "**Próximos pasos**"])
|
||||
if session.next_steps:
|
||||
for item in session.next_steps:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin pasos definidos")
|
||||
|
||||
lines.extend(["", "**Resumen**", summary])
|
||||
|
||||
return "\n".join(lines) + "\n\n"
|
||||
|
||||
def format_session_file(self, session: Session) -> str:
|
||||
"""Formatea archivo de sesion detalle en sessions/YYYY-MM-DD_HHMM.md.
|
||||
|
||||
Formato:
|
||||
# Sesion: 2026-03-23 10:00–11:20
|
||||
|
||||
## Objetivo
|
||||
...
|
||||
|
||||
## Notas
|
||||
...
|
||||
|
||||
## Trabajo realizado
|
||||
...
|
||||
|
||||
## Cambios
|
||||
...
|
||||
|
||||
## Decisiones
|
||||
...
|
||||
|
||||
## Bloqueos
|
||||
...
|
||||
|
||||
## Proximos pasos
|
||||
...
|
||||
|
||||
## Referencias
|
||||
...
|
||||
|
||||
## Duracion
|
||||
X minutos
|
||||
"""
|
||||
started = session.started_at.strftime("%Y-%m-%d %H:%M")
|
||||
ended = session.ended_at.strftime("%H:%M") if session.ended_at else "En Curso"
|
||||
|
||||
lines = [
|
||||
f"# Sesion: {started}–{ended}",
|
||||
"",
|
||||
"## Objetivo",
|
||||
f"{session.objective or 'No especificado'}",
|
||||
"",
|
||||
"## Notas",
|
||||
]
|
||||
|
||||
if session.raw_notes:
|
||||
for note in session.raw_notes:
|
||||
note_type = note.get("type", "work")
|
||||
note_text = note.get("text", "")
|
||||
lines.append(f"- [{note_type}] {note_text}")
|
||||
else:
|
||||
lines.append("- Sin notas")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Trabajo realizado",
|
||||
])
|
||||
if session.work_done:
|
||||
for item in session.work_done:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin trabajo realizado")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Cambios",
|
||||
])
|
||||
if session.changes:
|
||||
for item in session.changes:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin cambios")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Decisiones",
|
||||
])
|
||||
if session.decisions:
|
||||
for item in session.decisions:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin decisiones")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Bloqueos",
|
||||
])
|
||||
if session.blockers:
|
||||
for item in session.blockers:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin bloqueos")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Proximos pasos",
|
||||
])
|
||||
if session.next_steps:
|
||||
for item in session.next_steps:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin pasos definidos")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Referencias",
|
||||
])
|
||||
if session.references:
|
||||
for item in session.references:
|
||||
lines.append(f"- {item}")
|
||||
else:
|
||||
lines.append("- Sin referencias")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"## Duracion",
|
||||
f"{session.duration_minutes} minutos",
|
||||
])
|
||||
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
def format_autogen_section(self, content: str, section: str, new_content: str) -> str:
|
||||
"""Reemplaza o inserta una seccion AUTOGEN en contenido Markdown.
|
||||
|
||||
Busca <!-- AUTOGEN:{section}_START --> ... <!-- AUTOGEN:{section}_END -->
|
||||
Si existe, reemplaza el contenido entre los marcadores.
|
||||
Si no existe, inserta la seccion al final.
|
||||
|
||||
Returns el contenido modificado.
|
||||
"""
|
||||
start_marker = f"<!-- AUTOGEN:{section}_START -->"
|
||||
end_marker = f"<!-- AUTOGEN:{section}_END -->"
|
||||
|
||||
full_marker = f"{start_marker}\n{new_content}\n{end_marker}"
|
||||
|
||||
# Buscar si existe la seccion
|
||||
pattern = rf"{re.escape(start_marker)}.*?{re.escape(end_marker)}"
|
||||
if re.search(pattern, content, re.DOTALL):
|
||||
# Reemplazar seccion existente
|
||||
return re.sub(pattern, full_marker, content, flags=re.DOTALL)
|
||||
else:
|
||||
# Insertar al final
|
||||
return content + "\n" + full_marker + "\n"
|
||||
|
||||
def format_readme_section(self, section: str, content: str) -> str:
|
||||
"""Formatea una seccion de README.md.
|
||||
|
||||
Para usar con format_autogen_section.
|
||||
"""
|
||||
return content
|
||||
Reference in New Issue
Block a user