""" 应用更新节点 执行应用配置的更新操作，将设置保存到数据库 """ import json import logging from typing import Any, Dict, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from ..base import BaseNode, NodeContext, NodeResult from ..registry import NodeRegistry logger = logging.getLogger(__name__) @NodeRegistry.register class AppUpdateNode(BaseNode): """ 应用更新节点 接收应用设置配置，执行实际的更新操作 """ node_type = 'app_update' node_name = '应用更新' node_category = 'application' node_icon = 'save' node_description = '执行应用配置的更新操作，将设置保存到数据库' inputs = [ { 'name': 'settings', 'type': 'object', 'description': '应用设置配置（来自应用设置节点）', 'required': True, }, { 'name': 'application_id', 'type': 'string', 'description': '应用ID，不传则更新主应用配置', 'required': False, }, ] outputs = [ { 'name': 'success', 'type': 'boolean', 'description': '更新是否成功', }, { 'name': 'config_id', 'type': 'string', 'description': '配置记录ID', }, { 'name': 'message', 'type': 'string', 'description': '更新结果消息', }, ] async def execute_async(self, context: NodeContext) -> NodeResult: """异步执行节点""" try: # 获取输入参数 settings = self._get_settings(context) application_id = self._get_application_id(context) if not settings: return NodeResult( success=False, error='缺少应用设置配置', ) logger.info(f'AppUpdateNode - 开始更新应用配置, application_id={application_id}') # 获取数据库会话 db: AsyncSession = context.db_session if not db: return NodeResult( success=False, error='数据库会话不可用', ) # 执行更新 config_id = await self._update_preferences(db, settings, application_id) if config_id: logger.info(f'AppUpdateNode - 应用配置更新成功, config_id={config_id}') return NodeResult( success=True, output={ 'success': True, 'config_id': config_id, 'message': '应用配置更新成功', }, output_variables={ 'update_success': True, 'config_id': config_id, 'update_message': '应用配置更新成功', }, ) else: return NodeResult( success=False, error='应用配置更新失败', ) except Exception as e: logger.exception(f'应用更新节点执行失败: {e}') return NodeResult( success=False, error=f'节点执行失败: {str(e)}', ) def execute(self, context: NodeContext) -> NodeResult: """同步执行节点（不支持，需要数据库操作）""" return NodeResult( success=False, error='应用更新节点需要异步执行', ) def _get_settings(self, context: NodeContext) -> Optional[Dict[str, Any]]: """获取应用设置配置""" import re # 优先从配置中获取（支持变量引用） settings_config = self.config.get('settings', '') logger.info(f'AppUpdateNode - settings_config from config: {settings_config}') if settings_config: # 检查是否是变量引用格式 {{node_id.key}} var_pattern = r'\{\{([^}]+)\}\}' match = re.match(var_pattern, settings_config.strip()) if match: # 是变量引用，直接从上下文获取对象 var_ref = match.group(1).strip() logger.info(f'AppUpdateNode - 解析变量引用: {var_ref}') # 解析 node_id.key 格式 # 注意：node_id 可能包含连字符，如 app_settings-1769009671187 # 所以只按最后一个点分割 if '.' in var_ref: parts = var_ref.rsplit('.', 1) # 从右边分割，只分割一次 node_id = parts[0] key = parts[1] if len(parts) > 1 else None logger.info(f'AppUpdateNode - 查找节点输出: node_id={node_id}, key={key}') # 从节点输出命名空间获取 node_outputs = context.variables.get(f'_node_{node_id}') logger.info(f'AppUpdateNode - node_outputs type: {type(node_outputs)}, keys: {node_outputs.keys() if isinstance(node_outputs, dict) else "N/A"}') if isinstance(node_outputs, dict) and key and key in node_outputs: value = node_outputs[key] logger.info(f'AppUpdateNode - 找到值, type: {type(value)}') if isinstance(value, dict): return value if isinstance(value, str): try: return json.loads(value) except (json.JSONDecodeError, TypeError): logger.warning(f'AppUpdateNode - JSON解析失败: {value[:100]}') pass else: logger.warning(f'AppUpdateNode - 未找到节点输出或key不存在') else: # 直接变量名 value = context.variables.get(var_ref) logger.info(f'AppUpdateNode - 直接变量引用, type: {type(value)}') if isinstance(value, dict): return value if isinstance(value, str): try: return json.loads(value) except (json.JSONDecodeError, TypeError): pass else: # 不是变量引用，尝试解析为 JSON try: return json.loads(settings_config) except (json.JSONDecodeError, TypeError): pass # 从上下文变量获取 settings = context.variables.get('settings') logger.info(f'AppUpdateNode - 从上下文获取settings, type: {type(settings)}') if isinstance(settings, dict): return settings if isinstance(settings, str): try: return json.loads(settings) except (json.JSONDecodeError, TypeError): pass logger.warning(f'AppUpdateNode - 未能获取有效的settings配置') return None def _get_application_id(self, context: NodeContext) -> Optional[str]: """获取应用ID""" # 优先从配置中获取 app_id = self.config.get('application_id', '') if app_id: resolved = context.resolve_template(app_id) if resolved and resolved != 'main': return resolved # 从上下文变量获取 app_id = context.variables.get('application_id') if app_id and app_id != 'main': return app_id return None async def _update_preferences( self, db: AsyncSession, settings: Dict[str, Any], application_id: Optional[str] = None, ) -> Optional[str]: """更新偏好配置""" from core.ui_config.model import UIConfig # 配置键：使用与前端 API 一致的键名 config_key = 'frontend_preferences' # 查找现有配置 stmt = select(UIConfig).where( UIConfig.config_key == config_key, UIConfig.is_deleted == False, ) # 根据 application_id 过滤 if application_id: stmt = stmt.where(UIConfig.application_id == application_id) else: stmt = stmt.where(UIConfig.application_id.is_(None)) result = await db.execute(stmt) config = result.scalar_one_or_none() # 配置值 config_value = json.dumps(settings, ensure_ascii=False) if config: # 更新现有配置 config.config_value = config_value await db.commit() await db.refresh(config) return config.id else: # 创建新配置 new_config = UIConfig( application_id=application_id, config_key=config_key, config_value=config_value, config_type='preferences', description='前端UI偏好配置' if not application_id else '子应用UI偏好配置', status=True, sort=0, ) db.add(new_config) await db.commit() await db.refresh(new_config) return new_config.id @classmethod def get_config_schema(cls) -> Dict[str, Any]: """ 获取节点配置 Schema（供前端表单渲染） """ return { 'type': 'object', 'properties': { 'settings': { 'type': 'string', 'title': '应用设置', 'description': '应用设置配置，支持变量引用如 {{app_settings-1.settings}}', 'required': True, 'x-component': 'Input', 'x-component-props': { 'placeholder': '请输入应用设置或使用变量引用', }, }, 'application_id': { 'type': 'string', 'title': '应用ID', 'description': '目标应用ID，不填则更新主应用配置。支持变量引用如 {{application_id}}', 'required': False, 'x-component': 'Input', 'x-component-props': { 'placeholder': '留空则更新主应用配置', }, }, }, 'required': ['settings'], }