cfdaily
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
"""共享 helper 和常量"""
|
|
|
|
from pathlib import Path
|
|
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"})
|
|
|
|
|
|
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, 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]
|