cfdaily
cc2e5aa64c
- M1: expand=all 保持旧 list 格式（向后兼容 TaskModal .map()/.length）
- 细粒度 expand=comments,events 用新 {items,total_count,limit} 格式
- S1(PR#73): _toolchain 加入 _VIRTUAL_PROJECTS
- S1(PR#72): 移除 _validate_project 未使用 import
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
"""共享 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]
|