From b7bd8395eda8621d3b27474d3c79895cf4c701a0 Mon Sep 17 00:00:00 2001 From: Johnny Zhang Date: Mon, 24 Aug 2026 00:44:41 +0800 Subject: [PATCH] =?UTF-8?q?refactor(providers):=20=E6=8A=8A=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E5=8D=8F=E8=AE=AE=E6=8F=90=E5=8F=96=E4=B8=BA=E6=98=BE?= =?UTF-8?q?=E5=BC=8F=E6=8E=A5=E7=BC=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 请求构造与响应解析原先写死在 sufy.py 的条件分支里,路径、鉴权与轮询协议三样都由 adapter 自己拼。新增一个协议面因此只能继续往同一个文件里加分支 —— 上一次那样做的 412 行最终被整体删除,理由是接不进产品链路。 协议层只产出纯数据:build_submit/parse_submit/build_poll/parse_poll/build_fetch 都是纯 函数,不发请求、不重试、不休眠;轮询节奏与 52x 重发仍归 adapter。鉴权头由协议层产出 而不由厂商层统一注入,因为 Key 与 Bearer 写反时的响应与"模型不存在"难以区分。 edge_fingerprint 迁到 gateway.classify:视频面与图像面都要用,留在任一 provider 里会造成 循环导入。AdapterResult 加 result_url —— OpenAI 面的产物地址在轮询响应里,协议层需要 一个交回它的位置,否则只能另造一个平行的结果类型。 OpenAI 面行为不变,由既有 sufy 视频用例约束;新增 12 条脱网单测覆盖两层之间的契约 (未完成时 error_type 为 None)、鉴权头、以及 image_list 与 input_reference 的字段选择。 --- .../src/windup_framework/gateway/classify.py | 10 + .../src/windup_framework/gateway/types.py | 3 + .../providers/protocol/__init__.py | 10 + .../providers/protocol/openai_video.py | 199 +++++++++++++++++ .../providers/protocol/types.py | 53 +++++ .../src/windup_framework/providers/sufy.py | 205 ++++-------------- backend/tests/test_sufy_video_download.py | 8 +- backend/tests/test_video_protocol.py | 132 +++++++++++ 8 files changed, 448 insertions(+), 172 deletions(-) create mode 100644 backend/packages/framework/src/windup_framework/providers/protocol/__init__.py create mode 100644 backend/packages/framework/src/windup_framework/providers/protocol/openai_video.py create mode 100644 backend/packages/framework/src/windup_framework/providers/protocol/types.py create mode 100644 backend/tests/test_video_protocol.py diff --git a/backend/packages/framework/src/windup_framework/gateway/classify.py b/backend/packages/framework/src/windup_framework/gateway/classify.py index ab04e12c..1c9c3d39 100644 --- a/backend/packages/framework/src/windup_framework/gateway/classify.py +++ b/backend/packages/framework/src/windup_framework/gateway/classify.py @@ -79,6 +79,16 @@ def _looks_like_config_error(status: int, body: str | None) -> bool: return False +#: 52x 出自链路上哪一跳,只能从这几个头看。 +_DIAGNOSTIC_HEADERS = ("server", "cf-ray", "via", "x-served-by", "retry-after") + + +def edge_fingerprint(response: httpx.Response) -> str: + """不记下这几个头,线上就只剩一个状态码可复盘。""" + seen = {k: response.headers.get(k) for k in _DIAGNOSTIC_HEADERS} + return " ".join(f"{k}={v}" for k, v in seen.items() if v) or "无可辨识的边缘响应头" + + def classify_http_response( status: int, body: str | None = None, diff --git a/backend/packages/framework/src/windup_framework/gateway/types.py b/backend/packages/framework/src/windup_framework/gateway/types.py index 2566b51b..bbd2dab3 100644 --- a/backend/packages/framework/src/windup_framework/gateway/types.py +++ b/backend/packages/framework/src/windup_framework/gateway/types.py @@ -44,3 +44,6 @@ class AdapterResult: poll_ms: int | None = None download_ms: int | None = None poll_count: int | None = None + #: 产物地址。OpenAI 面在轮询响应里就给出它,协议层要有地方交回来, + #: 否则只能另造一个平行的结果类型。 + result_url: str | None = None diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py new file mode 100644 index 00000000..753e4202 --- /dev/null +++ b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py @@ -0,0 +1,10 @@ +from .openai_video import IMAGE_LIST_MODELS, OpenAIVideoProtocol +from .types import HttpCall, JobProtocol, VideoRequest + +__all__ = [ + "IMAGE_LIST_MODELS", + "HttpCall", + "JobProtocol", + "OpenAIVideoProtocol", + "VideoRequest", +] diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/openai_video.py b/backend/packages/framework/src/windup_framework/providers/protocol/openai_video.py new file mode 100644 index 00000000..bfa0ef8d --- /dev/null +++ b/backend/packages/framework/src/windup_framework/providers/protocol/openai_video.py @@ -0,0 +1,199 @@ +"""OpenAI 风格 ``/v1/videos`` 面。 + +首帧走 base64 dataURI,产物地址在轮询响应的 ``task_result.videos[0].url`` 里, +所以 ``build_fetch`` 恒为 ``None``。 +""" +from __future__ import annotations + +import base64 +import io + +import httpx + +from windup_common.enums.model import ModelErrorType +from windup_framework.gateway.billing import billing_flags +from windup_framework.gateway.classify import ( + classify_http_response, + edge_fingerprint, + retry_after_seconds, +) +from windup_framework.gateway.types import AdapterResult + +from .types import HttpCall, VideoRequest + +#: 只有 kling-video-o1 走 image_list;v2 系列 / sora 走 input_reference。 +#: 字段按模型选,塞错任务会 failed,而费用可能已产生。 +IMAGE_LIST_MODELS = ("kling-video-o1",) + +#: 透明首帧合成到不透明视频输入时的底色。中灰而不是黑:抠图靠主体与底色的距离判前景, +#: 黑底会把角色的暗部判成背景(#497 的方向已实测为"被抠掉的是最暗部"),白底对浅色角色同理。 +FIRST_FRAME_BG = (128, 128, 128) + + +def fit_first_frame( + frame: bytes, size: str, *, background: tuple[int, int, int] = FIRST_FRAME_BG +) -> bytes: + """首帧 bytes → 等比缩放(可放大) + 补边到目标尺寸 → JPG(RGB,q90) bytes。 + + 不强拉到目标尺寸(母版多为横幅,强压成方会把角色压成瘦长鬼影);JPG 因 PNG base64 + 会 VENDOR_FAILED(实测)。 + + 这一步同时是 kling 系"输出画幅"的唯一控制点:kling 的 i2v 端点没有 resolution/size + 字段,成片画幅跟随首帧,所以 ``size`` 只能在这里生效。 + + 小于目标画布的输入必须**放大**:128x128 的 sprite 原尺寸贴进 1280x720 只占 13% 高, + 等于自愿把主体有效分辨率砍掉七分之六,之后无论 i2v 还是重抠图都补不回来。 + + **放大用 NEAREST,缩小用 LANCZOS。** 放大是把一个源像素铺成一块,插值会在块边界 + 造出源图里没有的中间色:实测一张 256x256 的像素画母版放到 720x720,唯一色从 5982 + 涨到 32479(5.4 倍),硬边糊成渐变,而这张糊图正是喂给 i2v 的输入。缩小反过来, + NEAREST 会丢样出锯齿。交付侧的 ``_fit_to`` 早就是这条规则,这里与它对齐。 + """ + from PIL import Image + + w, h = (int(x) for x in size.split("x")) + im = Image.open(io.BytesIO(frame)) + if im.mode in ("RGBA", "LA") or (im.mode == "P" and "transparency" in im.info): + im = im.convert("RGBA") + flat = Image.new("RGB", im.size, background) + flat.paste(im, (0, 0), im) # 不能 convert("RGB"):透明像素的 RGB 未定义 + im, pad = flat, background + else: + im = im.convert("RGB") + pad = im.getpixel((0, 0)) # 不透明输入沿用角点色,补边与画面自身背景连成一片 + scale = min(w/im.width, h/im.height) + tw, th = max(1, round(im.width*scale)), max(1, round(im.height*scale)) + fitted = im.resize((tw, th), Image.NEAREST if scale > 1 else Image.LANCZOS) + canvas = Image.new("RGB", (w, h), pad) + canvas.paste(fitted, ((w - tw)//2, (h - th)//2)) + buf = io.BytesIO() + canvas.save(buf, "JPEG", quality=90) + return buf.getvalue() + + +def first_frame_datauri(frame: bytes, size: str) -> str: + """首帧 → base64 dataURI(本面专用;FAL 队列面不吃 dataURI)。""" + return "data:image/jpeg;base64," + base64.b64encode(fit_first_frame(frame, size)).decode() + + +def http_error( + resp: httpx.Response, *, job_id: str | None = None, phase: str = "submit" +) -> AdapterResult: + """非 2xx 的响应收成 AdapterResult。 + + 已建单之后的失败一律记 maybe_billed:单据存在就可能已计费,除非请求根本没到上游。 + """ + error_type = classify_http_response(resp.status_code, resp.text, phase=phase) + retry_after_header = resp.headers.get("Retry-After") + retry_after_s = retry_after_seconds(retry_after_header) if retry_after_header else None + maybe_billed = billing_flags(error_type=error_type, http_status=resp.status_code) + if job_id is not None and error_type not in { + ModelErrorType.UNREACHED, + ModelErrorType.NETWORK, + }: + maybe_billed = True + return AdapterResult( + ok=False, + error_type=error_type, + http_status=resp.status_code, + maybe_billed=maybe_billed, + edge_fingerprint=edge_fingerprint(resp), + retry_after_s=retry_after_s, + job_id=job_id, + ) + + +class OpenAIVideoProtocol: + """鉴权头由本层产出而不由厂商层统一注入 —— 写错时的响应与"模型不存在"难以区分。""" + + def __init__(self, api_key: str) -> None: + self._key = api_key + + @property + def _headers(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self._key}"} + + def build_submit(self, req: VideoRequest) -> HttpCall: + body: dict[str, object] = { + "model": req.model, + "prompt": req.prompt, + "size": req.size, + "seconds": str(req.seconds), + "mode": req.mode, + } + datauri = first_frame_datauri(req.first_frame, req.size) + if req.model in IMAGE_LIST_MODELS: + body["image_list"] = [{"image": datauri.split(",", 1)[1]}] + else: + body["input_reference"] = datauri + return HttpCall(method="POST", path="/videos", headers=self._headers, body=body) + + def parse_submit(self, resp: httpx.Response) -> AdapterResult: + if not (200 <= resp.status_code < 300): + return http_error(resp) + try: + payload = resp.json() + except ValueError: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + edge_fingerprint="响应不是 JSON", + ) + jid = payload.get("id") + if not jid: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + edge_fingerprint="响应没有 job id", + ) + return AdapterResult( + ok=True, + job_id=str(jid), + body=b"", + maybe_billed=True, + http_status=resp.status_code, + ) + + def build_poll(self, job_id: str) -> HttpCall: + return HttpCall(method="GET", path=f"/videos/{job_id}", headers=self._headers) + + def parse_poll(self, resp: httpx.Response, job_id: str) -> AdapterResult: + """未完成时 ``error_type`` 为 ``None`` 且 ``ok`` 为假 —— adapter 据此继续轮询。""" + if not (200 <= resp.status_code < 300): + return http_error(resp, job_id=job_id, phase="follow") + try: + st = resp.json() + except ValueError: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + job_id=job_id, + maybe_billed=True, + edge_fingerprint="轮询响应不是 JSON", + ) + status = st.get("status") + if status == "completed": + vids = (st.get("task_result") or {}).get("videos") or [] + return AdapterResult( + ok=True, + job_id=job_id, + maybe_billed=True, + job_status=status, + result_url=vids[0].get("url") if vids else None, + ) + if status in ("failed", "cancelled"): + return AdapterResult( + ok=False, + error_type=ModelErrorType.UPSTREAM_FAILED, + job_id=job_id, + maybe_billed=True, + job_status=status, + edge_fingerprint=str(st.get("error") or ""), + ) + return AdapterResult(ok=False, job_id=job_id, maybe_billed=True, job_status=status) + + def build_fetch(self, job_id: str) -> HttpCall | None: + return None diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/types.py b/backend/packages/framework/src/windup_framework/providers/protocol/types.py new file mode 100644 index 00000000..93292afd --- /dev/null +++ b/backend/packages/framework/src/windup_framework/providers/protocol/types.py @@ -0,0 +1,53 @@ +"""协议层的纯数据结构与接口。 + +一条规则:协议只知道字节怎么排,不发请求、不重试、不休眠。 +请求的构造与响应的解析是纯函数;发请求、轮询节奏、失败处理归 adapter 与 gateway。 +""" +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Protocol + +import httpx + +from windup_framework.gateway.types import AdapterResult + + +@dataclass(frozen=True) +class HttpCall: + """一次调用要发的全部内容。adapter 照着发,不再自己拼路径或补头。""" + + method: str + path: str + headers: Mapping[str, str] = field(default_factory=dict) + body: Mapping[str, object] | None = None + + +@dataclass(frozen=True) +class VideoRequest: + """图生视频的一次请求。首帧仍是 bytes —— 转成什么形状由协议面决定。""" + + model: str + prompt: str + seconds: int + size: str + mode: str + first_frame: bytes + + +class JobProtocol(Protocol): + """建单 → 轮询 → 取结果。各协议面的差别只在路径、鉴权与字段名,形状同构。 + + ``build_fetch`` 返回 ``None`` 表示该面的产物地址已在轮询响应里,无需再取一次。 + """ + + def build_submit(self, req: VideoRequest) -> HttpCall: ... + + def parse_submit(self, resp: httpx.Response) -> AdapterResult: ... + + def build_poll(self, job_id: str) -> HttpCall: ... + + def parse_poll(self, resp: httpx.Response, job_id: str) -> AdapterResult: ... + + def build_fetch(self, job_id: str) -> HttpCall | None: ... diff --git a/backend/packages/framework/src/windup_framework/providers/sufy.py b/backend/packages/framework/src/windup_framework/providers/sufy.py index 70cffd49..a5c9dacf 100644 --- a/backend/packages/framework/src/windup_framework/providers/sufy.py +++ b/backend/packages/framework/src/windup_framework/providers/sufy.py @@ -23,7 +23,6 @@ from __future__ import annotations import base64 -import io import json import logging import re @@ -34,101 +33,23 @@ from windup_common.enums.model import ModelErrorType from windup_framework.config.provider import AIProviderSettings, settings -from windup_framework.gateway.billing import billing_flags from windup_framework.gateway.classify import ( classify_exception, classify_http_response, + edge_fingerprint, retry_after_seconds as _retry_after_seconds, ) from windup_framework.gateway.types import AdapterResult from .interfaces import ImageProvider, VideoProvider +from .protocol import HttpCall, VideoRequest +from .protocol.openai_video import OpenAIVideoProtocol logger = logging.getLogger("windup.providers.sufy") -# 只有 kling-video-o1 走 image_list;v2 系列 / sora 走 input_reference(字段按模型选,塞错任务会 failed)。 -_IMAGE_LIST_MODELS = ("kling-video-o1",) DEFAULT_VIDEO_MODEL = "kling-v2-5-turbo" -#: 透明首帧合成到不透明视频输入时的底色。中灰而不是黑:抠图靠主体与底色的距离 -#: 判前景,黑底会把角色的暗部判成背景(#497 的方向已实测为"被抠掉的是最暗部"), -#: 白底对浅色角色同理。中灰对两端都不偏。 -_FIRST_FRAME_BG = (128, 128, 128) - - -def _fit_first_frame(frame: bytes, size: str, *, background: tuple[int, int, int] = _FIRST_FRAME_BG) -> bytes: - """首帧 bytes → 等比缩放(可放大) + 补边到目标尺寸 → JPG(RGB,q90) bytes。 - - 不强拉到目标尺寸(母版多为横幅,强压成方会把角色压成瘦长鬼影);JPG 因 PNG base64 - 会 VENDOR_FAILED(实测)。 - - 这一步同时是 kling 系"输出画幅"的唯一控制点:kling 的 i2v 端点没有 resolution/size - 字段,成片画幅跟随首帧,所以 ``size`` 只能在这里生效。 - - 小于目标画布的输入必须**放大**:128x128 的 sprite 原尺寸贴进 1280x720 只占 13% 高, - 等于自愿把主体有效分辨率砍掉七分之六,之后无论 i2v 还是重抠图都补不回来。 - - **放大用 NEAREST,缩小用 LANCZOS。** 放大是把一个源像素铺成一块,插值会在块边界 - 造出源图里没有的中间色:实测一张 256x256 的像素画母版放到 720x720,唯一色从 5982 - 涨到 32479(5.4 倍),硬边糊成渐变,而这张糊图正是喂给 i2v 的输入。缩小反过来, - NEAREST 会丢样出锯齿。交付侧的 ``_fit_to`` 早就是这条规则,这里与它对齐。 - """ - from PIL import Image - - w, h = (int(x) for x in size.split("x")) - im = Image.open(io.BytesIO(frame)) - if im.mode in ("RGBA", "LA") or (im.mode == "P" and "transparency" in im.info): - im = im.convert("RGBA") - flat = Image.new("RGB", im.size, background) - flat.paste(im, (0, 0), im) # 不能 convert("RGB"):透明像素的 RGB 未定义 - im, pad = flat, background - else: - im = im.convert("RGB") - pad = im.getpixel((0, 0)) # 不透明输入沿用角点色,补边与画面自身背景连成一片 - scale = min(w/im.width, h/im.height) - tw, th = max(1, round(im.width*scale)), max(1, round(im.height*scale)) - fitted = im.resize((tw, th), Image.NEAREST if scale > 1 else Image.LANCZOS) - canvas = Image.new("RGB", (w, h), pad) - canvas.paste(fitted, ((w - tw)//2, (h - th)//2)) - buf = io.BytesIO() - canvas.save(buf, "JPEG", quality=90) - return buf.getvalue() - - -def _first_frame_datauri(frame: bytes, size: str) -> str: - """首帧 → base64 dataURI(OpenAI 风格 ``/v1/videos`` 面专用;FAL 面不吃 dataURI)。""" - return "data:image/jpeg;base64," + base64.b64encode(_fit_first_frame(frame, size)).decode() - - -def _video_http_error( - resp: httpx.Response, - *, - job_id: str | None = None, - phase: str = "submit", -) -> AdapterResult: - error_type = classify_http_response(resp.status_code, resp.text, phase=phase) - retry_after_header = resp.headers.get("Retry-After") - retry_after_s = ( - _retry_after_seconds(retry_after_header) if retry_after_header else None - ) - maybe_billed = billing_flags(error_type=error_type, http_status=resp.status_code) - if job_id is not None and error_type not in { - ModelErrorType.UNREACHED, - ModelErrorType.NETWORK, - }: - maybe_billed = True - return AdapterResult( - ok=False, - error_type=error_type, - http_status=resp.status_code, - maybe_billed=maybe_billed, - edge_fingerprint=_edge_fingerprint(resp), - retry_after_s=retry_after_s, - job_id=job_id, - ) - - def _transport_result(exc: BaseException) -> AdapterResult: """POST 还没拿到状态行:收成 AdapterResult,让 Gateway 按 UNREACHED 决定是否重发。""" error_type, status, edge = classify_exception(exc) @@ -141,11 +62,14 @@ def _transport_result(exc: BaseException) -> AdapterResult: ) -def _poll_get(client: httpx.Client, job_id: str) -> httpx.Response: +def _poll_get(client: httpx.Client, call: HttpCall) -> httpx.Response: """轮询 GET;522/525(及同档未达上游码)该次再试 1 次,不新开单。""" - resp = client.get(f"/videos/{job_id}") + def once() -> httpx.Response: + return client.request(call.method, call.path, headers=dict(call.headers)) + + resp = once() if resp.status_code in (521, 522, 523, 525): - resp = client.get(f"/videos/{job_id}") + resp = once() return resp @@ -174,6 +98,11 @@ def __init__( self._max_min = max_min self._first_poll_after = min(first_poll_after, poll_interval) + @property + def _protocol(self) -> OpenAIVideoProtocol: + # 每次现取:key 由 config 注入,provider 建好之后 config 仍可能被换。 + return OpenAIVideoProtocol(self._cfg.api_key) + def _client(self) -> httpx.Client: return httpx.Client( base_url=self._cfg.normalized_base_url, @@ -190,51 +119,24 @@ def submit_video( model: str, ) -> AdapterResult: """一次 POST 建单。成功: ok=True, job_id, body=b"", maybe_billed=True。""" - body: dict = { - "model": model, - "prompt": prompt, - "size": size, - "seconds": str(seconds), - "mode": self._mode, - } - if model in _IMAGE_LIST_MODELS: - b64 = _first_frame_datauri(first_frame, size).split(",", 1)[1] - body["image_list"] = [{"image": b64}] - else: - body["input_reference"] = _first_frame_datauri(first_frame, size) - + call = self._protocol.build_submit( + VideoRequest( + model=model, + prompt=prompt, + seconds=seconds, + size=size, + mode=self._mode, + first_frame=first_frame, + ) + ) with self._client() as client: try: - resp = client.post("/videos", json=body) + resp = client.request( + call.method, call.path, json=call.body, headers=dict(call.headers) + ) except httpx.TransportError as exc: return _transport_result(exc) - - if 200 <= resp.status_code < 300: - try: - payload = resp.json() - except ValueError: - return AdapterResult( - ok=False, - error_type=ModelErrorType.INVALID_RESPONSE, - http_status=resp.status_code, - edge_fingerprint="响应不是 JSON", - ) - jid = payload.get("id") - if not jid: - return AdapterResult( - ok=False, - error_type=ModelErrorType.INVALID_RESPONSE, - http_status=resp.status_code, - edge_fingerprint="响应没有 job id", - ) - return AdapterResult( - ok=True, - job_id=str(jid), - body=b"", - maybe_billed=True, - http_status=resp.status_code, - ) - return _video_http_error(resp) + return self._protocol.parse_submit(resp) def follow_job(self, job_id: str) -> AdapterResult: """轮询已建单据 + 下载。poll GET 522/525 该次再试 1 次,不新开单。""" @@ -268,39 +170,15 @@ def with_poll( break time.sleep(wait) wait = min(wait * 2, self._poll) - resp = _poll_get(client, job_id) + resp = _poll_get(client, self._protocol.build_poll(job_id)) poll_count += 1 - if not (200 <= resp.status_code < 300): - return with_poll(_video_http_error(resp, job_id=job_id, phase="follow")) - try: - st = resp.json() - except ValueError: - return with_poll( - AdapterResult( - ok=False, - error_type=ModelErrorType.INVALID_RESPONSE, - http_status=resp.status_code, - job_id=job_id, - maybe_billed=True, - edge_fingerprint="轮询响应不是 JSON", - ) - ) - last_status = st.get("status") - if last_status == "completed": - vids = (st.get("task_result") or {}).get("videos") or [] - url = vids[0].get("url") if vids else None + parsed = self._protocol.parse_poll(resp, job_id) + if parsed.error_type is not None: + return with_poll(parsed) + last_status = parsed.job_status + if parsed.ok: + url = parsed.result_url break - if last_status in ("failed", "cancelled"): - return with_poll( - AdapterResult( - ok=False, - error_type=ModelErrorType.UPSTREAM_FAILED, - job_id=job_id, - maybe_billed=True, - job_status=last_status, - edge_fingerprint=str(st.get("error") or ""), - ) - ) poll_ms = int((time.monotonic() - poll_t0) * 1000) if not url: return replace( @@ -513,9 +391,6 @@ def _download(client: httpx.Client, url: str, tries: int = 3) -> bytes: _CLOUDFLARE_UNREACHED_STATUS = frozenset({521, 522, 523}) _UNREACHED_RESENDS = 2 -_DIAGNOSTIC_HEADERS = ("server", "cf-ray", "via", "x-served-by", "retry-after") - - class _ResendBudget: """跨 _post 的多次调用共享:叠乘的是循环次数,可重复计费的次数不该跟着叠乘。""" @@ -544,12 +419,6 @@ def _retry_exhausted_message(status: int, tries: int, fingerprint: str) -> str: ) -def _edge_fingerprint(response: httpx.Response) -> str: - """52x 出自链路上哪一跳,只能从这几个头看 —— 不记下来,线上就只剩一个状态码可复盘。""" - seen = {k: response.headers.get(k) for k in _DIAGNOSTIC_HEADERS} - return " ".join(f"{k}={v}" for k, v in seen.items() if v) or "无可辨识的边缘响应头" - - # 从响应里捞 data URI。模型把图放在 message.content 里,而不同网关的包裹层级不一样 # (有的 content 是字符串、有的是 parts 数组),故对整个响应 JSON 做一次正则, # 不去猜层级 —— 猜错的代价是"调用成功、费用已产生、但我们说没图"。 @@ -621,7 +490,7 @@ def _post(self, client: httpx.Client, body: dict, resends: _ResendBudget) -> dic for attempt in range(1, _POST_TRIES + 1): resp = client.post(self._cfg.chat_completions_path, json=body) code = resp.status_code - edge = _edge_fingerprint(resp) + edge = edge_fingerprint(resp) if code in _CLOUDFLARE_UNREACHED_STATUS and not resends.take(): raise RuntimeError(_retry_exhausted_message(code, resends.spent, edge)) retryable = code == 429 or code in _CLOUDFLARE_UNREACHED_STATUS @@ -683,7 +552,7 @@ def submit_image(self, prompt: str, refs: list[bytes], model: str) -> AdapterRes f"同一把 key 也是。原始响应:{resp.text[:200]}" ) else: - edge = _edge_fingerprint(resp) + edge = edge_fingerprint(resp) retry_after_header = resp.headers.get("Retry-After") retry_after_s = ( _retry_after_seconds(retry_after_header) if retry_after_header else None diff --git a/backend/tests/test_sufy_video_download.py b/backend/tests/test_sufy_video_download.py index 17526f35..87f49f9c 100644 --- a/backend/tests/test_sufy_video_download.py +++ b/backend/tests/test_sufy_video_download.py @@ -845,10 +845,10 @@ def test_image_list_models_use_a_different_first_frame_field(): 直到生成阶段才 failed "model is not supported"，而费用可能已经产生 （2026-07-29 实测）。 """ - from windup_framework.providers.sufy import _IMAGE_LIST_MODELS + from windup_framework.providers.protocol.openai_video import IMAGE_LIST_MODELS seen: dict = {} - p = _video_provider(_i2v_handler(seen), model=_IMAGE_LIST_MODELS[0], mode="pro") + p = _video_provider(_i2v_handler(seen), model=IMAGE_LIST_MODELS[0], mode="pro") p.i2v(_jpeg_first_frame(), "x") assert "image_list" in seen["body"] and "input_reference" not in seen["body"] assert not seen["body"]["image_list"][0]["image"].startswith("data:"), \ @@ -997,13 +997,13 @@ def test_transparent_first_frame_background_does_not_depend_on_undefined_rgb(): """ import numpy as _np - from windup_framework.providers.sufy import _FIRST_FRAME_BG + from windup_framework.providers.protocol.openai_video import FIRST_FRAME_BG black_void = _submitted_first_frame(_sprite(256, 256, alpha=True, void_rgb=(0, 0, 0))) red_void = _submitted_first_frame(_sprite(256, 256, alpha=True, void_rgb=(255, 0, 0))) corner = _np.asarray(black_void)[4, 4] - assert _np.allclose(corner, _FIRST_FRAME_BG, atol=12), f"角落底色 {corner}，应为声明的 {_FIRST_FRAME_BG}" + assert _np.allclose(corner, FIRST_FRAME_BG, atol=12), f"角落底色 {corner}，应为声明的 {FIRST_FRAME_BG}" assert not _np.allclose(corner, (0, 0, 0), atol=12), "透明背景又变成黑底了" diff = _np.abs(_np.asarray(black_void, dtype=float) - _np.asarray(red_void, dtype=float)) diff --git a/backend/tests/test_video_protocol.py b/backend/tests/test_video_protocol.py new file mode 100644 index 00000000..49095938 --- /dev/null +++ b/backend/tests/test_video_protocol.py @@ -0,0 +1,132 @@ +"""协议层脱网单测。 + +协议只产出纯数据,所以这一层不需要网络也不需要打桩 —— 断言的是"字节怎么排", +不是"发出去会怎样"。 +""" +from __future__ import annotations + +import base64 +import io + +import httpx +import pytest + +from windup_common.enums.model import ModelErrorType +from windup_framework.providers.protocol import OpenAIVideoProtocol, VideoRequest +from windup_framework.providers.protocol.openai_video import IMAGE_LIST_MODELS + + +def _png(size: tuple[int, int] = (64, 64)) -> bytes: + from PIL import Image + + buf = io.BytesIO() + Image.new("RGBA", size, (200, 30, 30, 255)).save(buf, "PNG") + return buf.getvalue() + + +def _req(model: str = "kling-v2-5-turbo") -> VideoRequest: + return VideoRequest( + model=model, prompt="向右走", seconds=5, size="1280x720", mode="std", + first_frame=_png(), + ) + + +def test_submit_is_a_post_to_videos_with_bearer(): + """鉴权头由协议层产出:Key 与 Bearer 互换后的响应与"模型不存在"难以区分。""" + call = OpenAIVideoProtocol("k-123").build_submit(_req()) + + assert (call.method, call.path) == ("POST", "/videos") + assert call.headers["Authorization"] == "Bearer k-123" + assert call.body["seconds"] == "5", "seconds 是字符串,不是整数" + assert call.body["mode"] == "std" + + +def test_first_frame_becomes_a_jpeg_data_uri(): + """PNG base64 实测会 VENDOR_FAILED,所以这一层必须转成 JPG。""" + body = OpenAIVideoProtocol("k").build_submit(_req()).body + + assert "image_list" not in body + ref = body["input_reference"] + assert ref.startswith("data:image/jpeg;base64,") + raw = base64.b64decode(ref.split(",", 1)[1]) + assert raw[:2] == b"\xff\xd8", "解出来必须是 JPEG" + + +def test_image_list_model_gets_bare_base64(): + """kling-video-o1 吃 image_list 且不带 data URI 前缀;塞错字段任务会 failed。""" + body = OpenAIVideoProtocol("k").build_submit(_req(IMAGE_LIST_MODELS[0])).body + + assert "input_reference" not in body + assert not body["image_list"][0]["image"].startswith("data:") + + +def test_submit_response_without_id_is_invalid_not_ok(): + """2xx 但没有单号不能当建单成功:后面轮询没有可跟的对象,而费用可能已产生。""" + p = OpenAIVideoProtocol("k") + + assert p.parse_submit(httpx.Response(200, json={"id": "job-9"})).job_id == "job-9" + assert p.parse_submit(httpx.Response(200, json={})).error_type is ModelErrorType.INVALID_RESPONSE + assert p.parse_submit(httpx.Response(200, text="")).error_type is ModelErrorType.INVALID_RESPONSE + + +def test_pending_poll_has_no_error_type(): + """adapter 靠"没有 error_type"判定该继续轮询,这条是两层之间的契约。""" + parsed = OpenAIVideoProtocol("k").parse_poll( + httpx.Response(200, json={"status": "processing"}), "job-9" + ) + + assert parsed.error_type is None and not parsed.ok + assert parsed.job_status == "processing" + + +def test_completed_poll_hands_back_the_url(): + parsed = OpenAIVideoProtocol("k").parse_poll( + httpx.Response(200, json={"status": "completed", + "task_result": {"videos": [{"url": "https://cdn/x.mp4"}]}}), + "job-9", + ) + + assert parsed.ok and parsed.result_url == "https://cdn/x.mp4" + + +def test_completed_without_video_is_not_a_url(): + """完成但没给地址,不能拿 None 当地址往下走。""" + parsed = OpenAIVideoProtocol("k").parse_poll( + httpx.Response(200, json={"status": "completed", "task_result": {}}), "job-9" + ) + + assert parsed.ok and parsed.result_url is None + + +@pytest.mark.parametrize("status", ["failed", "cancelled"]) +def test_upstream_failure_is_billed(status): + """单据已建,上游自己失败仍可能计费。""" + parsed = OpenAIVideoProtocol("k").parse_poll( + httpx.Response(200, json={"status": status, "error": "boom"}), "job-9" + ) + + assert parsed.error_type is ModelErrorType.UPSTREAM_FAILED + assert parsed.maybe_billed and parsed.job_status == status + assert "boom" in parsed.edge_fingerprint + + +def test_poll_http_error_keeps_the_job_id_and_the_edge(): + """轮询失败要带着单号回去,否则重试会新开一单、二次计费。""" + parsed = OpenAIVideoProtocol("k").parse_poll( + httpx.Response(500, text="oops", headers={"cf-ray": "r1", "server": "cf"}), "job-9" + ) + + assert parsed.job_id == "job-9" and parsed.maybe_billed + assert "cf-ray=r1" in parsed.edge_fingerprint + + +def test_fetch_is_none_because_the_url_is_already_in_the_poll(): + """本面不需要第三步取结果;返回 None 而不是一个空 HttpCall,免得 adapter 真去发它。""" + assert OpenAIVideoProtocol("k").build_fetch("job-9") is None + + +def test_poll_path_carries_the_job_id(): + call = OpenAIVideoProtocol("k").build_poll("job-9") + + assert (call.method, call.path) == ("GET", "/videos/job-9") + assert call.headers["Authorization"] == "Bearer k"