239 lines
9.7 KiB
Python
239 lines
9.7 KiB
Python
"""CRUD API for the global business → agent / chain mapping (业务链条管理).
|
|
|
|
Binds a fixed business code (3Q / 6BF / 3q-0…3q-6 / 6-1…6-4 / 7BF / 7-1…7-4 / 8BF / AFX) to an optional
|
|
智能体 and an optional 业务链条 (a **public** ``roundtable_chains`` row). Used to
|
|
auto-pick the business chain when a task page launches the multi-agent roundtable.
|
|
|
|
Global / shared — there is no user scoping; all accounts see and edit the same
|
|
rows. Fixed business rows are seeded on startup; this API only lists them and
|
|
updates the agent / chain columns (no create/delete).
|
|
|
|
Routes (prefix ``/api/business-mapping``):
|
|
GET "" list all business mappings (display order)
|
|
PUT "/{business_code}" update one row's agent / chain selection
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.catalog_cache import cached_catalog, invalidate_catalog
|
|
from app.gateway.deps import get_agent_store
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/business-mapping", tags=["business-mapping"])
|
|
|
|
_EIGHTBF_BUSINESS_CODE = "8BF"
|
|
_EIGHTBF_MANAGER_USERNAME = "lqq"
|
|
|
|
|
|
# ── schemas ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class BusinessMappingResponse(BaseModel):
|
|
business_code: str
|
|
label: str = ""
|
|
mapping_type: str = "内置业务"
|
|
is_custom: bool = False
|
|
agent_id: str | None = None
|
|
agent_name: str | None = None
|
|
chain_id: str | None = None
|
|
chain_title: str | None = None
|
|
sort_order: int = 0
|
|
created_at: datetime | str | None = None
|
|
updated_at: datetime | str | None = None
|
|
|
|
|
|
class BusinessMappingListResponse(BaseModel):
|
|
mappings: list[BusinessMappingResponse]
|
|
|
|
|
|
class BusinessMappingUpdateRequest(BaseModel):
|
|
# All optional → callers send only what changed; explicit null clears a field.
|
|
agent_id: str | None = Field(default=None, max_length=64)
|
|
agent_name: str | None = Field(default=None, max_length=191)
|
|
chain_id: str | None = Field(default=None, max_length=64)
|
|
chain_title: str | None = Field(default=None, max_length=512)
|
|
|
|
|
|
class BusinessMappingCreateRequest(BaseModel):
|
|
"""The name is both the display name and login ``goPath`` for a custom row."""
|
|
|
|
name: str = Field(min_length=1, max_length=32)
|
|
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _get_store(request: Request):
|
|
store = getattr(request.app.state, "business_mapping_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Business mapping store not available")
|
|
return store
|
|
|
|
|
|
def _is_admin(request: Request) -> bool:
|
|
"""Treat no-auth deployments as the single administrative workspace."""
|
|
user = getattr(request.state, "user", None)
|
|
return user is None or getattr(user, "system_role", None) == "admin"
|
|
|
|
|
|
def _username_of(request: Request) -> str:
|
|
"""Read the trusted login account name, not an externally supplied username."""
|
|
user = getattr(request.state, "user", None)
|
|
email = str(getattr(user, "email", "") or "")
|
|
return email.split("@", 1)[0].strip().lower()
|
|
|
|
|
|
def _can_update_mapping(request: Request, business_code: str) -> bool:
|
|
"""8BF's public chain is maintained exclusively by the lqq account."""
|
|
if business_code.strip().upper() == _EIGHTBF_BUSINESS_CODE:
|
|
return _username_of(request) == _EIGHTBF_MANAGER_USERNAME
|
|
return _is_admin(request)
|
|
|
|
|
|
def _current_user_id(request: Request) -> str:
|
|
user = getattr(request.state, "user", None)
|
|
if user is not None:
|
|
return str(user.id)
|
|
return get_effective_user_id()
|
|
|
|
|
|
_RESERVED_GOPATHS = frozenset({"3qfx", "rwfx", "xdfx", "agentfx", "b5-1"})
|
|
|
|
|
|
def _normalise_custom_mapping_input(body: BusinessMappingCreateRequest) -> str:
|
|
"""Validate a custom goPath without rewriting the administrator's label."""
|
|
name = re.sub(r"\s+", " ", body.name).strip()
|
|
if not name:
|
|
raise HTTPException(status_code=422, detail="名称不能为空")
|
|
# ``-A`` is a mode suffix, not part of a configurable name: ``name`` is
|
|
# single-agent and ``name-A`` is the same mapping's business-chain entry.
|
|
if name.lower().endswith("-a"):
|
|
raise HTTPException(status_code=422, detail="自定义名称不能以 -A 结尾")
|
|
if name.lower() in _RESERVED_GOPATHS:
|
|
raise HTTPException(status_code=422, detail="该名称已被内置 goPath 保留")
|
|
return name
|
|
|
|
|
|
async def _validate_public_chain_mapping(
|
|
request: Request,
|
|
fields: dict[str, object],
|
|
) -> None:
|
|
"""Require globally mapped chains to remain readable by every account."""
|
|
if "chain_id" not in fields:
|
|
return
|
|
|
|
raw_chain_id = fields["chain_id"]
|
|
chain_id = raw_chain_id.strip() if isinstance(raw_chain_id, str) else ""
|
|
if not chain_id:
|
|
fields["chain_id"] = None
|
|
fields["chain_title"] = None
|
|
return
|
|
|
|
chain_store = getattr(request.app.state, "roundtable_chain_store", None)
|
|
if chain_store is None:
|
|
raise HTTPException(status_code=503, detail="Business chain store not available")
|
|
|
|
chain = await chain_store.get_chain(
|
|
chain_id,
|
|
_current_user_id(request),
|
|
is_admin=True,
|
|
)
|
|
if chain is None:
|
|
raise HTTPException(status_code=400, detail="Business chain not found")
|
|
if not chain.get("is_public"):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Global business mappings require a public business chain",
|
|
)
|
|
|
|
# 公共业务映射会被任意账号启动，链条中的席位也必须对任意账号可见。
|
|
# 仅把链条本身发布、却把其中某个自定义智能体保留为未发布/私有，会导致
|
|
# 前端只能解析到不足两个席位，最终被误判为“没有可用业务链”。
|
|
seats = chain.get("seats")
|
|
if isinstance(seats, list):
|
|
agent_store = get_agent_store(request)
|
|
unavailable_names: list[str] = []
|
|
for raw_seat in seats:
|
|
if not isinstance(raw_seat, dict):
|
|
continue
|
|
agent_id = str(raw_seat.get("agent_id") or "").strip()
|
|
if not agent_id:
|
|
continue
|
|
agent = await agent_store.get_any(agent_id)
|
|
is_builtin = agent is not None and agent.get("user_id") is None
|
|
is_public_agent = bool(agent and agent.get("published")) and not bool(
|
|
agent and agent.get("is_private")
|
|
)
|
|
if is_builtin or is_public_agent:
|
|
continue
|
|
name = str(raw_seat.get("name") or agent_id).strip()
|
|
unavailable_names.append(name)
|
|
if unavailable_names:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
"Global business mappings require every seat to be published and public; "
|
|
f"unavailable agents: {', '.join(unavailable_names)}"
|
|
),
|
|
)
|
|
fields["chain_id"] = chain_id
|
|
fields["chain_title"] = str(chain.get("title") or "") or None
|
|
|
|
|
|
# ── routes ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("", response_model=BusinessMappingListResponse)
|
|
async def list_mappings(request: Request) -> BusinessMappingListResponse:
|
|
store = _get_store(request)
|
|
|
|
async def load() -> BusinessMappingListResponse:
|
|
rows = await store.list_mappings()
|
|
return BusinessMappingListResponse(mappings=[BusinessMappingResponse(**r) for r in rows])
|
|
|
|
return await cached_catalog(request, "business_mapping", "all", load, BusinessMappingListResponse)
|
|
|
|
|
|
@router.post("", response_model=BusinessMappingResponse, status_code=201)
|
|
async def create_mapping(
|
|
request: Request,
|
|
body: BusinessMappingCreateRequest,
|
|
) -> BusinessMappingResponse:
|
|
"""Create a globally shared custom mapping (administrator only)."""
|
|
if not _is_admin(request):
|
|
raise HTTPException(status_code=403, detail="Only administrators can create business mappings")
|
|
name = _normalise_custom_mapping_input(body)
|
|
store = _get_store(request)
|
|
row = await store.create_custom_mapping(
|
|
business_code=name,
|
|
mapping_type="自定义类型",
|
|
)
|
|
if row is None:
|
|
raise HTTPException(status_code=409, detail="该名称已存在，请换一个名称")
|
|
await invalidate_catalog(request, "business_mapping")
|
|
return BusinessMappingResponse(**row)
|
|
|
|
|
|
@router.put("/{business_code}", response_model=BusinessMappingResponse)
|
|
async def update_mapping(request: Request, business_code: str, body: BusinessMappingUpdateRequest) -> BusinessMappingResponse:
|
|
if not _can_update_mapping(request, business_code):
|
|
detail = "Only the lqq account can update the 8BF business mapping" if business_code.strip().upper() == _EIGHTBF_BUSINESS_CODE else "Only administrators can update business mappings"
|
|
raise HTTPException(status_code=403, detail=detail)
|
|
store = _get_store(request)
|
|
# Only forward fields the caller explicitly sent (distinguish omitted vs null).
|
|
fields = body.model_dump(exclude_unset=True)
|
|
await _validate_public_chain_mapping(request, fields)
|
|
row = await store.update_mapping(business_code, **fields)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Business mapping not found")
|
|
await invalidate_catalog(request, "business_mapping")
|
|
return BusinessMappingResponse(**row)
|