Metadata-Version: 2.5
Name: aigc-pipeline
Version: 0.1.4
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. 视频按命名规则存储 + 推送飞书

---

## 安装

### 方式一：从源码开发安装

```bash
git clone <repo>
cd aigc-pipeline
pip install -e .
```

### 方式二：从 whl 安装

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

---

## 快速开始

### 方式一：CLI 命令行（最直接）

```bash
# 准备：config.toml + generation_tasks.json 已存在

# 1. 跑指定 JSON 的完整流程（图片+飞书审核+视频+推送）
aigc-pipeline --tasks-json /path/to/generation_tasks.json

# 2. 或者用模块方式（同上）
python -m aigc_pipeline --tasks-json /path/to/generation_tasks.json

# 3. 第一次接入推荐先 dry-run（不真发飞书）
aigc-pipeline --tasks-json /path/to/generation_tasks.json --skip-feishu

# 4. 指定输出目录（CLI 默认从 config.toml 读）
aigc-pipeline --tasks-json /path/to/tasks.json  # 输出到 config.toml [paths].output_dir
```

### 方式二：Python import——推荐（Facade API）

AIGCPipeline 隐藏了 base_url / HTTP 细节，调用者只关心任务内容。

#### 1. 跑指定 JSON + 默认输出目录（顺序模式）

```python
from pathlib import Path
from aigc_pipeline import AIGCPipeline

pipeline = AIGCPipeline.from_config("./config.toml")
summary = pipeline.run_from_json("./tasks.json")
print(summary)
```

#### 2. 跑指定 JSON + 指定输出目录（顺序模式）

```python
from aigc_pipeline import AIGCPipeline

pipeline = AIGCPipeline.from_config("./config.toml")
summary = pipeline.run_from_json(
    json_path="./tasks.json",
    output_dir="./my_outputs",      # 覆盖 config.toml 里的 [paths].output_dir
)
```

#### 3. 跑指定 JSON + **并发模式**（高吞吐）

```python
from aigc_pipeline import AIGCPipeline

pipeline = AIGCPipeline.from_config("./config.toml")
summary = pipeline.run_from_json(
    json_path="./tasks.json",
    output_dir="./my_outputs",
    parallel=True,                  # 走方案 B：多进程并发 tasks
    max_processes=8,                # 进程数上限（None = 自动 = cpu_count）
)
print(summary)
# 内部走 parallel.run_tasks_parallel()：ProcessPoolExecutor + 内嵌 ThreadPool(8)
```

#### 4. 代码里动态指定 tasks_json + output_dir

```python
pipeline = AIGCPipeline.from_config("./config.toml")
pipeline.tasks_json_path = "./tasks_v2.json"   # 设属性
pipeline.output_dir = "./out_v2"              # setter
summary = pipeline.run_from_json(
    pipeline.tasks_json_path,
    parallel=True,
    max_processes=8,
)
```

#### 5. 只生成单张 / 多张图片

```python
paths = pipeline.generate_image(
    prompt="让门店表现廉价感",
    reference_paths=["./ref1.png", "./ref2.png"],
    quantity=3,                                  # 生成 3 张
    segment_name="玉湖公园-1",                    # 用于命名
)
print(paths)  # [Path('outputs/玉湖公园-1-8.23-（1）.png'), ...]
```

#### 6. 从一张图生成视频

```python
video = pipeline.generate_video(
    image_path="./output/玉湖公园-1-8.23-（1）.png",
    prompt="门店表现出城市烟火气",
)
print(video)
```

### 方式三：Python import——进阶（直接用底层）

适用于需要完全控制 AIGC 调用的场景（能看到 base_url，自行管理 HTTP 细节）。

#### 1. 跑完整流程 + 自定义 output_dir / tasks_json

```python
from pathlib import Path
from aigc_pipeline import AppConfig, load_config, log_loaded_config
from aigc_pipeline.core import load_tasks, run
from aigc_pipeline.feishu import FeishuBot
from aigc_pipeline.workflow import WorkflowContext, run_full_workflow
from aigc_pipeline.naming import today_str

cfg = load_config(Path("./config.toml"))
log_loaded_config(cfg)

# 代码中覆盖（不写进 config.toml）
cfg.batch.tasks_json_path = "./tasks.json"
cfg.paths.output_dir = "./my_outputs"   # 修改输出目录

bot = FeishuBot(
    webhook_url=cfg.feishu.webhook_url,
    reviewer_user_ids=list(cfg.feishu.reviewer_user_ids),
    dry_run=cfg.feishu.dry_run,
)

ctx = WorkflowContext(
    cfg=cfg,
    bot=bot,
    user_mapping=dict(cfg.users),
    output_root=Path(cfg.paths.output_dir),
    date_str=today_str(),
)
summary = run_full_workflow(ctx, Path("./tasks.json"))
print(summary)
```

#### 2. 编程式精细控制

```python
from dataclasses import replace
cfg = load_config(Path("./config.toml"))
tasks = load_tasks(Path("./tasks.json"))

for task in tasks:
    if task.record_id != "recvsxt4rtJ4Wd":
        continue
    for template in task.templates:
        proxy_task = replace(task, total_quantity=1)  # 临时 1 张
        result = run(cfg, proxy_task, template)
```

#### 3. 只用底层 AIGC 接口（不跑工作流）

```python
from pathlib import Path
from aigc_pipeline.core import (
    TokenManager, upload_one_reference,
    step_submit_job_multi, step_poll_until_done, step_download_assets,
    make_client_job_id,
)
import httpx

cfg = load_config(Path("./config.toml"))
tm = TokenManager(
    token_file=Path(cfg.paths.token_file),
    username=cfg.auth.username,
    password=cfg.auth.password,
    base_url=cfg.server.base_url,
)
token = tm.get()

with httpx.Client(timeout=cfg.server.timeout_seconds) as client:
    ref_url, ref_name = upload_one_reference(client, token, Path("./ref.png"), cfg)
    submit = step_submit_job_multi(
        client, token, [ref_url], [ref_name],
        prompt="让门店表现廉价感",
        client_job_id=make_client_job_id(1700000000000, 0),
        cfg=cfg,
    )
    finished = step_poll_until_done(client, token, submit["id"], cfg)
    saved = step_download_assets(finished, Path("./out/"))
    print("saved:", saved)
```

### 方式四：并发模式（方案 B）

适用于 **短期3-5万 task/天** 或多 task 同时处理场景：
- **进程级**：多进程并发跑 tasks（`ProcessPoolExecutor`）
- **线程级**：每个 task 内 quantity 个 jobs 并发（`ThreadPoolExecutor(8)`，已内嵌在 `core.run`）

#### 1. CLI 并发跑指定 JSON

```bash
# 默认进程数 = min(len(tasks), cpu_count)
aigc-pipeline --parallel --tasks-json /path/to/tasks.json

# 指定最大进程数（不超过 cpu_count，硬上限 16）
aigc-pipeline --parallel --max-processes 8 --tasks-json /path/to/tasks.json

# 并发 + dry-run + 单条 task
aigc-pipeline --parallel --skip-feishu --record-id recvsxt4rtJ4Wd
```

#### 2. Python import 并发模式

```python
from pathlib import Path
from aigc_pipeline import parallel, core

cfg = core.load_config(Path("./config.toml"))
tasks = core.load_tasks(Path("./path/to/tasks.json"))

# 直接调用底层 API
summaries = parallel.run_tasks_parallel(
    tasks=tasks,
    cfg=cfg,
    config_path=Path("./config.toml"),  # 传给 worker 重建
    max_processes=8,
)
```

#### 3. 适用场景对照

| 场景 | 推荐 |
|---|---|
| <100 task / 一次性脚本 | **顺序模式**（默认） |
| 100-3000 task / 每天 | **顺序模式 + 多脚本** |
| 3000-30000 task / 每天 | **并发模式**（方案 B） |
| 30000-100000 task / 每天 | **并发模式 + 多机** |
| >10万 task / 每天 | Dramatiq + RabbitMQ（见 [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md)） |

#### 4. 并发模式的内部机制

```
 Main Process
 ├─预登录拿 token (一次)
 ├─ProcessPoolExecutor (max_workers=N)
 │ ├─Worker #1 (pid=1001)
 │ │ ├─ 写私有 token.worker-1001.txt
 │ │ ├─ load_config 重建 cfg
 │ │ ├─ 独立 logger 文件
 │ │ ├─ 接收 task_dict
 │ │ └─ core.run(cfg, task, template)
 │ │    └─ ThreadPoolExecutor(8) 并发 quantity 个 jobs
 │ └─Worker #2 (pid=1002) ...
 └─as_completed() 收集结果
```

**重要约束**：
- 不含审核循环（不扫描 `rejected/`）—— 并发模式只做"批量处理"
- 输出目录从 `cfg.paths.output_dir` 读，要改可以编辑 config.toml 或代码设 `cfg.paths.output_dir`
- 产物目录命名仍按 `naming.py` 规则（`{片段名}-{日期}-{署名}/`）

---

## CLI 参数

| 参数 | 作用 |
|---|---|
| `--config PATH` | 配置文件路径（TOML），默认 `./config.toml` |
| `--tasks-json PATH` | 批量任务 JSON 路径，覆盖 config |
| `--skip-video` | 跳过视频生成阶段 |
| `--skip-feishu` | 不实际推送飞书（dry_run=true） |
| `--max-rounds N` | 最多几轮“审核-补生成”循环（默认 5） |
| `--wait-seconds N` | 推送后阻塞 N 秒让审核员操作（默认 0） |
| `--record-id ID` | 只跑指定 `record_id` 的 task |
| `--mode image\|video` | 覆盖 `cfg.task.mode` |
| `--quantity-mode loop\|single` | 覆盖 `cfg.batch.quantity_mode` |
| `--parallel` | **启用并发模式（方案 B）**：多进程跑 tasks + 内置 ThreadPool(8) 并发 jobs |
| `--max-processes N` | 并发进程数上限（默认 = min(len(tasks), cpu_count)，硬上限 16） |

---

## 对外集成：脱敏策略

本包采用 **脱敏 facade** 设计：集成方调 `AIGCPipeline` 看不到底层细节。

| 暴露面 | 状态 |
|---|---|
| `from aigc_pipeline import AIGCPipeline` | ✅ 默认导出 |
| `AIGCPipeline.run_from_json(...)` | ✅ 公开方法 |
| `AIGCPipeline.generate_image(...)` | ✅ 公开方法 |
| `AIGCPipeline.output_dir` 属性 | ✅ 可读可写 |
| `AIGCPipeline.tasks_json_path` 属性 | ✅ 可读可写 |
| `base_url` / `cfg` / `httpx.Client` | ❌ 不暴露 |
| `core` / `workflow` / `feishu` 等内部模块 | 需明确 import 才能用 |

**日志脱敏（默认开启）**：
```
原始 URL: POST http://14.103.124.30/api/auth/login
脱敏后:   POST /api/auth/login              ← host 完全消失
Authorization: Bearer abc123def
       → Bearer ********                  ← token 被 mask
```

脱敏**仅限日志输出**，实际发到服务端的 URL 仍是完整地址（保证功能）。

高级用户如需直接调底层（明确知道自己在做什么）：

```python
from aigc_pipeline.config import AppConfig, load_config
from aigc_pipeline.core import run, load_tasks, make_logged_client
from aigc_pipeline.feishu import FeishuBot
from aigc_pipeline.workflow import WorkflowContext, run_full_workflow
```

---

## 典型场景

### 1. 第一次接入：dry-run 测通流程

```bash
aigc-pipeline --tasks-json /path/to/generation_tasks.json --skip-feishu
```

不真发飞书，跑通图片+审核+视频，看日志和本地产物。

### 2. 给审核员时间标记不合格图

```bash
aigc-pipeline --tasks-json /path/to/generation_tasks.json --wait-seconds 60
```

推送后阻塞 60 秒，期间审核员把不合格图移到 `<output_dir>/<片段名>/rejected/`。程序自动检测并补生成。

### 3. 只生成图片，不生成视频

```bash
aigc-pipeline --skip-video
```

用于审图阶段快速迭代。

### 4. 切换视频模式

在 `config.toml [task]` 里设 `mode = "video"`，或 CLI 一次性覆盖：

```bash
aigc-pipeline --mode video
```

### 5. 调试单条 task

```bash
aigc-pipeline --record-id recvsxt4rtJ4Wd --skip-feishu
```

只跑指定 record_id，节省时间。

### 6. 嵌入你的业务脚本

```python
import json
from pathlib import Path
from aigc_pipeline import AppConfig, load_config, load_tasks, run
from aigc_pipeline.workflow import run_full_workflow, WorkflowContext
from aigc_pipeline.feishu import FeishuBot
from aigc_pipeline.naming import today_str

cfg = load_config(Path("./config.toml"))
bot = FeishuBot(
    webhook_url=cfg.feishu.webhook_url,
    reviewer_user_ids=list(cfg.feishu.reviewer_user_ids),
    dry_run=cfg.feishu.dry_run,
)
ctx = WorkflowContext(
    cfg=cfg,
    bot=bot,
    user_mapping=dict(cfg.users),
    output_root=Path(cfg.paths.output_dir),
    date_str=today_str(),
)
summary = run_full_workflow(ctx, Path("./your_tasks.json"))
with open("summary.json", "w", encoding="utf-8") as f:
    json.dump(summary, f, ensure_ascii=False, indent=2)
```

### 7. 并发批量跑（高吞吐场景）

```bash
# 启动 8 worker 进程跑一批 task
aigc-pipeline --parallel --max-processes 8 --tasks-json /path/to/big_tasks.json

# 跳过飞书推送（防止误推）
aigc-pipeline --parallel --skip-feishu --tasks-json /path/to/tasks.json

# 指定输出目录：CLI 参数或环境变量（当前版本需改 config.toml [paths].output_dir）
```

Python 嵌入：

```python
from pathlib import Path
from aigc_pipeline import parallel, core

cfg = core.load_config(Path("./config.toml"))
tasks = core.load_tasks(Path("./tasks.json"))

# 修改输出目录（运行时覆盖）
cfg.paths.output_dir = "/data/aigc_outputs"

summaries = parallel.run_tasks_parallel(
    tasks=tasks,
    cfg=cfg,
    config_path=Path("./config.toml"),
    max_processes=8,
)

# 汇总
ok = sum(1 for s in summaries if s.get("run_status", "ok") == "ok")
err = sum(1 for s in summaries if s.get("run_status") == "error")
print(f"ok={ok}  err={err}")
```

### 8. 指定输出目录的三种方式

| 场景 | 做法 |
|---|---|
| **临时改单次** | CLI：当前需改 `config.toml`；Python：`cfg.paths.output_dir = "/path"` |
| **永久改** | 改 `config.toml [paths].output_dir` |
| **AIGCPipeline facade** | `pipeline.output_dir = "/path"` (setter 已实现) |

> 注意：当前 `--parallel` CLI 不接 `--output-dir` 参数（如需可后续加）。

---

## 配置文件（`config.toml`）

```toml
[server]
base_url = "http://14.103.124.30"

[auth]
username = "D-39JLlay"
password = "39JLlay"

[task]
mode = "image"            # image | video
prompt = "让门店表现廉价感"
model_image = "auto-image"
model_video = "auto-video"

[batch]
tasks_json_path = "/path/to/generation_tasks.json"
quantity_mode = "loop"    # loop | single

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

[review]
rejected_subdir = "rejected"
wait_seconds = 0          # 推送后阻塞秒数
```

任何字段缺失会回退到代码默认值。详见 `config.toml`。

---

## 用户账号映射（`config.toml` 的 `[users]` section）

```toml
[users]
"D-39JLlay": "张三"
"D-99xxx": "李四"
```

把 `generation_tasks.json` 的 `assignee` 字段映射到中文署名（用于产物命名）。

**fallback 规则**：如果账号未在 `[users]` 中，自动提取账号字符串中的中文字符（如 `D-39姜` → `姜`）。

---

## 命名规则

| 类型 | 文件夹 | 文件 |
|---|---|---|
| 通用素材 | `{片段名}-{日期}-{署名}` | `{片段名}-{日期}-（N）.{ext}` |
| 门店素材 | `{片段名}-{门店编号}-{日期}-{署名}` | `{片段名}-{门店编号}-{日期}-（N）.{ext}` |
| 视频目录 | `{图片目录}_video` | `{图片 basename}.mp4` |

- 日期格式 `8.23` / `12.5`（不带前导 0）
- 序号用全角括号 `（1）` `（2）`
- 通用 vs 门店由 JSON 的 `category` 字段判断（`门店素材` / `门店` / `store` 任一关键词）
- 门店素材的 segment_name 第一段是门店编号（如 `130014WL-玉湖公园`）

示例：
```
outputs/
├── 玉湖公园-1-8.23-姜/
│   ├── 玉湖公园-1-8.23-（1）.png
│   ├── 玉湖公园-1-8.23-（2）.png
│   └── rejected/                # 审核员把不合格图放这里
│       └── 玉湖公园-1-8.23-（3）.png
└── 玉湖公园-1-8.23-姜_video/
    └── 玉湖公园-1-8.23-（1）.mp4
```

---

## 审核反馈机制

- **方案 C（文件标记）**：审核员把不合格图移到 `<image_dir>/rejected/` 子目录
- 程序下次运行（或本次 `--wait-seconds` 到期后）扫描 `rejected/`
- 发现不合格 → 删除对应位置 → 补生成 N 张 → 重新推送全量
- 最多循环 `--max-rounds` 轮（默认 5）

---

## 项目结构

```
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
├── naming.py           # 命名规则 + users.yaml 加载
├── feishu.py           # 飞书 webhook 推送
└── workflow.py         # 审核工作流编排
```

---

## 开发

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

# 跑测试
pytest

# 构建 whl + sdist
python -m build

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

---

## License

MIT

## 进阶文档

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