---

# 📘功能说明文档（FSD）  
## 模块名称：Python 在线脚本编辑器（Python Script Editor）

---

### 一、模块定位
为平台提供 **可扩展的 Python 脚本执行引擎** 的能力，支持：
- **文件创建与管理**：支持目录和脚本文件的创建、编辑、删除操作
- **脚本运行控制**：提供脚本启动、停止、重启等生命周期管理
- **实时日志推送**：通过 WebSocket 实时推送脚本执行日志和输出结果
- **日志记录存储**：将脚本执行过程、结果、异常等信息持久化存储
- **接口集成功能**：与 VWED 任务系统深度集成，支持任务流程中调用脚本
- **自定义模块扩展**：支持注册自定义接口、函数、事件监听器等
- **函数注册调用**：支持在 VWED 任务中通过脚本模块引用和运行注册的函数
- **多脚本并发服务**：每个脚本独立运行，支持多脚本同时提供服务
- **企业级安全沙箱**：提供安全的脚本执行环境和权限控制

---

### 二、基于现有架构的脚本引擎框架

```
VWED脚本引擎架构（基于现有项目结构）
├── 🌐 API接口层 (routes/)
│   ├── script_engine_api.py        # REST API接口
│   ├── script_websocket_api.py     # WebSocket API接口  
│   └── model/script_engine_model.py # 请求/响应模型
├── 🔧 业务服务层 (services/)
│   ├── script_engine_service.py      # 核心引擎服务
│   ├── script_registry_service.py    # 注册中心服务
│   ├── script_builtin_functions.py   # 内置函数服务
│   ├── script_security_sandbox.py    # 安全沙箱服务
│   └── script_websocket_logger.py    # WebSocket日志服务
├── 🔗 任务集成层 (services/execution/handlers/)
│   └── script_engine_integration.py  # 任务流程集成
├── 💾 数据模型层 (data/models/)
│   ├── script_project.py             # 脚本项目模型
│   ├── script_file.py                # 脚本文件模型
│   ├── script_api_registration.py    # API注册记录模型
│   ├── script_function_registration.py # 函数注册记录模型
│   ├── script_execution_log.py       # 执行日志模型
│   └── script_registry.py            # 脚本注册模型
├── ⚙️ 配置层 (config/)
│   └── components/script.json        # 脚本组件配置
└── 📊 任务流程集成
    ├── SCRIPT_ENGINE_EXECUTE（执行脚本）
    ├── SCRIPT_ENGINE_CALL_FUNCTION（调用注册函数）
    └── SCRIPT_ENGINE_HTTP_REQUEST（HTTP请求组件）
```

---

### 三、脚本引擎核心特性

| 特性名称 | 描述 | 技术实现 |
|----------|------|----------|
| 文件管理系统 | 支持脚本文件和目录的CRUD操作 | 基于文件系统的目录结构管理，支持多层级目录 |
| 脚本生命周期管理 | 脚本的启动、停止、重启等状态控制 | ScriptEngine服务管理脚本进程生命周期 |
| 实时日志推送 | WebSocket实时推送脚本执行日志和结果 | ScriptWebSocketLogger服务实现实时通信 |
| 日志持久化存储 | 脚本执行过程和结果的数据库存储 | 执行日志保存到数据库，支持历史查询 |
| 任务系统集成 | 与VWED任务流程深度集成调用 | script.py处理器支持任务中调用脚本函数 |
| 函数动态加载 | 动态加载并执行脚本中的指定函数 | importlib.util实现模块动态加载和函数调用 |
| 参数传递机制 | 支持多种类型参数传递给脚本函数 | 支持dict、list、str等类型参数智能解析 |
| 异步执行支持 | 支持异步函数的执行和结果等待 | inspect检测协程并使用asyncio.await处理 |
| 变量上下文管理 | 脚本执行结果保存到任务上下文 | TaskContext管理任务变量，支持跨组件传递 |
| 错误处理机制 | 完善的异常捕获和错误信息反馈 | 异常信息记录到日志，返回详细错误描述 |

---

# 📗需求文档（PRD）  
## 一、目标用户
- 实施工程师
- 自动化工程师
- 会 Python 的业务开发人员

---

## 二、基于现有架构的功能需求

| 编号 | 需求名称 | 描述 | 当前实现状态 |
|------|----------|------|-------------|
| BR01 | 脚本文件创建管理 | 支持脚本文件和目录的创建、编辑、删除 | ✅ 已实现 - 基于文件系统的CRUD操作 |
| BR02 | 脚本执行控制 | 脚本的启动、停止、状态监控 | ✅ 已实现 - ScriptEngine生命周期管理 |
| BR03 | 函数动态调用 | 动态加载脚本文件并执行指定函数 | ✅ 已实现 - importlib.util动态加载机制 |
| BR04 | 参数传递支持 | 支持多种数据类型的参数传递 | ✅ 已实现 - dict/list/str参数智能解析 |
| BR05 | 异步执行支持 | 支持异步函数的执行和结果等待 | ✅ 已实现 - inspect+asyncio异步处理 |
| BR06 | 实时日志推送 | WebSocket实时推送执行日志和结果 | ✅ 已实现 - WebSocket实时通信 |
| BR07 | 日志持久化 | 执行过程和结果的数据库存储 | ✅ 已实现 - 执行日志数据库记录 |
| BR08 | 任务系统集成 | 在VWED任务流程中调用脚本函数 | ✅ 已实现 - script.py处理器集成 |
| BR09 | 变量上下文管理 | 脚本执行结果保存到任务变量 | ✅ 已实现 - TaskContext变量管理 |
| BR10 | 错误处理机制 | 完善的异常捕获和错误信息反馈 | ✅ 已实现 - 多层异常处理和日志记录 |

---

## 三、非功能需求（不变）

| 类型 | 要求 |
|------|------|
| 性能 | 脚本加载 < 1s，接口响应 < 200ms |
| 安全 | 脚本沙箱运行，禁止访问系统文件、网络（除白名单） |
| 可维护性 | 脚本统一存储，支持版本控制（Git） |
| 可扩展性 | 后续可支持 Python3 虚拟环境、pip 安装依赖 |
| 兼容性 | 支持 Chrome、Edge、Firefox |

---

# 📙开发文档（TDD）  
## 一、基于现有架构的技术选型

| 架构层级 | 技术选型 | 具体实现 |
|----------|----------|----------|
| API接口层 | FastAPI + WebSocket | routes/script_engine_api.py + routes/script_websocket_api.py |
| 业务服务层 | Service Pattern + 依赖注入 | services/script_engine_service.py 等服务模块 |
| 数据模型层 | SQLAlchemy ORM + Enum | data/models/script_*.py + data/enum/ |
| 任务集成层 | VWED任务执行系统 | services/execution/handlers/script_engine_integration.py |
| 安全防护层 | Python沙箱 + 权限控制 | services/script_security_sandbox.py |
| 配置管理层 | JSON配置 + 组件系统 | config/components/script.json |
| 日志监控层 | WebSocket实时推送 + 结构化日志 | services/script_websocket_logger.py |
| 数据库层 | MySQL/PostgreSQL + 连接池 | data/session.py + migrations/ |

---

## 二、基于现有架构的脚本引擎设计

### 2.1 可扩展框架整体架构

```text
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│                       前端编辑界面 (Monaco Editor)                        │
│  ┌───────────────┬───────────────┬───────────────┬───────────────┐    │
│  │  脚本编辑器   │   实时日志     │   调试控台     │   性能监控     │    │
│  │- Python语法高亮│- WebSocket日志│- remote-pdb   │- CPU/内存      │    │
│  │- 代码智能补全  │- 等级过滤     │- 断点调试     │- 脚本执行状态  │    │
│  │- 语法检查&提示 │- 日志导出     │- 变量监控     │- 资源使用统计  │    │
│  └───────────────┴───────────────┴───────────────┴───────────────┘    │
└───────────────────────────────────────────────────────────────────────────────────────────┘
                                      │ HTTP/WebSocket API
                                      ▼
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│           🔌 接口服务层 (scripts/interfaces/) - 可扩展协议支持           │
│  ┌───────────────┬───────────────┬───────────────┬───────────────┐  │
│  │ ScriptEngineAPI │ScriptWebSocketAPI│ScriptManagementAPI│   协议扩展   │  │
│  │- 脚本执行控制  │- 实时日志推送  │- 文件管理       │- gRPC/MQTT     │  │
│  │- 状态监控      │- 调试通信      │- 权限管理       │- TCP/UDP       │  │
│  │- API注册管理   │- 性能指标      │- 版本控制       │- 自定义协议   │  │
│  └───────────────┴───────────────┴───────────────┴───────────────┘  │
└───────────────────────────────────────────────────────────────────────────────────────────┘
                                      │ 服务调用
                                      ▼
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│        🔧 业务服务层 (scripts/services/) - 服务注册与发现          │
│  ┌─────────────┬────────────┬────────────┬────────────┐  │
│  │ScriptEngine │ Registry    │BuiltinFunctions│WebSocketLogger│  │
│  │- 多脚本并发 │- 资源注册   │- VWED对象模块│- 日志监控    │  │
│  │- 生命周期管理│- script_id隔离│- 扩展模块加载│- 实时推送    │  │
│  │- 资源清理    │- 动态注销   │- 热更新支持  │- 日志聚合    │  │
│  └─────────────┴────────────┴────────────┴────────────┘  │
└───────────────────────────────────────────────────────────────────────────────────────────┘
                                      │ 模块加载
                                      ▼
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│     🧩 模块扩展层 (scripts/modules/) - 每个模块独立可扩展      │
│  ┌──────────────────────────────────────────────────────────────────────────────┐  │
│  │                      内置模块生态 (VWED对象系统)                      │  │
│  │  ┌────────────┬───────────┬───────────┬────────────┐    │  │
│  │  │ Core模块   │ Task模块  │Device模块 │ Comm模块  │    │  │
│  │  │- api        │- task       │- device     │- http       │    │  │
│  │  │- function   │- timer      │- robot      │- websocket  │    │  │
│  │  │- event      │- scheduler  │- modbus     │- tcp        │    │  │
│  │  └────────────┴───────────┴───────────┴────────────┘    │  │
│  │  ┌────────────┬───────────┬───────────┬────────────┐    │  │
│  │  │Storage模块│ Log模块   │Utility模块│ 自定义模块 │    │  │
│  │  │- storage    │- log        │- mail       │- AI模块     │    │  │
│  │  │- database   │- alert      │- crypto     │- 视觉模块   │    │  │
│  │  │- cache      │- audit      │- file       │- 算法模块   │    │  │
│  │  └────────────┴───────────┴───────────┴────────────┘    │  │
│  └──────────────────────────────────────────────────────────────────────────────┘  │
└───────────────────────────────────────────────────────────────────────────────────────────┘
                                      │
                                      ▼
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│          ⚙️ 核心引擎层 (scripts/engine/) - 多脚本并发管理           │
│  ┌───────────────────────────────────────────────────────────────────────────┐  │
│  │                     VWEDGlobalRegistry (全局注册中心)                    │  │
│  │  ┌────────────┬────────────┬────────────┬────────────┐    │  │
│  │  │MultiScript   │SecuritySandbox│LifecycleManager│ ResourceTracker│    │  │
│  │  │Engine        │  安全沙箱     │  生命周期管理  │   资源跟踪   │    │  │
│  │  │-多脚本并发 │- 模块白名单   │- boot()入口函数│- script_id隔离│    │  │
│  │  │-script_id隔离│- 资源限制     │- 异常处理      │- 资源自动回收│    │  │
│  │  │-状态管理   │- 权限控制     │- 重启机制      │- 性能监控    │    │  │
│  │  └────────────┴────────────┴────────────┴────────────┘    │  │
│  └───────────────────────────────────────────────────────────────────────────┐  │
└───────────────────────────────────────────────────────────────────────────────────────────┘
                                      │
                                      ▼
┌───────────────────────────────────────────────────────────────────────────────────────────┐
│           💾 数据模型层 (scripts/models/) - 持久化与存储            │
│  ┌───────────────┬───────────────┬───────────────┬───────────────┐  │
│  │ ScriptProject │  ScriptFile   │ExecutionLog   │APIRegistration │  │
│  │- 项目管理     │- 文件存储     │- 执行记录     │- API注册记录  │  │
│  │- 目录结构     │- 版本管理     │- 错误日志     │- 函数注册记录│  │
│  │- 权限控制     │- 内容检索     │- 性能指标     │- 路由注册记录│  │
│  └───────────────┴───────────────┴───────────────┴───────────────┘  │
└───────────────────────────────────────────────────────────────────────────────────────────┘
```

### 2.2 注册中心详细机制

#### 2.2.1 API接口注册与访问流程

```text
【在线脚本注册阶段】
┌─────────────────────────────────────────────────────────┐
│  在线脚本: VWED.api.register_route()                     │
│  def boot():                                           │
│      VWED.api.register_route(                         │
│          path="/script-api/custom",                   │
│          method="POST",                               │
│          handler=my_custom_handler                    │
│      )                                                │
└─────────────────────┬───────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│               全局注册中心存储                           │
│  api_routes = {                                        │
│      "/script-api/custom": {                          │
│          "method": "POST",                            │
│          "handler": my_custom_handler,                │
│          "script_id": "script_001",                   │
│          "registered_at": "2024-01-01 12:00:00"       │
│      }                                                │
│  }                                                    │
└─────────────────────┬───────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│              FastAPI 动态路由注册                        │
│  @app.route("/script-api/{path:path}", methods=ALL)    │
│  async def dynamic_script_route(path, request):        │
│      # 查找注册表                                       │
│      route_info = registry.get_api_route(path)        │
│      if route_info:                                   │
│          return await route_info["handler"](request)  │
│      else:                                            │
│          return {"error": "接口未注册"}                  │
└─────────────────────────────────────────────────────────┘

【HTTP请求块访问阶段】
┌─────────────────────────────────────────────────────────┐
│  任务流程中的HTTP请求块 (htttp_request.py)               │
│  async def execute():                                  │
│      url = "http://localhost:8000/script-api/custom"  │
│      response = await http_client.post(url, data={})  │
│      # 这会触发上面的动态路由                            │
└─────────────────────┬───────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│              执行用户注册的函数                          │
│  def my_custom_handler(request):                       │
│      # 用户自定义的业务逻辑                              │
│      data = request.json()                            │
│      result = process_business_logic(data)            │
│      return {"success": True, "data": result}         │
└─────────────────────────────────────────────────────────┘

【脚本执行块调用阶段】
┌─────────────────────────────────────────────────────────┐
│  任务流程中的脚本执行块 (script.py)                      │
│  async def _execute_script():                          │
│      # 除了加载文件函数外，还可以调用注册的函数            │
│      registered_func = registry.get_registered_function(│
│          function_name                                │
│      )                                                │
│      if registered_func:                              │
│          return await registered_func(function_args)  │
└─────────────────────────────────────────────────────────┘
```

#### 2.2.2 自定义函数注册机制

```text
【函数注册】
VWED.function.register(name="calculate_distance", handler=calc_func)
            │
            ▼
┌─────────────────────────────────────────────────────────┐
│               函数注册表                                │
│  registered_functions = {                              │
│      "calculate_distance": {                          │
│          "handler": calc_func,                        │
│          "script_id": "script_001",                   │
│          "description": "计算两点距离",                 │
│          "params": ["point1", "point2"]               │
│      }                                                │
│  }                                                    │
└─────────────────────┬───────────────────────────────────┘
                     │
                     ▼
【函数调用】
┌─────────────────────────────────────────────────────────┐
│  script.py 执行块                                      │
│  function_name = "calculate_distance"                  │
│  registered_func = registry.get_function(function_name)│
│  result = await registered_func(function_args)        │
└─────────────────────────────────────────────────────────┘
```

#### 2.2.3 事件系统集成

```text
【事件注册】
VWED.event.listen(event_name="task_completed", handler=on_complete)
            │
            ▼
┌─────────────────────────────────────────────────────────┐
│              事件监听器注册表                            │
│  event_listeners = {                                   │
│      "task_completed": [                              │
│          {                                            │
│              "handler": on_complete,                  │
│              "script_id": "script_001",               │
│              "priority": 1                            │
│          }                                            │
│      ]                                                │
│  }                                                    │
└─────────────────────┬───────────────────────────────────┘
                     │
                     ▼
【事件触发】
┌─────────────────────────────────────────────────────────┐
│  系统其他模块触发事件                                    │
│  from services.script_engine import script_registry   │
│  await script_registry.emit_event(                    │
│      "task_completed",                                │
│      {"task_id": 123, "status": "success"}           │
│  )                                                    │
└─────────────────────────────────────────────────────────┘
```

---

## 三、脚本服务生命周期与功能实现

### 3.1 文件创建与管理功能
1. **目录创建**：
   - 支持在 `scripts/user_save/` 目录下创建多层级目录结构
   - 提供目录重命名、删除等管理功能
   
2. **脚本文件创建**：
   - 在指定目录下创建 `.py` 脚本文件
   - 提供文件重命名、删除、复制等操作
   - 支持文件内容的在线编辑和保存

3. **脚本模板生成**：
   - 新建脚本时自动生成标准模板代码
   - 包含常用的函数定义和注释结构

### 3.2 脚本执行控制功能
1. **脚本启动运行**：
   - 通过 API 接口启动指定脚本文件执行
   - 支持传递启动参数和配置信息
   - 实时监控脚本运行状态

2. **脚本停止控制**：
   - 支持强制停止正在运行的脚本
   - 优雅关闭和资源清理机制
   - 停止后状态反馈和日志记录

3. **脚本重启功能**：
   - 支持脚本的重启操作
   - 保持配置参数的重启机制

### 3.3 实时日志推送与记录
1. **WebSocket 实时推送**：
   - 脚本执行过程中的标准输出实时推送
   - 错误信息和异常堆栈的实时推送
   - 执行状态变化的实时通知

2. **日志分级管理**：
   - 支持 DEBUG、INFO、WARNING、ERROR 等日志级别
   - 按级别过滤和显示日志信息
   - 日志格式标准化和时间戳记录

3. **日志持久化存储**：
   - 执行日志保存到数据库表中
   - 支持历史日志查询和检索
   - 执行结果和异常信息的结构化存储

### 3.4 任务系统集成调用
1. **VWED 任务流程集成**：
   - 在任务流程中通过脚本组件调用 Python 函数
   - 支持 `SCRIPT` 和 `SET_TASK_VARIABLES` 两种调用模式
   - 与任务执行引擎无缝集成

2. **函数动态调用机制**：
   ```python
   # 当前实现：固定加载 scripts/user_save/test1.py
   script_file = "scripts/user_save/test1.py"
   
   # 动态加载模块
   spec = importlib.util.spec_from_file_location("user_script", script_file)
   module = importlib.util.module_from_spec(spec)
   spec.loader.exec_module(module)
   
   # 调用指定函数
   func = getattr(module, function_name)
   result = func(function_args)
   ```

3. **参数传递机制**：
   - **字典参数**：`{"key": "value"}` → 作为关键字参数传递
   - **列表参数**：`[arg1, arg2]` → 作为位置参数传递
   - **字符串参数**：支持 `eval()` 解析为 Python 对象
   - **智能参数判断**：根据函数签名自动选择传递方式

4. **异步函数支持**：
   ```python
   # 检测异步函数并等待结果
   if inspect.iscoroutine(result_value):
       result_value = await result_value
   ```

### 3.5 变量上下文管理
1. **任务变量设置**：
   - `SET_TASK_VARIABLES` 模式：将函数返回值设置为任务变量
   - 支持字典结果的键值对分别设置为变量
   - 非字典结果设置为 `scriptResult` 变量

2. **变量持久化**：
   - 任务变量保存到数据库的 `taskrecord` 表
   - JSON 格式存储变量数据
   - 支持跨组件的变量传递和引用

3. **上下文传递**：
   - `TaskContext` 管理任务执行过程中的变量状态
   - 变量在任务流程中的传递和更新
   - 变量作用域和生命周期管理

---

## 四、统一 VWED 对象模块系统（Python 版）

### 4.1 设计理念
所有功能都通过统一的 `VWED.内置对象模块.具体内置对象(参数1，参数2)` 形式调用，包括：
- 内置函数注册
- 接口注册  
- 事件监听
- 定时任务
- 设备通信
- 系统操作

### 4.2 核心对象模块结构

| 对象模块 | 说明 | 示例 |
|----------|------|------|
| `VWED.api` | 接口注册模块 | `VWED.api.register_route()` |
| `VWED.function` | 自定义函数注册模块 | `VWED.function.register()` |
| `VWED.event` | 事件监听模块 | `VWED.event.listen()` |
| `VWED.task` | 任务管理模块 | `VWED.task.create()` |
| `VWED.device` | 设备通信模块 | `VWED.device.modbus()` |
| `VWED.robot` | 机器人控制模块 | `VWED.robot.get_status()` |
| `VWED.timer` | 定时任务模块 | `VWED.timer.schedule()` |
| `VWED.log` | 日志输出模块 | `VWED.log.info()` |
| `VWED.http` | HTTP 请求模块 | `VWED.http.post()` |
| `VWED.storage` | 库位管理模块 | `VWED.storage.get_location()` |
| `VWED.mail` | 邮件发送模块 | `VWED.mail.send()` |

### 4.3 统一对象模块 API

#### 4.3.1 接口注册模块 (VWED.api)
```python
# 注册 HTTP 接口
VWED.api.register_route(path="/script-api/move", method="POST", handler=move_handler)

# 注册 WebSocket 接口  
VWED.api.register_websocket(path="/ws/script", handler=ws_handler)

# 注册 TCP 接口
VWED.api.register_tcp(port=9999, handler=tcp_handler)
```

#### 4.3.2 自定义函数注册模块 (VWED.function)
```python
# 注册自定义函数
VWED.function.register(name="calculate_distance", handler=calc_distance_func, 
                      description="计算两点间距离", params=["point1", "point2"])

# 注册业务逻辑函数
VWED.function.register(name="warehouse_allocation", handler=allocate_warehouse,
                      description="仓库分配逻辑", params=["goods_type", "quantity"])

# 注册数据处理函数
VWED.function.register(name="format_sensor_data", handler=format_data_func,
                      description="格式化传感器数据", params=["raw_data"])

# 获取已注册函数列表
functions = VWED.function.list_all()

# 调用注册的函数（内部使用，通常在script.py中调用）
result = VWED.function.call(name="calculate_distance", args={"point1": [0,0], "point2": [3,4]})
```

#### 4.3.3 事件监听模块 (VWED.event)
```python
# 监听系统事件
VWED.event.listen(event_name="task_done", handler=on_task_done)

# 监听自定义事件
VWED.event.listen(event_name="custom_event", handler=on_custom)

# 触发事件
VWED.event.emit(event_name="custom_event", data={"msg": "test"})
```

#### 4.3.3 任务管理模块 (VWED.task)
```python
# 创建任务
VWED.task.create(task_label="move", input_params={"to": "A001"})

# 获取任务状态
VWED.task.get_status(task_id=123)

# 停止任务
VWED.task.stop(task_id=123)
```

#### 4.3.4 设备通信模块 (VWED.device)
```python
# Modbus 通信
VWED.device.modbus(ip="192.168.1.100", port=502, register=100, value=1)

# OPC 通信
VWED.device.opc(server_url="opc.tcp://127.0.0.1:4840", node_id="ns=2;i=2")

# PLC 通信
VWED.device.plc(ip="192.168.1.50", rack=0, slot=1, db_number=1, start=0, size=4)
```

#### 4.3.5 机器人控制模块 (VWED.robot)
```python
# 获取机器人状态
VWED.robot.get_status(robot_id="robot_001")

# 获取所有机器人
VWED.robot.get_all()

# 调度机器人
VWED.robot.dispatch(robot_id="robot_001", task_id=123)
```

#### 4.3.6 定时任务模块 (VWED.timer)
```python
# 定时执行
VWED.timer.schedule(interval=5, handler=timer_handler, repeat=True)

# 延时执行
VWED.timer.delay(seconds=10, handler=delay_handler)

# 取消定时任务
VWED.timer.cancel(timer_id="timer_001")
```

#### 4.3.7 日志模块 (VWED.log)
```python
# 各种日志级别
VWED.log.info(tag="script", message="脚本启动成功")
VWED.log.debug(tag="debug", message="调试信息")
VWED.log.warning(tag="warn", message="警告信息")
VWED.log.error(tag="error", message="错误信息")
```

#### 4.3.8 HTTP 请求模块 (VWED.http)
```python
# GET 请求
VWED.http.get(url="http://api.example.com/data", headers={})

# POST 请求
VWED.http.post(url="http://api.example.com/submit", json={"key": "value"})

# PUT/DELETE 请求
VWED.http.put(url="http://api.example.com/update", json={})
VWED.http.delete(url="http://api.example.com/delete")
```

#### 4.3.9 库位管理模块 (VWED.storage)
```python
# 获取库位信息
VWED.storage.get_location(location_code="A001")

# 锁定库位
VWED.storage.lock_location(location_code="A001", task_id=123)

# 释放库位
VWED.storage.unlock_location(location_code="A001")
```

#### 4.3.10 邮件模块 (VWED.mail)
```python
# 发送邮件
VWED.mail.send(to="admin@example.com", subject="告警", body="系统异常")

# 发送带附件邮件
VWED.mail.send_with_attachment(to="admin@example.com", subject="报告", 
                              body="请查看附件", attachment_path="/tmp/report.pdf")
```

### 4.4 注册清理机制
```python
# 脚本停止时自动清理所有注册项
def script_stop_cleanup():
    """脚本停止时调用，清理所有注册的内容"""
    VWED.api.clear_all()      # 清理接口注册
    VWED.event.clear_all()    # 清理事件监听
    VWED.timer.clear_all()    # 清理定时任务
    VWED.log.info("script", "脚本注册项已全部清理")
```

---

## 五、调试能力（Python 版）

- 支持 `remote_pdb.set_trace()`
- 启动脚本时自动开启调试端口（如 5555）
- 前端弹出调试地址：`telnet 127.0.0.1 5555`
- 支持断点、变量查看、单步执行

---

## 六、脚本目录结构（Python 版）

```
data/VWED/python_script/
├── common/                    # 通用工具目录
│   ├── utils.py               # 通用工具脚本
│   └── config.py              # 配置管理脚本
├── warehouse/                 # 仓库管理目录
│   ├── controller.py          # 仓库控制脚本
│   └── allocation.py          # 库位分配脚本
├── robot/                     # 机器人控制目录
│   ├── dispatcher.py          # 机器人调度脚本
│   └── monitor.py             # 机器人监控脚本
├── sensors/                   # 传感器处理目录
│   ├── data_processor.py      # 数据处理脚本
│   └── alarm_handler.py       # 告警处理脚本
└── boot.py                    # 主启动脚本
```

### 创建规则
- **创建目录**：`create_script(directory="新目录名")`
- **创建文件**：`create_script(directory="所属目录", filename="脚本名.py")`

注意：所有用户共享同一套脚本文件，任何用户的修改都会影响所有用户。

```

## 八、共享脚本架构

### 8.1 单用户共享模式架构

```text
┌─────────────────────────────────────────────────────────────┐
│                    全局注册中心                              │
│  - 统一API路由        - 统一函数注册      - 统一事件监听      │
│  - 所有用户共享       - 脚本版本控制      - 最后修改覆盖      │
└─────────────────────┬───────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────────┐
│                  统一脚本存储                                │
│  data/VWED/python_script/                                  │
│  ├── boot.py                                               │
│  ├── task_manager.py                                       │  
│  ├── sensor_processor.py                                   │
│  └── utils.py                                              │
└─────────────────────────────────────────────────────────────┘
```

### 8.2 多脚本服务引擎

```python
class VWEDGlobalRegistry:
    """VWED 全局注册中心（支持多脚本独立服务）"""
    def __init__(self):
        self.api_routes = {}            # HTTP接口注册: {route_key: route_info}
        self.websocket_routes = {}      # WebSocket接口注册
        self.tcp_servers = {}           # TCP服务注册
        self.event_listeners = {}       # 事件监听器注册: {event_name: [listener_info]}
        self.timers = {}               # 定时任务注册
        self.registered_functions = {}  # 自定义函数注册: {function_name: function_info}
        self.active_scripts = {}        # 活跃脚本: {script_id: script_info}
        self.current_script_id = None   # 当前执行的脚本ID
    
    def register_api_route(self, path: str, method: str, handler, script_id: str = None):
        """注册API接口（按脚本隔离）"""
        script_id = script_id or self.current_script_id
        route_key = f"{method}:{path}"
        self.api_routes[route_key] = {
            "path": path,
            "method": method,
            "handler": handler,
            "script_id": script_id,
            "registered_at": datetime.now().isoformat()
        }
        
    def register_function(self, name: str, handler, script_id: str = None, 
                         description: str = "", params: list = None):
        """注册自定义函数（按脚本隔离）"""
        script_id = script_id or self.current_script_id
        self.registered_functions[name] = {
            "handler": handler,
            "script_id": script_id,
            "description": description,
            "params": params or [],
            "registered_at": datetime.now().isoformat()
        }
    
    def register_event_listener(self, event_name: str, handler, script_id: str = None, priority: int = 1):
        """注册事件监听器（按脚本隔离）"""
        script_id = script_id or self.current_script_id
        if event_name not in self.event_listeners:
            self.event_listeners[event_name] = []
        
        self.event_listeners[event_name].append({
            "handler": handler,
            "script_id": script_id,
            "priority": priority,
            "registered_at": datetime.now().isoformat()
        })
    
    def get_api_route(self, path: str, method: str = "GET"):
        """获取API路由"""
        route_key = f"{method}:{path}"
        return self.api_routes.get(route_key)
    
    def get_registered_function(self, name: str):
        """获取注册的函数"""
        func_info = self.registered_functions.get(name)
        return func_info["handler"] if func_info else None
    
    def clear_script_registrations(self, script_id: str):
        """清理特定脚本的所有注册项"""
        # 清理API路由
        api_routes_to_remove = [k for k, v in self.api_routes.items() if v.get("script_id") == script_id]
        for route_key in api_routes_to_remove:
            del self.api_routes[route_key]
        
        # 清理函数注册
        functions_to_remove = [k for k, v in self.registered_functions.items() if v.get("script_id") == script_id]
        for func_name in functions_to_remove:
            del self.registered_functions[func_name]
        
        # 清理事件监听器
        for event_name in self.event_listeners:
            self.event_listeners[event_name] = [
                listener for listener in self.event_listeners[event_name] 
                if listener.get("script_id") != script_id
            ]
        
        # 清理定时任务
        timers_to_remove = [k for k, v in self.timers.items() if v.get("script_id") == script_id]
        for timer_id in timers_to_remove:
            del self.timers[timer_id]
        
        # 从活跃脚本列表移除
        if script_id in self.active_scripts:
            del self.active_scripts[script_id]
    
    def clear_all(self):
        """清理所有注册项"""
        self.api_routes.clear()
        self.websocket_routes.clear()
        self.tcp_servers.clear()
        self.event_listeners.clear()
        self.timers.clear()
        self.registered_functions.clear()
        self.active_scripts.clear()
    
    def set_current_script(self, script_id: str):
        """设置当前执行的脚本ID"""
        self.current_script_id = script_id
    
    def add_active_script(self, script_id: str, script_path: str):
        """添加活跃脚本记录"""
        self.active_scripts[script_id] = {
            "script_path": script_path,
            "started_at": datetime.now().isoformat(),
            "status": "running"
        }
    
    def get_script_registrations(self, script_id: str):
        """获取特定脚本的注册统计"""
        api_count = len([v for v in self.api_routes.values() if v.get("script_id") == script_id])
        function_count = len([v for v in self.registered_functions.values() if v.get("script_id") == script_id])
        event_count = sum([
            len([l for l in listeners if l.get("script_id") == script_id]) 
            for listeners in self.event_listeners.values()
        ])
        timer_count = len([v for v in self.timers.values() if v.get("script_id") == script_id])
        
        return {
            "apis": api_count,
            "functions": function_count,
            "events": event_count,
            "timers": timer_count
        }

# 全局注册实例
_global_registry = VWEDGlobalRegistry()


class MultiScriptEngine:
    """多脚本服务引擎"""
    
    def __init__(self):
        self.script_store = GlobalScriptStore()
        self.running_scripts = {}  # {script_id: script_info}
    
    def start_script_service(self, script_path: str):
        """启动脚本服务 - 支持多脚本并发"""
        try:
            # 1. 验证脚本文件存在
            full_script_path = f"data/VWED/python_script/{script_path}"
            if not os.path.exists(full_script_path):
                return {"success": False, "error": f"脚本文件不存在: {script_path}"}
            
            # 2. 生成唯一脚本ID
            script_id = f"{script_path}_{int(time.time() * 1000)}"
            
            # 3. 检查是否已有同名脚本在运行
            existing_script = self.find_running_script_by_path(script_path)
            if existing_script:
                return {
                    "success": False, 
                    "error": f"脚本 {script_path} 已在运行中，script_id: {existing_script['script_id']}"
                }
            
            # 4. 设置全局注册中心上下文
            _global_registry.set_current_script(script_id)
            _global_registry.add_active_script(script_id, script_path)
            
            # 5. 注入VWED模块
            script_globals = {
                'VWED': VWEDObjectFactory(),
                '__name__': '__main__'
            }
            
            # 6. 执行脚本
            with open(full_script_path, 'r', encoding='utf-8') as f:
                script_code = f.read()
            
            exec(script_code, script_globals)
            
            # 7. 调用boot函数注册服务
            if 'boot' in script_globals:
                script_globals['boot']()
            else:
                _global_registry.clear_script_registrations(script_id)
                return {"success": False, "error": "脚本缺少boot()入口函数"}
            
            # 8. 记录运行中的脚本
            self.running_scripts[script_id] = {
                "script_path": script_path,
                "started_at": datetime.now().isoformat(),
                "status": "running"
            }
            
            # 9. 保存脚本到数据库
            self.script_store.save_script(script_path, script_code)
            
            # 10. 获取注册统计
            registrations = _global_registry.get_script_registrations(script_id)
            
            return {
                "success": True, 
                "script_id": script_id,
                "message": f"脚本服务启动成功，已注册 {registrations['apis']} 个接口，{registrations['functions']} 个函数，{registrations['events']} 个事件监听器"
            }
            
        except Exception as e:
            if script_id:
                _global_registry.clear_script_registrations(script_id)
            return {"success": False, "error": f"脚本服务启动失败: {str(e)}"}
    
    def stop_script_service(self, script_id: str):
        """停止特定脚本服务 - 关键功能"""
        try:
            if script_id not in self.running_scripts:
                return {"success": False, "error": f"脚本 {script_id} 未运行"}
            
            # 获取注册统计
            registrations = _global_registry.get_script_registrations(script_id)
            
            # 清理该脚本的所有注册项 - 这是关键步骤
            _global_registry.clear_script_registrations(script_id)
            
            # 从运行列表移除
            script_info = self.running_scripts.pop(script_id)
            
            return {
                "success": True,
                "message": f"脚本服务已停止 ({script_info['script_path']})，已清理 {registrations['apis']} 个接口、{registrations['functions']} 个函数、{registrations['events']} 个事件监听器"
            }
            
        except Exception as e:
            return {"success": False, "error": f"停止脚本服务失败: {str(e)}"}
    
    def get_running_scripts(self):
        """获取所有运行中的脚本"""
        result = []
        for script_id, script_info in self.running_scripts.items():
            registrations = _global_registry.get_script_registrations(script_id)
            result.append({
                "script_id": script_id,
                "script_path": script_info["script_path"],
                "started_at": script_info["started_at"],
                "status": script_info["status"],
                "registrations": registrations
            })
        return result
    
    def find_running_script_by_path(self, script_path: str):
        """根据脚本路径查找运行中的脚本"""
        for script_id, script_info in self.running_scripts.items():
            if script_info["script_path"] == script_path:
                return {"script_id": script_id, **script_info}
        return None
    
    def create_script(self, directory: str, filename: str = None):
        """创建脚本或目录"""
        try:
            base_path = "data/VWED/python_script"
            
            if filename is None:
                # 创建目录
                dir_path = os.path.join(base_path, directory)
                os.makedirs(dir_path, exist_ok=True)
                return {"success": True, "message": f"目录创建成功: {directory}"}
            else:
                # 创建文件
                dir_path = os.path.join(base_path, directory)
                os.makedirs(dir_path, exist_ok=True)
                
                file_path = os.path.join(dir_path, filename)
                if not filename.endswith('.py'):
                    filename += '.py'
                    file_path += '.py'
                
                # 创建默认脚本模板
                default_content = f'''def boot():
    """脚本入口函数"""
    VWED.log.info("script", "脚本 {filename} 启动成功")
    
    # 在这里注册您的接口、函数和事件监听器
    # 示例：
    # VWED.api.register_route("/test", "GET", test_handler)
    # VWED.function.register("test_func", test_function)
    
    pass

def test_handler(request):
    """示例API处理函数"""
    return {{"message": "Hello from script API", "script": "{filename}"}}

def test_function(args):
    """示例自定义函数"""
    return {{"result": "Function executed", "args": args}}
'''
                
                with open(file_path, 'w', encoding='utf-8') as f:
                    f.write(default_content)
                
                return {"success": True, "message": f"文件创建成功: {directory}/{filename}"}
                
        except Exception as e:
            return {"success": False, "error": f"创建失败: {str(e)}"}


class GlobalScriptStore:
    """全局脚本存储管理"""
    
    def save_script(self, script_name: str, content: str):
        """保存脚本到文件系统和数据库"""
        # 1. 保存到文件系统
        script_path = f"data/VWED/python_script/{script_name}"
        with open(script_path, 'w', encoding='utf-8') as f:
            f.write(content)
        
        # 2. 保存到数据库（版本控制，但只保留最新版本）
        self._save_to_database(script_name, content)
    
    def _save_to_database(self, script_name: str, content: str):
        """保存到数据库（覆盖旧版本）"""
        # 这里实现数据库保存逻辑
        # 系统只保留最后一个版本，不考虑多版本管理
        pass
```

### 8.3 多脚本服务访问控制机制

```python
# FastAPI 多脚本路由处理器
@app.api_route("/script-api/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def multi_script_route(path: str, request: Request):
    """多脚本路由处理器"""
    
    # 1. 检查是否有任何脚本在运行
    if not multi_script_engine.running_scripts:
        return {"error": "没有脚本服务在运行，请先启动脚本服务"}
    
    # 2. 从全局注册中心获取路由
    api_path = "/" + path
    route_info = _global_registry.get_api_route(api_path, request.method)
    
    if route_info:
        # 3. 检查该路由对应的脚本是否仍在运行
        script_id = route_info.get("script_id")
        if script_id not in multi_script_engine.running_scripts:
            return {"error": f"路由 {api_path} 对应的脚本服务已停止"}
        
        try:
            handler = route_info["handler"]
            
            # 构建请求数据
            request_data = {
                "method": request.method,
                "path": path,
                "script_id": script_id,  # 添加脚本ID信息
                "headers": dict(request.headers),
                "query_params": dict(request.query_params)
            }
            
            if request.method in ["POST", "PUT", "PATCH"]:
                try:
                    request_data["json"] = await request.json()
                except:
                    request_data["body"] = await request.body()
            
            # 调用处理函数
            if asyncio.iscoroutinefunction(handler):
                result = await handler(request_data)
            else:
                result = handler(request_data)
                
            return result
            
        except Exception as e:
            return {"error": f"脚本接口执行失败: {str(e)}"}
    else:
        return {"error": f"接口 {request.method} {api_path} 未注册"}

# 脚本管理API
@app.get("/script-engine/running")
async def get_running_scripts():
    """获取所有运行中的脚本"""
    return multi_script_engine.get_running_scripts()

@app.post("/script-engine/start")
async def start_script_service(request: dict):
    """启动脚本服务"""
    script_path = request.get("script_path")
    if not script_path:
        return {"success": False, "error": "缺少script_path参数"}
    
    return multi_script_engine.start_script_service(script_path)

@app.post("/script-engine/stop")
async def stop_script_service(request: dict):
    """停止脚本服务"""
    script_id = request.get("script_id")
    if not script_id:
        return {"success": False, "error": "缺少script_id参数"}
    
    return multi_script_engine.stop_script_service(script_id)
```

## 九、关键功能说明

### 9.1 多脚本服务启动的重要性

多脚本服务启动是整个脚本引擎的核心功能：

1. **独立注册**：每个脚本启动时生成唯一的script_id，其接口、函数、事件带有该ID标记注册到全局注册中心
2. **并发运行**：支持多个脚本同时运行服务，每个脚本的资源相互独立
3. **外部访问控制**：只有运行中的脚本资源才能被外部系统访问
4. **防重复启动**：同一脚本路径不能重复启动，确保资源的唯一性

### 9.2 按脚本ID清理的关键作用

停止特定脚本服务是精确资源清理的关键：

1. **精确清理**：根据script_id只清理该脚本注册的接口、函数、事件和定时任务
2. **其他服务不受影响**：停止某个脚本不会影响其他正在运行的脚本服务
3. **访问控制**：该脚本的资源立即不可访问，但其他脚本的资源继续正常服务
4. **状态一致性**：确保注册中心的资源状态与实际运行脚本保持同步

### 9.3 多脚本资源隔离机制

每个脚本的资源完全独立：

1. **接口隔离**：不同脚本可以注册相同路径的接口，通过script_id区分
2. **函数隔离**：不同脚本的同名函数通过script_id管理，后注册的会覆盖前面的
3. **事件隔离**：事件监听器按script_id分组，停止脚本时精确清理对应监听器
4. **定时任务隔离**：定时任务带有script_id标记，支持按脚本独立管理

### 9.4 目录和文件创建机制

支持灵活的脚本组织结构：

1. **目录创建**：`create_script(directory="新目录名")`
2. **文件创建**：`create_script(directory="所属目录", filename="脚本名.py")`
3. **模板生成**：新创建的脚本文件包含默认的boot()函数和示例代码

## 十、多脚本服务示例

### 10.1 并发运行示例

```python
# 脚本A: warehouse/controller.py
def boot():
    VWED.api.register_route("/warehouse/status", "GET", get_warehouse_status)
    VWED.function.register("allocate_storage", allocate_storage_func)

# 脚本B: robot/dispatcher.py  
def boot():
    VWED.api.register_route("/robot/dispatch", "POST", dispatch_robot)
    VWED.function.register("get_robot_status", get_robot_status_func)

# 两个脚本可以同时运行，提供不同的服务
# A脚本: script_id = "warehouse/controller.py_1640000000000"
# B脚本: script_id = "robot/dispatcher.py_1640000001000"
```

### 10.2 脚本服务管理操作

```python
# 启动仓库控制脚本
POST /script-engine/start
{
    "script_path": "warehouse/controller.py"
}

# 启动机器人调度脚本
POST /script-engine/start  
{
    "script_path": "robot/dispatcher.py"
}

# 查看运行中的脚本
GET /script-engine/running
# 返回:
[
    {
        "script_id": "warehouse/controller.py_1640000000000",
        "script_path": "warehouse/controller.py", 
        "started_at": "2024-01-01T10:00:00",
        "status": "running",
        "registrations": {"apis": 1, "functions": 1, "events": 0, "timers": 0}
    },
    {
        "script_id": "robot/dispatcher.py_1640000001000",
        "script_path": "robot/dispatcher.py",
        "started_at": "2024-01-01T10:01:00", 
        "status": "running",
        "registrations": {"apis": 1, "functions": 1, "events": 0, "timers": 0}
    }
]

# 停止仓库控制脚本（机器人脚本继续运行）
POST /script-engine/stop
{
    "script_id": "warehouse/controller.py_1640000000000"
}
```

---

## 十一、框架扩展指南

### 11.1 新增内置模块扩展步骤

#### 步骤1：创建模块接口定义
```python
# scripts/modules/ai_module.py
from abc import ABC, abstractmethod

class BaseVWEDAIModule(ABC):
    """AI模块基础接口"""
    
    @abstractmethod
    def predict(self, data, model_name: str = "default"):
        """预测接口"""
        pass
    
    @abstractmethod
    def train(self, training_data, model_config: dict):
        """训练接口"""
        pass
    
    @abstractmethod
    def get_available_models(self):
        """获取可用模型列表"""
        pass

class VWEDAIModule(BaseVWEDAIModule):
    """AI模块实现"""
    
    def __init__(self, registry):
        self.registry = registry
        self.models = {}
    
    def predict(self, data, model_name: str = "default"):
        """执行预测"""
        if model_name not in self.models:
            raise ValueError(f"模型 {model_name} 未找到")
        
        # 实现预测逻辑
        return {"prediction": "result", "confidence": 0.95}
    
    def train(self, training_data, model_config: dict):
        """训练模型"""
        model_name = model_config.get("name", "trained_model")
        # 实现训练逻辑
        self.models[model_name] = {"status": "trained", "accuracy": 0.98}
        return {"success": True, "model_name": model_name}
    
    def get_available_models(self):
        """获取可用模型"""
        return list(self.models.keys())
```

#### 步骤2：注册到VWED对象工厂
```python
# scripts/services/script_builtin_functions.py
class VWEDObjectFactory:
    """VWED对象工厂 - 支持动态模块扩展"""
    
    def __init__(self):
        self._modules = {}
        self._load_core_modules()
        self._load_extended_modules()  # 加载扩展模块
    
    def _load_extended_modules(self):
        """加载扩展模块"""
        # AI模块
        from scripts.modules.ai_module import VWEDAIModule
        self._modules['ai'] = VWEDAIModule(self.registry)
        
        # 视觉模块
        from scripts.modules.vision_module import VWEDVisionModule
        self._modules['vision'] = VWEDVisionModule(self.registry)
        
        # 算法模块
        from scripts.modules.algorithm_module import VWEDAlgorithmModule
        self._modules['algorithm'] = VWEDAlgorithmModule(self.registry)
    
    def register_custom_module(self, module_name: str, module_instance):
        """注册自定义模块"""
        self._modules[module_name] = module_instance
    
    def __getattr__(self, name):
        if name in self._modules:
            return self._modules[name]
        raise AttributeError(f"'VWED' object has no attribute '{name}'")
```

#### 步骤3：脚本中使用新模块
```python
# 脚本示例：ai_processor.py
def boot():
    """AI处理脚本入口"""
    VWED.log.info("ai", "AI处理脚本启动")
    
    # 注册AI预测接口
    VWED.api.register_route("/ai/predict", "POST", ai_predict_handler)
    
    # 注册AI训练接口
    VWED.api.register_route("/ai/train", "POST", ai_train_handler)
    
    # 注册模型管理函数
    VWED.function.register("get_model_info", get_model_info_func)

def ai_predict_handler(request):
    """AI预测处理器"""
    data = request.get("json", {})
    model_name = data.get("model", "default")
    input_data = data.get("data")
    
    # 使用VWED.ai模块
    result = VWED.ai.predict(input_data, model_name)
    return {"success": True, "result": result}

def ai_train_handler(request):
    """AI训练处理器"""
    data = request.get("json", {})
    training_data = data.get("training_data")
    model_config = data.get("config", {})
    
    # 使用VWED.ai模块
    result = VWED.ai.train(training_data, model_config)
    return {"success": True, "result": result}

def get_model_info_func(args):
    """获取模型信息"""
    models = VWED.ai.get_available_models()
    return {"available_models": models}
```

### 11.2 协议扩展指南

#### 扩展gRPC协议支持
```python
# scripts/interfaces/grpc_interface.py
import grpc
from concurrent import futures

class ScriptGRPCAPI:
    """脚本gRPC接口"""
    
    def __init__(self, registry):
        self.registry = registry
        self.server = None
    
    def register_grpc_service(self, service_name: str, service_impl, port: int = 50051):
        """注册gRPC服务"""
        script_id = self.registry.current_script_id
        
        if not self.server:
            self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
        
        # 注册服务到gRPC服务器
        # add_ServicenameServicer_to_server(service_impl, self.server)
        
        # 记录到注册中心
        self.registry.grpc_services[f"{service_name}:{port}"] = {
            "service_impl": service_impl,
            "port": port,
            "script_id": script_id,
            "registered_at": datetime.now().isoformat()
        }
        
        self.server.add_insecure_port(f'[::]:{port}')
        self.server.start()
        
        return {"success": True, "service": service_name, "port": port}
```

### 11.3 安全策略扩展

#### 企业级安全扩展
```python
# scripts/engine/advanced_security.py
class EnterpriseSecuritySandbox(ScriptSecuritySandbox):
    """企业级安全沙箱"""
    
    def __init__(self):
        super().__init__()
        self.enterprise_policies = {
            "ip_whitelist": [],
            "domain_blacklist": [],
            "api_rate_limit": 100,  # 每分钟请求数
            "resource_quota": {
                "max_memory": "512MB",
                "max_cpu_percent": 30,
                "max_execution_time": 300  # 秒
            }
        }
    
    def validate_network_access(self, target_host: str, target_port: int) -> bool:
        """验证网络访问权限"""
        # 检查IP白名单
        if self.enterprise_policies["ip_whitelist"]:
            if target_host not in self.enterprise_policies["ip_whitelist"]:
                return False
        
        # 检查域名黑名单
        for blocked_domain in self.enterprise_policies["domain_blacklist"]:
            if blocked_domain in target_host:
                return False
        
        return super().validate_network_access(target_host, target_port)
    
    def check_api_rate_limit(self, script_id: str) -> bool:
        """检查API调用频率限制"""
        # 实现频率限制逻辑
        return True
    
    def monitor_resource_usage(self, script_id: str) -> dict:
        """监控资源使用情况"""
        # 实现资源监控逻辑
        return {
            "memory_usage": "128MB",
            "cpu_usage": "15%",
            "execution_time": 120
        }
```

### 11.4 日志目标扩展

#### 多目标日志系统
```python
# scripts/services/advanced_logger.py
class MultiTargetLogger(ScriptWebSocketLoggerService):
    """多目标日志服务"""
    
    def __init__(self):
        super().__init__()
        self.log_targets = {
            "file": FileLogTarget(),
            "database": DatabaseLogTarget(),
            "elasticsearch": ElasticsearchLogTarget(),
            "webhook": WebhookLogTarget()
        }
    
    def log(self, level: str, tag: str, message: str, script_id: str = None):
        """多目标日志输出"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "level": level,
            "tag": tag,
            "message": message,
            "script_id": script_id
        }
        
        # 发送到所有启用的日志目标
        for target_name, target in self.log_targets.items():
            if target.is_enabled():
                target.write_log(log_entry)
        
        # 继续原有的WebSocket推送
        super().log(level, tag, message, script_id)

class DatabaseLogTarget:
    """数据库日志目标"""
    
    def write_log(self, log_entry: dict):
        # 写入数据库
        pass
    
    def is_enabled(self) -> bool:
        return True

class ElasticsearchLogTarget:
    """Elasticsearch日志目标"""
    
    def write_log(self, log_entry: dict):
        # 写入Elasticsearch
        pass
    
    def is_enabled(self) -> bool:
        return False  # 可配置
```

---

## 十二、当前实现状态与技术细节

### 12.1 script.py 处理器实现分析

基于 `services/execution/handlers/script.py` 的当前实现，脚本引擎具备以下技术特性：

#### 12.1.1 双模式处理器设计
```python
# 两种脚本处理模式
@register_handler(ScriptBlockName.SET_TASK_VARIABLES)  # 变量设置模式
@register_handler(ScriptBlockName.SCRIPT)              # 脚本执行模式
```

#### 12.1.2 动态模块加载机制
```python
# 固定脚本路径（当前实现）
script_file = "scripts/user_save/test1.py"

# 动态加载实现
spec = importlib.util.spec_from_file_location("user_script", script_file)
module = importlib.util.module_from_spec(spec)
sys.modules["user_script"] = module
spec.loader.exec_module(module)
```

#### 12.1.3 智能参数传递算法
```python
# 参数类型智能识别
if isinstance(function_args, dict):
    kwargs = function_args
elif isinstance(function_args, list):
    args = function_args
elif isinstance(function_args, str):
    function_args = eval(function_args)

# 函数签名检测
sig = inspect.signature(func)
params = list(sig.parameters.keys())
if len(params) == 1 and isinstance(function_args, dict) and not args:
    result_value = func(function_args)  # 单参数字典传递
else:
    result_value = func(*args, **kwargs)  # 标准参数传递
```

#### 12.1.4 异步执行检测机制
```python
# 协程检测和等待
if inspect.iscoroutine(result_value):
    import asyncio
    result_value = await result_value
```

#### 12.1.5 变量上下文管理
```python
# SET_TASK_VARIABLES 模式：多变量设置
if isinstance(result_value, dict):
    for key, value in result_value.items():
        context.set_variable(key, value)
else:
    context.set_variable("scriptResult", result_value)

# SCRIPT 模式：单一结果变量
context.set_variable("scriptResult", result_value)
```

#### 12.1.6 数据库持久化机制
```python
# 变量保存到数据库
variables_json = json.dumps(context.variables, ensure_ascii=False)
stmt = update(VWEDTaskRecord).where(VWEDTaskRecord.id == task_record_id).values(
    variables=variables_json
)
```

### 12.2 当前架构的优势与局限

#### 优势：
1. **简单高效**：固定脚本路径，减少配置复杂度
2. **类型智能**：自动识别参数类型和传递方式
3. **异步支持**：完整的协程检测和执行机制
4. **双模式设计**：灵活的变量设置和脚本执行模式
5. **深度集成**：与 VWED 任务系统无缝集成

#### 当前局限：
1. **固定路径**：脚本路径硬编码为 `scripts/user_save/test1.py`
2. **单文件限制**：无法动态选择不同的脚本文件
3. **缺少文件管理**：没有脚本文件的创建、编辑、删除功能
4. **无日志推送**：缺少 WebSocket 实时日志推送功能
5. **无生命周期管理**：缺少脚本启动、停止、状态监控

### 12.3 扩展改进建议

#### 12.3.1 动态脚本路径支持
```python
# 建议改进：支持动态脚本路径
script_file = input_params.get("scriptPath", "scripts/user_save/test1.py")
```

#### 12.3.2 文件管理功能扩展
- 添加脚本文件 CRUD 操作 API
- 支持目录结构管理
- 实现在线编辑器集成

#### 12.3.3 实时日志推送集成
- 集成 WebSocket 日志推送服务
- 脚本执行过程的实时监控
- 执行状态和结果的实时反馈

#### 12.3.4 生命周期管理扩展
- 脚本进程的启动、停止控制
- 运行状态监控和管理
- 资源使用情况追踪

---

## 十三、最佳实践与注意事项

### 13.1 脚本开发最佳实践

#### 推荐的脚本结构
```python
# 标准脚本模板
import asyncio
from typing import Dict, Any

def boot():
    """脚本入口函数 - 必须实现"""
    VWED.log.info("script", "脚本启动中...")
    
    # 1. 注册API接口
    register_apis()
    
    # 2. 注册自定义函数
    register_functions()
    
    # 3. 注册事件监听器
    register_events()
    
    # 4. 启动定时任务
    start_timers()
    
    VWED.log.info("script", "脚本启动完成")

def register_apis():
    """注册API接口"""
    VWED.api.register_route("/health", "GET", health_check)
    VWED.api.register_route("/process", "POST", process_data)

def register_functions():
    """注册自定义函数"""
    VWED.function.register("calculate", calculate_func, "数据计算函数")
    VWED.function.register("validate", validate_func, "数据验证函数")

def register_events():
    """注册事件监听器"""
    VWED.event.listen("data_received", on_data_received)
    VWED.event.listen("task_completed", on_task_completed)

def start_timers():
    """启动定时任务"""
    VWED.timer.schedule(interval=60, handler=periodic_cleanup, repeat=True)

# API处理函数
def health_check(request: Dict[str, Any]) -> Dict[str, Any]:
    """健康检查接口"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

async def process_data(request: Dict[str, Any]) -> Dict[str, Any]:
    """数据处理接口（支持异步）"""
    data = request.get("json", {})
    
    try:
        # 数据验证
        if not validate_input(data):
            return {"error": "数据验证失败"}
        
        # 数据处理
        result = await process_business_logic(data)
        
        # 触发事件
        VWED.event.emit("data_processed", {"result": result})
        
        return {"success": True, "result": result}
        
    except Exception as e:
        VWED.log.error("script", f"数据处理失败: {str(e)}")
        return {"error": str(e)}

# 自定义函数
def calculate_func(args: Dict[str, Any]) -> Dict[str, Any]:
    """计算函数"""
    # 实现计算逻辑
    return {"result": "calculated_value"}

# 事件处理函数
def on_data_received(event_data: Dict[str, Any]):
    """数据接收事件处理"""
    VWED.log.info("event", f"收到数据: {event_data}")

# 定时任务函数
def periodic_cleanup():
    """定期清理任务"""
    VWED.log.info("timer", "执行定期清理")
```

### 13.2 性能优化建议

1. **异步编程**：优先使用异步函数处理IO密集型操作
2. **资源管理**：及时释放不需要的资源，避免内存泄漏
3. **缓存策略**：合理使用缓存减少重复计算
4. **批量处理**：对于大量数据处理，采用批量处理方式
5. **连接池**：使用连接池管理数据库和外部服务连接

### 13.3 安全注意事项

1. **输入验证**：严格验证所有外部输入
2. **权限控制**：实施最小权限原则
3. **敏感数据**：不在脚本中硬编码敏感信息
4. **网络访问**：只访问必要的外部资源
5. **日志安全**：避免在日志中记录敏感信息

### 13.4 故障排查指南

#### 常见问题及解决方案

| 问题类型 | 症状 | 可能原因 | 解决方案 |
|---------|------|----------|----------|
| 脚本启动失败 | boot()函数报错 | 语法错误、依赖缺失 | 检查脚本语法，确认依赖安装 |
| API无法访问 | 404错误 | 路由未注册、脚本已停止 | 检查脚本运行状态和API注册 |
| 内存泄漏 | 内存使用持续增长 | 循环引用、资源未释放 | 使用内存分析工具，检查资源释放 |
| 性能下降 | 响应时间变长 | 阻塞操作、资源竞争 | 优化算法，使用异步编程 |
| 并发问题 | 数据不一致 | 共享状态竞争 | 使用锁机制或消息队列 |

---

## 十四、总结

本文档详细描述了VWED高扩展性Python脚本执行引擎的完整架构设计。该框架具有以下核心优势：

### 14.1 核心优势

1. **高度可扩展**：模块化设计，支持任意功能模块扩展
2. **多脚本并发**：独立的脚本ID隔离机制，支持多脚本同时运行
3. **统一管理**：全局共享脚本，简化管理复杂度
4. **企业级安全**：多层安全防护，满足企业安全要求
5. **云原生支持**：Docker化部署，支持Kubernetes集群
6. **完善监控**：全方位的监控指标和告警机制

### 14.2 扩展能力

- **内置模块扩展**：AI、视觉、算法等专业模块
- **协议扩展**：gRPC、MQTT等多种通信协议
- **安全策略扩展**：企业级安全控制策略
- **日志目标扩展**：多种日志存储和分析方案
- **监控扩展**：Prometheus、Grafana等监控集成

### 14.3 应用场景

- 工业自动化控制系统
- 智能仓储管理系统
- 机器人调度控制系统
- IoT设备管理平台
- 数据处理与分析平台

该框架为企业提供了一个强大、灵活、安全的脚本执行环境，能够满足复杂业务场景的需求，并具备良好的扩展性和维护性。