lixi
8275a04903
CI / ai-gates (push) Successful in 38s
新增 app/providers/openai/：client 传输（JSON + multipart）、errors 错误映射、 models 厂商 DTO、capabilities 尺寸算术约束、tasks 同步→异步桥、image 图片业务。 装配与配置同步接入，缺 OPENAI_API_KEY 时跳过装配、服务仍可启动。 与火山视频的三处根本差异及其处置： - 接口同步（无任务 ID、无查询端点）→ SyncTaskRegistry 在本服务内部吸收， 对外仍是 ADR-023 §4 的两段式提交/轮询。代价：本服务从此持有状态，进程重启 丢在途任务，且横向扩容前须先解决任务 ID 只在签发进程内有效的问题。 - 产物是 base64 字节而非 URL → 无过期问题，但字节只在内存里。asset_url 放 data URI 仅为满足领域不变式，绝不能原样写进 media.assets.external_url。 - 图生图走 multipart → 先下载参考图再上传；下载失败与超限按参数错误同步拒绝， 不变成注定失败的后台任务。 两处需要记住的判断： - 尺寸上限「3840x2160」文档未说清是逐边还是按像素。逐边理解会禁掉所有高于 2160 的竖图，而 1024x1536 是文档明示的标准竖图，故必然过严。采用最长边 ≤3840 且总像素 ≤3840×2160，已在 capabilities.py 与取证页标为推断。 - 两份文档都没有错误码表，故 errors.py 以 HTTP 状态码为主、type/code 为辅， 顶部标注「未经文档证实」。429 无线索时按限流而非配额耗尽（猜成配额会把可 重试的限流判死）；400 无线索时归服务端错误而非内容被拒（错报会让用户去改 一个其实没问题的提示词）。 一处死锁已修并回归钉住：SyncTaskRegistry 原用 add_done_callback 记录完成时刻， 而快速完成的任务在注册时已 done，回调便同步地在持有非可重入锁的线程里执行。 改为首次被观察到时惰性记录。 测试 166 → 271（+105）。四道门禁逐条读退出码，全绿。 产物转存仍阻塞于存储决策（ADR-023 §7），图片模型因此未播种进 V7 目录。
124 lines
5.4 KiB
Python
124 lines
5.4 KiB
Python
"""内部接口路由。
|
||
|
||
**本层只做编排，不含任何厂商知识**：解析请求 → 按 `media_kind` 找适配器 → 调用 →
|
||
把领域结果序列化。厂商差异全在 `app/providers/<厂商>/` 内。
|
||
|
||
一处刻意的选择：路径参数是**服务商的任务 ID**，不是我方 `job_id`。本服务不持有
|
||
状态，唯一能用来问服务商的就是这个 ID；若用 `job_id` 就得存一份映射，「无状态」
|
||
不再成立，且会多出一个「Java 拿到 ID 但查不到任何东西」的中间态。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from fastapi import APIRouter, Depends, Request
|
||
|
||
from app.domain.generation import MediaKind
|
||
from app.providers.base import GenerationProvider
|
||
from app.providers.openai.client import OpenAiError
|
||
from app.providers.registry import ProviderRegistry
|
||
from app.providers.volcengine.client import ArkError
|
||
|
||
from .auth import verify_service_token
|
||
from .errors import ProviderError, ServiceError, ServiceErrorCode
|
||
from .schemas import SubmitRequest, SubmitResponse, TaskResponse
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(
|
||
prefix="/internal/generation",
|
||
dependencies=[Depends(verify_service_token)],
|
||
)
|
||
|
||
|
||
def _registry(request: Request) -> ProviderRegistry:
|
||
return request.app.state.registry
|
||
|
||
|
||
def _provider_for(
|
||
registry: ProviderRegistry, media_kind: MediaKind
|
||
) -> GenerationProvider:
|
||
"""按 `media_kind` 取适配器。
|
||
|
||
查不到即 `provider_unavailable`（503）而不是 500：这通常是凭证未配置导致适配器
|
||
没装配，属环境问题而非代码缺陷，且对调用方而言是**可重试**的——配好就能用。
|
||
"""
|
||
for provider in registry.all():
|
||
if provider.media_kind is media_kind:
|
||
return provider
|
||
raise ServiceError(
|
||
ServiceErrorCode.PROVIDER_UNAVAILABLE,
|
||
f"没有可用的 {media_kind.value} 生成适配器（检查服务商凭证是否已配置）",
|
||
)
|
||
|
||
|
||
@router.post("/tasks", response_model=SubmitResponse)
|
||
def submit_task(payload: SubmitRequest, request: Request) -> SubmitResponse:
|
||
"""提交生成任务，返回服务商任务 ID。
|
||
|
||
调用方**必须在上游产生任何计费动作之前**把该 ID 落库——火山不支持幂等键，
|
||
这是提交超时后唯一能判断「是否已提交」的依据。
|
||
"""
|
||
provider = _provider_for(_registry(request), payload.media_kind)
|
||
try:
|
||
provider_request_id = provider.submit(payload.to_domain())
|
||
except (ArkError, OpenAiError) as exc:
|
||
# 服务商调用失败：原样带上分类，调用方据 retry 决定重试。
|
||
raise ProviderError(exc.failure) from exc
|
||
except ValueError as exc:
|
||
# 本地预校验拦下的参数问题（超出该模型的时长/尺寸/ratio 限制、内容为空、
|
||
# 参考图不可下载）。属 invalid_request 而非 provider 错误——请求根本没发出去。
|
||
raise ServiceError(ServiceErrorCode.INVALID_REQUEST, str(exc)) from exc
|
||
except RuntimeError as exc:
|
||
# 同步型适配器的在途任务已达上限。这不是请求的问题而是容量问题，且稍后
|
||
# 重试会好，故归 provider_unavailable（503）而非 invalid_request——后者会
|
||
# 让调用方判为不可重试并直接失败掉一个本可成功的任务。
|
||
raise ServiceError(ServiceErrorCode.PROVIDER_UNAVAILABLE, str(exc)) from exc
|
||
|
||
log.info(
|
||
"已提交生成任务 job_id=%s provider_request_id=%s",
|
||
payload.job_id,
|
||
provider_request_id,
|
||
)
|
||
return SubmitResponse(provider_request_id=provider_request_id)
|
||
|
||
|
||
@router.get("/tasks/{provider_request_id}", response_model=TaskResponse)
|
||
def get_task(
|
||
provider_request_id: str,
|
||
request: Request,
|
||
media_kind: MediaKind = MediaKind.VIDEO,
|
||
) -> TaskResponse:
|
||
"""查询任务。
|
||
|
||
**任务失败也返回 200**：HTTP 调用成功了，失败的是任务，`state=failed` +
|
||
`failure` 表达它。做成非 2xx 会让「查不到任务」与「任务失败了」难以区分。
|
||
|
||
:param media_kind: 决定问哪个适配器。默认 video——当前只有视频适配器；
|
||
接入图片后调用方应显式传入。刻意不做「遍历所有适配器试一遍」：那会把
|
||
一个任务 ID 拿去问不相干的服务商。
|
||
"""
|
||
provider = _provider_for(_registry(request), media_kind)
|
||
try:
|
||
result = provider.fetch(provider_request_id)
|
||
except KeyError as exc:
|
||
# 同步型适配器的本地注册表里没有这个 ID：从未存在过，或进程重启后丢失
|
||
# （两者无法区分）。两种情形下调用方都该**停止轮询**而非退避重试。
|
||
raise ServiceError(
|
||
ServiceErrorCode.TASK_NOT_FOUND,
|
||
f"本服务无此任务（可能因进程重启而丢失）：{provider_request_id}",
|
||
) from exc
|
||
except ArkError as exc:
|
||
# 服务商说任务不存在（含超过 7 天保留期）→ 这是 service 类的 task_not_found，
|
||
# 而不是 provider 错误：调用方要做的是停止轮询，不是退避重试。
|
||
if exc.http_status == 404:
|
||
raise ServiceError(
|
||
ServiceErrorCode.TASK_NOT_FOUND,
|
||
f"服务商无此任务（或已超出其 7 天保留期）：{provider_request_id}",
|
||
) from exc
|
||
raise ProviderError(exc.failure) from exc
|
||
except OpenAiError as exc:
|
||
raise ProviderError(exc.failure) from exc
|
||
return TaskResponse.of(result)
|