303 lines
11 KiB
Python
303 lines
11 KiB
Python
"""
|
|
应用更新节点
|
|
|
|
执行应用配置的更新操作，将设置保存到数据库
|
|
"""
|
|
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'],
|
|
}
|