mirror of
https://github.com/ningzimu/codex-ppt-skill.git
synced 2026-09-14 20:06:52 +08:00
feat: add AtlasCloud image provider
This commit is contained in:
ningzimu
@@ -52,6 +52,7 @@ examples/
|
||||
!assets/slides_example.png
|
||||
!assets/paper-figures-example.png
|
||||
!assets/codex-ppt-community-qr.png
|
||||
!assets/atlas-cloud-logo.png
|
||||
!assets/good-skill-design.pptx
|
||||
!assets/skill_duo_intro.pdf
|
||||
!assets/style-previews/
|
||||
|
||||
@@ -2,6 +2,16 @@
|
||||
|
||||
Release notes are generated from this file. Keep changelog entries in English.
|
||||
|
||||
## Unreleased
|
||||
|
||||
### Features
|
||||
|
||||
- Add an AtlasCloud image provider adapter for GPT Image 2 generation and editing.
|
||||
|
||||
### Documentation
|
||||
|
||||
- Document OpenAI-compatible and AtlasCloud image provider configuration examples.
|
||||
|
||||
## 0.4.4
|
||||
|
||||
### Fixes
|
||||
|
||||
@@ -2,7 +2,18 @@
|
||||
|
||||
[](README_en.md) [](https://clawhub.ai/ningzimu/codex-ppt) [](https://github.com/ningzimu/codex-ppt-skill/stargazers) [](https://github.com/ningzimu/codex-ppt-skill/forks)
|
||||
|
||||
一个面向 Codex 的 PPT 生成 skill，也可在 Claude Code、OpenClaw、Hermes Agent 等支持 `SKILL.md` 的 agent 中使用；在这些非 Codex 环境中通常需要配置 `gpt-image-2` 或第三方 OpenAI 兼容格式的生图 API。它把文章、报告、论文、课程笔记等内容转换成“整页图片式”的演示文稿：先规划大纲和视觉风格，再生成每页幻灯片图片，最后用本地脚本组装为 `.pptx`。
|
||||
一个面向 Codex 的 PPT 生成 skill，也可在 Claude Code、OpenClaw、Hermes Agent 等支持 `SKILL.md` 的 agent 中使用；在这些非 Codex 环境中通常需要配置 `gpt-image-2`、第三方生图 API 或 OpenAI 兼容格式的生图接口。它把文章、报告、论文、课程笔记等内容转换成“整页图片式”的演示文稿：先规划大纲和视觉风格，再生成每页幻灯片图片，最后用本地脚本组装为 `.pptx`。
|
||||
|
||||
## 赞助
|
||||
|
||||
<table>
|
||||
<tr>
|
||||
<td width="180"><img src="assets/atlas-cloud-logo.png" alt="Atlas Cloud" width="160"></td>
|
||||
<td>感谢 <a href="https://www.atlascloud.ai/?utm_source=github&utm_medium=link&utm_campaign=codex-ppt-skill">Atlas Cloud</a> 赞助本项目。AtlasCloud 是多模态 AI 推理平台，提供统一 API 接入图片生成、视频生成和大语言模型等能力；本 skill 已支持通过现有 API key、base URL 和模型名配置接入 AtlasCloud 的 GPT Image 2 生图和编辑图接口，按量计费，开箱即用。完整模型列表可查看 <a href="https://www.atlascloud.ai/zh/models">Atlas Cloud 模型页</a>。</td>
|
||||
</tr>
|
||||
</table>
|
||||
|
||||
## 温馨提示
|
||||
|
||||
> [!TIP]
|
||||
> 本 skill 负责从文章、报告、大纲或想法生成图片式 PPT，适合强视觉表达，但页面元素本身不可直接编辑。如果你需要进一步转换成可编辑 PPT，可以在生成完成后尝试使用 [image-to-editable-ppt-skill](https://github.com/ningzimu/image-to-editable-ppt-skill) 进行转换。
|
||||
@@ -14,8 +25,6 @@
|
||||
> [!NOTE]
|
||||
> 想查看更多用户用这个 skill 做出的 PPT 效果，可以前往置顶 Issue 的案例展示区：[欢迎分享 codex-ppt 使用案例和 PPT 效果](https://github.com/ningzimu/codex-ppt-skill/issues/34)。
|
||||
|
||||
## 温馨提示
|
||||
|
||||
这个 skill 主要给大家提供一个还不错的 PPT 生成流程。为了尽量通用，它的流程设计会稍微复杂一些；复杂也会带来不稳定性或者冗余性。比如它同时兼容 Codex 内置生图和 API/CLI fallback 生图，也会兼容有无子 agent 可用这两种情况，但大部分人日常使用时其实只会固定走其中一条路线。
|
||||
|
||||
建议大家在走通自己常用的路线之后，让 AI 帮你改一下这个 skill，把你的偏好固定下来，省得每次都重新选择。比如固定使用内置生图或固定使用某个 API，固定是否使用子 agent，固定常用输出目录、风格、页数节奏等。
|
||||
@@ -27,7 +36,7 @@
|
||||
## 特点
|
||||
|
||||
- 多 agent 可用：支持 Codex、Claude Code、OpenClaw、Hermes Agent 等支持 `SKILL.md` 的环境；最推荐在 Codex 中使用，优先走内置生图和编辑图能力。
|
||||
- 第三方中转站接入：支持 OpenAI 兼容接口、`base URL` 和自定义模型名配置，方便通过 API/CLI fallback 使用 `gpt-image-2` 或兼容模型。
|
||||
- 第三方生图供应商接入：支持 OpenAI 兼容接口、AtlasCloud、`base URL` 和自定义模型名配置，方便通过 API/CLI fallback 使用 `gpt-image-2` 或兼容模型。
|
||||
- 稳定的阶段化流程：先确认大纲、页数、视觉风格、生图后端和样张，再进入整套生成，降低一次生成完整 PPT 时的返工和偏航。
|
||||
- 不是无脑生成：会先引导你确认 `outline.md`、每页要点、风格方向和样张效果，再按确认后的方案继续。
|
||||
- 低门槛输入：文章、报告、论文、课程笔记、Markdown、大纲、PDF、Word 等材料都可以作为起点。
|
||||
@@ -164,35 +173,9 @@ npx -y skills@latest add ningzimu/codex-ppt-skill \
|
||||
> 你可以先正常使用 Codex PPT 开始制作 PPT。一般不需要自己手动配置生图模型；当流程走到“选择生图后端”时，AI 会根据当前环境判断是否需要配置，并在需要时引导你提供相关信息。
|
||||
>
|
||||
> - 如果你使用的是 Codex 内置图片生成能力，通常不需要额外配置 API key。
|
||||
> - 如果你使用第三方 API 或 OpenAI 兼容中转站，请把中转站关于如何使用 `gpt-image-2` 的文档发给 AI，让它先阅读文档，再帮你配置相关脚本和参数。
|
||||
> - 如果你确定要使用第三方供应商或 OpenAI 兼容中转站，请让 AI 先阅读 [生图模型配置指南](skills/codex-ppt/docs/image-model-configuration.md)，再配置 API key、base URL 和模型名。
|
||||
|
||||
下面的手动配置说明主要用于 API/CLI fallback 场景。指定图片分辨率、提高质量或要求修改某一页，本身不会触发第三方 API 配置。典型需要配置的情况包括：
|
||||
|
||||
- 在 Codex 中使用第三方 API 或兼容中转站接入时，通常无法使用内置的图片生成工具。
|
||||
- 在 Claude Code、OpenClaw、Hermes Agent 等环境中使用该 skill。
|
||||
|
||||
如果你是通过 GPT 会员订阅使用 Codex，并且 Codex 内置图片生成工具可用，就不需要额外配置 `gpt-image-2`。即使你在提示词里写“使用 `gpt-image-2`”，通常也可以继续使用 Codex 内置生图能力，不需要准备 API key。
|
||||
|
||||
如果确实需要外部生图接口，配置文件会写入 `~/.codex-ppt-skill/.env`。使用第三方中转站时再填写 `base URL`；模型名默认是 `gpt-image-2`，除非中转站明确要求别的名称。配置完成后，Codex、Claude Code、OpenClaw、Hermes Agent 可以复用同一套配置。
|
||||
|
||||
确实需要手动配置或排查时，也可以直接运行配置命令：
|
||||
|
||||
```bash
|
||||
python3 /path/to/codex-ppt-skill/skills/codex-ppt/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-api-key" \
|
||||
--model gpt-image-2
|
||||
```
|
||||
|
||||
其中 `--api-key` 是你的 API key；`--model` 是图片模型名，默认可使用 `gpt-image-2`。配置会写入 `~/.codex-ppt-skill/.env`。不要把 API key 写进项目目录或提交到仓库。
|
||||
|
||||
如果使用第三方中转站，再加上 `--base-url`。如果中转站使用自定义模型名，就把 `--model` 改成中转站提供的名称：
|
||||
|
||||
```bash
|
||||
python3 /path/to/codex-ppt-skill/skills/codex-ppt/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-api-key" \
|
||||
--base-url "https://your-openai-compatible-endpoint/v1" \
|
||||
--model openai/gpt-image-2
|
||||
```
|
||||
指定图片分辨率、提高质量或要求修改某一页，本身不会触发第三方 API 配置。如果你是通过 GPT 会员订阅使用 Codex，并且 Codex 内置图片生成工具可用，通常可以继续使用内置生图能力，不需要准备 API key。
|
||||
|
||||
## 使用方式
|
||||
|
||||
|
||||
+15
-32
@@ -2,7 +2,18 @@
|
||||
|
||||
[](README.md) [](https://clawhub.ai/ningzimu/codex-ppt) [](https://github.com/ningzimu/codex-ppt-skill/stargazers) [](https://github.com/ningzimu/codex-ppt-skill/forks)
|
||||
|
||||
A Codex skill for generating PowerPoint decks. It can also be used in Claude Code, OpenClaw, Hermes Agent, and other agents that support `SKILL.md`; these non-Codex environments usually require configuring `gpt-image-2` or a third-party OpenAI-compatible image generation API. It turns articles, reports, papers, course notes, and other source materials into image-based presentations: first plan the outline and visual style, then generate each full-slide image, and finally assemble the images into a `.pptx` file with a local script.
|
||||
A Codex skill for generating PowerPoint decks. It can also be used in Claude Code, OpenClaw, Hermes Agent, and other agents that support `SKILL.md`; these non-Codex environments usually require configuring `gpt-image-2`, a third-party image API, or an OpenAI-compatible image generation endpoint. It turns articles, reports, papers, course notes, and other source materials into image-based presentations: first plan the outline and visual style, then generate each full-slide image, and finally assemble the images into a `.pptx` file with a local script.
|
||||
|
||||
## Sponsor
|
||||
|
||||
<table>
|
||||
<tr>
|
||||
<td width="180"><img src="assets/atlas-cloud-logo.png" alt="Atlas Cloud" width="160"></td>
|
||||
<td>Thanks to <a href="https://www.atlascloud.ai/?utm_source=github&utm_medium=link&utm_campaign=codex-ppt-skill">Atlas Cloud</a> for sponsoring this project. AtlasCloud is a multimodal AI inference platform that provides one API for image generation, video generation, LLMs, and more. This skill can use the existing API key, base URL, and model-name configuration to call AtlasCloud's GPT Image 2 generation and editing endpoints, with pay-as-you-go pricing and out-of-the-box setup. See the full model list on the <a href="https://www.atlascloud.ai/zh/models">Atlas Cloud models page</a>.</td>
|
||||
</tr>
|
||||
</table>
|
||||
|
||||
## Friendly Note
|
||||
|
||||
> [!TIP]
|
||||
> This skill generates image-based PPT decks from articles, reports, outlines, or ideas. It is suitable for strong visual expression, but slide elements are not directly editable. If you need a more editable PPT, you can try converting the generated deck with [image-to-editable-ppt-skill](https://github.com/ningzimu/image-to-editable-ppt-skill).
|
||||
@@ -14,8 +25,6 @@ A Codex skill for generating PowerPoint decks. It can also be used in Claude Cod
|
||||
> [!NOTE]
|
||||
> To see more PPT examples made by users with this skill, visit the pinned showcase issue: [欢迎分享 codex-ppt 使用案例和 PPT 效果](https://github.com/ningzimu/codex-ppt-skill/issues/34).
|
||||
|
||||
## Friendly Note
|
||||
|
||||
This skill is meant to provide a solid PPT generation workflow. To stay broadly useful, the workflow is a little more complex than most people need every day, and that complexity can sometimes add instability or redundant choices. For example, it supports both Codex built-in image generation and API/CLI fallback generation, and it also supports workflows with or without subagents. Most users will eventually use only one of those paths.
|
||||
|
||||
After you have a route working, consider asking an AI assistant to edit this skill and lock in your own preferences, such as your preferred image backend, whether to use subagents, output directory conventions, visual styles, or slide pacing. That way you do not need to make the same choices every time.
|
||||
@@ -27,7 +36,7 @@ For a basic introduction to skill design and usage, see [good-skill-design.pptx]
|
||||
## Features
|
||||
|
||||
- Works across multiple agents: supports Codex, Claude Code, OpenClaw, Hermes Agent, and other `SKILL.md`-based environments; Codex is the recommended environment because it can use the built-in image generation and image editing tools first.
|
||||
- Supports third-party proxy APIs: works with OpenAI-compatible endpoints, `base URL`, and custom model names, so API/CLI fallback can use `gpt-image-2` or compatible image models.
|
||||
- Supports third-party image providers: works with OpenAI-compatible endpoints, AtlasCloud, `base URL`, and custom model names, so API/CLI fallback can use `gpt-image-2` or compatible image models.
|
||||
- Stable staged workflow: confirms the outline, slide count, visual style, image backend, and sample slide before full-deck generation, reducing drift and rework when generating a complete PPT.
|
||||
- Guided instead of one-shot: the skill asks you to confirm `outline.md`, per-slide key points, style direction, and sample-slide quality before continuing.
|
||||
- Low setup effort: articles, reports, papers, course notes, Markdown files, outlines, PDFs, and Word documents can all be used as starting material.
|
||||
@@ -164,35 +173,9 @@ If you are developing this repository locally, you can use a symlink instead of
|
||||
> You can start using Codex PPT normally to make a deck. In most cases, you do not need to configure the image model by hand; when the workflow asks you to choose an image backend, the AI will check the current environment and guide you through any required information.
|
||||
>
|
||||
> - If you use Codex's built-in image generation, you usually do not need an extra API key.
|
||||
> - If you use a third-party API or an OpenAI-compatible proxy, send the AI that provider's documentation for using `gpt-image-2` first, then let it read the docs and configure the relevant scripts and parameters.
|
||||
> - If you have confirmed that a third-party provider or OpenAI-compatible proxy is needed, ask the AI to read the [image model configuration guide](skills/codex-ppt/docs/image-model-configuration.md) before configuring API key, base URL, and model name.
|
||||
|
||||
The manual configuration notes below mainly apply to API/CLI fallback. Asking for a specific resolution, higher quality, or edits to one slide does not by itself trigger third-party API configuration. Typical cases that require configuration include:
|
||||
|
||||
- Using a third-party API or OpenAI-compatible proxy in Codex, where the built-in image generation tool is usually unavailable.
|
||||
- Using this skill from Claude Code, OpenClaw, Hermes Agent, or similar agents.
|
||||
|
||||
If you use Codex through a GPT subscription and Codex's built-in image generation tool is available, you do not need to configure `gpt-image-2` separately. Even if your prompt says “use `gpt-image-2`”, you can usually keep using Codex's built-in image generation and do not need to prepare an API key.
|
||||
|
||||
If you do need an external image API, the config is written to `~/.codex-ppt-skill/.env`. Add a `base URL` only when using a third-party proxy. The model defaults to `gpt-image-2`; change it only if your proxy requires a different model name. Once configured, Codex, Claude Code, OpenClaw, and Hermes Agent can share the same settings.
|
||||
|
||||
If you need to configure or troubleshoot it manually, you can run the config command directly:
|
||||
|
||||
```bash
|
||||
python3 /path/to/codex-ppt-skill/skills/codex-ppt/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-api-key" \
|
||||
--model gpt-image-2
|
||||
```
|
||||
|
||||
`--api-key` is your API key. `--model` is the image model name, and `gpt-image-2` is the default choice. The config is written to `~/.codex-ppt-skill/.env`. Do not write API keys into the project directory or commit them to the repository.
|
||||
|
||||
If you use a third-party proxy, add `--base-url`. If the proxy uses a custom model name, replace `--model` with the name provided by that proxy:
|
||||
|
||||
```bash
|
||||
python3 /path/to/codex-ppt-skill/skills/codex-ppt/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-api-key" \
|
||||
--base-url "https://your-openai-compatible-endpoint/v1" \
|
||||
--model openai/gpt-image-2
|
||||
```
|
||||
Asking for a specific resolution, higher quality, or edits to one slide does not by itself trigger third-party API configuration. If you use Codex through a GPT subscription and Codex's built-in image generation tool is available, you can usually keep using the built-in image tool and do not need to prepare an API key.
|
||||
|
||||
## Usage
|
||||
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 131 KiB |
@@ -12,7 +12,7 @@ This skill supports two image backends:
|
||||
- Before recommending CLI/API fallback, actively check whether the built-in image generation tool is callable in the current environment. Do not infer availability only from the agent name or subscription context.
|
||||
- Prefer the built-in image tool when available. In Codex, this usually means the built-in `image_gen` tool. In OpenClaw, this may be `image_generate`. Resolution, quality, aspect ratio, slide-edit requests, or the user saying "use `gpt-image-2`" do not require CLI/API fallback.
|
||||
- In Codex, treat the built-in image tool as the preferred `gpt-image-2` path when it is available. If the user has a GPT subscription / Codex environment and asks for `gpt-image-2`, do not switch to `scripts/image_gen.py` only to satisfy the model name.
|
||||
- Use CLI/API fallback only when the built-in tool is unavailable, the built-in tool failed for a required capability, the user explicitly asks for API/CLI or a third-party OpenAI-compatible proxy, or the requested capability is unavailable in the built-in tool.
|
||||
- Use CLI/API fallback only when the built-in tool is unavailable, the built-in tool failed for a required capability, the user explicitly asks for API/CLI or a third-party image API/provider adapter, or the requested capability is unavailable in the built-in tool.
|
||||
- Do not recommend CLI/API fallback merely because it provides direct `--out` file paths, easier local file management, local config reuse, batch generation convenience, or simpler automation.
|
||||
- Before generating the first image, tell the user which tool availability you checked, which backend you plan to use, why fallback is or is not needed, and ask for confirmation. Do not treat being in a specific agent environment as proof that the built-in image tool is available.
|
||||
- CLI/API fallback loads `~/.codex-ppt-skill/.env` automatically. Run the CLI normally; do not manually parse `.env` or ask for configuration before an error.
|
||||
|
||||
@@ -24,10 +24,10 @@ If Codex is being used through a GPT subscription and the built-in image tool is
|
||||
## Required And Optional Values
|
||||
|
||||
- `OPENAI_API_KEY` is required for real API/CLI fallback calls.
|
||||
- `OPENAI_BASE_URL` is optional. When it is unset, the CLI uses the official OpenAI API. When it is set, the CLI treats the request as a third-party OpenAI-compatible proxy request.
|
||||
- `CODEX_PPT_IMAGE_MODEL` is optional. The default is `gpt-image-2`. Use a custom value only when the proxy provider requires one.
|
||||
- `OPENAI_BASE_URL` is optional. When it is unset, the CLI uses the official OpenAI API. When it is set, the CLI uses the configured third-party provider base URL.
|
||||
- `CODEX_PPT_IMAGE_MODEL` is optional. The default is `gpt-image-2`. Use a custom value only when the provider requires one.
|
||||
|
||||
Configure provided API settings with `scripts/codex_ppt_runtime.py config --api-key`.
|
||||
Configure provided API settings with `scripts/codex_ppt_runtime.py config --api-key`. The config command writes `~/.codex-ppt-skill/.env`.
|
||||
|
||||
## Official OpenAI Example
|
||||
|
||||
@@ -37,16 +37,39 @@ python3 {skill_root}/scripts/codex_ppt_runtime.py config \
|
||||
--model gpt-image-2
|
||||
```
|
||||
|
||||
## Third-Party Proxy Example
|
||||
## OpenAI-Compatible Provider Example
|
||||
|
||||
Use this shape for providers that implement the OpenAI Images API paths used by the fallback CLI.
|
||||
|
||||
```bash
|
||||
python3 {skill_root}/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-api-key" \
|
||||
--base-url "https://your-openai-compatible-endpoint/v1" \
|
||||
--model openai/gpt-image-2
|
||||
--api-key "your-provider-api-key" \
|
||||
--base-url "https://xxxx.example.com/v1" \
|
||||
--model gpt-image-2
|
||||
```
|
||||
|
||||
Replace `--base-url` and `--model` with the values from the proxy provider.
|
||||
This produces the same effective runtime config as:
|
||||
|
||||
```env
|
||||
OPENAI_API_KEY=your-provider-api-key
|
||||
OPENAI_BASE_URL=https://xxxx.example.com/v1
|
||||
CODEX_PPT_IMAGE_MODEL=gpt-image-2
|
||||
```
|
||||
|
||||
For OpenAI-compatible providers, `OPENAI_BASE_URL` should normally end at the provider's `/v1` root. Do not set it to `/images/generations`, `/images/edits`, or another terminal endpoint. The fallback CLI appends the image-generation or image-edit path through the OpenAI SDK.
|
||||
|
||||
Use the provider's model name only when the provider documents a custom name. Otherwise prefer `gpt-image-2`.
|
||||
|
||||
## AtlasCloud Example
|
||||
|
||||
For AtlasCloud, set `--model` to the base model name. The CLI chooses the matching generation or editing model route internally.
|
||||
|
||||
```bash
|
||||
python3 {skill_root}/scripts/codex_ppt_runtime.py config \
|
||||
--api-key "your-atlascloud-api-key" \
|
||||
--base-url "https://api.atlascloud.ai/api/v1/model" \
|
||||
--model openai/gpt-image-2
|
||||
```
|
||||
|
||||
## Runtime Config
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@ import subprocess
|
||||
import sys
|
||||
from typing import Dict, Iterable, Optional
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import venv
|
||||
|
||||
@@ -208,6 +209,11 @@ def _models_request(base_url: str, api_key: str, timeout: int) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _is_atlascloud_base_url(base_url: str) -> bool:
|
||||
hostname = urllib.parse.urlparse(base_url).hostname or ""
|
||||
return "atlascloud.ai" in hostname.lower()
|
||||
|
||||
|
||||
def _doctor(args: argparse.Namespace) -> int:
|
||||
home = _runtime_home()
|
||||
env_file = _env_path(home)
|
||||
@@ -237,6 +243,8 @@ def _doctor(args: argparse.Namespace) -> int:
|
||||
if not api_key:
|
||||
print("api check: skipped, OPENAI_API_KEY is unset")
|
||||
ok = False
|
||||
elif _is_atlascloud_base_url(base_url):
|
||||
print("api check: AtlasCloud provider detected; skipping /models probe")
|
||||
else:
|
||||
ok = _models_request(base_url, api_key, args.timeout) and ok
|
||||
|
||||
|
||||
@@ -6,8 +6,8 @@ opts into API mode, or when explicit transparent output requires the
|
||||
`gpt-image-1.5` fallback path.
|
||||
|
||||
Defaults to gpt-image-2 and a structured prompt augmentation workflow.
|
||||
Reads OPENAI_API_KEY, and optionally OPENAI_BASE_URL for OpenAI-compatible
|
||||
proxy providers.
|
||||
Reads OPENAI_API_KEY, and optionally OPENAI_BASE_URL for provider adapters or
|
||||
OpenAI-compatible proxy providers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -15,6 +15,7 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import asyncio
|
||||
import base64
|
||||
from io import BytesIO
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
@@ -22,8 +23,10 @@ import re
|
||||
import sys
|
||||
import time
|
||||
from typing import Any, Dict, Iterable, List, Optional, Tuple
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from io import BytesIO
|
||||
from image_providers import create_image_provider
|
||||
from image_providers.atlascloud import atlascloud_model_for_operation
|
||||
|
||||
DEFAULT_MODEL = "gpt-image-2"
|
||||
DEFAULT_SIZE = "2560x1440"
|
||||
@@ -95,10 +98,34 @@ def _api_base_url() -> Optional[str]:
|
||||
def _api_target_label() -> str:
|
||||
base_url = _api_base_url()
|
||||
if base_url:
|
||||
return f"OpenAI-compatible proxy (OPENAI_BASE_URL={base_url})"
|
||||
if _is_atlascloud_base_url(base_url):
|
||||
return f"AtlasCloud provider adapter (OPENAI_BASE_URL={base_url})"
|
||||
return f"third-party image API or OpenAI-compatible proxy (OPENAI_BASE_URL={base_url})"
|
||||
return "official OpenAI API (OPENAI_BASE_URL unset)"
|
||||
|
||||
|
||||
def _is_atlascloud_base_url(base_url: str) -> bool:
|
||||
hostname = urlparse(base_url).hostname or ""
|
||||
return "atlascloud.ai" in hostname.lower()
|
||||
|
||||
|
||||
def _preview_endpoint(kind: str) -> str:
|
||||
base_url = _api_base_url()
|
||||
if base_url and _is_atlascloud_base_url(base_url):
|
||||
return "/api/v1/model/generateImage"
|
||||
if kind == "edit":
|
||||
return "/v1/images/edits"
|
||||
return "/v1/images/generations"
|
||||
|
||||
|
||||
def _preview_model(model: str, kind: str) -> str:
|
||||
base_url = _api_base_url()
|
||||
if base_url and _is_atlascloud_base_url(base_url):
|
||||
operation = "edit" if kind == "edit" else "text-to-image"
|
||||
return atlascloud_model_for_operation(model, operation)
|
||||
return model
|
||||
|
||||
|
||||
def _runtime_python_path() -> str:
|
||||
home = _runtime_home()
|
||||
if os.name == "nt":
|
||||
@@ -478,37 +505,6 @@ def _decode_write_and_downscale(
|
||||
print(f"Wrote {derived}")
|
||||
|
||||
|
||||
def _create_client():
|
||||
try:
|
||||
from openai import OpenAI
|
||||
except ImportError:
|
||||
_die(f"openai SDK not installed in the active environment. {_dependency_hint('openai')}")
|
||||
return OpenAI(
|
||||
api_key=os.getenv("OPENAI_API_KEY"),
|
||||
base_url=os.getenv("OPENAI_BASE_URL") or None,
|
||||
)
|
||||
|
||||
|
||||
def _create_async_client():
|
||||
try:
|
||||
from openai import AsyncOpenAI
|
||||
except ImportError:
|
||||
try:
|
||||
import openai as _openai # noqa: F401
|
||||
except ImportError:
|
||||
_die(
|
||||
f"openai SDK not installed in the active environment. {_dependency_hint('openai')}"
|
||||
)
|
||||
_die(
|
||||
"AsyncOpenAI not available in this openai SDK version. "
|
||||
f"{_dependency_hint('openai', upgrade=True)}"
|
||||
)
|
||||
return AsyncOpenAI(
|
||||
api_key=os.getenv("OPENAI_API_KEY"),
|
||||
base_url=os.getenv("OPENAI_BASE_URL") or None,
|
||||
)
|
||||
|
||||
|
||||
def _slugify(value: str) -> str:
|
||||
value = value.strip().lower()
|
||||
value = re.sub(r"[^a-z0-9]+", "-", value)
|
||||
@@ -596,68 +592,6 @@ def _job_output_paths(
|
||||
]
|
||||
|
||||
|
||||
def _extract_retry_after_seconds(exc: Exception) -> Optional[float]:
|
||||
# Best-effort: openai SDK errors vary by version. Prefer a conservative fallback.
|
||||
for attr in ("retry_after", "retry_after_seconds"):
|
||||
val = getattr(exc, attr, None)
|
||||
if isinstance(val, (int, float)) and val >= 0:
|
||||
return float(val)
|
||||
msg = str(exc)
|
||||
m = re.search(r"retry[- ]after[:= ]+([0-9]+(?:\\.[0-9]+)?)", msg, re.IGNORECASE)
|
||||
if m:
|
||||
try:
|
||||
return float(m.group(1))
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _is_rate_limit_error(exc: Exception) -> bool:
|
||||
name = exc.__class__.__name__.lower()
|
||||
if "ratelimit" in name or "rate_limit" in name:
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return "429" in msg or "rate limit" in msg or "too many requests" in msg
|
||||
|
||||
|
||||
def _is_transient_error(exc: Exception) -> bool:
|
||||
if _is_rate_limit_error(exc):
|
||||
return True
|
||||
name = exc.__class__.__name__.lower()
|
||||
if "timeout" in name or "timedout" in name or "tempor" in name:
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return "timeout" in msg or "timed out" in msg or "connection reset" in msg
|
||||
|
||||
|
||||
async def _generate_one_with_retries(
|
||||
client: Any,
|
||||
payload: Dict[str, Any],
|
||||
*,
|
||||
attempts: int,
|
||||
job_label: str,
|
||||
) -> Any:
|
||||
last_exc: Optional[Exception] = None
|
||||
for attempt in range(1, attempts + 1):
|
||||
try:
|
||||
return await client.images.generate(**payload)
|
||||
except Exception as exc:
|
||||
last_exc = exc
|
||||
if not _is_transient_error(exc):
|
||||
raise
|
||||
if attempt == attempts:
|
||||
raise
|
||||
sleep_s = _extract_retry_after_seconds(exc)
|
||||
if sleep_s is None:
|
||||
sleep_s = min(60.0, 2.0**attempt)
|
||||
print(
|
||||
f"{job_label} attempt {attempt}/{attempts} failed ({exc.__class__.__name__}); retrying in {sleep_s:.1f}s",
|
||||
file=sys.stderr,
|
||||
)
|
||||
await asyncio.sleep(sleep_s)
|
||||
raise last_exc or RuntimeError("unknown error")
|
||||
|
||||
|
||||
async def _run_generate_batch(args: argparse.Namespace) -> int:
|
||||
jobs = _read_jobs_jsonl(args.input)
|
||||
out_dir = Path(args.out_dir)
|
||||
@@ -708,16 +642,19 @@ async def _run_generate_batch(args: argparse.Namespace) -> int:
|
||||
]
|
||||
_print_request(
|
||||
{
|
||||
"endpoint": "/v1/images/generations",
|
||||
"endpoint": _preview_endpoint("generate"),
|
||||
"job": i,
|
||||
"outputs": [str(p) for p in outputs],
|
||||
"outputs_downscaled": downscaled,
|
||||
**job_payload,
|
||||
**{
|
||||
**job_payload,
|
||||
"model": _preview_model(str(job_payload["model"]), "generate"),
|
||||
},
|
||||
}
|
||||
)
|
||||
return 0
|
||||
|
||||
client = _create_async_client()
|
||||
provider = create_image_provider(api_key=os.getenv("OPENAI_API_KEY"), base_url=_api_base_url())
|
||||
sem = asyncio.Semaphore(args.concurrency)
|
||||
|
||||
any_failed = False
|
||||
@@ -753,15 +690,13 @@ async def _run_generate_batch(args: argparse.Namespace) -> int:
|
||||
async with sem:
|
||||
print(f"{job_label} starting", file=sys.stderr)
|
||||
started = time.time()
|
||||
result = await _generate_one_with_retries(
|
||||
client,
|
||||
images = await provider.generate_batch(
|
||||
payload,
|
||||
attempts=args.max_attempts,
|
||||
job_label=job_label,
|
||||
)
|
||||
elapsed = time.time() - started
|
||||
print(f"{job_label} completed in {elapsed:.1f}s", file=sys.stderr)
|
||||
images = [item.b64_json for item in result.data]
|
||||
_decode_write_and_downscale(
|
||||
images,
|
||||
outputs,
|
||||
@@ -825,10 +760,13 @@ def _generate(args: argparse.Namespace) -> None:
|
||||
if args.dry_run:
|
||||
_print_request(
|
||||
{
|
||||
"endpoint": "/v1/images/generations",
|
||||
"endpoint": _preview_endpoint("generate"),
|
||||
"outputs": [str(p) for p in output_paths],
|
||||
"outputs_downscaled": downscaled,
|
||||
**payload,
|
||||
**{
|
||||
**payload,
|
||||
"model": _preview_model(str(payload["model"]), "generate"),
|
||||
},
|
||||
}
|
||||
)
|
||||
return
|
||||
@@ -838,12 +776,11 @@ def _generate(args: argparse.Namespace) -> None:
|
||||
file=sys.stderr,
|
||||
)
|
||||
started = time.time()
|
||||
client = _create_client()
|
||||
result = client.images.generate(**payload)
|
||||
provider = create_image_provider(api_key=os.getenv("OPENAI_API_KEY"), base_url=_api_base_url())
|
||||
images = provider.generate(payload)
|
||||
elapsed = time.time() - started
|
||||
print(f"Generation completed in {elapsed:.1f}s.", file=sys.stderr)
|
||||
|
||||
images = [item.b64_json for item in result.data]
|
||||
_decode_write_and_downscale(
|
||||
images,
|
||||
output_paths,
|
||||
@@ -898,10 +835,13 @@ def _edit(args: argparse.Namespace) -> None:
|
||||
payload_preview["mask"] = str(mask_path)
|
||||
_print_request(
|
||||
{
|
||||
"endpoint": "/v1/images/edits",
|
||||
"endpoint": _preview_endpoint("edit"),
|
||||
"outputs": [str(p) for p in output_paths],
|
||||
"outputs_downscaled": downscaled,
|
||||
**payload_preview,
|
||||
**{
|
||||
**payload_preview,
|
||||
"model": _preview_model(str(payload_preview["model"]), "edit"),
|
||||
},
|
||||
}
|
||||
)
|
||||
return
|
||||
@@ -911,18 +851,11 @@ def _edit(args: argparse.Namespace) -> None:
|
||||
file=sys.stderr,
|
||||
)
|
||||
started = time.time()
|
||||
client = _create_client()
|
||||
|
||||
with _open_files(image_paths) as image_files, _open_mask(mask_path) as mask_file:
|
||||
request = dict(payload)
|
||||
request["image"] = image_files if len(image_files) > 1 else image_files[0]
|
||||
if mask_file is not None:
|
||||
request["mask"] = mask_file
|
||||
result = client.images.edit(**request)
|
||||
provider = create_image_provider(api_key=os.getenv("OPENAI_API_KEY"), base_url=_api_base_url())
|
||||
images = provider.edit(payload, image_paths, mask_path)
|
||||
|
||||
elapsed = time.time() - started
|
||||
print(f"Edit completed in {elapsed:.1f}s.", file=sys.stderr)
|
||||
images = [item.b64_json for item in result.data]
|
||||
_decode_write_and_downscale(
|
||||
images,
|
||||
output_paths,
|
||||
@@ -933,60 +866,6 @@ def _edit(args: argparse.Namespace) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _open_files(paths: List[Path]):
|
||||
return _FileBundle(paths)
|
||||
|
||||
|
||||
def _open_mask(mask_path: Optional[Path]):
|
||||
if mask_path is None:
|
||||
return _NullContext()
|
||||
return _SingleFile(mask_path)
|
||||
|
||||
|
||||
class _NullContext:
|
||||
def __enter__(self):
|
||||
return None
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class _SingleFile:
|
||||
def __init__(self, path: Path):
|
||||
self._path = path
|
||||
self._handle = None
|
||||
|
||||
def __enter__(self):
|
||||
self._handle = self._path.open("rb")
|
||||
return self._handle
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
if self._handle:
|
||||
try:
|
||||
self._handle.close()
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
class _FileBundle:
|
||||
def __init__(self, paths: List[Path]):
|
||||
self._paths = paths
|
||||
self._handles: List[object] = []
|
||||
|
||||
def __enter__(self):
|
||||
self._handles = [p.open("rb") for p in self._paths]
|
||||
return self._handles
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
for handle in self._handles:
|
||||
try:
|
||||
handle.close()
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def _add_shared_args(parser: argparse.ArgumentParser) -> None:
|
||||
parser.add_argument("--model", default=_default_model())
|
||||
parser.add_argument("--prompt")
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
"""Image API providers for the codex-ppt fallback CLI."""
|
||||
|
||||
from .base import ImageProvider
|
||||
from .factory import create_image_provider
|
||||
|
||||
__all__ = ["ImageProvider", "create_image_provider"]
|
||||
@@ -0,0 +1,222 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import mimetypes
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import time
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
from urllib.parse import urlparse
|
||||
from urllib.request import Request, urlopen as default_urlopen
|
||||
|
||||
from .base import ImageProvider
|
||||
|
||||
|
||||
UrlOpen = Callable[..., Any]
|
||||
USER_AGENT = "codex-ppt-skill/0.1 (+https://github.com/ningzimu/codex-ppt-skill)"
|
||||
|
||||
|
||||
class AtlasCloudImageProvider(ImageProvider):
|
||||
_FINISHED_STATUSES = {"completed", "succeeded"}
|
||||
_FAILED_STATUSES = {"failed"}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
api_key: Optional[str],
|
||||
base_url: Optional[str],
|
||||
urlopen: UrlOpen = default_urlopen,
|
||||
sleep: Callable[[float], None] = time.sleep,
|
||||
poll_interval: float = 2.0,
|
||||
max_polls: int = 120,
|
||||
) -> None:
|
||||
self.api_key = api_key
|
||||
self.model_base_url = _model_base_url(base_url)
|
||||
self._urlopen = urlopen
|
||||
self._sleep = sleep
|
||||
self.poll_interval = poll_interval
|
||||
self.max_polls = max_polls
|
||||
|
||||
def generate(self, payload: Dict[str, Any]) -> List[str]:
|
||||
count = int(payload.get("n", 1))
|
||||
outputs: List[str] = []
|
||||
for _ in range(count):
|
||||
outputs.extend(self._submit_and_collect(payload, operation="text-to-image"))
|
||||
return outputs
|
||||
|
||||
def edit(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
image_paths: List[Path],
|
||||
mask_path: Optional[Path],
|
||||
) -> List[str]:
|
||||
if mask_path is not None:
|
||||
raise ValueError("AtlasCloud image edit does not support --mask.")
|
||||
|
||||
count = int(payload.get("n", 1))
|
||||
edit_payload = dict(payload)
|
||||
edit_payload["images"] = [_image_to_data_url(path) for path in image_paths]
|
||||
outputs: List[str] = []
|
||||
for _ in range(count):
|
||||
outputs.extend(self._submit_and_collect(edit_payload, operation="edit"))
|
||||
return outputs
|
||||
|
||||
async def generate_batch(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
*,
|
||||
attempts: int,
|
||||
job_label: str,
|
||||
) -> List[str]:
|
||||
last_exc: Optional[Exception] = None
|
||||
for attempt in range(1, attempts + 1):
|
||||
try:
|
||||
return await asyncio.to_thread(self.generate, payload)
|
||||
except Exception as exc:
|
||||
last_exc = exc
|
||||
if attempt == attempts:
|
||||
raise
|
||||
sleep_s = min(60.0, 2.0**attempt)
|
||||
print(
|
||||
f"{job_label} attempt {attempt}/{attempts} failed ({exc.__class__.__name__}); retrying in {sleep_s:.1f}s",
|
||||
file=sys.stderr,
|
||||
)
|
||||
await asyncio.sleep(sleep_s)
|
||||
raise last_exc or RuntimeError("unknown error")
|
||||
|
||||
def _submit_and_collect(self, payload: Dict[str, Any], *, operation: str) -> List[str]:
|
||||
request_payload = self._atlas_payload(payload, operation=operation)
|
||||
submitted = self._request_json(
|
||||
"POST",
|
||||
f"{self.model_base_url}/generateImage",
|
||||
request_payload,
|
||||
)
|
||||
prediction_id = _prediction_id(submitted)
|
||||
if not prediction_id:
|
||||
raise RuntimeError("AtlasCloud response did not include a prediction id.")
|
||||
|
||||
result_url = _prediction_result_url(submitted) or f"{self.model_base_url}/result/{prediction_id}"
|
||||
result = self._poll_prediction(result_url)
|
||||
outputs = result.get("outputs")
|
||||
if not isinstance(outputs, list) or not outputs:
|
||||
raise RuntimeError("AtlasCloud prediction completed without outputs.")
|
||||
return [self._output_to_b64(str(output)) for output in outputs]
|
||||
|
||||
def _poll_prediction(self, url: str) -> Dict[str, Any]:
|
||||
last: Dict[str, Any] = {}
|
||||
for _ in range(self.max_polls):
|
||||
last = self._request_json("GET", url)
|
||||
status = str(last.get("status", "")).lower()
|
||||
if status in self._FINISHED_STATUSES:
|
||||
return last
|
||||
if status in self._FAILED_STATUSES:
|
||||
raise RuntimeError(f"AtlasCloud prediction failed: {last}")
|
||||
self._sleep(self.poll_interval)
|
||||
raise TimeoutError(f"AtlasCloud prediction timed out: {last}")
|
||||
|
||||
def _atlas_payload(self, payload: Dict[str, Any], *, operation: str) -> Dict[str, Any]:
|
||||
output_format = payload.get("output_format")
|
||||
if output_format not in (None, "png", "jpeg"):
|
||||
raise ValueError("AtlasCloud supports output_format png or jpeg.")
|
||||
|
||||
body: Dict[str, Any] = {
|
||||
"model": atlascloud_model_for_operation(
|
||||
str(payload.get("model", "gpt-image-2")),
|
||||
operation,
|
||||
),
|
||||
"prompt": payload["prompt"],
|
||||
"enable_sync_mode": False,
|
||||
"enable_base64_output": True,
|
||||
}
|
||||
for key in ("size", "quality", "output_format"):
|
||||
value = payload.get(key)
|
||||
if value is not None and value != "auto":
|
||||
body[key] = value
|
||||
if operation == "edit":
|
||||
body["images"] = payload["images"]
|
||||
return body
|
||||
|
||||
def _request_json(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
payload: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Accept": "application/json",
|
||||
"User-Agent": USER_AGENT,
|
||||
}
|
||||
if payload is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
request = Request(url, data=data, headers=headers, method=method)
|
||||
with self._urlopen(request, timeout=60) as response:
|
||||
parsed = json.loads(response.read().decode("utf-8"))
|
||||
if not isinstance(parsed, dict):
|
||||
raise RuntimeError(f"Unexpected AtlasCloud response: {parsed}")
|
||||
code = parsed.get("code")
|
||||
if code is not None and code not in (0, 200, "0", "200"):
|
||||
raise RuntimeError(f"AtlasCloud API error: {parsed}")
|
||||
data_obj = parsed.get("data", parsed)
|
||||
if not isinstance(data_obj, dict):
|
||||
raise RuntimeError(f"Unexpected AtlasCloud response data: {parsed}")
|
||||
return data_obj
|
||||
|
||||
def _output_to_b64(self, value: str) -> str:
|
||||
if value.startswith("data:") and "," in value:
|
||||
return value.split(",", 1)[1]
|
||||
parsed = urlparse(value)
|
||||
if parsed.scheme in {"http", "https"}:
|
||||
request = Request(value, headers={"User-Agent": USER_AGENT}, method="GET")
|
||||
with self._urlopen(request, timeout=60) as response:
|
||||
return base64.b64encode(response.read()).decode("ascii")
|
||||
return value
|
||||
|
||||
|
||||
def _model_base_url(base_url: Optional[str]) -> str:
|
||||
if not base_url:
|
||||
return "https://api.atlascloud.ai/api/v1/model"
|
||||
parsed = urlparse(base_url)
|
||||
origin = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme and parsed.netloc else ""
|
||||
path = parsed.path.rstrip("/")
|
||||
marker = "/api/v1/model"
|
||||
if marker in path:
|
||||
prefix = path[: path.index(marker) + len(marker)]
|
||||
return f"{origin}{prefix}"
|
||||
if origin:
|
||||
return f"{origin}{marker}"
|
||||
return base_url.rstrip("/")
|
||||
|
||||
|
||||
def atlascloud_model_for_operation(model: str, operation: str) -> str:
|
||||
suffix = "edit" if operation == "edit" else "text-to-image"
|
||||
base = model.rstrip("/")
|
||||
for existing_suffix in ("/text-to-image", "/edit"):
|
||||
if base.endswith(existing_suffix):
|
||||
base = base[: -len(existing_suffix)]
|
||||
break
|
||||
if "/" not in base:
|
||||
base = f"openai/{base}"
|
||||
return f"{base}/{suffix}"
|
||||
|
||||
|
||||
def _prediction_id(data: Dict[str, Any]) -> Optional[str]:
|
||||
value = data.get("id") or data.get("prediction_id")
|
||||
return str(value) if value else None
|
||||
|
||||
|
||||
def _prediction_result_url(data: Dict[str, Any]) -> Optional[str]:
|
||||
urls = data.get("urls")
|
||||
if not isinstance(urls, dict):
|
||||
return None
|
||||
value = urls.get("get")
|
||||
return str(value) if value else None
|
||||
|
||||
|
||||
def _image_to_data_url(path: Path) -> str:
|
||||
mime = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
|
||||
encoded = base64.b64encode(path.read_bytes()).decode("ascii")
|
||||
return f"data:{mime};base64,{encoded}"
|
||||
@@ -0,0 +1,32 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
class ImageProvider(ABC):
|
||||
"""Common API shape used by image_gen.py command handlers."""
|
||||
|
||||
@abstractmethod
|
||||
def generate(self, payload: Dict[str, Any]) -> List[str]:
|
||||
"""Generate images and return base64-encoded image payloads."""
|
||||
|
||||
@abstractmethod
|
||||
def edit(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
image_paths: List[Path],
|
||||
mask_path: Optional[Path],
|
||||
) -> List[str]:
|
||||
"""Edit input images and return base64-encoded image payloads."""
|
||||
|
||||
@abstractmethod
|
||||
async def generate_batch(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
*,
|
||||
attempts: int,
|
||||
job_label: str,
|
||||
) -> List[str]:
|
||||
"""Generate one batch job with provider-specific retry behavior."""
|
||||
@@ -0,0 +1,21 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from .atlascloud import AtlasCloudImageProvider
|
||||
from .base import ImageProvider
|
||||
from .openai_compatible import OpenAICompatibleImageProvider
|
||||
|
||||
|
||||
def create_image_provider(*, api_key: Optional[str], base_url: Optional[str]) -> ImageProvider:
|
||||
if _is_atlascloud_base_url(base_url):
|
||||
return AtlasCloudImageProvider(api_key=api_key, base_url=base_url)
|
||||
return OpenAICompatibleImageProvider(api_key=api_key, base_url=base_url)
|
||||
|
||||
|
||||
def _is_atlascloud_base_url(base_url: Optional[str]) -> bool:
|
||||
if not base_url:
|
||||
return False
|
||||
hostname = urlparse(base_url).hostname or ""
|
||||
return "atlascloud.ai" in hostname.lower()
|
||||
@@ -0,0 +1,238 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
from .base import ImageProvider
|
||||
|
||||
|
||||
ClientFactory = Callable[[], Any]
|
||||
DEFAULT_RUNTIME_HOME = "~/.codex-ppt-skill"
|
||||
|
||||
|
||||
def _runtime_home() -> Path:
|
||||
return Path(os.getenv("CODEX_PPT_HOME", DEFAULT_RUNTIME_HOME)).expanduser()
|
||||
|
||||
|
||||
def _runtime_python_path() -> str:
|
||||
home = _runtime_home()
|
||||
if os.name == "nt":
|
||||
return str(home / ".venv" / "Scripts" / "python.exe")
|
||||
return str(home / ".venv" / "bin" / "python")
|
||||
|
||||
|
||||
def _skill_root() -> Path:
|
||||
return Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _dependency_hint(package: str, *, upgrade: bool = False) -> str:
|
||||
package_arg = f"-U {package}" if upgrade else package
|
||||
runtime_python = _runtime_python_path()
|
||||
requirements = _skill_root() / "requirements.txt"
|
||||
return (
|
||||
"Install codex-ppt dependencies in the shared runtime first, for example "
|
||||
f"`python3 {_skill_root() / 'scripts' / 'codex_ppt_runtime.py'} bootstrap`, "
|
||||
f"or install {package} directly with `{runtime_python} -m pip install "
|
||||
f"{package_arg}`. Requirements file: `{requirements}`."
|
||||
)
|
||||
|
||||
|
||||
def _extract_retry_after_seconds(exc: Exception) -> Optional[float]:
|
||||
for attr in ("retry_after", "retry_after_seconds"):
|
||||
val = getattr(exc, attr, None)
|
||||
if isinstance(val, (int, float)) and val >= 0:
|
||||
return float(val)
|
||||
msg = str(exc)
|
||||
m = re.search(r"retry[- ]after[:= ]+([0-9]+(?:\\.[0-9]+)?)", msg, re.IGNORECASE)
|
||||
if m:
|
||||
try:
|
||||
return float(m.group(1))
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _is_rate_limit_error(exc: Exception) -> bool:
|
||||
name = exc.__class__.__name__.lower()
|
||||
if "ratelimit" in name or "rate_limit" in name:
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return "429" in msg or "rate limit" in msg or "too many requests" in msg
|
||||
|
||||
|
||||
def _is_transient_error(exc: Exception) -> bool:
|
||||
if _is_rate_limit_error(exc):
|
||||
return True
|
||||
name = exc.__class__.__name__.lower()
|
||||
if "timeout" in name or "timedout" in name or "tempor" in name:
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return "timeout" in msg or "timed out" in msg or "connection reset" in msg
|
||||
|
||||
|
||||
async def _generate_one_with_retries(
|
||||
client: Any,
|
||||
payload: Dict[str, Any],
|
||||
*,
|
||||
attempts: int,
|
||||
job_label: str,
|
||||
) -> Any:
|
||||
last_exc: Optional[Exception] = None
|
||||
for attempt in range(1, attempts + 1):
|
||||
try:
|
||||
return await client.images.generate(**payload)
|
||||
except Exception as exc:
|
||||
last_exc = exc
|
||||
if not _is_transient_error(exc):
|
||||
raise
|
||||
if attempt == attempts:
|
||||
raise
|
||||
sleep_s = _extract_retry_after_seconds(exc)
|
||||
if sleep_s is None:
|
||||
sleep_s = min(60.0, 2.0**attempt)
|
||||
print(
|
||||
f"{job_label} attempt {attempt}/{attempts} failed ({exc.__class__.__name__}); retrying in {sleep_s:.1f}s",
|
||||
file=sys.stderr,
|
||||
)
|
||||
await asyncio.sleep(sleep_s)
|
||||
raise last_exc or RuntimeError("unknown error")
|
||||
|
||||
|
||||
class OpenAICompatibleImageProvider(ImageProvider):
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
api_key: Optional[str],
|
||||
base_url: Optional[str],
|
||||
client_factory: Optional[ClientFactory] = None,
|
||||
async_client_factory: Optional[ClientFactory] = None,
|
||||
) -> None:
|
||||
self.api_key = api_key
|
||||
self.base_url = base_url
|
||||
self._client_factory = client_factory
|
||||
self._async_client_factory = async_client_factory
|
||||
self._async_client: Optional[Any] = None
|
||||
|
||||
def generate(self, payload: Dict[str, Any]) -> List[str]:
|
||||
result = self._create_client().images.generate(**payload)
|
||||
return [item.b64_json for item in result.data]
|
||||
|
||||
def edit(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
image_paths: List[Path],
|
||||
mask_path: Optional[Path],
|
||||
) -> List[str]:
|
||||
with _open_files(image_paths) as image_files, _open_mask(mask_path) as mask_file:
|
||||
request = dict(payload)
|
||||
request["image"] = image_files if len(image_files) > 1 else image_files[0]
|
||||
if mask_file is not None:
|
||||
request["mask"] = mask_file
|
||||
result = self._create_client().images.edit(**request)
|
||||
return [item.b64_json for item in result.data]
|
||||
|
||||
async def generate_batch(
|
||||
self,
|
||||
payload: Dict[str, Any],
|
||||
*,
|
||||
attempts: int,
|
||||
job_label: str,
|
||||
) -> List[str]:
|
||||
result = await _generate_one_with_retries(
|
||||
self._create_async_client(),
|
||||
payload,
|
||||
attempts=attempts,
|
||||
job_label=job_label,
|
||||
)
|
||||
return [item.b64_json for item in result.data]
|
||||
|
||||
def _create_client(self) -> Any:
|
||||
if self._client_factory is not None:
|
||||
return self._client_factory()
|
||||
try:
|
||||
from openai import OpenAI
|
||||
except ImportError as exc:
|
||||
raise RuntimeError(
|
||||
f"openai SDK not installed in the active environment. {_dependency_hint('openai')}"
|
||||
) from exc
|
||||
return OpenAI(api_key=self.api_key, base_url=self.base_url)
|
||||
|
||||
def _create_async_client(self) -> Any:
|
||||
if self._async_client is not None:
|
||||
return self._async_client
|
||||
if self._async_client_factory is not None:
|
||||
self._async_client = self._async_client_factory()
|
||||
return self._async_client
|
||||
try:
|
||||
from openai import AsyncOpenAI
|
||||
except ImportError as exc:
|
||||
try:
|
||||
import openai as _openai # noqa: F401
|
||||
except ImportError:
|
||||
raise RuntimeError(
|
||||
f"openai SDK not installed in the active environment. {_dependency_hint('openai')}"
|
||||
) from exc
|
||||
raise RuntimeError(
|
||||
"AsyncOpenAI not available in this openai SDK version. "
|
||||
f"{_dependency_hint('openai', upgrade=True)}"
|
||||
) from exc
|
||||
self._async_client = AsyncOpenAI(api_key=self.api_key, base_url=self.base_url)
|
||||
return self._async_client
|
||||
|
||||
|
||||
def _open_files(paths: List[Path]):
|
||||
return _FileBundle(paths)
|
||||
|
||||
|
||||
def _open_mask(mask_path: Optional[Path]):
|
||||
if mask_path is None:
|
||||
return _NullContext()
|
||||
return _SingleFile(mask_path)
|
||||
|
||||
|
||||
class _NullContext:
|
||||
def __enter__(self):
|
||||
return None
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class _SingleFile:
|
||||
def __init__(self, path: Path):
|
||||
self._path = path
|
||||
self._handle = None
|
||||
|
||||
def __enter__(self):
|
||||
self._handle = self._path.open("rb")
|
||||
return self._handle
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
if self._handle:
|
||||
try:
|
||||
self._handle.close()
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
class _FileBundle:
|
||||
def __init__(self, paths: List[Path]):
|
||||
self._paths = paths
|
||||
self._handles: List[object] = []
|
||||
|
||||
def __enter__(self):
|
||||
self._handles = [p.open("rb") for p in self._paths]
|
||||
return self._handles
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
for handle in self._handles:
|
||||
try:
|
||||
handle.close()
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
Reference in New Issue
Block a user