Metadata-Version: 2.5
Name: aigc-pipeline
Version: 0.1.15
Summary: AIGC 接口批量驱动 + 飞书审核工作流（图片生成 → 飞书推送 → 审核补生成 → 视频）
Project-URL: Homepage, https://github.com/your-username/aigc-pipeline
Project-URL: Repository, https://github.com/your-username/aigc-pipeline
Project-URL: Issues, https://github.com/your-username/aigc-pipeline/issues
Author-email: xuchaohui <hfyhui@126.com>
License-Expression: MIT
License-File: LICENSE
Keywords: aigc,feishu,image-generation,video-generation,workflow
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Multimedia :: Graphics
Classifier: Topic :: Utilities
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Provides-Extra: dev
Requires-Dist: build>=1.0; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Description-Content-Type: text/markdown

# aigc-pipeline

> AIGC 接口批量驱动 + 飞书审核工作流（图片生成 → 飞书推送 → 审核 → 视频 → 百度网盘）

按用户的 8 步流程串联：

1. 读 `generation_tasks.json`
2. 按 `image_prompt_b` + `all_reference_images_d` 生成图片
3. 按命名规则保存到本地（`{片段名}-{日期}-{署名}/{片段名}-{日期}-（N）.png`）
4. 推送到飞书群 `@` 审核员（interactive 卡片 + 文件名列表）
5. 审核员把不合格图片移到 `<out_dir>/rejected/` 子目录
6. 程序检测 `rejected/` 触发补生成（最多 N 轮），重推全量
7. 通过图片 + `video_prompt_f` 生成视频（一图一视频）
8. 视频按命名规则存储 + 推送飞书

---

## 目录

- [安装](#安装)
- [快速开始](#快速开始)
  - [1. 安装](#1-安装)
  - [2. 准备 `config.toml`](#2-准备-configtoml)
  - [3. 最小可运行示例](#3-最小可运行示例)
  - [4. 三种使用方式](#4-三种使用方式)
  - [5. 审批工作流（重点）](#5-审批工作流重点)
  - [6. 命名规范](#6-命名规范)
- [完整生命周期（Phase 4）](#完整生命周期phase-4)
- [状态机一览](#状态机一览)
- [决策路由](#决策路由)
- [项目结构](#项目结构)
- [开发](#开发)
- [License](#license)

---

## 安装

### 方式一：从 whl 安装（推荐集成方）

```bash
pip install aigc_pipeline-0.1.4-py3-none-any.whl
```

### 方式二：从源码开发安装（推荐开发方）

```bash
git clone <repo>
cd aigc-pipeline
pip install -e ".[dev]"
```

### 方式三：从 PyPI（发布后）

```bash
pip install aigc-pipeline
```

---

## 快速开始

### 1. 安装

参见上一节。

### 2. 准备 `config.toml`

最小可工作的配置（运行 `examples/` 或 README 章节时把下面的字段填好）：

```toml
[server]
base_url = "http://your-aigc-server"        # AIGC 服务地址

[auth]
username = "D-39JLlay"                       # 登录账号
password = "39JLlay"

[paths]
output_dir = "./outputs"                      # 产物输出根目录
db_path = "./aigc_pipeline.db"                # SQLite 状态库
token_file = "./token.txt"                    # 登录 token 缓存
attachments_dir = "./attachments"             # ad-hoc 单文件模式的素材目录

[task]
mode = "image"                                # image | video
model_image = "auto-image"
model_video = "auto-video"
aspect_ratio = "9:16"                          # 9:16 / 1:1 / 16:9 ...
image_size = "2K"                              # 1K / 2K / 4K
batch_mode = "manual"                          # manual | folder | table

[review]
review_timeout_hours = 72                      # 超时 → abandoned
max_retry = 3                                  # rejected 重生上限

[feishu]
webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
reviewer_user_ids = ["ou_xxx", "ou_yyy"]
dry_run = true                                 # true=只 log 不真发

[baidu]
access_token = ""                              # 接入时填
share_expires_days = 7

[poll]
interval_seconds = 5                           # 轮询 AIGC server 的间隔
timeout_seconds = 600                          # 单 batch 超时

[users]
# 账号→中文署名映射（命名规则依赖）
"D-39JLlay" = "姜"
"D-39haitong" = "haitong"
```

### 3. 最小可运行示例

#### 3.1 命令行（5 秒上手）

```bash
# 顺序模式（默认，单进程）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json

# 并发模式（多进程 × 线程，推荐 3000+ task/天）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json \
    --parallel --max-processes 8

# Dry-run（不真发飞书）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json --skip-feishu
```

#### 3.2 Python facade（嵌入业务代码，**推荐**）

```python
from aigc_pipeline import AIGCPipeline

pipe = AIGCPipeline.from_config("./config.toml")

# 跑批（生成 image + 写 state）
summary = pipe.run_from_json(
    json_path="./tasks.json",
    output_dir="./outputs",
    parallel=True,           # ← 并发模式（多进程）
    max_processes=8,
)

# 查产物
pending = pipe.get_pending_reviews()
for p in pending:
    print(f"待审批: {p['target_type']} {p['target_id']}  file={p['file_path']}")

# 审批通过
pipe.submit_review_by_token(
    review_token=pending[0]["review_token"],
    decision="approved",
    comment="构图可以",
)

# 触发视频生成（支持并发）
pipe.generate_pending_videos(parallel=True, max_processes=4)
```

### 4. 三种使用方式

#### 4.1 CLI 跑批

```bash
# 顺序模式（默认）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json

# 并发模式（推荐 3000+ task/天）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json \
    --parallel --max-processes 8

# 指定输出目录
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json \
    --output-dir ./my_out

# Dry-run（不真发飞书）
aigc-pipeline --config ./config.toml --tasks-json ./tasks.json --skip-feishu

# 跑单张图（ad-hoc）
aigc-pipeline --config ./config.toml --mode image
```

#### 4.2 Python facade（**推荐**）

```python
from aigc_pipeline import AIGCPipeline

pipe = AIGCPipeline.from_config("./config.toml")

# ─────────── 跑批 ──────────
pipe.run_from_json(
    json_path="./tasks.json",
    output_dir="./outputs",
    parallel=True,            # 多进程并发
    max_processes=8,
)

# ─────────── 视频生成（approved images → video） ──────────
pipe.generate_pending_videos(parallel=True, max_processes=4)

# ─────────── 百度网盘（approved videos → shared） ──────────
pipe.process_approved_videos()

# ─────────── 审批 ──────────
pending = pipe.get_pending_reviews()
for p in pending:
    pipe.submit_review_by_token(
        review_token=p["review_token"],
        decision="approved",
    )

# ─────────── 巡查超时 ──────────
abandoned = pipe.run_dispatcher()
# {abandoned_images, abandoned_videos, count}

# ─────────── 查 task 全状态 ──────────
status = pipe.get_task_status("rec_xxx")
# {task, images[{...image, videos}]}
```

#### 4.3 并发模式对比

| task 量 / 天 | 推荐模式 | 进程数 |
|---|---|---|
| < 3,000 | 顺序模式（默认） | — |
| 3,000 ~ 30,000 | **并发模式** | `--max-processes 8` |
| 30,000 ~ 100,000 | 并发 + 多机 | 16/机 |
| > 100,000 | 升级到 [Dramatiq + RabbitMQ](docs/ARCHITECTURE.md) | — |

```bash
# 硬上限 16 (image) / 8 (video)，可调
aigc-pipeline --parallel --max-processes 8 --tasks-json ./tasks.json
```

### 5. 审批工作流（重点）

整套工作流核心是 **state 驱动 + 外部程序审阅**：

```
 ┌─────────────────────────────────────────────────────┐
 │              state-driven 全自动流程                  │
 └─────────────────────────────────────────────────────┘
         │
         ▼
   run_from_json()                ← 你跑：批量生成 image
         │
         ▼  状态机自动推进
   image.pending_review            ← 待审批
         │
         ▼  get_pending_reviews()
   外部程序定时拉                    ← 你集成：UI/邮件/IM 推给客户
         │
         ▼  submit_review_by_token(token, "approved")
   image.approved → 自动触发 video 生成
         │
         ▼  generate_pending_videos(parallel=True)
   video.pending_review
         │
         ▼  submit_review_by_token(token, "approved")
   video.approved → 自动触发百度网盘
         │
         ▼  process_approved_videos()
   video.shared + share_url
         │
         ▼
   image/video/task → completed
```

#### 完整 8 步集成示例

```python
from aigc_pipeline import AIGCPipeline

pipe = AIGCPipeline.from_config("./config.toml")

# ──── Step 1: 跑批（生成 image + 写 state） ────
pipe.run_from_json(
    json_path="./tasks.json",
    parallel=True,          # ← 多进程并发
    max_processes=8,
)

# ──── Step 2: 定时轮询拉取待审批（cron 触发） ────
pending = pipe.get_pending_reviews()
# pending = [
#   {"target_type": "image", "target_id": "img_001",
#    "record_id": "rec_xxx", "review_token": "abcdef...",
#    "file_path": ".../玉湖公园-1-8.28-（1）.png",
#    "retry_count": 0},
#   {"target_type": "video", ...},
# ]

# ──── Step 3: 推给客户（你自己接 UI/邮件/IM） ────
for p in pending:
    notify_client(p)   # 占位：调你自己的通道

# ──── Step 4: 客户审完，外部程序把决策回传 ────
result = pipe.submit_review_by_token(
    review_token=pending[0]["review_token"],
    decision="approved",     # 或 "rejected"
    comment="构图可以",
)
# result: {image_id, decision, new_status, retry_count, next_action}

# ──── Step 5: 触发视频生成（支持并发） ────
pipe.generate_pending_videos(parallel=True, max_processes=4)

# ──── Step 6: 视频审批通过后调百度网盘 ────
result = pipe.process_approved_videos()
# 空壳（NotImplementedError）→ skipped；接入后 → shared + share_url

# ──── Step 7: 定时巡查超时（推荐每 10 分钟跑一次） ────
abandoned = pipe.run_dispatcher()
# {abandoned_images, abandoned_videos, count}

# ──── Step 8: 查询全状态 ────
status = pipe.get_task_status("rec_xxx")
# {task, images[{...image, videos}]}
```

#### 极简审批入口

**客户端只要拿到 token 就能审批**，无需知道 image_id / video_id：

```python
# 客户点 "通过" 后，外部程序调一行
pipe.submit_review_by_token(
    review_token="abcdef1234...",   # 客户系统从 URL/邮件中拿到
    decision="approved",
)
```

### 6. 命名规范

#### 日期格式

- `3.27`（月份 / 日期都不带前导 0）

#### 通用素材

| 类型 | 格式 | 示例 |
|---|---|---|
| 文件夹 | `{片段名}-{日期}-{负责同学}` | `安保人群-3.27-张三` |
| 文件 | `{片段名}-{日期}-（{序号}）.{ext}` | `安保人群-3.27-（1）.png` |

#### 门店素材

| 类型 | 格式 | 示例 |
|---|---|---|
| 文件夹 | `{片段名}-{门店编号}-{日期}-{负责同学}` | `安保人群-130014WL-3.27-张三` |
| 文件 | `{片段名}-{门店编号}-{日期}-（{序号}）.{ext}` | `安保人群-130014WL-3.27-（1）.png` |

#### 视频产物

视频目录 = 图片目录 + `_video` 后缀：

```
outputs/
├── 玉湖公园-1-8.28-姜/                # image 目录
│   ├── 玉湖公园-1-8.28-（1）.png
│   └── 玉湖公园-1-8.28-（2）.png
└── 玉湖公园-1-8.28-姜_video/         # video 目录（同一规则 + _video）
    ├── 玉湖公园-1-8.28-（1）.mp4
    └── 玉湖公园-1-8.28-（2）.mp4
```

#### 通用 vs 门店的判定

- 由 JSON 的 `category` 字段判断（`门店素材` / `门店` / `store` 任一关键词）
- 门店素材的 `segment_name` 第一段是门店编号（如 `130014WL-玉湖公园`）

#### 批量命名技巧（飞书审核员）

1. `Ctrl+A` 全选
2. `F2` 进入命名模式
3. 给其中一个素材命名：`安保人群-工厂直销店-3.27-`
4. 系统会按规则自动补全编号

---

## 完整生命周期（Phase 4）

下面是本包支持的端到端流程，以客户审批为主轴：

```
 +-------------------+   +--------------------+   +-------------------+
 | 1. 读 generation_ |   | 3. 客户审批         |   | 5. 视频生成       |
 |    tasks.json     |   |                    |   |                    |
 +-------------------+   +--------------------+   +-------------------+
        |                       |                      |
        v                       |                      v
 +-------------------+          |             +-------------------+
 | 2. 创建 image     |          |             | 6. 视频审批         |
 |   记录 (generating)|          |             |   (pending_review) |
 |    并发跑 core.run |          |             +-------------------+
 +-------------------+          |                      |
        |                      |                      v
        v                      |             +-------------------+
 +-------------------+          |             | 7. 百度网盘上传     |
 | image pending_    |          |             |   + create_share    |
 |   review          |          |             +-------------------+
 +-------------------+          |                      |
        |                      |                      v
        +----------------------+             +-------------------+
                                  |             |  8. video.shared   |
                                  +------------>+ + image.completed |
                                                 | + task.completed  |
                                                 +-------------------+
```

---

## 状态机一览

| 对象 | 状态转换 |
|---|---|
| **image** | generating → pending_review → (approved\|rejected) → (generating\[重试\]) \| abandoned |
| **video** | pending → generating → pending_review → (approved\|rejected) → (generating\[重试\]) \| abandoned → uploaded → shared |
| **task** | pending → generating → reviewing → completed \| abandoned |

---

## 决策路由

| 决策 | 路由 |
|---|---|
| image approved | → 触发 video 生成（调 `generate_pending_videos`） |
| image rejected | retry < 3 → 回到 generating 重生；>= 3 → abandoned |
| video approved | → 触发百度网盘上传（调 `process_approved_videos`） |
| video rejected | retry < 3 → 回到 generating（用**同 image**重生）；>= 3 → abandoned |
| pending 超时 | dispatcher 巡查 → abandoned |

---

## 项目结构

```
src/aigc_pipeline/
├── __init__.py          # 公开 API 出口
├── __main__.py          # python -m aigc_pipeline
├── cli.py               # CLI 入口（argparse）
├── config.py            # AppConfig + load_config (TOML)
├── core.py              # AIGC 底层：upload/submit/poll/download
├── aigc_client.py       # AIGCClient（统一 base_url + TokenAuth + 401/402 refresh）
├── naming.py            # 命名规则 + NamingCtx
├── feishu.py            # 飞书 webhook 推送
├── workflow.py          # 老工作流（DEPRECATED）+ write_pre/post_state
├── runner.py            # AIGCRunner：image/video 统一引擎
├── parallel.py          # 并发执行（run_tasks_parallel / run_videos_parallel）
├── dispatcher.py        # 超时巡查
├── pipeline.py          # AIGCPipeline（facade）
├── data/
│   └── state.py         # SQLite StateStore（tasks / images / videos）
└── baidu_uploader.py    # 百度网盘（空壳）
```

---

## 开发

```bash
# 安装 dev 依赖
pip install -e ".[dev]"

# 跑测试
pytest                                       # 当前 199 个测试

# 加新测试
tests/test_<feature>.py

# 构建 whl + sdist
python -m build

# 产出文件
ls dist/
# aigc_pipeline-0.1.4-py3-none-any.whl
# aigc_pipeline-0.1.4.tar.gz
```

### 调试 httpx 内部日志

默认 `httpx` logger 已被设为 WARNING（不打印 AIGC server IP）。调试时手动打开：

```python
import logging
logging.getLogger("httpx").setLevel(logging.INFO)
```

---

## License

MIT

---

## 进阶文档

- [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) — 架构改造设计文档（从单进程到 10万 task/天的演进路线）
