deerflow-code/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py

"""Minimal Coze Studio workflow_api compatibility layer (phase 1).

Internal services always use DeerFlow draft DTOs. This module only translates
request/response shapes for the standalone canvas page: ``/canvas`` replies
with the Coze playground envelope and ``/save`` persists the Coze canvas into
``save_studio_draft`` without letting it anywhere near ``WorkflowGraph``
validation (the canvas is an opaque editing document; the executable graph is
compiled client-side and saved via the studio facade).
"""

from __future__ import annotations

import json
from typing import Any

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field

from app.gateway.deps import get_current_user, get_optional_user_from_request
from app.gateway.routers.workflow_studio import _parse_canvas
from deerflow.config.app_config import get_app_config
from deerflow.persistence.workflows import (
    WorkflowDraftConflictError,
    WorkflowNotFoundError,
    WorkflowStore,
)

router = APIRouter(prefix="/api/workflow_api", tags=["workflow-coze-compat"])
# Playground 启动还会打这些 Coze 壳接口；缺了它们画布和节点资源都不会挂载。
chrome_router = APIRouter(tags=["workflow-coze-compat"])

_DEFAULT_SPACE: dict[str, Any] = {
    "id": "1",
    "space_id": 1,
    "name": "默认空间",
    "description": "",
    # SpaceType.Personal = 1；Playground 找不到个人空间会去调 space/save。
    "space_type": 1,
    "space_mode": 0,
    "icon_uri": "",
    "icon_url": "",
    "owner_id": "1",
    "owner_user_id": "1",
    "role_type": 1,
    "has_permission": True,
}

# Coze playground node_template_list payload. The standalone canvas maps
# DeerFlow node kinds onto these StandardNodeType ids (see the frontend's
# node-catalog.ts); entries keep the Coze field names (node_type / icon_url /
# color) so the playground's template list renders without adaptation.
_COZE_NODE_TEMPLATES: list[dict[str, Any]] = [
    {"node_type": "1", "name": "开始", "icon_url": "", "color": "#5C62FF", "desc": "工作流入口，定义输入参数"},
    {"node_type": "2", "name": "结束", "icon_url": "", "color": "#5C62FF", "desc": "工作流出口，汇总输出"},
    {"node_type": "3", "name": "大模型", "icon_url": "", "color": "#5C62FF", "desc": "调用智能体处理任务", "support_batch": 1},
    {"node_type": "5", "name": "代码", "icon_url": "", "color": "#00B2B2", "desc": "在沙箱中执行代码", "support_batch": 0},
    {"node_type": "8", "name": "选择器", "icon_url": "", "color": "#00B2B2", "desc": "按条件分支执行", "support_batch": 0},
    {"node_type": "9", "name": "工作流", "icon_url": "", "color": "#00B83E", "desc": "调用已发布子工作流", "support_batch": 1},
    {"node_type": "13", "name": "输出", "icon_url": "", "color": "#5C62FF", "desc": "定义输出字段", "support_batch": 0},
    {"node_type": "15", "name": "文本处理", "icon_url": "", "color": "#3071F2", "desc": "模板/文本转换", "support_batch": 0},
    {"node_type": "18", "name": "问答", "icon_url": "", "color": "#3071F2", "desc": "人工介入确认", "support_batch": 0},
    {"node_type": "21", "name": "循环", "icon_url": "", "color": "#00B2B2", "desc": "循环执行子图", "support_batch": 0},
    {"node_type": "32", "name": "变量聚合", "icon_url": "", "color": "#00B2B2", "desc": "合并多分支变量", "support_batch": 0},
    {"node_type": "43", "name": "查询数据", "icon_url": "", "color": "#F2B600", "desc": "SQL 只读查询", "support_batch": 0},
    {"node_type": "45", "name": "HTTP 请求", "icon_url": "", "color": "#3071F2", "desc": "调用外部接口", "support_batch": 0},
]

_COZE_NODE_CATEGORIES: list[dict[str, Any]] = [
    {"name": "", "node_type_list": ["3", "9"]},
    {"name": "业务逻辑", "node_type_list": ["5", "8", "21", "32"]},
    {"name": "输入&输出", "node_type_list": ["1", "2", "13"]},
    {"name": "组件", "node_type_list": ["15", "18", "45"]},
    {"name": "数据库", "node_type_list": ["43"]},
]

# Mirrors the standalone page's built-in empty canvas (start → end), used when
# a workflow has never been opened in the Coze editor.
_DEFAULT_CANVAS_SCHEMA = json.dumps(
    {
        "nodes": [
            {
                "id": "100001",
                "type": "1",
                "meta": {"position": {"x": 0, "y": 0}},
                "data": {"outputs": [{"type": "string", "name": "", "required": True}]},
            },
            {
                "id": "900001",
                "type": "2",
                "meta": {"position": {"x": 1000, "y": 0}},
            },
        ],
        "edges": [{"sourceNodeID": "100001", "targetNodeID": "900001"}],
    },
    ensure_ascii=False,
)


class CozeCreateRequest(BaseModel):
    name: str = Field(default="未命名工作流")
    description: str = ""
    graph: dict[str, Any] | None = None


class CozeCanvasRequest(BaseModel):
    workflow_id: str | None = Field(default=None, alias="workflowId")
    workflow_id_snake: str | None = Field(default=None, alias="workflow_id")

    model_config = {"populate_by_name": True}

    def resolved_id(self) -> str | None:
        return self.workflow_id or self.workflow_id_snake


class CozeSaveRequest(BaseModel):
    workflow_id: str | None = Field(default=None, alias="workflowId")
    workflow_id_snake: str | None = Field(default=None, alias="workflow_id")
    expected_revision: int | None = Field(default=None, alias="expectedRevision")
    expected_revision_snake: int | None = Field(default=None, alias="expected_revision")
    # Coze posts the canvas as a JSON string under ``schema``.
    schema_: str | dict[str, Any] | None = Field(default=None, alias="schema")
    graph: str | dict[str, Any] | None = None
    data: dict[str, Any] | None = None

    model_config = {"populate_by_name": True}

    def resolved_id(self) -> str | None:
        return self.workflow_id or self.workflow_id_snake

    def resolved_revision(self) -> int | None:
        if self.expected_revision is not None:
            return self.expected_revision
        return self.expected_revision_snake

    def resolved_canvas(self) -> str | dict[str, Any] | None:
        if self.schema_ is not None:
            return self.schema_
        if isinstance(self.graph, (str, dict)):
            return self.graph
        if self.data is not None:
            for key in ("schema", "graph"):
                value = self.data.get(key)
                if isinstance(value, (str, dict)):
                    return value
            return self.data
        return None


def _require_enabled() -> None:
    if not get_app_config().workflows.enabled:
        raise HTTPException(status_code=503, detail="Workflow Studio is disabled")


def _get_store(request: Request) -> WorkflowStore:
    store = getattr(request.app.state, "workflow_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="Workflow store not available")
    return store


def _canvas_schema_text(definition: dict[str, Any]) -> str:
    raw = definition.get("draft_canvas_schema")
    if isinstance(raw, str) and raw.strip() not in ("", "{}", "null"):
        return raw
    if isinstance(raw, dict) and raw:
        return json.dumps(raw, ensure_ascii=False)
    return _DEFAULT_CANVAS_SCHEMA


def _to_coze_canvas(definition: dict[str, Any], *, is_published: bool = False) -> dict[str, Any]:
    owner = str(definition.get("owner_id") or "1")
    return {
        "code": 0,
        "msg": "success",
        "data": {
            "workflow": {
                "workflow_id": definition["id"],
                "space_id": "1",
                "name": definition.get("name"),
                "desc": definition.get("description") or "",
                "schema_json": _canvas_schema_text(definition),
                "status": 1,
                "collaborator_mode": "Open",
                # Playground derives editability from this flag even when the
                # VCS `can_edit` value is true.  Omitting it puts an owned
                # standalone workflow into preview/read-only mode, which also
                # disables dragging candidate nodes on the canvas.
                "creator": {"user_id": owner, "nickname": owner, "self": True},
                "is_published": is_published,
            },
            "vcs_data": {"can_edit": True},
            "db_data": {"status": 1},
            "operation_info": {},
            "is_bind_agent": False,
            "workflow_version": "1",
        },
    }


async def _load_owned_definition(request: Request, store: WorkflowStore, workflow_id: str, user_id: str) -> dict[str, Any]:
    definition = await store.get_definition(workflow_id, include_draft=True)
    if definition is None:
        raise HTTPException(status_code=404, detail="工作流不存在")
    if definition.get("owner_id") == user_id:
        return definition
    user = await get_optional_user_from_request(request)
    if user is not None and getattr(user, "system_role", None) == "admin":
        return definition
    raise HTTPException(status_code=403, detail="无权访问该工作流")


def _coze_ok(data: dict[str, Any]) -> dict[str, Any]:
    return {"code": 0, "msg": "success", "data": data}


def _account_payload(user: Any, user_id: str) -> dict[str, Any]:
    email = str(getattr(user, "email", "") or "")
    nickname = email.split("@", 1)[0] if email else user_id
    return {
        "user_id_str": user_id,
        "user_id": user_id,
        "nickname": nickname,
        "name": nickname,
        "email": email,
        "avatar_url": "",
        "login_status": "logined",
    }


async def _require_studio_user(request: Request) -> tuple[str, Any]:
    _require_enabled()
    user_id = await get_current_user(request)
    if not user_id:
        raise HTTPException(status_code=401, detail="未登录")
    user = await get_optional_user_from_request(request)
    return user_id, user


def _space_list_payload() -> dict[str, Any]:
    return {
        "has_personal_space": True,
        "bot_space_list": [_DEFAULT_SPACE],
        "team_space_num": 0,
        "max_team_space_num": 10,
        "recently_used_space_list": [],
    }


@chrome_router.api_route("/api/passport/account/info/v2/", methods=["GET", "POST"])
async def coze_account_info(request: Request) -> dict[str, Any]:
    user_id, user = await _require_studio_user(request)
    return _coze_ok(_account_payload(user, user_id))


@chrome_router.api_route("/api/space/list", methods=["GET", "POST"])
async def coze_space_list(request: Request) -> dict[str, Any]:
    await _require_studio_user(request)
    payload = _space_list_payload()
    # developer_api.SpaceList 把列表放在顶层；playground v2 读 data。
    return {"code": 0, "msg": "success", "data": payload, **payload}


@chrome_router.api_route(
    "/api/playground_api/space/list",
    methods=["GET", "POST", "PUT", "PATCH"],
)
@chrome_router.api_route(
    "/api/playground_api/space/list/",
    methods=["GET", "POST", "PUT", "PATCH"],
)
async def coze_playground_space_list(request: Request) -> dict[str, Any]:
    await _require_studio_user(request)
    return _coze_ok(_space_list_payload())


@chrome_router.api_route(
    "/api/playground_api/space/save",
    methods=["GET", "POST"],
)
async def coze_playground_space_save(request: Request) -> dict[str, Any]:
    await _require_studio_user(request)
    return _coze_ok({"id": _DEFAULT_SPACE["id"], "check_not_pass": False})


def _coze_model_type(name: str) -> int:
    """Stable positive int32 from the model name.

    The Coze LLM form stores the picked model as ``modelType`` (number) plus
    ``modleName`` (string, upstream typo) — the compiler reads the string, so
    the number only needs to be stable within a deployment, not reversible.
    """
    digest = 5381
    for char in name:
        digest = ((digest * 33) + ord(char)) & 0xFFFFFFFF
    return digest & 0x7FFFFFFF


@chrome_router.api_route("/api/bot/get_type_list", methods=["GET", "POST"])
async def coze_model_type_list(request: Request) -> dict[str, Any]:
    """LLM form model list → real DeerFlow models (names only, no secrets).

    The Coze ``Model`` shape is trimmed to what the workflow form consumes:
    ``model_type`` (stable number), ``name`` (the DeerFlow model name, stored
    verbatim into the node so the runtime can override the model), and empty
    ability/params maps — ``getLLMModels`` fills in response_format defaults
    client-side. Parameter tuning panels stay inert: model tuning happens in
    DeerFlow, the workflow node only picks which model to call.
    """
    await _require_studio_user(request)
    config = get_app_config()
    configured = list(getattr(config, 'models', None) or [])
    model_list = [
        {
            "model_type": _coze_model_type(model.name),
            "name": model.name,
            "display_name": model.display_name or model.name,
            "description": "",
            "model_ability": {},
            "model_params": [],
            "is_custom": False,
        }
        for model in configured
    ]
    return _coze_ok({"model_list": model_list})


@router.post("/create")
async def coze_create(request: Request, body: CozeCreateRequest) -> dict[str, Any]:
    _require_enabled()
    store = _get_store(request)
    user_id = await get_current_user(request)
    created = await store.create_definition(
        {
            "name": body.name,
            "description": body.description,
            "owner_id": user_id,
        }
    )
    return {"code": 0, "msg": "success", "data": {"workflow_id": created["id"]}}


@router.post("/canvas")
async def coze_canvas(request: Request, body: CozeCanvasRequest) -> dict[str, Any]:
    _require_enabled()
    workflow_id = body.resolved_id()
    if not workflow_id:
        raise HTTPException(status_code=400, detail="workflow_id required")
    store = _get_store(request)
    user_id = await get_current_user(request)
    definition = await _load_owned_definition(request, store, workflow_id, user_id)
    versions = await store.list_versions(workflow_id)
    return _to_coze_canvas(definition, is_published=bool(versions))


@router.post("/save")
async def coze_save(request: Request, body: CozeSaveRequest) -> dict[str, Any]:
    _require_enabled()
    workflow_id = body.resolved_id()
    if not workflow_id:
        raise HTTPException(status_code=400, detail="workflow_id required")
    canvas_raw = body.resolved_canvas()
    if canvas_raw is None or canvas_raw == "" or canvas_raw == {}:
        raise HTTPException(status_code=400, detail="schema required")
    canvas_text = _parse_canvas(canvas_raw)
    store = _get_store(request)
    user_id = await get_current_user(request)
    definition = await _load_owned_definition(request, store, workflow_id, user_id)
    expected = body.resolved_revision()
    if expected is None:
        expected = int(definition.get("draft_revision") or 0)
    try:
        await store.save_studio_draft(
            workflow_id,
            expected_revision=expected,
            # The execution graph is owned by the studio facade; the Coze save
            # only persists the canvas, so the current graph passes through.
            graph=definition.get("draft_graph") or {},
            canvas_schema=canvas_text,
            updated_by=user_id,
        )
    except WorkflowDraftConflictError as exc:
        raise HTTPException(
            status_code=409,
            detail={
                "code": "WORKFLOW_DRAFT_CONFLICT",
                "currentRevision": exc.current_revision,
            },
        ) from exc
    except WorkflowNotFoundError as exc:
        raise HTTPException(status_code=404, detail="工作流不存在") from exc
    return {"code": 0, "msg": "success", "data": {}}


@router.post("/node_template_list")
async def coze_node_template_list(request: Request) -> dict[str, Any]:
    _require_enabled()
    await get_current_user(request)
    # Coze playground envelope: data.template_list / cate_list (§ workflow-playground-context).
    return {
        "code": 0,
        "msg": "success",
        "data": {
            "template_list": _COZE_NODE_TEMPLATES,
            "cate_list": _COZE_NODE_CATEGORIES,
            "plugin_api_list": [],
            "plugin_category_list": [],
        },
    }


@router.post("/workflow_references")
async def coze_workflow_references(request: Request) -> dict[str, Any]:
    """Which workflows reference the current one via subworkflow nodes.

    Honest empty answer: the standalone page's subworkflow form is not wired
    yet, so no canvas can reference another — an empty list is the truthful
    state, not a placeholder for missing data.
    """
    _require_enabled()
    await get_current_user(request)
    return {"code": 0, "msg": "success", "data": {"workflow_list": []}}


@router.post("/nodeDebug")
async def coze_node_debug(request: Request) -> dict[str, Any]:
    """Single-node debug: NOT wired to the DeerFlow executor.

    Deliberately a business error (Coze envelope code != 0) so the playground
    surfaces an explicit message instead of pretending the debug succeeded.
    """
    _require_enabled()
    await get_current_user(request)
    return {
        "code": 4003101,
        "msg": "节点单点调试暂未接入 DeerFlow 执行器，请使用画布下方「试运行」运行整个工作流",
    }


@router.post("/llm_fc_setting_detail")
@router.post("/llm_fc_setting_merged")
async def coze_llm_fc_setting(request: Request) -> dict[str, Any]:
    """LLM function-call settings detail/merge for plugin/workflow lists.

    No Coze plugins are wired into the canvas, so every detail map is empty.
    """
    _require_enabled()
    await get_current_user(request)
    return {
        "code": 0,
        "msg": "success",
        "data": {
            "plugin_detail_map": {},
            "plugin_api_detail_map": {},
            "workflow_detail_map": {},
            "dataset_detail_map": {},
            "knowledge_detail_map": {},
        },
    }


@chrome_router.post("/api/devops/debugger/v1/coze/testcase/casedata/mget")
async def coze_testcase_casedata_mget(request: Request) -> dict[str, Any]:
    """Playground test-case fetch: no Coze test cases exist for this page."""
    await _require_studio_user(request)
    return {"code": 0, "msg": "success", "data": {"case_data_list": [], "total": 0}}