refactor(providers): 把视频协议提取为显式接缝

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,

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

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",
+]

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

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: ...