chengma/cmbot
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai-outfit): per-resolution timeout + _to_int default fallback

Browse Source 
- §8 dynamic timeout: add resolution_timeout() (512/1K/2K/4K ->
  180/240/360/600s); generate() uses it for the POST and image download.
  timeout_seconds now defaults to 0 = auto-by-resolution; an explicit
  positive value overrides. Validation allows 0, rejects negatives.
- _to_int now returns the supplied default on parse failure (was 0), so a
  bad connect_timeout_seconds falls back to 30 instead of failing as 0.
- Tests: +5 (resolution map, auto vs override timeout, connect fallback).
- tasks.md §19.1/§19.2: tick unit-test item, note runner is pure-logic.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
admin
2026-06-18 18:03:52 +08:00
co-authored by Claude Opus 4.8
parent 5b57a4d39a
commit 5ca615f7ae
3 changed files with 93 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/ai_image_service.py
+34 -5
View File
@@ -28,6 +28,16 @@ SUPPORTED_API_TYPES = {
_BASE64_KEYS = {"image_base64", "base64", "b64_json", "data"}
_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif"}
# Default read timeout (seconds) per output resolution (docs/11-ai-outfit.md §8).
# Higher resolutions take longer to generate; a fixed timeout_seconds in the
# model config overrides this mapping.
RESOLUTION_TIMEOUTS = {"512": 180, "1K": 240, "2K": 360, "4K": 600}
def resolution_timeout(resolution, default=240):
"""Return the default read timeout in seconds for *resolution*."""
return RESOLUTION_TIMEOUTS.get(str(resolution).strip().upper(), default)
class AiImageServiceError(RuntimeError):
"""Base error for AI image generation service failures."""
@@ -45,7 +55,7 @@ class AiModelConfig:
model: str
api_key: str
api_type: str = API_AUTO
timeout_seconds: int = 240
timeout_seconds: int = 0 # 0 = 按分辨率自动（resolution_timeout）
connect_timeout_seconds: int = 30
extra_body: Any = field(default_factory=dict)
@@ -58,7 +68,7 @@ class AiModelConfig:
model=str(data.get("model", "") or ""),
api_key=str(data.get("api_key", "") or ""),
api_type=str(data.get("api_type", API_AUTO) or API_AUTO),
timeout_seconds=_to_int(data.get("timeout_seconds", 240), 240),
timeout_seconds=_parse_optional_timeout(data.get("timeout_seconds")),
connect_timeout_seconds=_to_int(data.get("connect_timeout_seconds", 30), 30),
extra_body=data.get("extra_body") or {},
)
@@ -84,7 +94,7 @@ def api_config_errors(config):
errors.append("缺少 api_key")
if cfg.api_type not in SUPPORTED_API_TYPES:
errors.append("api_type 不支持: {}".format(cfg.api_type))
if cfg.timeout_seconds <= 0:
if cfg.timeout_seconds < 0: # 0 = 按分辨率自动；负数 = 解析失败/非法
errors.append("timeout_seconds 必须大于 0")
if cfg.connect_timeout_seconds <= 0:
errors.append("connect_timeout_seconds 必须大于 0")
@@ -285,7 +295,12 @@ class ImageApiClient:
url = url.replace("{model}", self.config.model)
headers = {"Authorization": "Bearer {}".format(self.config.api_key)}
timeout = (self.config.connect_timeout_seconds, self.config.timeout_seconds)
read_timeout = (
self.config.timeout_seconds
if self.config.timeout_seconds > 0
else resolution_timeout(resolution)
)
timeout = (self.config.connect_timeout_seconds, read_timeout)
if api_type == API_IMAGES_EDITS:
data, files = build_multipart_fields(self.config, prompt, image_path, resolution)
@@ -310,7 +325,7 @@ class ImageApiClient:
image_bytes = extract_image_from_response(
payload,
session=self.session,
timeout=self.config.timeout_seconds,
timeout=read_timeout,
)
if not image_bytes:
raise AiImageServiceError("AI 响应中未找到图片")
@@ -327,7 +342,21 @@ def _to_int(value, default):
try:
return int(value or default)
except (TypeError, ValueError):
return default
def _parse_optional_timeout(value):
"""Parse an optional read timeout.
Absent/blank -> 0 (auto by resolution); an explicit positive value overrides
the per-resolution default; an unparseable value -> -1 so validation flags it.
"""
if value is None or value == "":
return 0
try:
return int(value)
except (TypeError, ValueError):
return -1
def _join_url(base_url, suffix):
tasks.md
+4 -4
View File
@@ -1049,15 +1049,15 @@
- [x] `core/models.py` 新增 `OutfitTask` / `OutfitResult`（纯 dataclass，Python 3.7 兼容，不依赖 PySide6）
- [x] `services/excel_service.py`：读行 → `List[OutfitTask]`、写回 D/E/F、占用检测、跳过「完成」/按设置重试「失败」/空字段安全跳过、每行即存
- [x] `services/ai_image_service.py`：移植旧项目 `ImageApiClient`（多模型、多请求格式 `chat/gemini/images/images_edits`、传图 data-url、递归取图、URL 归一化、字段校验）；注意 PEP585 类型注解改 Python 3.7 写法
- [x] `core/ai_outfit.py`：单行生成纯逻辑编排（提示词渲染 + 调用 + 保存 JPG + 产出 `OutfitResult`）
- [ ] 单测：Excel 读写、提示词渲染、取图、命名去重（API 用 mock）；可在 Python 3.7 运行、不依赖 GUI
- [x] `core/ai_outfit.py`：单行生成纯逻辑编排（提示词渲染 + 调用 + 保存 JPG + 产出 `OutfitResult`）；§8「超时按分辨率动态决定（512/1K/2K/4K→180/240/360/600，可被 `timeout_seconds` 覆盖）」由 `ai_image_service.resolution_timeout` 实现
- [x] 单测：`test_excel_service` / `test_ai_image_service` / `test_ai_outfit` / `test_outfit_batch` 共 40+ 用例（Excel 读写、提示词渲染、取图、命名去重、超时映射；API 用 mock）；Python 3.7 通过、不依赖 GUI
### 19.2 批量编排 — docs/11 §14 阶段 2
前置阅读：`docs/11-ai-outfit.md`（§8）
- [x] `Worker(QObject)`：`ThreadPoolExecutor(并发数)` + `RateLimiter(请求间隔)` + 单任务冷却 + 阶梯重试 + 温和停止 + >30s 心跳；纯逻辑尽量可测
- [x] 子线程只经 signal 回主线程，不直接碰控件
- [x] `core/outfit_batch.py` 纯逻辑 `OutfitBatchRunner`：`ThreadPoolExecutor(并发数)` + `RateLimiter(请求间隔)` + 单任务冷却 + 阶梯重试 + 温和停止 + >30s 心跳；全可测（`QObject` 包装放 §19.3 UI 接线）
- [x] 子线程只经回调/signal 回主线程，不直接碰控件（runner 用 callback，UI 层转 signal）
### 19.3 UI 页签 — docs/11 §14 阶段 3
tests/test_ai_image_service.py
+55
View File
@@ -215,6 +215,59 @@ class TestImageApiClient(unittest.TestCase):
self.assertEqual(session.last_headers["Authorization"], "Bearer secret")
self.assertEqual(session.last_json["model"], "model-x")
def test_generate_uses_resolution_timeout_when_unset(self):
from services.ai_image_service import ImageApiClient
session = _FakeSession()
config = {"url": "https://api.example.test", "model": "m", "api_key": "k", "api_type": "chat"}
ImageApiClient(config, session=session).generate("p", self.image_path, resolution="4K")
# timeout_seconds absent -> (connect 30, read 600 for 4K)
self.assertEqual(session.last_timeout, (30, 600))
def test_generate_explicit_timeout_overrides_resolution(self):
from services.ai_image_service import ImageApiClient
session = _FakeSession()
config = {
"url": "https://api.example.test",
"model": "m",
"api_key": "k",
"api_type": "chat",
"timeout_seconds": 120,
}
ImageApiClient(config, session=session).generate("p", self.image_path, resolution="4K")
self.assertEqual(session.last_timeout, (30, 120))
class TestTimeoutHelpers(unittest.TestCase):
def test_resolution_timeout_mapping(self):
from services.ai_image_service import resolution_timeout
self.assertEqual(resolution_timeout("512"), 180)
self.assertEqual(resolution_timeout("1K"), 240)
self.assertEqual(resolution_timeout("2k"), 360) # case-insensitive
self.assertEqual(resolution_timeout("4K"), 600)
self.assertEqual(resolution_timeout("weird"), 240) # unknown -> default
def test_blank_timeout_is_auto_and_valid(self):
from services.ai_image_service import AiModelConfig, api_config_errors
cfg = AiModelConfig.from_dict({"url": "https://x", "model": "m", "api_key": "k"})
self.assertEqual(cfg.timeout_seconds, 0) # 0 = auto by resolution
self.assertEqual(api_config_errors(cfg), [])
def test_invalid_connect_timeout_falls_back_to_default(self):
from services.ai_image_service import AiModelConfig
cfg = AiModelConfig.from_dict(
{"url": "https://x", "model": "m", "api_key": "k", "connect_timeout_seconds": "bad"}
)
self.assertEqual(cfg.connect_timeout_seconds, 30)
class _FakeResponse:
def __init__(self, payload=None, content=b"downloaded"):
@@ -239,6 +292,7 @@ class _FakeSession:
self.last_post_url = None
self.last_headers = None
self.last_json = None
self.last_timeout = None
def get(self, url, timeout=60):
self.last_url = url
@@ -248,6 +302,7 @@ class _FakeSession:
self.last_post_url = url
self.last_headers = headers
self.last_json = json
self.last_timeout = timeout
return _FakeResponse()