"""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//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//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//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//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//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//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//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 ... 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//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()