# imagelab-worker

**Repository Path**: image-lab/imagelab-worker

## Basic Information

- **Project Name**: imagelab-worker
- **Description**: imagelab-worker 是 ImageLab 图像处理实验平台 的 Python 图像处理 Worker。它负责接收图像任务、读取图片、执行处理流水线、输出结果，并返回结构化执行状态。
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-06-29
- **Last Updated**: 2026-06-29

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# ImageLab Worker

`imagelab-worker` 是 **ImageLab 图像处理实验平台** 的 Python 图像处理 Worker。它负责接收图像任务、读取图片、执行处理流水线、输出结果，并返回结构化执行状态。

当前仓库先聚焦 Worker 本地闭环，并已提供 Redis Stream 消费入口。它暂不直接依赖 Go API、PostgreSQL 或 MinIO；这样可以先把图像处理与任务消费能力做稳，再逐步接入完整平台。

## 项目定位

ImageLab 整体建议拆成三个模块：

```text
imagelab-web      Vue3 + TypeScript，负责上传、预览、参数配置、结果展示
imagelab-api      Go + Gin/Fiber，负责业务 API、数据库、任务创建、权限控制
imagelab-worker   Python，负责图像处理、OCR、识别、相似图算法
```

当前 Worker 的职责：

1. 校验任务参数。
2. 从存储中读取输入图片。
3. 顺序执行多个图像处理操作。
4. 将结果写回存储。
5. 解析输出图片元信息。
6. 返回任务状态和处理记录。

## 当前能力

### 已实现

- 本地文件输入 / 输出
- 任务 JSON 校验
- 图片元信息解析：宽、高、格式、mime、文件大小、SHA-256
- CLI 单任务执行
- Redis Stream 异步任务消费
- pytest 自动化测试
- 基础图像处理：
  - `resize`：缩放
  - `crop`：裁剪
  - `rotate`：旋转
  - `grayscale`：灰度化
  - `brightness`：亮度调整
  - `contrast`：对比度调整
  - `thumbnail`：生成缩略图
  - `watermark`：文字水印
  - `compress`：压缩质量参数
  - `convert`：格式转换，支持 `jpg/jpeg/png/webp`

### 暂未实现

- RabbitMQ 队列消费
- Redis pending 消息恢复 / dead-letter / retry
- MinIO / S3 对象存储
- OCR
- 二维码识别
- 人脸检测
- YOLO / CLIP
- pHash / dHash 相似图检索
- 批量任务

## 环境要求

- Python >= 3.10
- Pillow
- Pydantic
- Pytest
- Redis server，可选，仅运行 `consume` 模式或手动联调时需要

建议使用虚拟环境：

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

如果使用 Homebrew Python，系统可能禁止全局安装依赖，请务必使用 `.venv`。

## 运行测试

```bash
python -m pytest
```

## 运行示例任务

项目中已经准备了一张示例图片：

```text
examples/input.jpg
```

执行 resize + convert：

```bash
python -m app.main run examples/jobs/resize_convert.json
```

执行裁剪、亮度、对比度、水印、格式转换：

```bash
python -m app.main run examples/jobs/basic_operations.json
```

执行缩略图 + 灰度化：

```bash
python -m app.main run examples/jobs/thumbnail_grayscale.json
```

输出文件默认写入：

```text
output/
```

## Redis Stream 异步消费

除了 `run` 单任务模式，Worker 也支持常驻消费 Redis Stream：

```bash
python -m app.main consume
```

指定 consumer 名称：

```bash
python -m app.main consume --consumer worker-1
```

### 队列架构

```text
Go API / producer / redis-cli
   ↓
Redis Stream: imagelab:jobs
   ↓
Python Worker: python -m app.main consume
   ↓
Redis Stream: imagelab:job_results
   ↓
Go API / result updater
```

### Redis 配置

| 环境变量 | 默认值 | 说明 |
|---|---|---|
| `IMAGELAB_REDIS_URL` | `redis://localhost:6379/0` | Redis 连接地址 |
| `IMAGELAB_REDIS_JOBS_STREAM` | `imagelab:jobs` | 任务 Stream |
| `IMAGELAB_REDIS_RESULTS_STREAM` | `imagelab:job_results` | 结果 Stream |
| `IMAGELAB_REDIS_CONSUMER_GROUP` | `imagelab-workers` | Consumer group |
| `IMAGELAB_REDIS_CONSUMER_NAME` | `hostname-pid` | Consumer 名称 |
| `IMAGELAB_REDIS_BLOCK_MS` | `5000` | 阻塞读取超时时间 |

### 任务消息格式

向 `imagelab:jobs` 写入消息时，推荐字段：

| 字段 | 说明 |
|---|---|
| `job_id` | 任务 ID，便于日志和结果关联 |
| `job` | 完整 ImageJob JSON 字符串 |

示例：

```bash
redis-cli XADD imagelab:jobs '*' \
  job_id job_001 \
  job '{"job_id":"job_001","input":{"type":"local_file","path":"examples/input.jpg"},"operations":[{"type":"resize","width":800,"height":600},{"type":"convert","format":"webp","quality":85}],"output":{"type":"local_file","path":"output/job_001.webp"}}'
```

### 结果消息格式

Worker 会向 `imagelab:job_results` 写入：

| 字段 | 说明 |
|---|---|
| `job_id` | 任务 ID |
| `status` | `success` 或 `failed` |
| `source_stream` | 原任务 Stream |
| `source_message_id` | 原任务消息 ID |
| `result` | 完整 JobResult JSON 字符串 |

查看结果：

```bash
redis-cli XRANGE imagelab:job_results - +
```

### Redis 消费可靠性

当前 MVP 的处理顺序：

```text
XREADGROUP 读取任务
  -> JobHandler.handle(job)
  -> XADD 写入结果 Stream
  -> XACK 确认原消息
```

原则：

- 只有结果成功写入后才 `XACK`。
- 图像处理失败也会写入 `failed` result，然后 ack，避免业务失败任务反复阻塞队列。
- Redis 连接失败或结果写入失败时不会 ack，后续可通过 pending 机制恢复。

当前暂未实现：

- `XAUTOCLAIM` pending 消息恢复
- retry count / max attempts
- dead-letter stream
- worker heartbeat
- 幂等结果去重

## 任务格式

一个任务由四部分组成：

```json
{
  "job_id": "job_001",
  "input": {
    "type": "local_file",
    "path": "examples/input.jpg"
  },
  "operations": [],
  "output": {
    "type": "local_file",
    "path": "output/job_001.webp"
  }
}
```

字段说明：

| 字段 | 说明 |
|---|---|
| `job_id` | 任务唯一 ID，后续可对应数据库任务表 ID 或队列消息 ID |
| `input` | 输入图片位置，当前只支持 `local_file` |
| `operations` | 图像处理流水线，按数组顺序执行 |
| `output` | 输出图片位置，当前只支持 `local_file` |

## 操作参数说明

### resize

按指定尺寸缩放。默认保持比例。

```json
{
  "type": "resize",
  "width": 800,
  "height": 600,
  "keep_aspect_ratio": true
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `width` | int | 否 | 目标宽度 |
| `height` | int | 否 | 目标高度 |
| `keep_aspect_ratio` | bool | 否 | 是否保持原图比例，默认 `true` |

说明：

- `width` 和 `height` 至少传一个。
- `keep_aspect_ratio=true` 且宽高都传时，会缩放到不超过该边界的最大尺寸。

### crop

裁剪图片区域。

```json
{
  "type": "crop",
  "x": 100,
  "y": 60,
  "width": 900,
  "height": 600
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `x` | int | 是 | 裁剪区域左上角 x 坐标 |
| `y` | int | 是 | 裁剪区域左上角 y 坐标 |
| `width` | int | 是 | 裁剪宽度 |
| `height` | int | 是 | 裁剪高度 |

如果裁剪区域超过原图边界，任务会失败并返回错误信息。

### rotate

旋转图片。

```json
{
  "type": "rotate",
  "angle": 90,
  "expand": true,
  "fill_color": [255, 255, 255]
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `angle` | float | 是 | 顺时针旋转角度 |
| `expand` | bool | 否 | 是否扩展画布避免裁切，默认 `true` |
| `fill_color` | RGB array | 否 | 旋转后空白区域填充色，默认白色 |

### grayscale

灰度化。

```json
{
  "type": "grayscale"
}
```

### brightness

调整亮度。

```json
{
  "type": "brightness",
  "factor": 1.2
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `factor` | float | 是 | 亮度倍率，`1.0` 表示不变，小于 1 变暗，大于 1 变亮 |

### contrast

调整对比度。

```json
{
  "type": "contrast",
  "factor": 1.1
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `factor` | float | 是 | 对比度倍率，`1.0` 表示不变 |

### thumbnail

生成缩略图。保持比例，结果会被限制在指定矩形内。

```json
{
  "type": "thumbnail",
  "width": 320,
  "height": 320
}
```

### watermark

添加文字水印。

```json
{
  "type": "watermark",
  "text": "ImageLab",
  "position": "bottom_right",
  "opacity": 180,
  "font_size": 36,
  "margin": 32,
  "fill_color": [255, 255, 255]
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `text` | string | 是 | 水印文字 |
| `position` | string | 否 | `top_left` / `top_right` / `bottom_left` / `bottom_right` / `center` |
| `opacity` | int | 否 | 透明度，范围 `0-255`，默认 `160` |
| `font_size` | int | 否 | 字号，默认 `32` |
| `margin` | int | 否 | 边距，默认 `24` |
| `fill_color` | RGB array | 否 | 字体颜色，默认白色 |

### compress

设置输出质量。对 `jpeg` 和 `webp` 生效。

```json
{
  "type": "compress",
  "quality": 80
}
```

### convert

转换输出格式。

```json
{
  "type": "convert",
  "format": "webp",
  "quality": 85
}
```

参数：

| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| `format` | string | 是 | `jpg` / `jpeg` / `png` / `webp` |
| `quality` | int | 否 | 输出质量，范围 `1-100`，默认 `85` |

## 输出结果

成功示例：

```json
{
  "job_id": "job_001",
  "status": "success",
  "output_path": "output/job_001.webp",
  "metadata": {
    "width": 800,
    "height": 600,
    "format": "WEBP",
    "mime_type": "image/webp",
    "file_size": 950,
    "sha256": "..."
  },
  "records": [
    {
      "operation": "resize",
      "status": "success",
      "params": {
        "type": "resize",
        "width": 800,
        "height": 600,
        "keep_aspect_ratio": true
      }
    }
  ]
}
```

失败示例：

```json
{
  "job_id": "job_001",
  "status": "failed",
  "records": [
    {
      "operation": "crop",
      "status": "failed",
      "params": {
        "type": "crop",
        "x": 1000,
        "y": 1000,
        "width": 900,
        "height": 600
      },
      "error_message": "crop area exceeds image bounds"
    }
  ],
  "error_message": "crop area exceeds image bounds"
}
```

## 模块设计

### JobHandler

位置：`app/worker/job_handler.py`

Interface：

```python
handle(job: ImageJob | dict) -> JobResult
```

职责：

- 校验任务
- 读取输入图片
- 调用 `ImageProcessor`
- 写出结果
- 解析输出元信息
- 返回结构化结果

### ImageProcessor

位置：`app/processors/image_processor.py`

Interface：

```python
process(image_data: bytes, operations: list[ImageOperation]) -> bytes
```

职责：

- 顺序执行图像处理流水线
- 隐藏 Pillow 具体实现
- 统一编码输出结果

### LocalStorage

位置：`app/storage/local_storage.py`

Interface：

```python
read(path: str) -> bytes
write(path: str, data: bytes) -> str
```

职责：

- 从本地路径读取图片
- 将结果写入本地路径
- 限制访问范围不能逃逸出项目根目录

后续可以增加 `MinioStorage`，保持 `JobHandler` 不变。

### inspect_image

位置：`app/utils/image_metadata.py`

Interface：

```python
inspect_image(data: bytes) -> ImageMetadata
```

职责：

- 解析图片宽高、格式、mime
- 计算文件大小
- 计算 SHA-256 hash

## 目录结构

```text
imagelab-worker
├── app
│   ├── main.py
│   ├── config.py
│   ├── models.py
│   ├── processors
│   │   └── image_processor.py
│   ├── storage
│   │   └── local_storage.py
│   ├── utils
│   │   └── image_metadata.py
│   └── worker
│       ├── job_handler.py
│       └── redis_consumer.py
├── examples
│   ├── input.jpg
│   └── jobs
├── tests
├── README.md
├── requirements.txt
└── pyproject.toml
```

## 平台整体实施路线

### 阶段 1：Worker 本地闭环，已完成

- 任务 JSON
- 本地图片输入输出
- 基础图像处理
- 元信息解析
- CLI
- 测试

### 阶段 2：补齐更多传统图像处理

候选能力：

- 图片格式参数细化
- PNG 压缩
- WebP lossless
- 图片 EXIF 方向修正
- 图片尺寸限制与安全校验
- 多水印样式
- 图片处理性能基准

### 阶段 3：接入任务队列，Redis Stream MVP 已完成

当前已实现：

```text
Go API / producer -> Redis Stream -> Python Worker -> Result Stream
```

已具备：

- Redis Stream consumer
- Consumer group 自动创建
- 任务结果写回 result stream
- 业务失败结果回传
- CLI `consume` 模式

后续增强：

- pending 消息恢复
- 失败重试
- dead-letter stream
- 幂等 job_id
- worker heartbeat

### 阶段 4：OCR 与识别

建议顺序：

1. Tesseract / EasyOCR OCR
2. 二维码识别
3. OpenCV 人脸检测
4. YOLO / ONNX Runtime 物体检测
5. CLIP 标签生成

### 阶段 5：相似图检索

建议顺序：

1. aHash / dHash / pHash
2. Hamming distance 近似重复判断
3. Go API 保存 hash
4. CLIP 向量
5. PostgreSQL + pgvector 检索

### 阶段 6：批量任务

建议顺序：

1. 单任务稳定后封装 batch job
2. 子任务独立失败
3. 进度统计
4. 失败重试
5. 前端展示进度

## 下一步建议

当前基础处理能力已经够支撑前端 MVP。下一步建议优先做：

1. `imagelab-api`：Go API，图片上传、图片列表、创建处理任务、查询任务结果。
2. Redis Stream 增强：pending 消息恢复、失败重试、dead-letter、worker heartbeat。
3. `imagelab-web`：上传图片、配置处理参数、查看输出结果。