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. 安装
  • 2. 准备 config.toml
  • 3. 最小可运行示例
  • 4. 三种使用方式
  • 5. 审批工作流（重点）
  • 6. 命名规范
• 完整生命周期（Phase 4）
• 状态机一览
• 决策路由
• 项目结构
• 开发
• License

安装

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

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

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

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

方式三：从 PyPI（发布后）

pip install aigc-pipeline

快速开始

1. 安装

参见上一节。

2. 准备 config.toml

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

[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 秒上手）

# 顺序模式（默认，单进程）
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（嵌入业务代码，推荐）

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 跑批

# 顺序模式（默认）
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（推荐）

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 并发模式对比

# 硬上限 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 步集成示例

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：

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

6. 命名规范

日期格式

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

通用素材

门店素材

视频产物

视频目录 = 图片目录 + _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  |
                                                 +-------------------+

状态机一览

决策路由

项目结构

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    # 百度网盘（空壳）

开发

# 安装 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）。调试时手动打开：

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

License

MIT

进阶文档

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