""" 仪表盘创建节点 将仪表盘配置保存到数据库 """ import json import logging from typing import Any, Dict from ..base import BaseNode, NodeContext, NodeResult from ..registry import NodeRegistry logger = logging.getLogger(__name__) @NodeRegistry.register class DashboardCreateNode(BaseNode): """ 仪表盘创建节点 将仪表盘基础信息和页面配置保存到数据库 """ node_type = 'dashboard_create' node_name = '仪表盘创建' node_category = 'dashboard' node_icon = 'file-plus' node_description = '将仪表盘配置保存到数据库' inputs = [ { 'name': 'dashboard_basic_info', 'type': 'object', 'description': '仪表盘基础信息（来自基础信息节点）', 'required': True, }, { 'name': 'page_config', 'type': 'object', 'description': '页面设计配置（来自设计节点）', 'required': True, }, { 'name': 'application_id', 'type': 'string', 'description': '所属应用ID（可选）', 'required': False, }, ] outputs = [ { 'name': 'dashboard_id', 'type': 'string', 'description': '创建的仪表盘ID', }, { 'name': 'dashboard_code', 'type': 'string', 'description': '仪表盘编码', }, { 'name': 'page_meta', 'type': 'object', 'description': '完整的页面元数据', }, ] def execute(self, context: NodeContext) -> NodeResult: """执行节点（同步方法，通过运行异步方法实现）""" import asyncio loop = asyncio.get_event_loop() if loop.is_running(): import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit(asyncio.run, self.execute_async(context)) return future.result() else: return loop.run_until_complete(self.execute_async(context)) async def execute_async(self, context: NodeContext) -> NodeResult: """异步执行节点""" try: # 获取输入参数 dashboard_basic_info_raw = self.config.get('dashboard_basic_info', '{}') page_config_raw = self.config.get('page_config', '{}') # 解析变量引用 dashboard_basic_info = self._resolve_variable_value(context, dashboard_basic_info_raw) page_config = self._resolve_variable_value(context, page_config_raw) application_id = context.resolve_template(self.config.get('application_id', '')) # 配置选项 update_if_exists = self.config.get('update_if_exists', True) logger.info(f'DashboardCreateNode - 开始创建仪表盘') logger.info(f'DashboardCreateNode - 原始配置: dashboard_basic_info_raw={dashboard_basic_info_raw}, page_config_raw={page_config_raw}') logger.info(f'DashboardCreateNode - 上下文变量: {list(context.variables.keys())}') logger.info(f'DashboardCreateNode - dashboard_basic_info type: {type(dashboard_basic_info)}, value: {dashboard_basic_info}') logger.info(f'DashboardCreateNode - page_config type: {type(page_config)}, value: {str(page_config)[:200]}') # 确保 dashboard_basic_info 是字典 if isinstance(dashboard_basic_info, str): try: dashboard_basic_info = json.loads(dashboard_basic_info) logger.info(f'DashboardCreateNode - 成功解析 dashboard_basic_info 字符串为字典') except json.JSONDecodeError as e: logger.error(f'DashboardCreateNode - 解析 dashboard_basic_info 失败: {e}') return NodeResult( success=False, error=f'仪表盘基础信息格式错误: {str(e)}', ) # 确保 page_config 是字典 if isinstance(page_config, str): try: page_config = json.loads(page_config) logger.info(f'DashboardCreateNode - 成功解析 page_config 字符串为字典') except json.JSONDecodeError as e: logger.error(f'DashboardCreateNode - 解析 page_config 失败: {e}') return NodeResult( success=False, error=f'页面配置格式错误: {str(e)}', ) # 验证必要配置 if not dashboard_basic_info: return NodeResult( success=False, error='仪表盘基础信息不能为空', ) if not page_config: return NodeResult( success=False, error='页面设计配置不能为空', ) # 提取基础信息 name = dashboard_basic_info.get('name', '') code_base = dashboard_basic_info.get('code', '') category = dashboard_basic_info.get('category', 'dashboard') description = dashboard_basic_info.get('description', '') sort = dashboard_basic_info.get('sort', 0) if not name or not code_base: return NodeResult( success=False, error='仪表盘名称和编码不能为空', ) # 保存到数据库 from app.database import AsyncSessionLocal from online_dev.page_manager.model import PageMeta from sqlalchemy import select async with AsyncSessionLocal() as db: # 生成唯一的 code（如果已存在则自动添加后缀） code = await self._get_unique_code(db, code_base) logger.info(f'DashboardCreateNode - 使用 code: {code} (原始: {code_base})') # 检查是否已存在（理论上不应该存在，因为已经生成了唯一 code） stmt = select(PageMeta).where(PageMeta.code == code, PageMeta.is_deleted == False) result = await db.execute(stmt) existing_page = result.scalar_one_or_none() if existing_page: if update_if_exists: # 更新现有记录 existing_page.name = name existing_page.category = category existing_page.description = description existing_page.sort = sort existing_page.page_config = page_config if application_id: existing_page.application_id = application_id existing_page.version = (existing_page.version or 1) + 1 await db.commit() await db.refresh(existing_page) page_meta = self._page_to_dict(existing_page) logger.info(f'DashboardCreateNode - 更新仪表盘成功: {code}') else: return NodeResult( success=False, error=f'仪表盘编码已存在: {code}', ) else: # 创建新记录 new_page = PageMeta( name=name, code=code, category=category, description=description, sort=sort, status='draft', version=1, page_config=page_config, application_id=application_id if application_id else None, ) db.add(new_page) await db.commit() await db.refresh(new_page) page_meta = self._page_to_dict(new_page) logger.info(f'DashboardCreateNode - 创建仪表盘成功: {code}') output_data = { 'dashboard_id': page_meta['id'], 'dashboard_code': page_meta['code'], 'page_meta': page_meta, } return NodeResult( success=True, output=output_data, output_variables=output_data, ) except Exception as e: logger.error(f'DashboardCreateNode - 执行失败: {str(e)}') return NodeResult( success=False, error=f'创建仪表盘失败: {str(e)}', ) def _resolve_variable_value(self, context: NodeContext, value: Any) -> Any: """解析变量引用，保持原始类型""" import re import ast # 如果已经是字典，直接返回 if isinstance(value, dict): return value # 如果不是字符串，返回原值 if not isinstance(value, str): return value value = value.strip() if not value or value == '{}': return {} # 检查是否是单个变量引用 {{node_id.variable_name}} single_var_pattern = r'^\{\{([^}]+)\}\}$' match = re.match(single_var_pattern, value) if match: var_ref = match.group(1).strip() # 解析 node_id.variable_name 格式 if '.' in var_ref: parts = var_ref.split('.', 1) node_id = parts[0] key = parts[1] # 从节点输出命名空间获取 node_outputs = context.variables.get(f'_node_{node_id}') if isinstance(node_outputs, dict) and key in node_outputs: return node_outputs[key] # 直接从变量中获取 if key in context.variables: return context.variables[key] logger.warning(f'变量引用未找到: {var_ref}') else: # 直接变量引用 if var_ref in context.variables: return context.variables[var_ref] logger.warning(f'变量引用未找到: {var_ref}') # 不是变量引用，尝试解析为 JSON 或 Python dict # 尝试 JSON 解析 try: return json.loads(value) except json.JSONDecodeError: pass # 尝试 Python literal_eval（处理单引号格式） try: result = ast.literal_eval(value) if isinstance(result, dict): return result return result except (ValueError, SyntaxError) as e: logger.warning(f'配置解析失败: {e}, 值: {value[:200]}') return value async def _get_unique_code(self, db, base_code: str) -> str: """ 生成唯一的 code，如果已存在则自动添加数字后缀 Args: db: 数据库会话 base_code: 基础 code Returns: 唯一的 code """ from online_dev.page_manager import PageService code = base_code counter = 1 while True: try: existing = await PageService.get_by_code(db, code) if not existing: # code 不存在，可以使用 break # code 已存在，添加后缀 code = f"{base_code}_{counter}" counter += 1 except Exception: # 查询出错，认为 code 可用 break return code def _page_to_dict(self, page: Any) -> Dict[str, Any]: """将PageMeta对象转换为字典""" return { 'id': page.id, 'name': page.name, 'code': page.code, 'category': page.category, 'description': page.description, 'status': page.status, 'version': page.version, 'sort': page.sort, 'page_config': page.page_config, 'application_id': page.application_id, }