版本：Next

PythonScriptActivity

分类: 通用步骤
命名空间: CMS.Plugin.FlowManagement.Domain.FlowBusiness.Activitys
基类: ActionActivity
模块: FlowManagement.Domain

概述

PythonScriptActivity（Python脚本）是用于在流程中直接执行 Python 脚本的节点。它继承自 ActionActivity，使用 IronPython 引擎在流程运行时执行内联的 Python 脚本代码。该节点支持参数映射和结果提取，可以灵活地进行数据计算、转换和处理。

与 ScriptCallActivity 不同，PythonScriptActivity 允许直接在节点配置中编写 Python 代码，无需预先在脚本管理中配置，适合简单的数据处理和计算场景。

业务场景

适用场景

数据计算: 执行数学计算、统计分析等操作
数据转换: 字符串处理、格式转换、编码解码
条件判断: 复杂的业务规则判断
数据验证: 数据格式验证、范围检查
快速原型: 快速实现和测试业务逻辑

在系统中的作用

PythonScriptActivity 在 LMES 流程系统中扮演着内联脚本执行器的角色：

直接执行配置的 Python 脚本
支持从流程上下文传入参数
支持将执行结果写回流程上下文
使用 IronPython 引擎，兼容 Python 2.x 语法

与其他节点的协作

BusinessActivity: 在脚本执行前准备参数，在脚本执行后使用结果
ActionActivity: 作为动作节点，可以配置执行时机
ScriptCallActivity: 如需调用预配置的脚本，使用 ScriptCallActivity
MethodCallActivity: 如需调用 C# 方法，使用 MethodCallActivity

配置说明

基本配置

属性名	类型	必填	默认值	说明
Script	string	是	-	执行脚本，编辑用于计算结果的 Python 脚本
Mapping	ParameterMapCollection	否	空集合	参数匹配，将流程参数作为调用参数传入
ResultParameter	ResultParameterCollection	否	空集合	结果参数，动作执行后结果存储于流程参数的名称
Action	ActionType	否	Execute	执行时机（继承自 ActionActivity）
IsSynchronous	bool	否	true	是否同步执行（继承自 ActionActivity）

配置项详解

Script

说明: Python 脚本代码，在流程执行时运行。脚本使用 IronPython 引擎执行，兼容 Python 2.x 语法。

取值范围: 字符串，有效的 Python 代码

注意事项:

使用 IronPython 引擎，部分 Python 3.x 特性可能不支持
脚本会自动添加当前工作目录到 sys.path
可以使用 import 导入标准库和自定义模块

Mapping

说明: 参数映射配置，定义如何将流程上下文中的参数传递给 Python 脚本。

配置结构: ParameterMapCollection

参数映射模型:

Source: 流程上下文中的参数名（数据来源）
Target: Python 脚本中的变量名（目标变量）

注意事项:

映射的参数会作为 Python 变量在脚本中可用
参数类型会自动转换为 Python 对应类型

ResultParameter

说明: 结果参数配置，定义如何将 Python 脚本的执行结果写回流程上下文。

配置结构: ResultParameterCollection

结果参数模型:

Name: Python 脚本中的变量名，同时也是写入流程上下文的键名

注意事项:

脚本执行后，会从 Python 作用域中获取指定变量的值
结果会按顺序写入流程上下文

流程上下文

输入参数

参数名	类型	说明
Mapping.Source	any	通过 Mapping 配置传入脚本的参数

输出参数

参数名	类型	说明
ResultParameter.Name	any	通过 ResultParameter 配置从脚本获取的结果

数据流转说明

执行前:
- 创建 IronPython 引擎和作用域
- 根据 Mapping 配置，从流程上下文读取参数并设置为 Python 变量
- 设置 curWorkDirectory 变量为当前工作目录
脚本执行:
- 自动添加 import sys; sys.path.append(curWorkDirectory) 到脚本开头
- 执行 Python 脚本
执行后:
- 根据 ResultParameter 配置，从 Python 作用域获取变量值
- 将结果写入流程上下文

业务逻辑说明

处理流程

PythonScriptActivity 的执行流程如下：

创建引擎:
- 使用 Python.CreateEngine() 创建 IronPython 引擎
- 创建脚本执行作用域
设置参数:
- 遍历 Mapping 配置
- 从流程上下文获取源参数值
- 使用 scope.SetVariable 设置 Python 变量
设置工作目录:
- 获取当前程序集所在目录
- 设置 curWorkDirectory 变量
执行脚本:
- 在脚本开头添加 sys.path 配置
- 创建脚本源并执行
获取结果:
- 遍历 ResultParameter 配置
- 使用 scope.GetVariable 获取 Python 变量值
- 将结果写入流程上下文

依赖服务

服务接口	用途	说明
IronPython.Hosting.Python	Python 引擎	创建和执行 Python 脚本
Flow.Logger	日志记录器	记录脚本执行日志

日志记录

PythonScriptActivity 记录以下关键日志：

工作目录: curWorkDirectory={curWorkDirectory}
脚本内容: Script={newScript}
执行结果: Result={result}

使用示例

基本示例：简单计算

{
  "Type": "PythonScriptActivity",
  "Name": "计算总价",
  "Alias": "CalcTotal",
  "Script": "total = price * quantity",
  "Mapping": [
    { "Source": "Price_Value", "Target": "price" },
    { "Source": "Quantity_Value", "Target": "quantity" }
  ],
  "ResultParameter": [
    { "Name": "total" }
  ]
}

字符串处理示例

{
  "Type": "PythonScriptActivity",
  "Name": "格式化条码",
  "Alias": "FormatBarcode",
  "Script": "formatted = prefix + barcode.zfill(10) + suffix",
  "Mapping": [
    { "Source": "Barcode_Value", "Target": "barcode" },
    { "Source": "Prefix_Value", "Target": "prefix" },
    { "Source": "Suffix_Value", "Target": "suffix" }
  ],
  "ResultParameter": [
    { "Name": "formatted" }
  ]
}

条件判断示例

{
  "Type": "PythonScriptActivity",
  "Name": "质量判定",
  "Alias": "QualityCheck",
  "Script": "if value >= min_val and value <= max_val:\n    result = 'OK'\nelse:\n    result = 'NG'",
  "Mapping": [
    { "Source": "MeasuredValue", "Target": "value" },
    { "Source": "MinValue", "Target": "min_val" },
    { "Source": "MaxValue", "Target": "max_val" }
  ],
  "ResultParameter": [
    { "Name": "result" }
  ]
}

多结果输出示例

{
  "Type": "PythonScriptActivity",
  "Name": "数据分析",
  "Alias": "DataAnalysis",
  "Script": "total = sum(values)\naverage = total / len(values)\nmax_val = max(values)\nmin_val = min(values)",
  "Mapping": [
    { "Source": "DataList_Value", "Target": "values" }
  ],
  "ResultParameter": [
    { "Name": "total" },
    { "Name": "average" },
    { "Name": "max_val" },
    { "Name": "min_val" }
  ]
}

完整流程示例

{
  "Name": "产品检验流程",
  "Activities": [
    {
      "Type": "BusinessActivity",
      "Name": "获取检验数据",
      "Alias": "GetInspectionData"
    },
    {
      "Type": "PythonScriptActivity",
      "Name": "计算合格率",
      "Alias": "CalcPassRate",
      "Script": "pass_rate = (pass_count / total_count) * 100\nif pass_rate >= 95:\n    status = 'PASS'\nelse:\n    status = 'FAIL'",
      "Mapping": [
        { "Source": "PassCount_Value", "Target": "pass_count" },
        { "Source": "TotalCount_Value", "Target": "total_count" }
      ],
      "ResultParameter": [
        { "Name": "pass_rate" },
        { "Name": "status" }
      ]
    },
    {
      "Type": "BusinessActivity",
      "Name": "记录检验结果",
      "Alias": "SaveResult"
    }
  ]
}

使用外部模块示例

{
  "Type": "PythonScriptActivity",
  "Name": "日期计算",
  "Alias": "DateCalc",
  "Script": "from datetime import datetime, timedelta\nstart = datetime.strptime(start_date, '%Y-%m-%d')\nend = start + timedelta(days=days)\nend_date = end.strftime('%Y-%m-%d')",
  "Mapping": [
    { "Source": "StartDate_Value", "Target": "start_date" },
    { "Source": "Days_Value", "Target": "days" }
  ],
  "ResultParameter": [
    { "Name": "end_date" }
  ]
}

扩展开发指南

继承层次

Activity (SYC.Flow.Kernel)
  └── BusinessActivity
        └── ActionActivity
              └── PythonScriptActivity

可重写方法

方法名	用途	何时重写
ExecuteActionAsync	执行脚本	需要自定义执行流程
ExecuteScript	执行 Python 脚本	需要自定义脚本执行逻辑

自定义 Python 脚本节点示例

[Serializable]
[Design("安全Python脚本", "带超时和异常处理的Python脚本节点", Sort = 1)]
[Category("自定义")]
public class SafePythonScriptActivity : PythonScriptActivity
{
    [Design("超时时间", "脚本执行超时时间（秒）", Sort = 10)]
    [Category("脚本配置")]
    public int TimeoutSeconds { get; set; } = 30;
    
    protected override object[] ExecuteScript(ProcessflowEventArgs args)
    {
        try
        {
            var task = Task.Run(() => base.ExecuteScript(args));
            if (task.Wait(TimeSpan.FromSeconds(TimeoutSeconds)))
            {
                return task.Result;
            }
            else
            {
                Flow.Logger.LogErrorMessage($"脚本执行超时: {TimeoutSeconds}秒", Name);
                return null;
            }
        }
        catch (Exception ex)
        {
            Flow.Logger.LogExceptionMessage(ex, Name);
            return null;
        }
    }
}

注意事项

⚠️ IronPython 限制: 使用 IronPython 引擎，部分 Python 3.x 特性不支持
⚠️ 性能考虑: 每次执行都会创建新的 Python 引擎，频繁调用可能影响性能
⚠️ 类型转换: 参数在 .NET 和 Python 之间传递时会自动转换类型
⚠️ 异常处理: 脚本执行异常会导致流程中断，建议在脚本中添加异常处理
⚠️ 安全性: 脚本可以访问系统资源，注意安全风险
💡 最佳实践:
- 保持脚本简洁，复杂逻辑建议使用 ScriptCallActivity
- 使用 Mapping 传递参数，避免硬编码
- 使用 ResultParameter 获取结果，便于后续节点使用
- 添加适当的日志记录，便于调试
- 对于复杂计算，考虑使用 MethodCallActivity 调用 C# 方法

更新历史

日期	版本	说明
2025-11-28	1.0	初始版本

本文档最后更新时间: 2025-11-28

概述​

业务场景​

适用场景​

在系统中的作用​

与其他节点的协作​

配置说明​

基本配置​

配置项详解​

Script​

Mapping​

ResultParameter​

流程上下文​

输入参数​

输出参数​

数据流转说明​

业务逻辑说明​

处理流程​

依赖服务​

日志记录​

使用示例​

基本示例：简单计算​

字符串处理示例​

条件判断示例​

多结果输出示例​

完整流程示例​

使用外部模块示例​

扩展开发指南​

继承层次​

可重写方法​

自定义 Python 脚本节点示例​

注意事项​

相关节点​

更新历史​

概述

业务场景

适用场景

在系统中的作用

与其他节点的协作

配置说明

基本配置

配置项详解

Script

Mapping

ResultParameter

流程上下文

输入参数

输出参数

数据流转说明

业务逻辑说明

处理流程

依赖服务

日志记录

使用示例

基本示例：简单计算

字符串处理示例

条件判断示例

多结果输出示例

完整流程示例

使用外部模块示例

扩展开发指南

继承层次

可重写方法

自定义 Python 脚本节点示例

注意事项

相关节点

更新历史