first commit

.env

@@ -0,0 +1,67 @@
+# 日志级别: DEBUG / INFO / WARNING / ERROR
+AI_FLOW_LOG_LEVEL=INFO
+
+# Redis
+AI_FLOW_REDIS_HOST=127.0.0.1
+AI_FLOW_REDIS_PORT=6379
+AI_FLOW_REDIS_PASSWORD=123456
+AI_FLOW_REDIS_DB=0
+AI_FLOW_REDIS_TIMEOUT=2000
+AI_FLOW_STORE_BACKEND=redis
+
+# Redis Pub/Sub
+AI_FLOW_REDIS_PUBLISH_ENABLED=true
+AI_FLOW_REDIS_PUBLISH_CHANNEL=WORKFLOW_TASK
+
+# 工作流线程池
+AI_FLOW_EXECUTOR_POOL_SIZE=5
+
+# 模板缓存
+AI_FLOW_TEMPLATE_CACHE_ENABLED=true
+AI_FLOW_TEMPLATE_CACHE_DIR=data/templates
+AI_FLOW_TEMPLATE_DOWNLOAD_TIMEOUT=30000
+AI_FLOW_TEMPLATE_DOWNLOAD_MAX_SIZE=104857600
+
+# API Key 认证
+AI_FLOW_API_AUTH_ENABLED=false
+AI_FLOW_API_AUTH_KEYS=["sk-ai-flow-123456789","sk-ai-flow-987654321"]
+
+# 本地存储
+AI_FLOW_TASK_STORE_DATA_DIR=data/tasks
+
+# 阿里云 OSS
+AI_FLOW_ALIYUN_ENDPOINT=oss-accelerate.aliyuncs.com
+AI_FLOW_ALIYUN_ACCESS_KEY_ID=LTAI5tKMTVZ9x4Sohkgoy8Jm
+AI_FLOW_ALIYUN_ACCESS_KEY_SECRET=HSfY4wunwquX1jq3qXWFveCH4s1hqf
+AI_FLOW_ALIYUN_BUCKET_NAME=aitori-jp
+
+# Google Cloud Storage
+AI_FLOW_GCS_PROJECT_ID=app-aitori-jp
+AI_FLOW_GCS_BUCKET_NAME=aitori-public
+AI_FLOW_GCS_CREDENTIALS_JSON=
+AI_FLOW_GCS_CREDENTIALS_FILE=data/gcs-credentials.json
+
+# Gemini API
+AI_FLOW_GEMINI_API_KEY=AIzaSyBVU4YfAzZjNcIX35BPFV1GZ2xfLtAHPYc
+AI_FLOW_GEMINI_BASE_URL=https://generativelanguage.googleapis.com/v1beta
+AI_FLOW_GEMINI_NANO_BANANA_MODEL=gemini-3.1-flash-image-preview
+AI_FLOW_GEMINI_VEO3_MODEL=veo-3.0-generate-001
+
+# Kontext (BFL) API
+AI_FLOW_KONTEXT_API_KEY=0d830bd5-03e4-4580-b6c7-666081990cd4
+AI_FLOW_KONTEXT_PRO_URL=https://api.bfl.ai/v1/flux-kontext-pro
+AI_FLOW_KONTEXT_MAX_URL=https://api.bfl.ai/v1/flux-kontext-max
+
+# Keling API
+AI_FLOW_KELING_APP_KEY=AYBekdtKmr3RQmRkJC3yRfABFt8tNf9Q
+AI_FLOW_KELING_APP_SECRET=9CGKFFLFnpJfnbgBhyfKPRB3RPd4pthD
+AI_FLOW_KELING_IMAGE2VIDEO_URL=https://api-singapore.klingai.com/v1/videos/image2video
+AI_FLOW_KELING_VIDEO2EFFECT_URL=https://api-singapore.klingai.com/v1/videos/effects
+
+# SeedDance / SeedDream API
+AI_FLOW_SEED_API_KEY=2dcc8f7a-6944-486f-9f20-9e0f2755ffab
+AI_FLOW_SEEDANCE_VIDEO_URL=https://ark.ap-southeast.bytepluses.com/api/v3/contents/generations/tasks
+AI_FLOW_SEEDREAM_IMAGE_URL=https://ark.ap-southeast.bytepluses.com/api/v3/images/generations
+
+# 资源路径
+AI_FLOW_RESOURCES_PATH=resources

.env.example

@@ -0,0 +1,67 @@
+# 日志级别: DEBUG / INFO / WARNING / ERROR
+AI_FLOW_LOG_LEVEL=INFO
+
+# Redis
+AI_FLOW_REDIS_HOST=localhost
+AI_FLOW_REDIS_PORT=6379
+AI_FLOW_REDIS_PASSWORD=
+AI_FLOW_REDIS_DB=0
+AI_FLOW_REDIS_TIMEOUT=2000
+AI_FLOW_STORE_BACKEND=redis
+
+# Redis Pub/Sub
+AI_FLOW_REDIS_PUBLISH_ENABLED=true
+AI_FLOW_REDIS_PUBLISH_CHANNEL=WORKFLOW_TASK
+
+# 工作流线程池
+AI_FLOW_EXECUTOR_POOL_SIZE=8
+
+# 模板缓存
+AI_FLOW_TEMPLATE_CACHE_ENABLED=true
+AI_FLOW_TEMPLATE_CACHE_DIR=data/templates
+AI_FLOW_TEMPLATE_DOWNLOAD_TIMEOUT=30000
+AI_FLOW_TEMPLATE_DOWNLOAD_MAX_SIZE=104857600
+
+# API Key 认证
+AI_FLOW_API_AUTH_ENABLED=false
+AI_FLOW_API_AUTH_KEYS=[]
+
+# 本地存储
+AI_FLOW_TASK_STORE_DATA_DIR=data/tasks
+
+# 阿里云 OSS
+AI_FLOW_ALIYUN_ENDPOINT=oss-accelerate.aliyuncs.com
+AI_FLOW_ALIYUN_ACCESS_KEY_ID=your-access-key-id
+AI_FLOW_ALIYUN_ACCESS_KEY_SECRET=your-access-key-secret
+AI_FLOW_ALIYUN_BUCKET_NAME=your-bucket-name
+
+# Google Cloud Storage
+AI_FLOW_GCS_PROJECT_ID=your-gcs-project-id
+AI_FLOW_GCS_BUCKET_NAME=your-gcs-bucket-name
+AI_FLOW_GCS_CREDENTIALS_JSON=
+AI_FLOW_GCS_CREDENTIALS_FILE=data/gcs-credentials.json
+
+# Gemini API
+AI_FLOW_GEMINI_API_KEY=your-gemini-api-key
+AI_FLOW_GEMINI_BASE_URL=https://generativelanguage.googleapis.com/v1beta
+AI_FLOW_GEMINI_NANO_BANANA_MODEL=gemini-3.1-flash-image-preview
+AI_FLOW_GEMINI_VEO3_MODEL=veo-3.0-generate-001
+
+# Kontext (BFL) API
+AI_FLOW_KONTEXT_API_KEY=your-kontext-api-key
+AI_FLOW_KONTEXT_PRO_URL=https://api.bfl.ai/v1/flux-kontext-pro
+AI_FLOW_KONTEXT_MAX_URL=https://api.bfl.ai/v1/flux-kontext-max
+
+# Keling API
+AI_FLOW_KELING_APP_KEY=your-keling-app-key
+AI_FLOW_KELING_APP_SECRET=your-keling-app-secret
+AI_FLOW_KELING_IMAGE2VIDEO_URL=https://api-singapore.klingai.com/v1/videos/image2video
+AI_FLOW_KELING_VIDEO2EFFECT_URL=https://api-singapore.klingai.com/v1/videos/effects
+
+# SeedDance / SeedDream API
+AI_FLOW_SEED_API_KEY=your-seed-api-key
+AI_FLOW_SEEDANCE_VIDEO_URL=https://ark.ap-southeast.bytepluses.com/api/v3/contents/generations/tasks
+AI_FLOW_SEEDREAM_IMAGE_URL=https://ark.ap-southeast.bytepluses.com/api/v3/images/generations
+
+# 资源路径
+AI_FLOW_RESOURCES_PATH=resources

.gitignore

@@ -0,0 +1 @@
+.venv

.idea/.gitignore

@@ -0,0 +1,10 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Ignored default folder with query files
+/queries/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml
+# Editor-based HTTP Client requests
+/httpRequests/

.idea/ai-flow-python.iml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="ai-flow-python" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PyDocumentationSettings">
+    <option name="format" value="PLAIN" />
+    <option name="myDocStringFormat" value="Plain" />
+  </component>
+  <component name="TemplatesService">
+    <option name="TEMPLATE_FOLDERS">
+      <list>
+        <option value="$MODULE_DIR$/data/templates" />
+      </list>
+    </option>
+  </component>
+</module>

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/misc.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="Python 3.9 (ai-flow-python)" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="ai-flow-python" project-jdk-type="Python SDK" />
+</project>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/ai-flow-python.iml" filepath="$PROJECT_DIR$/.idea/ai-flow-python.iml" />
+    </modules>
+  </component>
+</project>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+RUN mkdir -p data/data data/tasks data/templates
+
+EXPOSE 8080
+
+HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')" || exit 1
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

README.md

@@ -0,0 +1,488 @@
+# AI Flow (Python)
+
+基于 FastAPI 的 AI 工作流执行引擎，支持多种 AI 任务（图像处理、视频编辑、智能美颜等）。
+
+## 项目结构
+
+```
+ai-flow-python/
+├── app/
+│   ├── main.py                    # FastAPI 应用入口
+│   ├── config.py                  # 配置管理 (Pydantic Settings)
+│   ├── controller/
+│   │   └── flow_controller.py     # REST API 路由
+│   ├── middleware/
+│   │   └── api_key.py             # API Key 认证中间件
+│   ├── repository/
+│   │   ├── record_interface.py    # 存储抽象接口
+│   │   ├── task_record.py         # 任务记录数据模型
+│   │   ├── redis_store.py         # Redis 存储实现
+│   │   └── local_store.py         # 本地文件存储实现
+│   ├── service/
+│   │   ├── engine/                # 工作流引擎
+│   │   │   ├── workflow_engine.py
+│   │   │   ├── execution_context.py
+│   │   │   ├── parameter_resolver.py
+│   │   │   └── workflow_result.py
+│   │   ├── spec/                  # 工作流规范定义
+│   │   │   ├── workflow.py
+│   │   │   ├── step.py
+│   │   │   └── workflow_loader.py
+│   │   ├── task/                  # 任务执行系统
+│   │   │   ├── base_task.py
+│   │   │   ├── task_executor.py
+│   │   │   ├── task_param.py
+│   │   │   ├── task_result.py
+│   │   │   └── impl/             # 15 个任务实现
+│   │   └── template/
+│   │       └── template_service.py
+│   └── util/                      # 工具类
+├── data/                          # 数据目录
+├── resources/                     # 资源文件 (美颜/编辑器)
+├── requirements.txt
+├── Dockerfile
+├── docker-compose.yml
+└── .env.example
+```
+
+## 快速开始
+
+### 本地运行
+
+```bash
+# 安装依赖
+pip install -r requirements.txt
+
+# 复制环境变量配置
+cp .env.example .env
+# 编辑 .env 填入实际的 API Key
+
+# 启动服务
+uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
+```
+
+### Docker 运行
+
+```bash
+docker compose up -d
+```
+
+## API 接口
+
+> 若启用了 API Key 认证（`AI_FLOW_API_AUTH_ENABLED=true`），除 `/health` 外所有接口需携带请求头：
+> `Authorization: Bearer <your-api-key>`
+
+### 接口总览
+
+| 方法 | 路径 | 说明 |
+|------|------|------|
+| POST | /execute | 提交并执行工作流 |
+| GET  | /result | 查询单个任务结果 |
+| GET  | /tasks | 获取所有任务列表 |
+| POST | /cleanup | 清理已完成/失败的过期任务 |
+| GET  | /health | 健康检查 |
+| GET  | /executor/status | 线程池状态 |
+| GET  | /template/cache/status | 模板缓存状态 |
+| POST | /template/cache/cleanup | 清理过期模板缓存 |
+
+---
+
+### POST /execute
+
+提交工作流任务，异步执行。提交后立即返回 `taskId`，通过 `/result` 轮询结果或通过 Redis Pub/Sub（channel: `WORKFLOW_TASK`）订阅结果。
+
+**请求体 (JSON)**
+
+| 字段 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| template | string | 是 | 工作流模板地址（URL，支持 zip 包） |
+| inputs | object | 否 | 工作流输入参数，键值对形式 |
+
+**请求示例**
+
+```bash
+curl -X POST http://localhost:8080/execute \
+  -H "Content-Type: application/json" \
+  -d '{
+    "template": "https://example.com/workflow-template.zip",
+    "inputs": {
+      "prompt": "a beautiful sunset",
+      "image_url": "https://example.com/photo.jpg"
+    }
+  }'
+```
+
+**响应 — 成功**
+
+```json
+{
+  "state": "successes",
+  "taskId": "f47ac10b-58cc-4372-a567-0e02b2c3d479"
+}
+```
+
+**响应 — 参数缺失**
+
+```json
+{
+  "state": "failed",
+  "message": "template parameter is required"
+}
+```
+
+---
+
+### GET /result
+
+查询单个任务的执行结果。
+
+**查询参数**
+
+| 参数 | 类型 | 必填 | 说明 |
+|------|------|------|------|
+| taskId | string | 是 | 任务 ID（由 `/execute` 返回） |
+
+**请求示例**
+
+```bash
+curl http://localhost:8080/result?taskId=f47ac10b-58cc-4372-a567-0e02b2c3d479
+```
+
+**响应 — 执行中**
+
+```json
+{
+  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
+  "status": "processing",
+  "message": "",
+  "type": "workflow",
+  "data": null,
+  "request": {
+    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
+    "inputs": {
+      "images": [
+        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
+      ]
+    }
+  },
+  "createTime": 1773313868333,
+  "updateTime": 1773313868500
+}
+```
+
+**响应 — 成功**
+
+```json
+{
+  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
+  "status": "successes",
+  "message": "",
+  "type": "workflow",
+  "data": {
+    "images": [
+      "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
+    ]
+  },
+  "request": {
+    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
+    "inputs": {
+      "images": [
+        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
+      ]
+    }
+  },
+  "createTime": 1773313868333,
+  "updateTime": 1773313889665
+}
+```
+
+**响应 — 失败**
+
+```json
+{
+  "taskId": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
+  "status": "failed",
+  "message": "Failed to load workflow: template not found",
+  "type": "workflow",
+  "data": null,
+  "request": {
+    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
+    "inputs": {
+      "images": [
+        "https://aitori-jp.oss-accelerate.aliyuncs.com/img/upload/2026-03-12/f61e1be77e114c1e84b1dd9bdaa38e02.jpg"
+      ]
+    }
+  },
+  "createTime": 1773313868333,
+  "updateTime": 1773313870000
+}
+```
+
+**响应 — 任务不存在**
+
+```json
+{
+  "error": "task not found"
+}
+```
+
+**status 状态说明**
+
+| 值 | 说明 |
+|----|------|
+| submitted | 已提交，等待执行 |
+| processing | 执行中 |
+| successes | 执行成功 |
+| failed | 执行失败 |
+
+---
+
+### GET /tasks
+
+获取所有任务列表。
+
+**请求示例**
+
+```bash
+curl http://localhost:8080/tasks
+```
+
+**响应**
+
+```json
+{
+  "total": 2,
+  "tasks": {
+    "68710045-e95c-4bd9-9103-6853cdcfd4fd": {
+      "task_id": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
+      "status": "successes",
+      "message": "",
+      "type": "workflow",
+      "data": {
+        "images": [
+          "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
+        ]
+      },
+      "request": {
+        "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
+        "inputs": { "images": ["https://..."] }
+      },
+      "create_time": 1773313868333,
+      "update_time": 1773313889665
+    },
+    "a1b2c3d4-5e6f-7890-abcd-ef1234567890": {
+      "task_id": "a1b2c3d4-5e6f-7890-abcd-ef1234567890",
+      "status": "processing",
+      "message": "",
+      "type": "workflow",
+      "data": null,
+      "request": {
+        "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/another-template.zip",
+        "inputs": { "images": ["https://..."] }
+      },
+      "create_time": 1773314000000,
+      "update_time": 1773314001000
+    }
+  }
+}
+```
+
+---
+
+### POST /cleanup
+
+清理超过指定时间的已完成（successes）和失败（failed）任务记录。
+
+**请求体 (JSON)**
+
+| 字段 | 类型 | 必填 | 默认值 | 说明 |
+|------|------|------|--------|------|
+| maxAgeHours | int | 否 | 24 | 保留时长（小时），超过此时间的终态任务将被删除 |
+
+**请求示例**
+
+```bash
+curl -X POST http://localhost:8080/cleanup \
+  -H "Content-Type: application/json" \
+  -d '{"maxAgeHours": 48}'
+```
+
+**响应**
+
+```json
+{
+  "message": "Cleanup completed",
+  "maxAgeHours": 48
+}
+```
+
+---
+
+### GET /health
+
+健康检查接口，无需认证。
+
+**请求示例**
+
+```bash
+curl http://localhost:8080/health
+```
+
+**响应**
+
+```json
+{
+  "status": "UP",
+  "service": "ai-flow",
+  "timestamp": 1740600000000
+}
+```
+
+---
+
+### GET /executor/status
+
+查看工作流线程池状态。
+
+**请求示例**
+
+```bash
+curl http://localhost:8080/executor/status
+```
+
+**响应**
+
+```json
+{
+  "maxWorkers": 8,
+  "isShutdown": false
+}
+```
+
+---
+
+### GET /template/cache/status
+
+查看模板缓存统计信息。
+
+**请求示例**
+
+```bash
+curl http://localhost:8080/template/cache/status
+```
+
+**响应**
+
+```json
+{
+  "cacheEnabled": true,
+  "cacheDir": "data/templates",
+  "cachedTemplates": 3,
+  "cacheSizeBytes": 52428800,
+  "cacheSizeMB": 50,
+  "templateDirectories": 3
+}
+```
+
+---
+
+### POST /template/cache/cleanup
+
+清理过期的模板缓存目录。
+
+**请求体 (JSON)**
+
+| 字段 | 类型 | 必填 | 默认值 | 说明 |
+|------|------|------|--------|------|
+| maxAgeHours | int | 否 | 24 | 保留时长（小时），超过此时间的模板缓存将被删除 |
+
+**请求示例**
+
+```bash
+curl -X POST http://localhost:8080/template/cache/cleanup \
+  -H "Content-Type: application/json" \
+  -d '{"maxAgeHours": 72}'
+```
+
+**响应**
+
+```json
+{
+  "message": "Template cache cleanup completed",
+  "maxAgeHours": 72
+}
+```
+
+---
+
+### Redis Pub/Sub 事件通知
+
+当任务到达终态（`successes` / `failed`）时，若使用 Redis 存储后端且启用了发布功能，会自动将任务记录发布到 Redis channel。
+
+| 配置项 | 默认值 | 说明 |
+|--------|--------|------|
+| AI_FLOW_REDIS_PUBLISH_ENABLED | true | 是否启用发布 |
+| AI_FLOW_REDIS_PUBLISH_CHANNEL | WORKFLOW_TASK | channel 名称 |
+
+**消息格式（JSON）**
+
+与 `/result` 接口返回的 `TaskRecord` 结构一致：
+
+```json
+{
+  "task_id": "68710045-e95c-4bd9-9103-6853cdcfd4fd",
+  "status": "successes",
+  "message": "",
+  "type": "workflow",
+  "data": {
+    "images": [
+      "https://aitori-jp.oss-accelerate.aliyuncs.com/0000/yyyy-MM-dd/2db32a61-9859-4f36-94d8-76f19ca65515.png"
+    ]
+  },
+  "request": {
+    "template": "https://aitori-jp.oss-accelerate.aliyuncs.com/img/style/606e80a19bb545238c5eab47f15d3bda.zip",
+    "inputs": { "images": ["https://..."] }
+  },
+  "create_time": 1773313868333,
+  "update_time": 1773313889665
+}
+```
+
+**订阅示例（Python）**
+
+```python
+import redis, json
+
+r = redis.Redis(host="10.146.0.3", port=6379, password="xxx", decode_responses=True)
+pubsub = r.pubsub()
+pubsub.subscribe("WORKFLOW_TASK")
+
+for message in pubsub.listen():
+    if message["type"] == "message":
+        record = json.loads(message["data"])
+        print(f"Task {record['task_id']} -> {record['status']}")
+```
+
+## 支持的任务类型
+
+| 类型 | 说明 |
+|------|------|
+| kontext_pro | Flux Kontext Pro AI 图像生成 |
+| kontext_max | Flux Kontext Max AI 图像生成 |
+| keling_image2video | 可灵图生视频 |
+| keling_video2effect | 可灵视频特效 |
+| nano_banana | Gemini 图像生成 |
+| google_veo3 | Google Veo3 视频生成 |
+| seedance_video | SeedDance 视频生成 |
+| seedream_image | SeedDream 图像生成 |
+| smart_beauty | 智能美颜 |
+| image_editor | 图像编辑 |
+| video_editor | 视频编辑 |
+| file_download | 文件下载 |
+| file_upload | 文件上传 (阿里云 OSS / Google Cloud Storage) |
+| base64_encode | Base64 编码 |
+| base64_decode | Base64 解码 |
+
+## 配置说明
+
+所有配置通过环境变量注入，前缀为 `AI_FLOW_`。详见 `.env.example`。

app/__init__.py

@@ -0,0 +1 @@
+

app/config.py

@@ -0,0 +1,84 @@
+from pydantic_settings import BaseSettings
+from typing import List
+
+
+class Settings(BaseSettings):
+    app_name: str = "ai-flow"
+
+    # 日志级别
+    log_level: str = "INFO"
+
+    # Redis
+    redis_host: str
+    redis_port: int
+    redis_password: str
+    redis_db: int = 0
+    redis_timeout: int = 2000
+
+    # 存储后端: "redis" 或 "local"
+    store_backend: str = "redis"
+
+    # Redis Pub/Sub
+    redis_publish_enabled: bool = True
+    redis_publish_channel: str = "WORKFLOW_TASK"
+
+    # 工作流线程池
+    executor_pool_size: int = 8
+
+    # 模板缓存
+    template_cache_enabled: bool = True
+    template_cache_dir: str = "data/templates"
+    template_download_timeout: int = 30000
+    template_download_max_size: int = 104857600
+
+    # API Key 认证
+    api_auth_enabled: bool = False
+    api_auth_keys: List[str] = []
+    api_auth_header_name: str = "Authorization"
+    api_auth_prefix: str = "Bearer"
+    api_auth_error_message: str = "Invalid API Key"
+
+    # 本地存储
+    task_store_data_dir: str = "data/tasks"
+
+    # 阿里云 OSS
+    aliyun_endpoint: str
+    aliyun_access_key_id: str
+    aliyun_access_key_secret: str
+    aliyun_bucket_name: str
+
+    # Google Cloud Storage
+    gcs_project_id: str
+    gcs_bucket_name: str
+    gcs_credentials_json: str = ""
+    gcs_credentials_file: str = ""
+
+    # Gemini API
+    gemini_api_key: str
+    gemini_base_url: str
+    gemini_nano_banana_model: str
+    gemini_veo3_model: str
+
+    # Kontext (BFL) API
+    kontext_api_key: str
+    kontext_pro_url: str
+    kontext_max_url: str
+
+    # Keling API
+    keling_app_key: str
+    keling_app_secret: str
+    keling_image2video_url: str
+    keling_video2effect_url: str
+
+    # SeedDance / SeedDream API
+    seed_api_key: str
+    seedance_video_url: str
+    seedream_image_url: str
+
+    # 资源路径
+    resources_path: str = "resources"
+
+    model_config = {"env_prefix": "AI_FLOW_", "env_file": ".env"}
+
+
+settings = Settings()

app/controller/__init__.py

@@ -0,0 +1 @@
+

app/controller/flow_controller.py

@@ -0,0 +1,163 @@
+import logging
+import time
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
+
+from fastapi import APIRouter, Request
+
+from app.config import settings
+from app.repository.record_interface import RecordInterface
+from app.service.engine.workflow_engine import WorkflowEngine
+from app.service.spec.workflow_loader import load_from_directory
+from app.service.task.task_executor import TaskExecutor
+from app.service.template.template_service import TemplateService
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# 全局单例（在 main.py 中初始化并注入）
+_record_store: RecordInterface | None = None
+_template_service: TemplateService | None = None
+_engine: WorkflowEngine | None = None
+_executor: ThreadPoolExecutor | None = None
+
+
+def init_controller(
+    record_store: RecordInterface,
+    template_service: TemplateService,
+):
+    global _record_store, _template_service, _engine, _executor
+    _record_store = record_store
+    _template_service = template_service
+    _engine = WorkflowEngine(TaskExecutor.create_default())
+    _executor = ThreadPoolExecutor(
+        max_workers=settings.executor_pool_size,
+        thread_name_prefix="workflow-executor-",
+    )
+    logger.info("FlowController initialized with pool size: %d", settings.executor_pool_size)
+
+
+def shutdown_controller():
+    global _executor
+    if _executor:
+        _executor.shutdown(wait=True, cancel_futures=False)
+        logger.info("FlowController thread pool shut down")
+
+
+@router.post("/execute")
+async def execute(request: Request) -> dict[str, Any]:
+    req: dict = await request.json()
+    template_url = req.get("template")
+    inputs: dict = req.get("inputs") if isinstance(req.get("inputs"), dict) else {}
+
+    logger.info("POST /execute  template=%s, input_keys=%s", template_url, list(inputs.keys()))
+
+    if not template_url or not str(template_url).strip():
+        logger.warning("POST /execute  rejected: template parameter is required")
+        return {"state": "failed", "message": "template parameter is required"}
+
+    task_id = _record_store.create_record(req)
+    logger.info("[%s] Task accepted, submitting to thread pool", task_id)
+
+    def _run():
+        t0 = time.time()
+        try:
+            _record_store.set_processing(task_id)
+
+            logger.info("[%s] Loading template: %s", task_id, template_url)
+            try:
+                template_path = _template_service.get_template_path(template_url)
+                flow = load_from_directory(template_path)
+                logger.info("[%s] Workflow loaded: version=%s, steps=%s, order=%s",
+                            task_id, flow.version, len(flow.steps), flow.order)
+            except Exception as e:
+                logger.error("[%s] Failed to load workflow: %s", task_id, e)
+                _record_store.set_failed(task_id, f"Failed to load workflow: {e}")
+                return
+
+            logger.info("[%s] Starting workflow execution", task_id)
+            result = _engine.execute(task_id, flow, inputs)
+            elapsed = time.time() - t0
+
+            if result.success:
+                logger.info("[%s] Workflow completed successfully in %.2fs", task_id, elapsed)
+                _record_store.set_success(task_id, result.outputs)
+            else:
+                logger.warning("[%s] Workflow failed in %.2fs: %s", task_id, elapsed, result.error_message)
+                _record_store.set_failed(task_id, result.error_message)
+        except Exception as e:
+            elapsed = time.time() - t0
+            logger.error("[%s] Workflow execution exception in %.2fs: %s", task_id, elapsed, e, exc_info=True)
+            _record_store.set_failed(task_id, str(e))
+
+    _executor.submit(_run)
+    return {"state": "successes", "taskId": task_id}
+
+
+@router.get("/result")
+async def result(taskId: str) -> dict[str, Any]:
+    logger.info("GET /result  taskId=%s", taskId)
+    r = _record_store.get_record(taskId)
+    if r is None:
+        logger.warning("GET /result  taskId=%s not found", taskId)
+        return {"error": "task not found"}
+    return {
+        "taskId": r.task_id,
+        "status": r.status,
+        "message": r.message,
+        "type": r.type,
+        "data": r.data,
+        "request": r.request or {},
+        "createTime": r.create_time,
+        "updateTime": r.update_time,
+    }
+
+
+@router.get("/tasks")
+async def get_all_tasks() -> dict[str, Any]:
+    all_tasks = _record_store.get_all_records()
+    return {
+        "total": len(all_tasks),
+        "tasks": {tid: rec.to_dict() for tid, rec in all_tasks.items()},
+    }
+
+
+@router.post("/cleanup")
+async def cleanup_tasks(request: Request) -> dict[str, Any]:
+    req: dict = await request.json()
+    max_age = int(req.get("maxAgeHours", 24))
+    _record_store.delete_records(max_age)
+    return {"message": "Cleanup completed", "maxAgeHours": max_age}
+
+
+@router.get("/health")
+async def health() -> dict[str, Any]:
+    return {
+        "status": "UP",
+        "service": settings.app_name,
+        "timestamp": int(time.time() * 1000),
+    }
+
+
+@router.get("/executor/status")
+async def executor_status() -> dict[str, Any]:
+    if _executor is None:
+        return {"error": "executor not initialized"}
+    return {
+        "maxWorkers": _executor._max_workers,
+        "isShutdown": _executor._shutdown,
+    }
+
+
+@router.get("/template/cache/status")
+async def template_cache_status() -> dict[str, Any]:
+    return _template_service.get_cache_stats()
+
+
+@router.post("/template/cache/cleanup")
+async def template_cache_cleanup(request: Request) -> dict[str, Any]:
+    req: dict = await request.json()
+    max_age = int(req.get("maxAgeHours", 24))
+    _template_service.cleanup_expired_templates(max_age)
+    return {"message": "Template cache cleanup completed", "maxAgeHours": max_age}

app/main.py

@@ -0,0 +1,65 @@
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from app.config import settings
+from app.controller.flow_controller import (
+    init_controller,
+    router,
+    shutdown_controller,
+)
+from app.middleware.api_key import ApiKeyMiddleware
+from app.repository.local_store import LocalRecordStore
+from app.repository.redis_store import RedisRecordStore
+from app.service.template.template_service import TemplateService
+
+_LOG_FORMAT = "%(asctime)s  %(levelname)-5s  [%(name)s]  %(message)s"
+_LOG_DATEFMT = "%Y-%m-%d %H:%M:%S"
+
+logger = logging.getLogger(__name__)
+
+
+def setup_logging() -> None:
+    """配置根日志。使用 force=True 确保在 uvicorn dictConfig 之后仍能生效。"""
+    level = getattr(logging, settings.log_level.upper(), logging.INFO)
+    logging.basicConfig(
+        level=level,
+        format=_LOG_FORMAT,
+        datefmt=_LOG_DATEFMT,
+        force=True,
+    )
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    setup_logging()
+
+    logger.info("========================================")
+    logger.info("Starting %s ...", settings.app_name)
+    logger.info("Log level     : %s", settings.log_level.upper())
+    logger.info("Store backend : %s", settings.store_backend)
+    logger.info("Executor pool : %d", settings.executor_pool_size)
+    logger.info("========================================")
+
+    if settings.store_backend == "redis":
+        store = RedisRecordStore()
+        logger.info("Redis record store initialized (host=%s, port=%d)",
+                     settings.redis_host, settings.redis_port)
+    else:
+        store = LocalRecordStore()
+        logger.info("Local record store initialized")
+
+    template_service = TemplateService()
+    init_controller(store, template_service)
+
+    logger.info("%s is ready to accept requests", settings.app_name)
+    yield
+
+    shutdown_controller()
+    logger.info("%s stopped.", settings.app_name)
+
+
+app = FastAPI(title="AI Flow", lifespan=lifespan)
+app.add_middleware(ApiKeyMiddleware)
+app.include_router(router)

app/middleware/__init__.py

@@ -0,0 +1 @@
+

app/middleware/api_key.py

@@ -0,0 +1,43 @@
+import logging
+import time
+from typing import Callable
+
+from fastapi import Request, Response
+from fastapi.responses import JSONResponse
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class ApiKeyMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        method = request.method
+        path = request.url.path
+        client_ip = request.client.host if request.client else "unknown"
+
+        logger.info("--> %s %s  (client=%s)", method, path, client_ip)
+        t0 = time.time()
+
+        if settings.api_auth_enabled and path != "/health":
+            header_value = request.headers.get(settings.api_auth_header_name, "")
+            prefix = settings.api_auth_prefix
+            if prefix and header_value.startswith(prefix + " "):
+                token = header_value[len(prefix) + 1:]
+            else:
+                token = header_value
+
+            if token not in settings.api_auth_keys:
+                elapsed = time.time() - t0
+                logger.warning("<-- %s %s  401 Unauthorized  (%.3fs, client=%s)",
+                                method, path, elapsed, client_ip)
+                return JSONResponse(
+                    status_code=401,
+                    content={"error": settings.api_auth_error_message},
+                )
+
+        response = await call_next(request)
+        elapsed = time.time() - t0
+        logger.info("<-- %s %s  %d  (%.3fs)", method, path, response.status_code, elapsed)
+        return response

app/repository/__init__.py

@@ -0,0 +1 @@
+

app/repository/local_store.py

@@ -0,0 +1,86 @@
+import json
+import logging
+import os
+import time
+import threading
+from typing import Any, Optional
+
+from app.config import settings
+from app.repository.record_interface import RecordInterface
+from app.repository.task_record import TaskRecord
+
+logger = logging.getLogger(__name__)
+
+
+class LocalRecordStore(RecordInterface):
+    def __init__(self):
+        self._storage: dict[str, TaskRecord] = {}
+        self._lock = threading.Lock()
+        self._data_dir = settings.task_store_data_dir
+        os.makedirs(self._data_dir, exist_ok=True)
+        self._load_from_file()
+
+    def _load_from_file(self):
+        filepath = os.path.join(self._data_dir, "tasks.json")
+        if os.path.isfile(filepath) and os.path.getsize(filepath) > 0:
+            try:
+                with open(filepath, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                for tid, rec in data.items():
+                    self._storage[tid] = TaskRecord.from_dict(rec)
+                logger.info("Loaded %d tasks from file system", len(data))
+            except Exception as e:
+                logger.error("Failed to load tasks from file: %s", e)
+
+    def _save_to_file(self):
+        try:
+            tmp = os.path.join(self._data_dir, "tasks.tmp")
+            filepath = os.path.join(self._data_dir, "tasks.json")
+            with open(tmp, "w", encoding="utf-8") as f:
+                json.dump(
+                    {tid: r.to_dict() for tid, r in self._storage.items()},
+                    f,
+                    indent=2,
+                    ensure_ascii=False,
+                )
+            os.replace(tmp, filepath)
+        except Exception as e:
+            logger.error("Failed to save tasks to file: %s", e)
+
+    def _on_save(self, record: TaskRecord):
+        with self._lock:
+            self._storage[record.task_id] = record
+            self._save_to_file()
+
+    def _on_get(self, task_id: str) -> Optional[TaskRecord]:
+        return self._storage.get(task_id)
+
+    def _on_get_all(self) -> dict[str, TaskRecord]:
+        return dict(self._storage)
+
+    def _on_delete(self, max_age_hours: int):
+        cutoff = int(time.time() * 1000) - max_age_hours * 3600 * 1000
+        with self._lock:
+            to_delete = [
+                tid
+                for tid, r in self._storage.items()
+                if r.status in ("successes", "failed") and r.update_time < cutoff
+            ]
+            for tid in to_delete:
+                del self._storage[tid]
+            self._save_to_file()
+
+    def _on_update_status(
+        self, task_id: str, status: str, message: str, data: Optional[dict]
+    ):
+        try:
+            record = self._on_get(task_id)
+            if record:
+                record.status = status
+                record.message = message
+                record.touch()
+                if data is not None:
+                    record.data = data
+                self._on_save(record)
+        except Exception as e:
+            logger.error("Failed to update task status: %s", e)

app/repository/record_interface.py

@@ -0,0 +1,68 @@
+import uuid
+import logging
+import time
+from abc import ABC, abstractmethod
+from typing import Any, Optional
+
+from app.repository.task_record import TaskRecord
+
+logger = logging.getLogger(__name__)
+
+
+class RecordInterface(ABC):
+    """任务记录存储的抽象接口"""
+
+    def create_record(self, request: dict[str, Any]) -> str:
+        record_id = str(uuid.uuid4())
+        record = TaskRecord(
+            task_id=record_id,
+            status="submitted",
+            message="",
+            request=request,
+        )
+        self._on_save(record)
+        logger.info("[%s] Record created, status=submitted", record_id)
+        return record_id
+
+    def set_processing(self, task_id: str):
+        logger.info("[%s] Status -> processing", task_id)
+        self._on_update_status(task_id, "processing", "", None)
+
+    def set_failed(self, task_id: str, message: str):
+        logger.warning("[%s] Status -> failed, message=%s", task_id, message)
+        self._on_update_status(task_id, "failed", message or "", None)
+        self._on_publish(task_id)
+
+    def set_success(self, task_id: str, outputs: dict[str, Any]):
+        logger.info("[%s] Status -> successes, output_keys=%s", task_id, list(outputs.keys()) if outputs else [])
+        self._on_update_status(task_id, "successes", "", outputs)
+        self._on_publish(task_id)
+
+    def get_record(self, task_id: str) -> Optional[TaskRecord]:
+        return self._on_get(task_id)
+
+    def get_all_records(self) -> dict[str, TaskRecord]:
+        return self._on_get_all()
+
+    def delete_records(self, max_age_hours: int):
+        self._on_delete(max_age_hours)
+
+    @abstractmethod
+    def _on_save(self, record: TaskRecord): ...
+
+    @abstractmethod
+    def _on_get(self, task_id: str) -> Optional[TaskRecord]: ...
+
+    @abstractmethod
+    def _on_get_all(self) -> dict[str, TaskRecord]: ...
+
+    @abstractmethod
+    def _on_delete(self, max_age_hours: int): ...
+
+    @abstractmethod
+    def _on_update_status(
+        self, task_id: str, status: str, message: str, data: Optional[dict]
+    ): ...
+
+    def _on_publish(self, task_id: str):
+        """任务到达终态时发布事件，子类可覆盖实现"""

app/repository/redis_store.py

@@ -0,0 +1,95 @@
+import json
+import logging
+import time
+from typing import Any, Optional
+
+import redis
+
+from app.config import settings
+from app.repository.record_interface import RecordInterface
+from app.repository.task_record import TaskRecord
+
+logger = logging.getLogger(__name__)
+
+TASK_KEY_PREFIX = "task:"
+
+
+class RedisRecordStore(RecordInterface):
+    def __init__(self):
+        self._client = redis.Redis(
+            host=settings.redis_host,
+            port=settings.redis_port,
+            password=settings.redis_password,
+            db=settings.redis_db,
+            socket_timeout=settings.redis_timeout / 1000,
+            decode_responses=True,
+        )
+
+    def _on_save(self, record: TaskRecord):
+        try:
+            self._client.set(
+                TASK_KEY_PREFIX + record.task_id, json.dumps(record.to_dict())
+            )
+        except Exception as e:
+            logger.error("Failed to save task record: %s", e)
+
+    def _on_get(self, task_id: str) -> Optional[TaskRecord]:
+        try:
+            data = self._client.get(TASK_KEY_PREFIX + task_id)
+            if data:
+                return TaskRecord.from_dict(json.loads(data))
+            return None
+        except Exception as e:
+            logger.error("Failed to get task record: %s", e)
+            return None
+
+    def _on_get_all(self) -> dict[str, TaskRecord]:
+        result: dict[str, TaskRecord] = {}
+        try:
+            keys = self._client.keys(TASK_KEY_PREFIX + "*")
+            for key in keys or []:
+                task_id = key[len(TASK_KEY_PREFIX):]
+                data = self._client.get(key)
+                if data:
+                    result[task_id] = TaskRecord.from_dict(json.loads(data))
+        except Exception as e:
+            logger.error("Failed to get all tasks: %s", e)
+        return result
+
+    def _on_delete(self, max_age_hours: int):
+        try:
+            cutoff = int(time.time() * 1000) - max_age_hours * 3600 * 1000
+            all_records = self._on_get_all()
+            for task_id, record in all_records.items():
+                if record.status in ("successes", "failed") and record.update_time < cutoff:
+                    self._client.delete(TASK_KEY_PREFIX + task_id)
+        except Exception as e:
+            logger.error("Failed to cleanup completed tasks: %s", e)
+
+    def _on_update_status(
+        self, task_id: str, status: str, message: str, data: Optional[dict]
+    ):
+        try:
+            record = self._on_get(task_id)
+            if record:
+                record.status = status
+                record.message = message
+                record.touch()
+                if data is not None:
+                    record.data = data
+                self._on_save(record)
+        except Exception as e:
+            logger.error("Failed to update task status: %s", e)
+
+    def _on_publish(self, task_id: str):
+        if not settings.redis_publish_enabled:
+            return
+        try:
+            record = self._on_get(task_id)
+            if record:
+                payload = json.dumps(record.to_dict(), ensure_ascii=False)
+                receivers = self._client.publish(settings.redis_publish_channel, payload)
+                logger.info("[%s] Published to channel '%s', receivers=%d",
+                            task_id, settings.redis_publish_channel, receivers)
+        except Exception as e:
+            logger.error("[%s] Failed to publish task event: %s", task_id, e)

app/repository/task_record.py

@@ -0,0 +1,43 @@
+import time
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+
+@dataclass
+class TaskRecord:
+    task_id: str = ""
+    status: str = "submitted"  # submitted | processing | successes | failed
+    message: str = ""
+    type: str = "workflow"
+    data: Optional[dict[str, Any]] = None
+    request: Optional[dict[str, Any]] = None
+    create_time: int = field(default_factory=lambda: int(time.time() * 1000))
+    update_time: int = field(default_factory=lambda: int(time.time() * 1000))
+
+    def touch(self):
+        self.update_time = int(time.time() * 1000)
+
+    def to_dict(self) -> dict:
+        return {
+            "task_id": self.task_id,
+            "status": self.status,
+            "message": self.message,
+            "type": self.type,
+            "data": self.data,
+            "request": self.request,
+            "create_time": self.create_time,
+            "update_time": self.update_time,
+        }
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "TaskRecord":
+        return cls(
+            task_id=d.get("task_id", ""),
+            status=d.get("status", "submitted"),
+            message=d.get("message", ""),
+            type=d.get("type", "workflow"),
+            data=d.get("data"),
+            request=d.get("request"),
+            create_time=d.get("create_time", int(time.time() * 1000)),
+            update_time=d.get("update_time", int(time.time() * 1000)),
+        )

app/service/__init__.py

@@ -0,0 +1 @@
+

app/service/engine/__init__.py

@@ -0,0 +1 @@
+

app/service/engine/execution_context.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import re
+from typing import Any, Optional
+
+_ARRAY_INDEX_RE = re.compile(r"^(.+)\[(\d+)\]$")
+
+
+class ExecutionContext:
+    def __init__(self, inputs: dict[str, Any] | None = None):
+        self._inputs: dict[str, Any] = inputs or {}
+        self._step_results: dict[str, dict[str, Any]] = {}
+
+    @property
+    def inputs(self) -> dict[str, Any]:
+        return self._inputs
+
+    def put_step_result(self, step_id: str, outputs: dict[str, Any]):
+        self._step_results[step_id] = outputs or {}
+
+    def get_step_result(self, step_id: str) -> dict[str, Any] | None:
+        return self._step_results.get(step_id)
+
+    def lookup(self, path: str) -> Any:
+        if not path:
+            return None
+
+        parts = path.split(".")
+        if parts[0] == "inputs":
+            current: Any = self._inputs
+            idx = 1
+        else:
+            current = self._step_results.get(parts[0])
+            idx = 1
+
+        for i in range(idx, len(parts)):
+            if current is None:
+                return None
+
+            key = parts[i]
+            m = _ARRAY_INDEX_RE.match(key)
+            if m:
+                array_key = m.group(1)
+                array_index = int(m.group(2))
+                if isinstance(current, dict):
+                    current = current.get(array_key)
+                else:
+                    return None
+                if isinstance(current, list):
+                    current = current[array_index] if 0 <= array_index < len(current) else None
+                else:
+                    return None
+            else:
+                if isinstance(current, dict):
+                    current = current.get(key)
+                elif isinstance(current, list):
+                    try:
+                        index = int(key)
+                        current = current[index] if 0 <= index < len(current) else None
+                    except (ValueError, IndexError):
+                        return None
+                else:
+                    return None
+
+        return current

app/service/engine/parameter_resolver.py

@@ -0,0 +1,43 @@
+import re
+from typing import Any
+
+from app.service.engine.execution_context import ExecutionContext
+
+_ARRAY_COMBINE_RE = re.compile(r"^\[(.*)?\]$")
+
+
+class ParameterResolver:
+    def resolve(self, value: Any, context: ExecutionContext) -> Any:
+        if value is None:
+            return None
+
+        if isinstance(value, str):
+            m = _ARRAY_COMBINE_RE.match(value)
+            if m:
+                content = (m.group(1) or "").strip()
+                if not content:
+                    return []
+                elements = content.split(",")
+                result = []
+                for elem in elements:
+                    elem = elem.strip()
+                    if elem:
+                        resolved = self.resolve(elem, context)
+                        if resolved is not None:
+                            result.append(resolved)
+                return result
+
+            if value.startswith("$"):
+                path = value[1:]
+                return context.lookup(path)
+            return value
+
+        if isinstance(value, dict):
+            return {
+                str(k): self.resolve(v, context) for k, v in value.items()
+            }
+
+        if isinstance(value, list):
+            return [self.resolve(v, context) for v in value]
+
+        return value

app/service/engine/workflow_engine.py

@@ -0,0 +1,145 @@
+import logging
+import os
+import time
+from typing import Any
+
+from app.service.engine.execution_context import ExecutionContext
+from app.service.engine.parameter_resolver import ParameterResolver
+from app.service.engine.workflow_result import WorkflowResult
+from app.service.spec.step import Step
+from app.service.spec.workflow import Workflow
+from app.service.task.task_executor import TaskExecutor
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.task_util import create_task_workspace, cleanup_workspace
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowEngine:
+    def __init__(self, task_executor: TaskExecutor):
+        self._task_executor = task_executor
+        self._resolver = ParameterResolver()
+
+    def execute(
+        self, task_id: str, workflow: Workflow, inputs: dict[str, Any]
+    ) -> WorkflowResult:
+        total_steps = len(workflow.order)
+        logger.info("[%s] ========== Workflow start ==========  version=%s, total_steps=%d",
+                     task_id, workflow.version, total_steps)
+        wf_t0 = time.time()
+
+        ctx = ExecutionContext(inputs)
+
+        try:
+            workspace = create_task_workspace(task_id, workflow.template_path)
+            logger.info("[%s] Workspace created: %s", task_id, workspace)
+        except Exception as e:
+            logger.error("[%s] Failed to create workspace: %s", task_id, e)
+            return WorkflowResult.failure(f"failed: {e}")
+
+        for idx, step_id in enumerate(workflow.order, 1):
+            step = workflow.steps.get(step_id)
+            if step is None:
+                logger.warning("[%s] Step '%s' not found in steps definition, skipping", task_id, step_id)
+                continue
+
+            logger.info("[%s] ------ Step %d/%d [%s] type=%s start ------",
+                         task_id, idx, total_steps, step.id, step.type)
+            logger.info("[%s] [%s] raw_inputs=%s", task_id, step.id, step.inputs)
+
+            step_t0 = time.time()
+            try:
+                step_dir = os.path.join(workspace, step.id)
+                os.makedirs(step_dir, exist_ok=True)
+                result = self._execute_step_with_retry(task_id, workspace, step, ctx)
+                step_elapsed = time.time() - step_t0
+
+                if result.is_failure():
+                    logger.error("[%s] [%s] FAILED in %.2fs after retries. Error: %s",
+                                  task_id, step_id, step_elapsed, result.error_message)
+                    return WorkflowResult.failure(
+                        f"Step {step_id} failed: {result.error_message}"
+                    )
+            except Exception as e:
+                step_elapsed = time.time() - step_t0
+                logger.error("[%s] [%s] EXCEPTION in %.2fs: %s", task_id, step_id, step_elapsed, e, exc_info=True)
+                return WorkflowResult.failure(f"Step {step_id} failed: {e}")
+
+            logger.info("[%s] [%s] SUCCESS in %.2fs  outputs=%s",
+                         task_id, step.id, step_elapsed, _shorten(result.outputs))
+            ctx.put_step_result(step_id, result.outputs)
+
+        logger.info("[%s] All steps completed, resolving final outputs ...", task_id)
+        outputs: dict[str, Any] = {}
+        if workflow.outputs:
+            for key, value in workflow.outputs.items():
+                outputs[key] = self._resolver.resolve(value, ctx)
+        logger.info("[%s] Final output_keys=%s", task_id, list(outputs.keys()))
+
+        cleanup_workspace(workspace, task_id)
+        total_elapsed = time.time() - wf_t0
+        logger.info("[%s] ========== Workflow end ==========  total_elapsed=%.2fs, success=True",
+                     task_id, total_elapsed)
+        return WorkflowResult.ok(outputs)
+
+    def _execute_step_with_retry(
+        self, task_id: str, workspace: str, step: Step, ctx: ExecutionContext
+    ) -> TaskResult:
+        max_retries = step.retry_count
+        last_result: TaskResult | None = None
+
+        for attempt in range(max_retries + 1):
+            step_inputs = self._resolve_inputs(step.inputs, ctx)
+            task = self._task_executor.resolve(step.type)
+            param = TaskParam(
+                workspace=workspace,
+                step_id=step.id,
+                type=step.type,
+                resources=step.resources,
+                inputs=step_inputs,
+            )
+
+            if attempt > 0:
+                logger.info("[%s] [%s] Retry attempt %d/%d",
+                             task_id, step.id, attempt, max_retries)
+
+            logger.info("[%s] [%s] Executing task type=%s, resolved_input_keys=%s",
+                         task_id, step.id, step.type, list(step_inputs.keys()))
+
+            attempt_t0 = time.time()
+            last_result = task.execute(param)
+            attempt_elapsed = time.time() - attempt_t0
+
+            if last_result.is_success():
+                logger.info("[%s] [%s] Task returned SUCCESS in %.2fs",
+                             task_id, step.id, attempt_elapsed)
+                return last_result
+
+            if attempt < max_retries:
+                logger.warning(
+                    "[%s] [%s] Task returned FAILURE in %.2fs (attempt %d/%d), will retry. Error: %s",
+                    task_id, step.id, attempt_elapsed,
+                    attempt + 1, max_retries + 1, last_result.error_message,
+                )
+                time.sleep(1)
+            else:
+                logger.error(
+                    "[%s] [%s] Task returned FAILURE in %.2fs (final attempt %d/%d). Error: %s",
+                    task_id, step.id, attempt_elapsed,
+                    attempt + 1, max_retries + 1, last_result.error_message,
+                )
+
+        return last_result  # type: ignore[return-value]
+
+    def _resolve_inputs(
+        self, inputs: dict[str, Any], ctx: ExecutionContext
+    ) -> dict[str, Any]:
+        return {k: self._resolver.resolve(v, ctx) for k, v in inputs.items()} if inputs else {}
+
+
+def _shorten(d: dict | None, max_len: int = 120) -> str:
+    if d is None:
+        return "null"
+    s = str(d)
+    return s if len(s) <= max_len else s[:max_len] + "..."

app/service/engine/workflow_result.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+
+@dataclass
+class WorkflowResult:
+    success: bool
+    outputs: dict[str, Any] = field(default_factory=dict)
+    error_message: str = ""
+
+    @classmethod
+    def ok(cls, outputs: dict[str, Any]) -> "WorkflowResult":
+        return cls(success=True, outputs=outputs)
+
+    @classmethod
+    def failure(cls, message: str) -> "WorkflowResult":
+        return cls(success=False, error_message=message)

app/service/spec/__init__.py

@@ -0,0 +1 @@
+

app/service/spec/step.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+
+@dataclass
+class Step:
+    id: str = ""
+    type: str = ""
+    resources: str = ""
+    inputs: dict[str, Any] = field(default_factory=dict)
+    outputs: dict[str, Any] = field(default_factory=dict)
+    retry_count: int = 0

app/service/spec/workflow.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+from app.service.spec.step import Step
+
+
+@dataclass
+class Workflow:
+    version: str = ""
+    inputs: dict[str, Any] = field(default_factory=dict)
+    outputs: dict[str, Any] = field(default_factory=dict)
+    order: list[str] = field(default_factory=list)
+    steps: dict[str, Step] = field(default_factory=dict)
+    template_path: str = ""

app/service/spec/workflow_loader.py

@@ -0,0 +1,45 @@
+import json
+import os
+
+from app.service.spec.step import Step
+from app.service.spec.workflow import Workflow
+
+
+def load_from_directory(template_dir: str) -> Workflow:
+    content_path = os.path.join(template_dir, "content.json")
+    if not os.path.isfile(content_path):
+        raise FileNotFoundError(f"content.json not found in {template_dir}")
+
+    with open(content_path, "r", encoding="utf-8") as f:
+        root = json.load(f)
+
+    workflow = _parse_workflow(root)
+    workflow.template_path = template_dir
+    return workflow
+
+
+def _parse_workflow(root: dict) -> Workflow:
+    flow = Workflow(
+        version=root.get("version", ""),
+        inputs=root.get("inputs") or {},
+        outputs=root.get("outputs") or {},
+        order=root.get("order") or [],
+    )
+
+    steps: dict[str, Step] = {}
+    for step_id in flow.order:
+        node = root.get(step_id)
+        if node is None:
+            continue
+        step = Step(
+            id=step_id,
+            type=node.get("type", ""),
+            resources=node.get("resources", ""),
+            inputs=node.get("inputs") or {},
+            outputs=node.get("outputs") or {},
+            retry_count=node.get("retryCount", 0),
+        )
+        steps[step_id] = step
+
+    flow.steps = steps
+    return flow

app/service/task/__init__.py

@@ -0,0 +1 @@
+

app/service/task/base_task.py

@@ -0,0 +1,14 @@
+from abc import ABC, abstractmethod
+
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+
+
+class BaseTask(ABC):
+    @abstractmethod
+    def get_type(self) -> str:
+        ...
+
+    @abstractmethod
+    def execute(self, param: TaskParam) -> TaskResult:
+        ...

app/service/task/impl/__init__.py

@@ -0,0 +1 @@
+

app/service/task/impl/base64_decode_task.py

@@ -0,0 +1,51 @@
+import logging
+import os
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.file_util import generate_unique_filename, base64_to_file
+
+logger = logging.getLogger(__name__)
+
+
+class Base64DecodeTask(BaseTask):
+    def get_type(self) -> str:
+        return "base64_decode"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        try:
+            files_obj = param.inputs.get("files")
+            files: list[str] = files_obj if isinstance(files_obj, list) else []
+
+            if not files:
+                return TaskResult.failure("No files provided for base64 decoding")
+            if not param.workspace:
+                return TaskResult.failure("Workspace path is not provided")
+
+            logger.info("[base64_decode] Starting: %d files to decode", len(files))
+            t0 = time.time()
+            decoded_files: list[str] = []
+            for i, b64 in enumerate(files):
+                try:
+                    name = generate_unique_filename(".png")
+                    rel_path = os.path.join(param.step_id, name)
+                    abs_path = os.path.join(param.workspace, rel_path)
+                    base64_to_file(b64, abs_path)
+                    decoded_files.append(rel_path)
+                    logger.info("[base64_decode] Decoded file %d/%d -> %s (base64_len=%d)",
+                                 i + 1, len(files), rel_path, len(b64))
+                except Exception as e:
+                    logger.error("[base64_decode] Failed to decode file %d/%d: %s", i + 1, len(files), e)
+
+            if not decoded_files:
+                return TaskResult.failure("No files were successfully decoded")
+
+            logger.info("[base64_decode] Completed: %d/%d decoded in %.2fs",
+                         len(decoded_files), len(files), time.time() - t0)
+            return TaskResult.success({"files": decoded_files})
+        except Exception as e:
+            logger.error("[base64_decode] Error: %s", e, exc_info=True)
+            return TaskResult.failure(f"Base64 decoding failed: {e}")

app/service/task/impl/base64_encode_task.py

@@ -0,0 +1,53 @@
+import logging
+import os
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.file_util import base64_content
+
+logger = logging.getLogger(__name__)
+
+
+class Base64EncodeTask(BaseTask):
+    def get_type(self) -> str:
+        return "base64_encode"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        try:
+            files_obj = param.inputs.get("files")
+            files: list[str] = files_obj if isinstance(files_obj, list) else []
+
+            if not files:
+                return TaskResult.failure("No files provided for base64 encoding")
+            if not param.workspace:
+                return TaskResult.failure("Workspace path is not provided")
+
+            logger.info("[base64_encode] Starting: %d files to encode", len(files))
+            t0 = time.time()
+            encoded: list[str] = []
+            for i, fp in enumerate(files):
+                try:
+                    full_path = os.path.join(param.workspace, fp)
+                    if not os.path.isfile(full_path):
+                        logger.warning("[base64_encode] File %d/%d does not exist: %s", i + 1, len(files), full_path)
+                        continue
+                    file_size = os.path.getsize(full_path)
+                    b64 = base64_content(full_path)
+                    encoded.append(b64)
+                    logger.info("[base64_encode] Encoded file %d/%d: %s (file_size=%d, base64_len=%d)",
+                                 i + 1, len(files), fp, file_size, len(b64))
+                except Exception as e:
+                    logger.error("[base64_encode] Failed to encode file %d/%d %s: %s", i + 1, len(files), fp, e)
+
+            if not encoded:
+                return TaskResult.failure("No files were successfully encoded")
+
+            logger.info("[base64_encode] Completed: %d/%d encoded in %.2fs",
+                         len(encoded), len(files), time.time() - t0)
+            return TaskResult.success({"files": encoded})
+        except Exception as e:
+            logger.error("[base64_encode] Error: %s", e, exc_info=True)
+            return TaskResult.failure(f"Base64 encoding failed: {e}")

app/service/task/impl/file_download_task.py

@@ -0,0 +1,41 @@
+import logging
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.download_util import download_files
+
+logger = logging.getLogger(__name__)
+
+
+class FileDownloadTask(BaseTask):
+    def get_type(self) -> str:
+        return "file_download"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        try:
+            files_obj = param.inputs.get("files")
+            files: list[str] = files_obj if isinstance(files_obj, list) else []
+
+            if not files:
+                return TaskResult.failure("No files provided for download")
+            if not param.workspace:
+                return TaskResult.failure("Workspace path is not provided")
+
+            logger.info("[file_download] Starting: %d files to download, workspace=%s", len(files), param.workspace)
+            t0 = time.time()
+            downloaded = download_files(files, param.workspace, param.step_id)
+            elapsed = time.time() - t0
+
+            if not downloaded:
+                logger.warning("[file_download] No files were successfully downloaded in %.2fs", elapsed)
+                return TaskResult.failure("No files were successfully downloaded")
+
+            logger.info("[file_download] Completed: %d/%d files downloaded in %.2fs",
+                         len(downloaded), len(files), elapsed)
+            return TaskResult.success({"files": downloaded})
+        except Exception as e:
+            logger.error("[file_download] Error: %s", e, exc_info=True)
+            return TaskResult.failure(f"File download failed: {e}")

app/service/task/impl/file_upload_task.py

@@ -0,0 +1,132 @@
+import logging
+import os
+import time
+from datetime import datetime
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util import aliyun_util, google_storage_util
+from app.util.download_util import download_file
+from app.util.file_util import generate_unique_filename
+
+logger = logging.getLogger(__name__)
+
+
+class FileUploadTask(BaseTask):
+    def get_type(self) -> str:
+        return "file_upload"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        files_obj = param.inputs.get("files")
+        files: list[str] = files_obj if isinstance(files_obj, list) else []
+
+        try:
+            provider = str(param.inputs.get("provider", ""))
+            prefix = str(param.inputs.get("prefix", ""))
+            date_folder = param.inputs.get("date_folder")
+
+            logger.info("[file_upload] Starting: %d files, provider=%s, prefix=%s, date_folder=%s",
+                         len(files), provider, prefix, date_folder)
+
+            if not files:
+                return TaskResult.failure("No files to upload")
+            if provider not in ("aliyun", "google"):
+                return TaskResult.failure(f"Unsupported target provider: {provider}")
+            if not prefix or not prefix.strip():
+                return TaskResult.failure("Key prefix cannot be empty")
+            if not prefix.endswith("/"):
+                prefix += "/"
+
+            final_prefix = self._build_final_prefix(prefix, str(date_folder) if date_folder else None)
+            logger.info("[file_upload] Final prefix: %s", final_prefix)
+
+            t0 = time.time()
+            local_files = self._process_files(files, param.workspace, param.step_id)
+            logger.info("[file_upload] Resolved %d local files in %.2fs", len(local_files), time.time() - t0)
+
+            uploaded_urls: list[str] = []
+            upload_t0 = time.time()
+            if provider == "aliyun":
+                uploaded_urls = self._upload_to_aliyun(local_files, final_prefix)
+            elif provider == "google":
+                uploaded_urls = self._upload_to_google(local_files, final_prefix)
+
+            logger.info("[file_upload] Completed: %d/%d files uploaded to %s in %.2fs",
+                         len(uploaded_urls), len(local_files), provider, time.time() - upload_t0)
+            return TaskResult.success({"files": uploaded_urls})
+        except Exception as e:
+            logger.error("[file_upload] Execution failed: %s", e, exc_info=True)
+            return TaskResult.failure(f"File copy failed: {e}")
+
+    @staticmethod
+    def _build_final_prefix(prefix: str, date_folder: str | None) -> str:
+        if not date_folder:
+            return prefix
+        try:
+            date_path = datetime.now().strftime(date_folder)
+            if not date_path.endswith("/"):
+                date_path += "/"
+            return prefix + date_path
+        except Exception:
+            return prefix
+
+    @staticmethod
+    def _process_files(files: list[str], workspace: str, step_id: str) -> list[str]:
+        local_files: list[str] = []
+        for f in files:
+            try:
+                if _is_url(f):
+                    logger.info("[file_upload] Downloading URL: %s", f)
+                    local_path = download_file(f, workspace + "/" + step_id)
+                    if local_path:
+                        local_files.append(local_path)
+                else:
+                    local_files.append(workspace + "/" + f)
+            except Exception as e:
+                logger.error("[file_upload] Error processing file: %s, %s", f, e)
+        return local_files
+
+    @staticmethod
+    def _upload_to_aliyun(files: list[str], prefix: str) -> list[str]:
+        uploaded: list[str] = []
+        for path in files:
+            if not os.path.isfile(path):
+                logger.warning("[file_upload] File not found, skipping: %s", path)
+                continue
+            try:
+                unique_name = generate_unique_filename(os.path.basename(path))
+                key = prefix + unique_name
+                logger.info("[file_upload] Uploading to Aliyun OSS: %s -> %s", os.path.basename(path), key)
+                url = aliyun_util.upload_file(path, key)
+                if url:
+                    uploaded.append(url)
+            except Exception as e:
+                logger.error("[file_upload] Exception uploading file: %s, %s", path, e)
+        return uploaded
+
+    @staticmethod
+    def _upload_to_google(files: list[str], prefix: str) -> list[str]:
+        uploaded: list[str] = []
+        for path in files:
+            if not os.path.isfile(path):
+                logger.warning("[file_upload] File not found, skipping: %s", path)
+                continue
+            try:
+                unique_name = generate_unique_filename(os.path.basename(path))
+                key = prefix + unique_name
+                logger.info("[file_upload] Uploading to Google GCS: %s -> %s", os.path.basename(path), key)
+                url = google_storage_util.upload_file(path, key)
+                if url:
+                    uploaded.append(url)
+            except Exception as e:
+                logger.error("[file_upload] Exception uploading file: %s, %s", path, e)
+        return uploaded
+
+
+def _is_url(s: str) -> bool:
+    if not s:
+        return False
+    lower = s.strip().lower()
+    return any(lower.startswith(p) for p in ("http://", "https://", "ftp://", "s3://", "gs://"))

app/service/task/impl/google_veo3_task.py

@@ -0,0 +1,105 @@
+import logging
+import os
+import time
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.file_util import generate_unique_filename, write_bytes
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleVeo3Task(BaseTask):
+    def get_type(self) -> str:
+        return "google_veo3"
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "x-goog-api-key": settings.gemini_api_key,
+            "Content-Type": "application/json",
+        }
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        t0 = time.time()
+        try:
+            op_name = self._create_request(param.inputs)
+            logger.info("[google_veo3] Operation created: %s", op_name)
+
+            video_uri = self._poll_until_done(op_name)
+            if not video_uri:
+                return TaskResult.failure("Failed to get video download link")
+
+            logger.info("[google_veo3] Downloading video: %s", video_uri)
+            video_bytes = self._download_video(video_uri)
+            video_name = generate_unique_filename(".mp4")
+            video_rel = os.path.join(param.step_id, video_name)
+            write_bytes(os.path.join(param.workspace, video_rel), video_bytes)
+
+            logger.info("[google_veo3] Completed: video=%s, size=%d bytes, total_elapsed=%.2fs",
+                         video_rel, len(video_bytes), time.time() - t0)
+            return TaskResult.success({"videos": [video_rel]})
+        except Exception as e:
+            logger.error("[google_veo3] Execution failed in %.2fs: %s", time.time() - t0, e, exc_info=True)
+            return TaskResult.failure(f"GoogleVeo3Task execution failed: {e}")
+
+    def _create_request(self, inputs: dict) -> str:
+        endpoint = f"{settings.gemini_base_url}/models/{settings.gemini_veo3_model}:predictLongRunning"
+        logger.info("[google_veo3] Submitting generation request: POST %s", endpoint)
+        with logged_client(timeout=120) as client:
+            resp = client.post(endpoint, headers=self._headers(), json=inputs)
+            logger.info("[google_veo3] Submit response: status=%d", resp.status_code)
+            resp.raise_for_status()
+            data = resp.json()
+        op_name = data.get("name")
+        if not op_name:
+            raise RuntimeError(f"Operation name not found: {data}")
+        return op_name
+
+    def _poll_until_done(self, operation_name: str) -> str | None:
+        url = f"{settings.gemini_base_url}/{operation_name}"
+        max_retries = 60
+        poll_t0 = time.time()
+
+        for i in range(max_retries):
+            try:
+                with logged_client(timeout=60) as client:
+                    resp = client.get(url, headers=self._headers())
+                    if not resp.is_success:
+                        logger.warning("[google_veo3] Poll #%d: HTTP %d, elapsed=%.1fs",
+                                        i + 1, resp.status_code, time.time() - poll_t0)
+                        time.sleep(10)
+                        continue
+                    data = resp.json()
+            except Exception as e:
+                logger.warning("[google_veo3] Poll #%d: exception=%s, elapsed=%.1fs",
+                                i + 1, e, time.time() - poll_t0)
+                time.sleep(10)
+                continue
+
+            done = data.get("done")
+            logger.info("[google_veo3] Poll #%d: done=%s, elapsed=%.1fs",
+                         i + 1, done, time.time() - poll_t0)
+            if done is None:
+                time.sleep(10)
+                continue
+            if done:
+                try:
+                    samples = data["response"]["generateVideoResponse"]["generatedSamples"]
+                    return samples[0]["video"]["uri"]
+                except (KeyError, IndexError):
+                    raise RuntimeError("Task completed but video URI not found")
+            time.sleep(10)
+
+        raise RuntimeError("Polling timeout after %d attempts" % max_retries)
+
+    def _download_video(self, video_uri: str) -> bytes:
+        t0 = time.time()
+        with logged_client(timeout=120, follow_redirects=True) as client:
+            resp = client.get(video_uri, headers=self._headers())
+            resp.raise_for_status()
+            logger.info("[google_veo3] Video downloaded: %d bytes in %.2fs",
+                         len(resp.content), time.time() - t0)
+            return resp.content

app/service/task/impl/image_editor_task.py

@@ -0,0 +1,75 @@
+import logging
+import os
+import subprocess
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.file_util import copy_file
+from app.util.res_util import get_resources_path
+
+logger = logging.getLogger(__name__)
+
+
+class ImageEditorTask(BaseTask):
+    def get_type(self) -> str:
+        return "image_editor"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        images_obj = param.inputs.get("images")
+        images: list[str] = images_obj if isinstance(images_obj, list) else []
+
+        logger.info("[image_editor] Starting: %d images, resources=%s", len(images), param.resources)
+        t0 = time.time()
+
+        try:
+            src = os.path.join(get_resources_path(), "image_editor", "p2pconvertor")
+            dst = os.path.join(param.workspace, "p2pconvertor")
+            copy_file(src, dst)
+            os.chmod(dst, 0o755)
+            logger.info("[image_editor] Copied p2pconvertor to workspace")
+
+            output_path = f"{param.step_id}/output.jpg"
+            self._execute_p2p(images, param.resources, output_path, param.workspace)
+
+            out_file = os.path.join(param.workspace, output_path)
+            if os.path.isfile(out_file):
+                file_size = os.path.getsize(out_file)
+                logger.info("[image_editor] Completed: output=%s (%.1f KB) in %.2fs",
+                             output_path, file_size / 1024, time.time() - t0)
+            else:
+                logger.warning("[image_editor] Completed but output file not found: %s", output_path)
+
+            return TaskResult.success({"images": [output_path]})
+        except Exception as e:
+            logger.error("[image_editor] Failed in %.2fs: %s", time.time() - t0, e, exc_info=True)
+            return TaskResult.failure(str(e))
+
+    @staticmethod
+    def _execute_p2p(images: list[str], effect_dir: str, output_path: str, workspace: str):
+        cmd_parts = ["./p2pconvertor"]
+        for img in images:
+            cmd_parts.extend(["-i", img])
+        cmd_parts.extend(["-e", effect_dir, "-o", output_path])
+
+        cmd_str = " ".join(cmd_parts)
+        logger.info("[image_editor] Executing: %s  (cwd=%s)", cmd_str, workspace)
+
+        cmd_t0 = time.time()
+        result = subprocess.run(
+            ["sh", "-c", cmd_str],
+            cwd=workspace, capture_output=True, text=True,
+        )
+        cmd_elapsed = time.time() - cmd_t0
+
+        if result.stdout:
+            logger.info("[image_editor] stdout: %s", result.stdout.strip())
+        if result.stderr:
+            logger.warning("[image_editor] stderr: %s", result.stderr.strip())
+
+        if result.returncode != 0:
+            logger.error("[image_editor] p2pconvertor failed: exit_code=%d in %.2fs", result.returncode, cmd_elapsed)
+        else:
+            logger.info("[image_editor] p2pconvertor completed: exit_code=0 in %.2fs", cmd_elapsed)

app/service/task/impl/keling_image2video_task.py

@@ -0,0 +1,107 @@
+import logging
+import time
+
+import jwt
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class KelingImage2VideoTask(BaseTask):
+    def get_type(self) -> str:
+        return "keling_image2video"
+
+    @property
+    def execute_url(self) -> str:
+        return settings.keling_image2video_url
+
+    @property
+    def search_url(self) -> str:
+        return settings.keling_image2video_url
+
+    def _generate_jwt(self) -> str:
+        now = time.time()
+        payload = {
+            "iss": settings.keling_app_key,
+            "exp": int(now) + 1800,
+            "nbf": int(now) - 5,
+        }
+        return jwt.encode(payload, settings.keling_app_secret, algorithm="HS256")
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self._generate_jwt()}",
+        }
+
+    def _retry_execute(self, param: TaskParam, remaining: int = 30) -> dict | None:
+        logger.info("[%s] Submitting task: POST %s  (remaining_retries=%d)",
+                     self.get_type(), self.execute_url, remaining)
+        try:
+            with logged_client(timeout=60) as client:
+                resp = client.post(self.execute_url, headers=self._headers(), json=param.inputs)
+                logger.info("[%s] Submit response: status=%d", self.get_type(), resp.status_code)
+                bean = resp.json()
+        except Exception as e:
+            raise RuntimeError(f"Request failed: {e}")
+
+        code = bean.get("code", 0)
+        if code in (1302, 1303) and remaining > 0:
+            logger.warning("[%s] Rate limited (code=%d), retrying in 5s ... remaining=%d",
+                            self.get_type(), code, remaining - 1)
+            time.sleep(5)
+            return self._retry_execute(param, remaining - 1)
+        return bean
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        t0 = time.time()
+        bean = self._retry_execute(param)
+        if not bean:
+            return TaskResult.failure("submit task fail !")
+        if not bean.get("data"):
+            msg = bean.get("message", "unknown error")
+            logger.error("[%s] Submit failed: %s", self.get_type(), msg)
+            return TaskResult.failure(msg)
+
+        task_id = bean["data"].get("task_id")
+        logger.info("[%s] Task submitted, remote_id=%s", self.get_type(), task_id)
+
+        poll_count = 0
+        poll_t0 = time.time()
+        while True:
+            poll_count += 1
+            poll = self._poll(task_id)
+            poll_data = poll.get("data", {})
+            status = poll_data.get("task_status")
+            logger.info("[%s] Poll #%d: remote_id=%s, status=%s, elapsed=%.1fs",
+                         self.get_type(), poll_count, task_id, status, time.time() - poll_t0)
+
+            if status in ("submitted", "processing"):
+                time.sleep(3)
+            elif status == "succeed":
+                task_result = poll_data.get("task_result", {})
+                break
+            else:
+                logger.warning("[%s] Unexpected status '%s', treating as failure", self.get_type(), status)
+                time.sleep(3)
+                return TaskResult.failure(str(poll_data))
+
+        videos_data = task_result.get("videos")
+        if not videos_data:
+            return TaskResult.failure("No video results found")
+
+        videos = [v["url"] for v in videos_data if v.get("url", "").strip()]
+        logger.info("[%s] Completed: %d videos, total_elapsed=%.2fs",
+                     self.get_type(), len(videos), time.time() - t0)
+        return TaskResult.success({"videos": videos})
+
+    def _poll(self, task_id: str) -> dict:
+        url = f"{self.search_url}/{task_id}"
+        with logged_client(timeout=60) as client:
+            resp = client.get(url, headers=self._headers())
+            return resp.json()

app/service/task/impl/keling_video2effect_task.py

@@ -0,0 +1,19 @@
+import logging
+
+from app.config import settings
+from app.service.task.impl.keling_image2video_task import KelingImage2VideoTask
+
+logger = logging.getLogger(__name__)
+
+
+class KelingVideo2EffectTask(KelingImage2VideoTask):
+    def get_type(self) -> str:
+        return "keling_video2effect"
+
+    @property
+    def execute_url(self) -> str:
+        return settings.keling_video2effect_url
+
+    @property
+    def search_url(self) -> str:
+        return settings.keling_video2effect_url

app/service/task/impl/kontext_max_task.py

@@ -0,0 +1,15 @@
+import logging
+
+from app.config import settings
+from app.service.task.impl.kontext_pro_task import KontextProTask
+
+logger = logging.getLogger(__name__)
+
+
+class KontextMaxTask(KontextProTask):
+    def get_type(self) -> str:
+        return "kontext_max"
+
+    @property
+    def execute_url(self) -> str:
+        return settings.kontext_max_url

app/service/task/impl/kontext_pro_task.py

@@ -0,0 +1,85 @@
+import logging
+import time
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class KontextProTask(BaseTask):
+    def get_type(self) -> str:
+        return "kontext_pro"
+
+    @property
+    def execute_url(self) -> str:
+        return settings.kontext_pro_url
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "x-key": settings.kontext_api_key,
+        }
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        logger.info("[%s] Submitting task: POST %s", self.get_type(), self.execute_url)
+        t0 = time.time()
+        try:
+            with logged_client(timeout=60) as client:
+                resp = client.post(self.execute_url, headers=self._headers(), json=param.inputs)
+                elapsed = time.time() - t0
+                logger.info("[%s] Submit response: status=%d, elapsed=%.2fs",
+                            self.get_type(), resp.status_code, elapsed)
+                resp.raise_for_status()
+                bean = resp.json()
+        except Exception as e:
+            elapsed = time.time() - t0
+            logger.error("[%s] Submit failed in %.2fs: %s", self.get_type(), elapsed, e)
+            return TaskResult.failure(f"submit task fail: {e}")
+
+        if not bean:
+            return TaskResult.failure("submit task fail !")
+
+        task_id = bean.get("id")
+        polling_url = bean.get("polling_url")
+        logger.info("[%s] Task submitted, remote_id=%s, polling_url=%s",
+                     self.get_type(), task_id, polling_url)
+
+        poll_count = 0
+        poll_t0 = time.time()
+        while True:
+            poll_count += 1
+            poll = self._poll(polling_url)
+            status = poll.get("status")
+            logger.info("[%s] Poll #%d: remote_id=%s, status=%s, elapsed=%.1fs",
+                         self.get_type(), poll_count, task_id, status, time.time() - poll_t0)
+
+            if status == "Pending":
+                time.sleep(3)
+            elif status == "Ready":
+                result = poll.get("result", {})
+                break
+            else:
+                logger.warning("[%s] Unexpected status '%s', treating as failure", self.get_type(), status)
+                time.sleep(3)
+                return TaskResult.failure(str(poll))
+
+        if not result:
+            return TaskResult.failure("No result found")
+
+        images: list[str] = []
+        sample = result.get("sample")
+        if sample and sample.strip():
+            images.append(sample)
+
+        logger.info("[%s] Completed: %d images, total_elapsed=%.2fs",
+                     self.get_type(), len(images), time.time() - t0)
+        return TaskResult.success({"images": images})
+
+    def _poll(self, polling_url: str) -> dict:
+        with logged_client(timeout=60) as client:
+            resp = client.get(polling_url, headers=self._headers())
+            return resp.json()

app/service/task/impl/nano_banana_task.py

@@ -0,0 +1,55 @@
+import logging
+import time
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class NanoBananaTask(BaseTask):
+    def get_type(self) -> str:
+        return "nano_banana"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        endpoint = f"{settings.gemini_base_url}/models/{settings.gemini_nano_banana_model}:generateContent"
+        headers = {
+            "x-goog-api-key": settings.gemini_api_key,
+            "Content-Type": "application/json",
+        }
+
+        logger.info("[nano_banana] Calling Gemini API: POST %s", endpoint)
+        t0 = time.time()
+        try:
+            with logged_client(timeout=120) as client:
+                resp = client.post(endpoint, headers=headers, json=param.inputs)
+                elapsed = time.time() - t0
+                logger.info("[nano_banana] Response: status=%d, elapsed=%.2fs, body_size=%d",
+                            resp.status_code, elapsed, len(resp.content))
+                resp.raise_for_status()
+                data = resp.json()
+        except Exception as e:
+            elapsed = time.time() - t0
+            logger.error("[nano_banana] Request failed in %.2fs: %s", elapsed, e)
+            return TaskResult.failure(f"submit execute fail: {e}")
+
+        candidates = data.get("candidates")
+        if not candidates:
+            logger.warning("[nano_banana] No candidates in response")
+            return TaskResult.failure("submit execute fail !")
+
+        encoded_files: list[str] = []
+        parts = candidates[0].get("content", {}).get("parts", [])
+        for part in parts:
+            text = part.get("text")
+            if text:
+                logger.info("[nano_banana] Text part: %s", text[:200])
+            inline_data = part.get("inlineData")
+            if inline_data and inline_data.get("data"):
+                encoded_files.append(inline_data["data"])
+
+        logger.info("[nano_banana] Completed: %d images generated", len(encoded_files))
+        return TaskResult.success({"images": encoded_files})

app/service/task/impl/seedance_video_task.py

@@ -0,0 +1,89 @@
+import logging
+import time
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class SeedanceVideoTask(BaseTask):
+    def get_type(self) -> str:
+        return "seedance_video"
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {settings.seed_api_key}",
+        }
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        execute_url = settings.seedance_video_url
+        t0 = time.time()
+
+        logger.info("[seedance_video] Submitting task: POST %s", execute_url)
+        try:
+            with logged_client(timeout=60) as client:
+                resp = client.post(execute_url, headers=self._headers(), json=param.inputs)
+                logger.info("[seedance_video] Submit response: status=%d, elapsed=%.2fs",
+                            resp.status_code, time.time() - t0)
+                data = resp.json()
+            task_id = data.get("id")
+            if not task_id:
+                logger.error("[seedance_video] No task ID in response: %s", data)
+                return TaskResult.failure("Failed to submit task: no task ID received")
+            logger.info("[seedance_video] Task submitted, remote_id=%s", task_id)
+        except Exception as e:
+            logger.error("[seedance_video] Submit failed: %s", e)
+            return TaskResult.failure(f"Failed to submit task: {e}")
+
+        result = self._poll(task_id)
+        if result is None:
+            return TaskResult.failure("Failed to poll task status")
+
+        if result.get("status") != "succeeded":
+            return TaskResult.failure(f"Task failed with status: {result.get('status')}")
+
+        videos: list[str] = []
+        content = result.get("content", {})
+        video_url = content.get("video_url")
+        if video_url:
+            videos.append(video_url)
+
+        logger.info("[seedance_video] Completed: %d videos, total_elapsed=%.2fs",
+                     len(videos), time.time() - t0)
+        return TaskResult.success({"videos": videos})
+
+    def _poll(self, task_id: str) -> dict | None:
+        search_url = f"{settings.seedance_video_url}/{task_id}"
+        max_attempts = 30
+        poll_t0 = time.time()
+
+        for attempt in range(1, max_attempts + 1):
+            try:
+                with logged_client(timeout=60) as client:
+                    resp = client.get(search_url, headers=self._headers())
+                    data = resp.json()
+
+                status = data.get("status")
+                logger.info("[seedance_video] Poll #%d/%d: remote_id=%s, status=%s, elapsed=%.1fs",
+                             attempt, max_attempts, task_id, status, time.time() - poll_t0)
+
+                if status == "succeeded":
+                    return data
+                if status in ("failed", "cancelled"):
+                    logger.error("[seedance_video] Task %s terminated with status=%s", task_id, status)
+                    return None
+
+                if attempt < max_attempts:
+                    time.sleep(2)
+            except Exception as e:
+                logger.error("[seedance_video] Poll #%d error for %s: %s", attempt, task_id, e)
+                if attempt < max_attempts:
+                    time.sleep(2)
+
+        logger.warning("[seedance_video] Max polling attempts (%d) reached for %s", max_attempts, task_id)
+        return None

app/service/task/impl/seedream_image_task.py

@@ -0,0 +1,55 @@
+import logging
+import time
+
+from app.config import settings
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.http_util import logged_client
+
+logger = logging.getLogger(__name__)
+
+
+class SeedreamImageTask(BaseTask):
+    def get_type(self) -> str:
+        return "seedream_image"
+
+    def _headers(self) -> dict[str, str]:
+        return {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {settings.seed_api_key}",
+        }
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        url = settings.seedream_image_url
+        logger.info("[seedream_image] Calling API: POST %s", url)
+        t0 = time.time()
+
+        try:
+            with logged_client(timeout=60) as client:
+                resp = client.post(url, headers=self._headers(), json=param.inputs)
+                elapsed = time.time() - t0
+                logger.info("[seedream_image] Response: status=%d, elapsed=%.2fs",
+                            resp.status_code, elapsed)
+                data = resp.json()
+
+            if "error" in data:
+                err = data["error"]
+                code = err.get("code", "")
+                msg = err.get("message", "")
+                logger.error("[seedream_image] API error: code=%s, message=%s", code, msg)
+                return TaskResult.failure(f"API error [{code}]: {msg}")
+
+            images: list[str] = []
+            for item in data.get("data", []):
+                url_val = item.get("url")
+                if url_val:
+                    images.append(url_val)
+
+            logger.info("[seedream_image] Completed: model=%s, %d images, elapsed=%.2fs",
+                         data.get("model"), len(images), time.time() - t0)
+            return TaskResult.success({"images": images})
+        except Exception as e:
+            elapsed = time.time() - t0
+            logger.error("[seedream_image] Failed in %.2fs: %s", elapsed, e, exc_info=True)
+            return TaskResult.failure(f"Failed to execute task: {e}")

app/service/task/impl/smart_beauty_task.py

@@ -0,0 +1,125 @@
+import logging
+import os
+import subprocess
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.res_util import get_resources_path
+
+logger = logging.getLogger(__name__)
+
+
+class SmartBeautyTask(BaseTask):
+    def get_type(self) -> str:
+        return "smart_beauty"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        try:
+            images_obj = param.inputs.get("images")
+            images_in: list[str] = images_obj if isinstance(images_obj, list) else []
+
+            if not images_in:
+                return TaskResult.failure("No images provided for processing")
+
+            workspace = param.workspace
+            resources = param.resources
+            if not workspace or not os.path.isdir(workspace):
+                return TaskResult.failure("Workspace path is not provided")
+
+            logger.info("[smart_beauty] Starting: %d images, resources=%s", len(images_in), resources)
+            t0 = time.time()
+            images_out: list[str] = []
+            for i, inp in enumerate(images_in):
+                try:
+                    base_name = _get_base_name(inp)
+                    output_path = f"{param.step_id}/{base_name}_beauty.jpg"
+                    logger.info("[smart_beauty] Processing image %d/%d: %s -> %s",
+                                 i + 1, len(images_in), inp, output_path)
+                    if self._execute_beauty(workspace, inp, output_path, resources):
+                        images_out.append(output_path)
+                        logger.info("[smart_beauty] Image %d/%d processed successfully", i + 1, len(images_in))
+                    else:
+                        logger.warning("[smart_beauty] Image %d/%d processing failed", i + 1, len(images_in))
+                except Exception as e:
+                    logger.error("[smart_beauty] Error processing image %d/%d %s: %s",
+                                  i + 1, len(images_in), inp, e)
+
+            if not images_out:
+                return TaskResult.failure("No images were successfully processed")
+
+            logger.info("[smart_beauty] Completed: %d/%d images in %.2fs",
+                         len(images_out), len(images_in), time.time() - t0)
+            return TaskResult.success({"images": images_out})
+        except Exception as e:
+            logger.error("[smart_beauty] Error: %s", e, exc_info=True)
+            return TaskResult.failure(f"Smart beauty processing failed: {e}")
+
+    @staticmethod
+    def _execute_beauty(workspace: str, inp: str, output: str, config: str) -> bool:
+        try:
+            base_dir = os.path.join(get_resources_path(), "beautify")
+            executor_path = os.path.join(base_dir, "beautify")
+            lib_path = os.path.join(base_dir, "libs")
+            licence_path = os.path.join(base_dir, "res", "license.lic")
+            face_model = os.path.join(base_dir, "res", "Face_Video_Template_p_4.0.0.model")
+            hair_model = os.path.join(base_dir, "res", "Segment_Hair_p_4.4.0.model")
+
+            if not os.path.isfile(executor_path):
+                logger.error("[smart_beauty] Beauty program not found: %s", executor_path)
+                return False
+
+            cmd = [
+                executor_path,
+                f"-i={inp}",
+                f"-o={output}",
+                f"-p={config}",
+                f"-l={licence_path}",
+                f"-face={face_model}",
+                f"-hair={hair_model}",
+            ]
+
+            env = os.environ.copy()
+            ld = env.get("LD_LIBRARY_PATH", "")
+            env["LD_LIBRARY_PATH"] = f"{ld}:{lib_path}" if ld else lib_path
+
+            logger.info("[smart_beauty] Executing: %s", " ".join(cmd))
+            cmd_t0 = time.time()
+            result = subprocess.run(
+                cmd, cwd=workspace, env=env,
+                capture_output=True, text=True, timeout=60,
+            )
+            cmd_elapsed = time.time() - cmd_t0
+
+            if result.stdout:
+                logger.info("[smart_beauty] stdout: %s", result.stdout.strip())
+            if result.stderr:
+                logger.warning("[smart_beauty] stderr: %s", result.stderr.strip())
+
+            if result.returncode == 0:
+                out_file = os.path.join(workspace, output)
+                if os.path.isfile(out_file):
+                    file_size = os.path.getsize(out_file)
+                    logger.info("[smart_beauty] Output generated: %s (%.1f KB) in %.2fs",
+                                 output, file_size / 1024, cmd_elapsed)
+                    return True
+                logger.warning("[smart_beauty] Output file not generated: %s", output)
+                return False
+            else:
+                logger.error("[smart_beauty] Process failed: exit_code=%d in %.2fs",
+                              result.returncode, cmd_elapsed)
+                return False
+        except subprocess.TimeoutExpired:
+            logger.error("[smart_beauty] Process timeout (60s) for input: %s", inp)
+            return False
+        except Exception as e:
+            logger.error("[smart_beauty] Error executing command: %s", e)
+            return False
+
+
+def _get_base_name(path: str) -> str:
+    name = os.path.basename(path)
+    dot = name.rfind(".")
+    return name[:dot] if dot > 0 else name

app/service/task/impl/video_editor_task.py

@@ -0,0 +1,111 @@
+import json
+import logging
+import os
+import subprocess
+import time
+from typing import Any
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+from app.util.download_util import get_relative_path
+from app.util.file_util import copy_file, move_file
+from app.util.res_util import get_resources_path
+
+logger = logging.getLogger(__name__)
+
+
+class VideoEditorTask(BaseTask):
+    def get_type(self) -> str:
+        return "video_editor"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        images = param.inputs.get("images") if isinstance(param.inputs.get("images"), list) else []
+        videos = param.inputs.get("videos") if isinstance(param.inputs.get("videos"), list) else []
+
+        logger.info("[video_editor] Starting: %d images, %d videos, resources=%s",
+                     len(images), len(videos), param.resources)
+        t0 = time.time()
+
+        try:
+            src = os.path.join(get_resources_path(), "video_editor", "compile")
+            dst = os.path.join(param.workspace, "compile")
+            copy_file(src, dst)
+            os.chmod(dst, 0o755)
+            logger.info("[video_editor] Copied compile to workspace")
+
+            self._update_json(param.workspace, param.resources, videos, images)
+            logger.info("[video_editor] Updated JSON config with media paths")
+
+            self._execute_compile(param.workspace, param.resources)
+            output_path = self._move_output(param.workspace, param.resources, param.step_id)
+
+            logger.info("[video_editor] Completed: output=%s in %.2fs", output_path, time.time() - t0)
+            return TaskResult.success({"videos": [output_path]})
+        except Exception as e:
+            logger.error("[video_editor] Failed in %.2fs: %s", time.time() - t0, e, exc_info=True)
+            return TaskResult.failure(str(e))
+
+    @staticmethod
+    def _update_json(workspace: str, resources: str, videos: list, images: list):
+        json_path = os.path.join(workspace, resources)
+        with open(json_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        video_track = data.get("videoTrack", [])
+        vi, ii = 0, 0
+        for track in video_track:
+            t = track.get("type")
+            if t == "video" and vi < len(videos):
+                logger.info("[video_editor] Assigning video[%d] -> track path: %s", vi, videos[vi])
+                track["path"] = videos[vi]
+                vi += 1
+            elif t == "image" and ii < len(images):
+                logger.info("[video_editor] Assigning image[%d] -> track path: %s", ii, images[ii])
+                track["path"] = images[ii]
+                ii += 1
+
+        with open(json_path, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=2, ensure_ascii=False)
+
+    @staticmethod
+    def _execute_compile(workspace: str, resources: str):
+        cmd = f"./compile -m {resources}"
+        logger.info("[video_editor] Executing: %s  (cwd=%s)", cmd, workspace)
+
+        cmd_t0 = time.time()
+        result = subprocess.run(
+            ["sh", "-c", cmd], cwd=workspace, capture_output=True, text=True,
+        )
+        cmd_elapsed = time.time() - cmd_t0
+
+        if result.stdout:
+            logger.info("[video_editor] stdout: %s", result.stdout.strip())
+        if result.stderr:
+            logger.warning("[video_editor] stderr: %s", result.stderr.strip())
+
+        if result.returncode != 0:
+            logger.error("[video_editor] compile failed: exit_code=%d in %.2fs", result.returncode, cmd_elapsed)
+        else:
+            logger.info("[video_editor] compile completed: exit_code=0 in %.2fs", cmd_elapsed)
+
+    @staticmethod
+    def _move_output(workspace: str, resources: str, step_id: str) -> str | None:
+        json_path = os.path.join(workspace, resources)
+        with open(json_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        output_cfg = data.get("output", {})
+        output_path = output_cfg.get("path")
+        if not output_path:
+            logger.warning("[video_editor] No output path found in config")
+            return None
+
+        src = os.path.join(workspace, output_path)
+        dest_dir = os.path.join(workspace, step_id)
+        os.makedirs(dest_dir, exist_ok=True)
+        dest = os.path.join(dest_dir, os.path.basename(output_path))
+        move_file(src, dest)
+        rel_path = get_relative_path(dest, workspace)
+        logger.info("[video_editor] Output moved: %s -> %s", output_path, rel_path)
+        return rel_path

app/service/task/task_executor.py

@@ -0,0 +1,76 @@
+import logging
+from typing import Callable
+
+from app.service.task.base_task import BaseTask
+from app.service.task.task_param import TaskParam
+from app.service.task.task_result import TaskResult
+
+logger = logging.getLogger(__name__)
+
+
+class _NoopTask(BaseTask):
+    def get_type(self) -> str:
+        return "noop"
+
+    def execute(self, param: TaskParam) -> TaskResult:
+        logger.warning("[noop] No task implementation found for type=%s, returning empty success", param.type)
+        return TaskResult.success({})
+
+
+class TaskExecutor:
+    def __init__(self):
+        self._registry: dict[str, BaseTask] = {}
+
+    def register(self, type_name: str, task: BaseTask) -> "TaskExecutor":
+        self._registry[type_name] = task
+        return self
+
+    def resolve(self, type_name: str) -> BaseTask:
+        task = self._registry.get(type_name)
+        if task is None:
+            logger.warning("No registered task for type '%s', using NoopTask", type_name)
+            return _NoopTask()
+        return task
+
+    def registered_types(self) -> set[str]:
+        return set(self._registry.keys())
+
+    @classmethod
+    def create_default(cls) -> "TaskExecutor":
+        from app.service.task.impl.kontext_pro_task import KontextProTask
+        from app.service.task.impl.kontext_max_task import KontextMaxTask
+        from app.service.task.impl.keling_image2video_task import KelingImage2VideoTask
+        from app.service.task.impl.keling_video2effect_task import KelingVideo2EffectTask
+        from app.service.task.impl.smart_beauty_task import SmartBeautyTask
+        from app.service.task.impl.video_editor_task import VideoEditorTask
+        from app.service.task.impl.image_editor_task import ImageEditorTask
+        from app.service.task.impl.file_download_task import FileDownloadTask
+        from app.service.task.impl.file_upload_task import FileUploadTask
+        from app.service.task.impl.base64_decode_task import Base64DecodeTask
+        from app.service.task.impl.base64_encode_task import Base64EncodeTask
+        from app.service.task.impl.nano_banana_task import NanoBananaTask
+        from app.service.task.impl.google_veo3_task import GoogleVeo3Task
+        from app.service.task.impl.seedance_video_task import SeedanceVideoTask
+        from app.service.task.impl.seedream_image_task import SeedreamImageTask
+
+        executor = cls()
+        tasks: list[BaseTask] = [
+            KontextProTask(),
+            KontextMaxTask(),
+            KelingImage2VideoTask(),
+            KelingVideo2EffectTask(),
+            SmartBeautyTask(),
+            VideoEditorTask(),
+            ImageEditorTask(),
+            FileDownloadTask(),
+            FileUploadTask(),
+            Base64DecodeTask(),
+            Base64EncodeTask(),
+            NanoBananaTask(),
+            GoogleVeo3Task(),
+            SeedanceVideoTask(),
+            SeedreamImageTask(),
+        ]
+        for t in tasks:
+            executor.register(t.get_type(), t)
+        return executor

app/service/task/task_param.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class TaskParam:
+    workspace: str = ""
+    step_id: str = ""
+    type: str = ""
+    resources: str = ""
+    inputs: dict[str, Any] = field(default_factory=dict)

app/service/task/task_result.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class TaskResult:
+    _success: bool
+    outputs: dict[str, Any] = field(default_factory=dict)
+    error_message: str = ""
+
+    def is_success(self) -> bool:
+        return self._success
+
+    def is_failure(self) -> bool:
+        return not self._success
+
+    @classmethod
+    def success(cls, outputs: dict[str, Any] | None = None) -> "TaskResult":
+        return cls(_success=True, outputs=outputs or {})
+
+    @classmethod
+    def failure(cls, error_message: str) -> "TaskResult":
+        return cls(_success=False, error_message=error_message)

app/service/template/__init__.py

@@ -0,0 +1 @@
+