auto-sync: 2026-05-17 00:44:10

src/blackboard/registry.py

@@ -0,0 +1,124 @@
+"""多项目管理 — _registry.yaml + 项目 CRUD"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import yaml
+
+logger = logging.getLogger("moziplus-v2.registry")
+
+
+class ProjectRegistry:
+    """项目注册表（_registry.yaml 读写）"""
+
+    def __init__(self, root: Path):
+        self.root = root
+        self.registry_path = root / "_registry.yaml"
+        self._cache: Optional[Dict[str, Any]] = None
+
+    def _load(self) -> Dict[str, Any]:
+        if self._cache is not None:
+            return self._cache
+        if self.registry_path.exists():
+            with open(self.registry_path) as f:
+                self._cache = yaml.safe_load(f) or {}
+        else:
+            self._cache = {"projects": {}}
+        return self._cache
+
+    def _save(self, data: Dict[str, Any]) -> None:
+        self.registry_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.registry_path, "w") as f:
+            yaml.dump(data, f, default_flow_style=False, allow_unicode=True)
+        self._cache = data
+
+    def create_project(self, project_id: str, name: str,
+                       agents: Optional[List[str]] = None,
+                       description: str = "") -> Dict[str, Any]:
+        """创建项目"""
+        data = self._load()
+        if project_id in data.get("projects", {}):
+            raise ValueError(f"Project '{project_id}' already exists")
+
+        project_dir = self.root / project_id
+        project_dir.mkdir(parents=True, exist_ok=True)
+
+        project_info = {
+            "name": name,
+            "description": description,
+            "agents": agents or [],
+            "created_at": _now_iso(),
+            "status": "active",
+        }
+
+        data.setdefault("projects", {})[project_id] = project_info
+        self._save(data)
+
+        # Write per-project config skeleton
+        config_dir = project_dir / "config"
+        config_dir.mkdir(exist_ok=True)
+        project_yaml = config_dir / "project.yaml"
+        if not project_yaml.exists():
+            with open(project_yaml, "w") as f:
+                yaml.dump({
+                    "project": {
+                        "name": name,
+                        "description": description,
+                        "agents": agents or [],
+                    }
+                }, f, default_flow_style=False, allow_unicode=True)
+
+        # Create artifacts/experiences/skills dirs
+        for subdir in ("artifacts", "experiences", "skills"):
+            (project_dir / subdir).mkdir(exist_ok=True)
+
+        logger.info("Project created: %s (%s)", project_id, name)
+        return project_info
+
+    def get_project(self, project_id: str) -> Optional[Dict[str, Any]]:
+        data = self._load()
+        return data.get("projects", {}).get(project_id)
+
+    def list_projects(self) -> Dict[str, Dict[str, Any]]:
+        data = self._load()
+        return data.get("projects", {})
+
+    def archive_project(self, project_id: str) -> bool:
+        """归档项目"""
+        data = self._load()
+        proj = data.get("projects", {}).get(project_id)
+        if not proj:
+            return False
+        proj["status"] = "archived"
+        proj["archived_at"] = _now_iso()
+        self._save(data)
+
+        # Move to _archived/
+        src = self.root / project_id
+        archived_dir = self.root / "_archived"
+        archived_dir.mkdir(exist_ok=True)
+        dst = archived_dir / project_id
+        if src.exists() and not dst.exists():
+            src.rename(dst)
+        return True
+
+    def delete_project(self, project_id: str) -> bool:
+        """删除项目（仅从注册表移除，不删目录）"""
+        data = self._load()
+        if project_id not in data.get("projects", {}):
+            return False
+        del data["projects"][project_id]
+        self._save(data)
+        return True
+
+    def reload(self) -> None:
+        """清除缓存，下次读取重新加载"""
+        self._cache = None
+
+
+def _now_iso() -> str:
+    from datetime import datetime
+    return datetime.utcnow().isoformat()