from __future__ import annotations from dataclasses import dataclass from pathlib import Path import re _PROFILE_MONTHLY_RE = re.compile(r"^monthly-(\d{4}-\d{2})$") @dataclass(frozen=True) class ResolvedWorkspace: branch: str workspace_name: str workspace_path: Path qmd_collection: str qmd_index: str class WorkspaceManager: def __init__(self, workspaces_root: Path, default_branch: str = "main", qmd_index_prefix: str = "ws") -> None: self.workspaces_root = workspaces_root self.default_branch = default_branch self.qmd_index_prefix = qmd_index_prefix self.workspaces_root.mkdir(parents=True, exist_ok=True) def resolve_branch(self, branch: str | None, memory_profile: str | None) -> str: if branch and branch.strip(): return branch.strip() if memory_profile and memory_profile.strip(): return self._profile_to_branch(memory_profile.strip()) return self.default_branch def resolve_workspace(self, branch: str | None, memory_profile: str | None) -> ResolvedWorkspace: resolved_branch = self.resolve_branch(branch=branch, memory_profile=memory_profile) workspace_name = self.workspace_name_for_branch(resolved_branch) workspace_path = self.workspaces_root / workspace_name qmd_collection = self.collection_name_for_workspace(workspace_name) qmd_index = self.index_name_for_workspace(workspace_name) return ResolvedWorkspace( branch=resolved_branch, workspace_name=workspace_name, workspace_path=workspace_path, qmd_collection=qmd_collection, qmd_index=qmd_index, ) def _profile_to_branch(self, memory_profile: str) -> str: if memory_profile == "stable": return "main" monthly = _PROFILE_MONTHLY_RE.match(memory_profile) if monthly: return f"memory/{monthly.group(1)}" if memory_profile.startswith("task-") and len(memory_profile) > len("task-"): return f"task/{memory_profile[len('task-') :]}" raise ValueError( "unsupported memory_profile, expected stable | monthly-YYYY-MM | task-" ) def workspace_name_for_branch(self, branch: str) -> str: if branch in {"main", "stable"}: return "main" if branch.startswith("memory/"): return f"memory-{branch.split('/', 1)[1].replace('/', '-') }" if branch.startswith("task/"): return f"task-{branch.split('/', 1)[1].replace('/', '-') }" return "branch-" + self._slugify(branch) def collection_name_for_workspace(self, workspace_name: str) -> str: return self._slugify(workspace_name) def index_name_for_workspace(self, workspace_name: str) -> str: return f"{self.qmd_index_prefix}_{self._slugify(workspace_name)}" def _slugify(self, value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "-", value) cleaned = cleaned.strip("-") return cleaned or "default"