from __future__ import annotations from typing import Any from ..spec_models import EntitySpec, WorkflowNodeType WORKFLOW_TYPE_MAP = { WorkflowNodeType.start: {"type": 0, "dealType": 3}, WorkflowNodeType.branch: {"type": 1, "dealType": None}, WorkflowNodeType.audit: {"type": 0, "dealType": 0}, WorkflowNodeType.fill: {"type": 0, "dealType": 1}, WorkflowNodeType.copy: {"type": 0, "dealType": 2}, WorkflowNodeType.webhook: {"type": 3, "dealType": 10}, WorkflowNodeType.condition: {"type": 2, "dealType": None}, } def _default_audit_user_infos() -> dict[str, Any]: return { "member": [], "depart": [], "role": [], "dynamic": [], "includeSubDeparts": None, "externalMemberList": [], "externalDepartList": [], "role_refs": [], } def compile_workflow(entity: EntitySpec) -> dict[str, Any] | None: workflow = entity.workflow if workflow is None or not workflow.enabled: return None actions: list[dict[str, Any]] = [] seen_node_ids: set[str] = set() created_extra_branch_lanes: set[str] = set() start_node_ids = { node.node_id for node in workflow.nodes if node.node_type == WorkflowNodeType.start } for node in workflow.nodes: if node.node_type == WorkflowNodeType.start: seen_node_ids.add(node.node_id) continue if node.parent_node_id and node.parent_node_id not in seen_node_ids: raise ValueError(f"workflow node '{node.node_id}' must appear after parent node '{node.parent_node_id}'") if node.branch_parent_id and node.branch_parent_id not in seen_node_ids: raise ValueError(f"workflow node '{node.node_id}' must appear after branch node '{node.branch_parent_id}'") branch_index = _branch_index(node) branch_lane_ref = _branch_lane_ref(node.branch_parent_id, branch_index) if node.branch_parent_id else None if branch_lane_ref and branch_index > 2 and branch_lane_ref not in created_extra_branch_lanes: actions.append( { "action": "create_sub_branch", "node_id": branch_lane_ref, "payload": { "editVersionNo": 1, "auditNodeRef": node.branch_parent_id, }, } ) created_extra_branch_lanes.add(branch_lane_ref) lane_only = bool((node.config or {}).get("__lane_only__")) and node.node_type == WorkflowNodeType.condition and branch_lane_ref if lane_only: lane_payload = {key: value for key, value in node.config.items() if key != "__lane_only__"} actions.append( { "action": "update_node", "node_id": branch_lane_ref, "node_name": node.name, "node_type": node.node_type.value, "payload": { "editVersionNo": 1, "auditNodeName": node.name, "type": WORKFLOW_TYPE_MAP[node.node_type]["type"], "dealType": WORKFLOW_TYPE_MAP[node.node_type]["dealType"], **lane_payload, }, } ) seen_node_ids.add(node.node_id) continue actions.append( { "action": "add_node", "node_id": node.node_id, "node_name": node.name, "node_type": node.node_type.value, "payload": { "editVersionNo": 1, "auditNodeName": node.name, "type": WORKFLOW_TYPE_MAP[node.node_type]["type"], "dealType": WORKFLOW_TYPE_MAP[node.node_type]["dealType"], "prevNodeRef": _prev_node_ref(node, branch_lane_ref, start_node_ids), "auditUserInfos": _build_audit_user_infos(node) if node.node_type in {WorkflowNodeType.audit, WorkflowNodeType.fill, WorkflowNodeType.copy} else None, **node.config, }, } ) seen_node_ids.add(node.node_id) return { "global_settings": { "editVersionNo": 1, **workflow.global_settings, }, "actions": actions, } def _build_audit_user_infos(node) -> dict[str, Any]: audit_user_infos = _default_audit_user_infos() assignees = node.assignees or {} member_uids = assignees.get("member_uids") or [] if member_uids: audit_user_infos["member"] = [ {"uid": uid, "beingFrontendConfig": True} for uid in member_uids if isinstance(uid, int) and uid > 0 ] role_refs = assignees.get("role_refs") or [] if role_refs: audit_user_infos["role_refs"] = [role_ref for role_ref in role_refs if role_ref] role_entries = assignees.get("role_entries") or [] if role_entries: audit_user_infos["role"] = [ { "roleId": int(entry.get("roleId") or entry.get("role_id")), "roleName": entry.get("roleName") or entry.get("role_name") or str(entry.get("roleId") or entry.get("role_id")), "roleIcon": entry.get("roleIcon") or entry.get("role_icon") or "ex-user-outlined", "beingFrontendConfig": True, } for entry in role_entries if isinstance(entry, dict) and isinstance(entry.get("roleId") or entry.get("role_id"), int) and int(entry.get("roleId") or entry.get("role_id")) > 0 ] include_sub_departs = assignees.get("include_sub_departs") if include_sub_departs is not None: audit_user_infos["includeSubDeparts"] = bool(include_sub_departs) return audit_user_infos def _prev_node_ref(node, branch_lane_ref: str | None, start_node_ids: set[str]) -> str: if branch_lane_ref: if node.parent_node_id and node.parent_node_id != node.branch_parent_id: return node.parent_node_id return branch_lane_ref if node.parent_node_id in start_node_ids: return "__applicant__" return node.parent_node_id or "__applicant__" def _branch_index(node) -> int: config = node.config or {} raw_value = getattr(node, "branch_index", None) if raw_value is None: raw_value = config.get("branch_index", config.get("branchIndex", config.get("lane_index", config.get("laneIndex", 1)))) try: branch_index = int(raw_value) except (TypeError, ValueError) as exc: raise ValueError(f"workflow node '{node.node_id}' has invalid branch index '{raw_value}'") from exc if branch_index <= 0: raise ValueError(f"workflow node '{node.node_id}' branch index must be positive") return branch_index def _branch_lane_ref(branch_parent_id: str | None, branch_index: int) -> str: if not branch_parent_id: raise ValueError("branch_parent_id is required for branch lane references") return f"__branch_lane__{branch_parent_id}__{branch_index}"