"""共享 helper 和常量""" from typing import Any, Dict from fastapi import HTTPException from src.blackboard.operations import Blackboard from src.blackboard.queries import Queries from src.blackboard.models import Task from src.blackboard.registry import ProjectRegistry from src.utils import get_data_root # 虚拟项目白名单 _VIRTUAL_PROJECTS = frozenset({"_general", "_mail", "_toolchain"}) def _validate_project(project_id: str) -> str: """校验 project_id""" if project_id in _VIRTUAL_PROJECTS: return project_id reg = ProjectRegistry(get_data_root()) if reg.get_project(project_id): return project_id raise HTTPException(400, { "ok": False, "error": "project_not_found", "detail": f"Project '{project_id}' is not registered.", "suggestions": [ f"Register first: POST /api/projects with id='{project_id}'", "Or use '_general' for tasks without a specific project", ], }) def _bb(project_id: str) -> Blackboard: _validate_project(project_id) return Blackboard(get_data_root() / project_id / "blackboard.db") def _q(project_id: str) -> Queries: _validate_project(project_id) return Queries(get_data_root() / project_id / "blackboard.db") def _task_to_dict(t: Task) -> Dict[str, Any]: d = {k: v for k, v in t.__dict__.items() if v is not None} return d _KNOWN_AGENT_IDS: list = [] def _init_agent_ids(): """从配置文件加载 Agent ID 列表""" global _KNOWN_AGENT_IDS if _KNOWN_AGENT_IDS: return try: import yaml import os cfg_path = os.path.join(os.path.dirname(__file__), "..", "..", "config", "default.yaml") with open(cfg_path) as f: cfg = yaml.safe_load(f) _KNOWN_AGENT_IDS = list(cfg.get("daemon", {}).get("agent_profiles", {}).keys()) except Exception: _KNOWN_AGENT_IDS = [] def _extract_mentions(text: str) -> list: """从文本中自动提取 @agent-id 格式的 mention""" import re _init_agent_ids() candidates = set(re.findall(r'@([a-z][a-z0-9]*(?:-[a-z][a-z0-9]*)+)', text)) return [a for a in candidates if a in _KNOWN_AGENT_IDS]