Metadata-Version: 2.5
Name: p0aiapi
Version: 0.2.1
Summary: Python SDK for the P0 AI API with sync and async media and task clients
Keywords: aiapi,openapi,p0,sdk
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == 'dev'
Requires-Dist: pytest-asyncio<2,>=0.24; extra == 'dev'
Requires-Dist: pytest<10,>=8; extra == 'dev'
Requires-Dist: readme-renderer[md]>=44; extra == 'dev'
Requires-Dist: ruff>=0.9; extra == 'dev'
Requires-Dist: twine>=6; extra == 'dev'
Description-Content-Type: text/markdown

# p0aiapi

[English](#english) | [简体中文](#chinese)

## English

Python SDK for the P0 API. Synchronous and asynchronous clients, image and video generation, task queries, and automatic polling. Supports Python 3.10+ with `httpx` as the only runtime dependency.

**Version:** `0.2.1` · **Default API:** `https://p0-api.inaiai.com`

### Installation

```bash
python -m pip install --upgrade p0aiapi==0.2.1
export P0_API_KEY='your-api-key'
export P0_API_SECRET_KEY='your-api-secret'
```

Use an API key and secret for v2 authentication. Credentials can also be passed to `Client(api_key="...", secret_key="...")` or `AsyncClient(...)`. Omitting the secret preserves v1 compatibility; authentication selection and fallback are handled by the server.

Set `P0_API_BASE_URL` or pass `base_url=` to use another deployment. Explicit arguments take precedence over environment variables. The SDK preserves gateway path prefixes.

### Quickstart

Replace the example URLs with your own accessible images. The source supplies the face; the target is the image to edit.

```python
from p0aiapi import Client

with Client() as client:
    task = client.face_swap.image(
        "https://your-cdn.example.com/face.jpg",
        "https://your-cdn.example.com/portrait.jpg",
    )
    result = client.tasks.wait(task["task_id"])
    print(result.get("out_data"))
```

Resource methods work directly without loading a schema or looking up operation IDs. Generation returns an initial task dictionary; use its `task_id` to query or wait for completion. Context managers close client connections automatically.

### Task queries

Use `tasks.get()` for one task, `tasks.batch()` for multiple tasks, and `tasks.wait()` for automatic polling.

```python
from p0aiapi import Client

with Client() as client:
    task = client.tasks.get("task-1")
    print(task.get("status"), task.get("out_data"))

    for task in client.tasks.batch(["task-1", "task-2"]):
        print(task.get("task_id"), task.get("status"), task.get("out_data"))

    result = client.tasks.wait("task-1", timeout=600, poll_interval=10)
```

Async methods accept the same arguments:

```python
import asyncio
from p0aiapi import AsyncClient


async def main():
    async with AsyncClient() as client:
        task = await client.tasks.get("task-1")
        tasks = await client.tasks.batch(["task-1", "task-2"])
        result = await client.tasks.wait("task-1", timeout=600, poll_interval=10)
        print(task, tasks, result.get("out_data"))


asyncio.run(main())
```

Pass the exact `task_id` returned by submission, rather than `uuid` or `bid`. IDs must contain at least one non-whitespace character and are preserved without trimming. Batch input must be nonempty. Batch responses are lists of task dictionaries: missing IDs are omitted, order is unspecified, and one ID may match multiple records. A missing single task or a batch with no matches raises `NotFoundError` (404).

Status `2` means completed, `-1` means failed, and `0` / `1` / `3` continue polling. `tasks.wait()` defaults to a 600-second timeout and 10-second polling interval. It raises `TaskFailedError` on failure or `TaskTimeoutError` on timeout; use `raise_on_failure=False` to return a failed task. A timeout ends local polling without cancelling the server task. `client.wait_for_task()` remains available.

### Resources

All methods below support both `Client` and `AsyncClient`.

| Method | Purpose |
| --- | --- |
| `client.face_swap.image(source_url, target_url)` | Replace a face in a target image using a source face image |
| `client.face_swap.video(source_url, target_url, output_format="video")` | Replace a face in a target video using a source face image |
| `client.videos.transfer_motion(source_url, motion_url)` | Animate a source image using a reference motion video |
| `client.scenes.list()` | Retrieve scene metadata as a list of dictionaries |
| `client.scenes.figurine(source_url, reference_url, prompt=...)` | Generate a figurine image using a character image and pose reference |
| `client.images.generate(prompt, **fields)` | Generate images from text |
| `client.novels.generate(description, **fields)` | Submit novel generation |
| `client.novels.continue_from(file_url, **fields)` | Continue writing from a public TXT or Markdown document |
| `client.audio.list_voices()` / `client.audio.generate(text, **fields)` | List voices and generate speech |
| `client.audio.upload_voice(spk_id, prompt_text, path)` | Upload a voice sample with its transcript |

Motion transfer and figurine generation:

```python
from p0aiapi import Client

with Client() as client:
    video = client.videos.transfer_motion(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/walking.mp4",
        width=800,
        height=600,
        output_format="video",
        video_format="video/h264-mp4",
    )
    figurine = client.scenes.figurine(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/pose.png",
        prompt="Create a desktop figurine of the original character, using the reference pose.",
    )
    tasks = client.tasks.batch([video["task_id"], figurine["task_id"]])
    print(tasks)
```

Media inputs must be absolute HTTP(S) URLs accessible to the server. Local paths and URLs containing credentials are rejected; signed query strings are preserved. Optional `**fields` pass through supported API parameters such as `bid`, `task_id`, and `notify_url`. `None` is omitted; `False` and `0` are retained.

Use the named URL and `prompt` arguments with media methods. The underlying fields `source_path`, `target_path`, `incoming_prompt`, and `scene_name` cannot be overridden through `**fields`. `scenes.figurine()` uses the fixed `person_model` scene and produces images. The general scene list is a separate configuration source.

### Types and errors

`Task` and `TaskStatus` provide type hints and editor support. Responses remain ordinary dictionaries, preserving unknown fields and the server's original `out_data` value.

HTTP errors inherit from `APIError`, with `status_code` and `body` available for inspection. Connection failures raise `TransportError`. Bodyless read requests retry up to twice by default; generation POSTs are never automatically retried. If the server returns `submission_uncertain=true`, query the task before submitting again.

### Development

```bash
python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .
python scripts/check_readme.py
```

Tests use mock HTTP transports. Run `python examples/offline_demo.py` for an offline example.

Full documentation, the changelog, and contribution guidelines are included in the [source distribution](https://pypi.org/project/p0aiapi/0.2.1/#files): `docs/advanced.md`, `CHANGELOG.md`, `CONTRIBUTING.md`, and `docs/publishing.md`.

The supported endpoints are based on [YApi project 8](https://yapi.inaiai.com/project/8/interface/api) and backend contracts. Availability depends on the deployment. Verification scope is recorded in `docs/backend-compatibility.md` and `docs/live-verification-2026-10-09.md` in the source distribution.

---

## Chinese

P0 API 的 Python SDK，提供同步和 asyncio 客户端、普通图像与视频任务、任务查询及自动等待。支持 Python 3.10+，运行时仅依赖 `httpx`。

发行包与导入名均为 `p0aiapi`，当前版本 `0.2.1`。默认服务地址为 **`https://p0-api.inaiai.com`**。以下服务接口见于 [YApi 项目 8](https://yapi.inaiai.com/project/8/interface/api)，实际可用能力取决于部署版本。

### 安装与鉴权

```bash
python -m pip install --upgrade p0aiapi==0.2.1
export P0_API_KEY='你的 API key'
export P0_API_SECRET_KEY='你的 API secret'
```

推荐同时配置 key 和 secret 使用鉴权 v2，分别发送 `apikey` 和 `X-SECRET-KEY`。也可传入 `Client(api_key="...", secret_key="...")`，`AsyncClient` 用法相同。未提供 secret 时兼容 v1，鉴权版本选择及回退由后端处理。SDK 不内置凭据。

其他部署可设置 `P0_API_BASE_URL` 或 `base_url=`，显式参数优先于环境变量；反向代理路径前缀会保留。使用 `with` / `async with` 管理连接。

### 快速开始

使用已获授权的普通非色情素材，将示例地址替换为 backend 能访问的图片 URL。第一张提供人脸，第二张是目标图片：

```python
from p0aiapi import Client

# 替换为你的授权素材 URL。
with Client() as client:
    task = client.face_swap.image(
        "https://your-cdn.example.com/face.jpg",
        "https://your-cdn.example.com/portrait.jpg",
    )
    result = client.tasks.wait(task["task_id"])
    print(result.get("out_data"))
```

普通方法可以直接使用，无需加载接口配置或查找 operationId。生成方法返回初始任务字典，通过 `task_id` 查询或等待结果。

### 查询任务

**一个任务用 `tasks.get()`，多个任务用 `tasks.batch()`；需要自动轮询时用 `tasks.wait()`。** 将下面的 ID 换成提交任务时返回的 `task_id`。

```python
from p0aiapi import Client

with Client() as client:
    task = client.tasks.get("task-1")
    print(task.get("status"), task.get("out_data"))

    for task in client.tasks.batch(["task-1", "task-2"]):
        print(task.get("task_id"), task.get("status"), task.get("out_data"))

    result = client.tasks.wait("task-1", timeout=600, poll_interval=10)
```

异步方法使用相同参数，并加 `await`：

```python
import asyncio
from p0aiapi import AsyncClient


async def main():
    async with AsyncClient() as client:
        task = await client.tasks.get("task-1")
        tasks = await client.tasks.batch(["task-1", "task-2"])
        result = await client.tasks.wait("task-1", timeout=600, poll_interval=10)
        print(task, tasks, result.get("out_data"))


asyncio.run(main())
```

`task_id` 必须是非空且不全为空白的字符串，SDK 保留原值，不自动去掉空白；不能用内部 `uuid` 或业务编号 `bid` 替代。`batch()` 接收非空任务 ID 列表或其他可迭代对象，发送裸 JSON 数组，返回任务字典数组。未命中的 ID 被省略，结果不保证输入顺序，同一 ID 可能有多条记录；单独查询未命中，或批量全部未命中时抛出 `NotFoundError`（404）。

状态 `2` 成功、`-1` 失败；`0` / `1` / `3` 继续等待。`tasks.wait()` 默认等待 600 秒、每 10 秒查询一次，失败抛出 `TaskFailedError`，超时抛出 `TaskTimeoutError`；`raise_on_failure=False` 可返回失败记录。超时停止本地等待，后端任务继续执行。原来的 `client.wait_for_task()` 仍可使用。

### 常用能力

下面的方法同时支持 `Client` 和 `AsyncClient`，不依赖 OpenAPI 配置。

| 方法 | 用途与输入 |
| --- | --- |
| `client.face_swap.image(source_url, target_url)` | 换脸图片：人脸图片 → 目标图片 |
| `client.face_swap.video(source_url, target_url, output_format="video")` | 换脸视频：人脸图片 → 目标视频 |
| `client.videos.transfer_motion(source_url, motion_url)` | 动作迁移：角色图片 + 动作参考视频 → 视频 |
| `client.scenes.list()` | 查询通用场景元信息，返回字典数组 |
| `client.scenes.figurine(source_url, reference_url, prompt=...)` | 角色原图 + 姿势参考图 → 手办图片，固定 `person_model` 场景 |
| `client.images.generate(prompt, **fields)` | 文生图 |
| `client.novels.generate(description, **fields)` | 小说生成 |
| `client.novels.continue_from(file_url, **fields)` | 从公网 TXT / Markdown 文件续写 |
| `client.audio.list_voices()` / `generate(text, **fields)` | 音色列表与语音生成 |
| `client.audio.upload_voice(spk_id, prompt_text, path)` | 上传音色样本及对应文本 |

动作迁移和手办生成示例：

```python
from p0aiapi import Client

with Client() as client:
    video = client.videos.transfer_motion(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/walking.mp4",
        width=800,
        height=600,
        output_format="video",
        video_format="video/h264-mp4",
    )
    figurine = client.scenes.figurine(
        "https://your-cdn.example.com/character.png",
        "https://your-cdn.example.com/pose.png",
        prompt="将原创角色制作成桌面收藏手办，参考第二张图片的姿势。",
    )
    tasks = client.tasks.batch([video["task_id"], figurine["task_id"]])
    print(tasks)
```

媒体方法接收绝对 HTTP(S) URL，拒绝本机路径和含用户名、密码的 URL，保留签名查询参数。SDK 在本地校验 URL 格式，素材是否能下载由 backend 检查。

生成方法接受 `**fields` 传入具体接口支持的可选字段，例如 `bid`、`task_id`、`notify_url`。`None` 省略，`False` 和 `0` 保留，默认值由服务端决定。新媒体方法使用 `source_url`、`target_url` / `motion_url` / `reference_url` 和 `prompt` 参数；不要在 `**fields` 中重复传旧的 `source_path`、`target_path`、`incoming_prompt` 或 `scene_name`。

`scenes.list()` 是通用场景清单，与手办配置属于不同集合。`scenes.figurine()` 固定使用已核对的 `person_model`，生成图片；无需从清单中选择模板。

### 类型与错误

公开的 `Task` 类型和 `TaskStatus` 枚举用于类型提示与编辑器补全。任务仍是服务端返回的原始字典，保留未知字段；`out_data` 的形态取决于任务类型，不进行强制转换。

HTTP 错误继承 `APIError`，可检查 `status_code` 和 `body`；连接错误为 `TransportError`。任务失败和等待超时分别为 `TaskFailedError`、`TaskTimeoutError`。默认无请求体的读取请求最多重试 2 次，生成 POST 不自动重试；返回 `submission_uncertain=true` 时先查询任务，避免重复提交。

完整进阶文档、设计及适配检查记录随[源码发行包](https://pypi.org/project/p0aiapi/0.2.1/#files)提供，位于 `docs/` 目录。

### 开发

```bash
python -m pip install -e '.[dev]'
pytest
ruff check .
ruff format --check .
python scripts/check_readme.py
```

测试使用模拟 HTTP 验证请求合同和错误处理，不提交真实生成任务。离线演示：`python examples/offline_demo.py`。

源码发行包同时包含 `CHANGELOG.md`、`CONTRIBUTING.md`、`docs/publishing.md` 和独立的 `README.zh-CN.md`。
