feat:backend dev

backend/packages/ai_engine/pyproject.toml

@@ -10,6 +10,8 @@ dependencies = [
     "langchain-core>=0.3",
     "pillow>=10.4",
     "numpy>=1.26",
+    "imageio>=2.36",
+    "av>=14.0",              # imageio pyav 后端(视频抽帧)
     # "rembg",             # 抠图(按需启用)
 ]
 

backend/packages/ai_engine/src/windup_ai_engine/impl/__init__.py

@@ -0,0 +1,5 @@
+"""impl:CharacterGeneratorPort 的装配实现(串联 strategy + 最后一公里)。"""
+
+from .character_generator import CharacterGenerator
+
+__all__ = ["CharacterGenerator"]

backend/packages/ai_engine/src/windup_ai_engine/impl/character_generator.py

@@ -0,0 +1,88 @@
+"""CharacterGenerator —— 装配 strategy + 最后一公里,串起整条生产线(架构串联点)。
+
+这是 CharacterGeneratorPort 的实现;server 经 port 调它、不碰这里。
+串联:选路线(ROUTE_MATRIX)→ strategy.derive 出帧 → 最后一公里(脚线对齐)→ GeneratedAction。
+
+MVP 边界(与作者对齐):**只出帧 bytes + 逐帧时长**,不打包 sprite sheet、不落存储——
+上传对象存储、写 character_data、拼图集/多格式导出由 server / export 侧做(#22)。
+"""
+from __future__ import annotations
+
+import io
+
+from PIL import Image
+
+from windup_common.models import ActionSpec, CharacterCard, GenRoute
+
+from windup_ai_engine.ports import (
+    CharacterGeneratorPort,
+    GeneratedAction,
+    ProgressPort,
+)
+from windup_ai_engine.postprocess import align_bottom_center, frame_durations
+from windup_ai_engine.strategy.base import ROUTE_MATRIX, DerivationStrategy
+
+
+def _png(img: Image.Image) -> bytes:
+    buf = io.BytesIO()
+    img.convert("RGBA").save(buf, "PNG")
+    return buf.getvalue()
+
+
+def _img(png: bytes) -> Image.Image:
+    return Image.open(io.BytesIO(png)).convert("RGBA")
+
+
+class CharacterGenerator(CharacterGeneratorPort):
+    """由 bootstrap 注入 {GenRoute: DerivationStrategy} 装配表。"""
+
+    def __init__(self, strategies: dict[GenRoute, DerivationStrategy]) -> None:
+        self._by_route = strategies
+
+    def generate(
+        self,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+    ) -> GeneratedAction:
+        # ① 选路线(架构决策矩阵)
+        route = ROUTE_MATRIX[action.action]
+        progress.step("route", 0, 3, f"{action.action} → {route.value}")
+        strategy = self._by_route[route]
+
+        # ② 生成帧(交给 strategy —— 串联)
+        frames = strategy.derive(card, action, master, progress)
+
+        # ③ 最后一公里:脚线对齐成原地序列帧
+        frames = self._lastmile(frames, progress)
+
+        # ④ 出参:帧 + 逐帧时长(上传 / 落库在 server 侧)
+        progress.step("package", 2, 3, f"{len(frames)} 帧 + 逐帧时长")
+        return GeneratedAction(
+            frames=frames,
+            durations=frame_durations(action.action.value, len(frames)),
+            fps=action.fps,
+        )
+
+    def _lastmile(self, frames: list[bytes], progress: ProgressPort) -> list[bytes]:
+        """脚线对齐:把各帧对齐成原地序列帧(消除逐帧画布漂移,Issue #21)。
+
+        位移轨道(root_motion)MVP 先不做(见 #63 / character_data.frames 暂无该字段):
+        序列帧保持原地即可,位移留给后续 export / playtest 阶段再算。
+        """
+        progress.step("lastmile", 1, 3, "脚线对齐(原地)")
+        if not frames or not all(frames):   # 含空桩帧(未开发路线)→ 跳过
+            return frames
+        imgs = [_img(f) for f in frames]
+        # 参考姿态高 = 各帧包围盒高的中位数:比"最高帧"稳(不被举过头顶的武器带偏),
+        # 各动作都以自身中位姿态定标,本体尺寸跨动作一致。
+        import numpy as _np
+        _hs = []
+        for _im in imgs:
+            _ys, _ = _np.where(_np.asarray(_im)[:, :, 3] > 128)
+            if len(_ys):
+                _hs.append(float(_ys.max() - _ys.min()))
+        aligned = align_bottom_center(imgs, ref_height=(float(_np.median(_hs)) if _hs else None))
+        # TODO(dev, #21): tail_match 循环闭合(净位移动作先锚点再匹配帧)
+        return [_png(im) for im in aligned]

backend/packages/ai_engine/src/windup_ai_engine/master_prep.py

@@ -0,0 +1,82 @@
+"""母版规格与预处理:每个动作需要什么样的母版。
+
+**核心规律(三次实测验证,写死为契约):母版姿态决定动作,提示词只能微调。**
+  - walk:母版**朝侧向**才不转身;正面母版配侧走词 → 模型靠转身调和图文矛盾。
+  - jump:母版**顶部留白**才不被视频画面裁掉。
+  - attack:必须给**极限蓄力母版**(武器已拉到身后腰际)。用站立母版时,即使提示词写死
+    "武器不过头顶 / 不转身 / 只做一次",模型仍会抡过头顶、转到背面、劈两次 —— 强动作
+    先验压不住;换蓄力母版后模型只能"接着往前挥",没有再抡起的空间。
+
+
+实测教训:母版里角色居中、占 ~70% 画面高时,i2v 跳跃会让角色**头顶顶出视频画面上沿**
+被裁掉(生成本身没错,是构图没留够空间)。规则同 MasterSpec 的"运动方向多留白":
+  - jump:向上运动 → 顶部补空间,角色坐低
+  - dash / walk / run:向右位移 → 前进方向多留白(由母版生成时构图保证,此处不改)
+
+纯 PIL,零 API。背景色取母版四角中位色,补出来的边与母版底色一致。
+"""
+
+from __future__ import annotations
+
+import io
+
+import numpy as np
+from PIL import Image
+
+__all__ = ["add_headroom", "prepare_master", "MASTER_POSES"]
+
+# 各动作所需的母版姿态(生成专用母版时的姿势描述)。空=可直接用中性站立母版。
+MASTER_POSES = {
+    "walk": "",     # 中性站立即可,但必须朝侧向
+    "run": "",
+    "idle": "",
+    # jump:与 attack 同理——重甲带剑角色的"跳跃"强动作先验压不住(站立母版会让模型摆
+    # 造型、只举剑不腾空,实测)。给**极限蓄力半蹲母版**,模型只能"接着往上蹬"。顶部留白
+    # 由 prepare_master(add_headroom)保证。
+    "jump": (
+        "deep crouch coiled to spring straight upward: the knees bent low and the hips sunk down, "
+        "both arms drawn back behind the body, the weight loaded onto both legs at the very moment "
+        "before springing straight up, the weapon kept in a fixed grip; "
+        "leave generous empty space above the head"
+    ),
+    "attack": (
+        "extreme wind-up stance for a horizontal slash: the weapon drawn far BACK behind the body "
+        "at WAIST height, the torso twisted back and coiled, weight fully loaded on the back leg, "
+        "both arms low and pulled back, the weapon staying BELOW the shoulders; "
+        "leave generous empty space on the swing side"
+    ),
+}
+
+
+def _bg_color(img: Image.Image) -> tuple[int, int, int]:
+    """取四角中位色当背景色(母版通常是纯色底)。"""
+    rgb = np.asarray(img.convert("RGB"))
+    corners = np.stack([rgb[0, 0], rgb[0, -1], rgb[-1, 0], rgb[-1, -1]])
+    return tuple(int(v) for v in np.median(corners, axis=0))
+
+
+def add_headroom(master: bytes, ratio: float = 0.6) -> bytes:
+    """在母版上方补空间,让角色坐到画面下部,给腾空留出余量。
+
+    Args:
+        master: 母版图 bytes。
+        ratio: 处理后角色所占的画面高度比例(越小头顶空间越多)。0.6 表示角色高度
+            约占新画面的 60%,上方留约 40%。
+    """
+    if not 0.1 < ratio < 1.0:
+        raise ValueError("ratio 需在 (0.1, 1.0) 之间")
+    img = Image.open(io.BytesIO(master)).convert("RGB")
+    new_h = max(img.height + 1, int(round(img.height / ratio)))
+    canvas = Image.new("RGB", (img.width, new_h), _bg_color(img))
+    canvas.paste(img, (0, new_h - img.height))       # 原图贴底,空间加在顶部
+    buf = io.BytesIO()
+    canvas.save(buf, "PNG")
+    return buf.getvalue()
+
+
+def prepare_master(master: bytes, action: str) -> bytes:
+    """按动作类型预处理母版;不需要处理的动作原样返回。"""
+    if action in ("jump", "attack"):
+        # jump 向上腾空、attack 挥砍过头顶,都会顶出视频画面上沿(实测 attack 15/72 帧触顶)
+        return add_headroom(master, ratio=0.62 if action == "jump" else 0.70)
+    return master

backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py

@@ -0,0 +1,59 @@
+"""ai_engine 对外契约(ports)—— server 只 import 这里,不碰 slicing / strategy / impl。
+
+CI 的 import-linter 分层门禁会强制:app.server 依赖只到 ai_engine.ports。
+换掉内部实现(strategy / provider)时 server 零改动。
+
+MVP 边界(与作者对齐):ai_engine **只产出帧 bytes + 进度**,不碰存储 / DB。
+母版(master)由 server 侧从 ``Character.reference_image_url`` 取好、以 bytes 传入;
+产出的帧由 server 侧上传对象存储、落 ``character_data``。故本层无 ArtifactStore 依赖。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Protocol, runtime_checkable
+
+from windup_common.models import ActionSpec, CharacterCard
+
+
+# ---- server 实现、注入给 ai_engine 的进度回调 port ----
+class ProgressPort(Protocol):
+    """进度上报 —— server 转 SSE / 轮询状态(取代管线里的 print)。"""
+
+    def step(self, stage: str, i: int, total: int, note: str = "") -> None: ...
+
+
+# ---- ai_engine 出参(不含存储引用:上传 / 落库在 server 侧)----
+@dataclass
+class GeneratedAction:
+    """一个动作的生成产物:对齐后的原地序列帧 + 逐帧时长。
+
+    frames / durations **等长**;server 侧把每帧上传对象存储得 URL,组成
+    ``CharacterActionOutput.frames[{index, image_url, duration_ms}]`` 回填 character_data。
+    """
+
+    frames: list[bytes] = field(default_factory=list)   # RGBA PNG,按播放序
+    durations: list[int] = field(default_factory=list)  # 逐帧时长(ms),与 frames 等长
+    fps: int = 10
+
+
+# ---- ai_engine 暴露给 server(server 调用的唯一入口)----
+@runtime_checkable
+class CharacterGeneratorPort(Protocol):
+    """生成入口:角色卡 + 动作规格 + 母版 → 帧序列产物。
+
+    不关心租户 / 配额 / 任务状态 / 存储(那些在 app.server)。
+
+    Args:
+        card: 角色卡(身份 / 画风 / 朝向)。
+        action: 动作规格(类型 / 帧数 / 风格化 / 朝向)。
+        master: 定妆母版图 bytes(server 从 reference_image_url 取)。
+        progress: 进度回调。
+    """
+
+    def generate(
+        self,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+    ) -> GeneratedAction: ...

backend/packages/ai_engine/src/windup_ai_engine/postprocess/__init__.py

@@ -0,0 +1,28 @@
+"""后处理:把选好的帧落地成交付级序列帧(像素化 / 对齐 / 打包)。
+
+抽帧 / 选帧见 :mod:`..slicing`。逐帧时长 ``frame_durations`` 在 :mod:`.rootmotion`。
+"""
+
+from .rootmotion import DEFAULT_FPS_MS, extract_root_motion, frame_durations
+from .pixelate import (
+    detect_pixel_size,
+    extract_palette,
+    master_pixel_spec,
+    pixelate_frames,
+    to_pixel_art,
+)
+from .pack import align_bottom_center, save_gif, sprite_sheet
+
+__all__ = [
+    "to_pixel_art",
+    "pixelate_frames",
+    "detect_pixel_size",
+    "extract_palette",
+    "master_pixel_spec",
+    "extract_root_motion",
+    "frame_durations",
+    "DEFAULT_FPS_MS",
+    "align_bottom_center",
+    "sprite_sheet",
+    "save_gif",
+]

backend/packages/ai_engine/src/windup_ai_engine/postprocess/pack.py

@@ -0,0 +1,95 @@
+"""对齐 / 打包（后处理的收尾：脚线对齐 → sprite sheet / gif）。
+
+抽帧 / 选帧见 :mod:`..slicing`，像素化见 :mod:`.pixelate`，抠图见 framework 的
+MatteProvider（#20）。本模块把对齐后的帧拼成交付物。
+"""
+
+from __future__ import annotations
+
+from PIL import Image
+
+__all__ = ["align_bottom_center", "sprite_sheet", "save_gif"]
+
+
+def align_bottom_center(
+    frames: list[Image.Image],
+    cell: int = 256,
+    foot_line: float = 0.92,
+    fill_h: float = 0.62,
+    preserve_lift: bool = False,
+    ref_height: float | None = None,
+) -> list[Image.Image]:
+    """按脚线对齐到统一画布,消除逐帧画布漂移(Issue #21)。
+
+    **整段共用一个缩放系数**(取全序列最高帧定标),不逐帧归一化 —— 逐帧各自缩放到等高
+    会把走路自然的身高起伏(实测约 4%)反向变成"忽大忽小":蹲下的帧被放大、伸展的帧被
+    缩小。统一缩放后帧间只剩真实姿态差,尺度稳定。
+
+    水平方向按**主体水平中心**对齐(不含挥出的武器会更好,当前用整体包围盒中心兜底);
+    垂直方向按**脚线**(包围盒底边)对齐到 ``foot_line``。
+
+    ``ref_height``:**跨动作一致性的关键**,单位=传入帧的像素高。给定时按它定标,否则按本
+    序列最高帧。按最高帧定标会让"举过头顶"的动作整段被缩小去迁就那一帧 —— 实测攻击时
+    斧头高举使 bbox 从 485 涨到 660,角色本体因此明显变小;跳跃顶点同理。故传入**参考姿态**
+    (站立)的高度,各动作即共用同一本体尺寸。``fill_h`` 默认 0.62,给举过头顶留出余量。
+
+    ``preserve_lift``:腾空位移**默认不烘进像素**(业界:位移交引擎 root motion)。仅在要把
+    位移画进序列帧时才开;开启后以序列里最低的脚线为地面基准,保留每帧相对地面的抬升量。
+    """
+    import numpy as np
+
+    boxes: list[tuple[int, int, int, int] | None] = []
+    for f in frames:
+        ys, xs = np.where(np.asarray(f)[:, :, 3] > 128)
+        boxes.append(
+            (int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1)
+            if len(ys)
+            else None
+        )
+    heights = [b[3] - b[1] for b in boxes if b]
+    if not heights:
+        return [Image.new("RGBA", (cell, cell), (0, 0, 0, 0)) for _ in frames]
+    # 腾空模式:以最低脚线(数值最大 = 站在地上)为地面基准,保留每帧的抬升量
+    ground = max(b[3] for b in boxes if b) if preserve_lift else 0
+    # 定标要把抬升量算进去,否则跳到最高时头顶会顶出画布被切掉
+    if preserve_lift:
+        need = max((ground - b[3]) + (b[3] - b[1]) for b in boxes if b)
+        scale = (cell * fill_h) / max(1, need)
+    elif ref_height:
+        scale = (cell * fill_h) / ref_height     # 参考姿态定标(跨动作一致)
+    else:
+        scale = (cell * fill_h) / max(heights)   # 回退:本序列最高帧
+
+    out = []
+    for f, box in zip(frames, boxes):
+        if box is None:
+            out.append(Image.new("RGBA", (cell, cell), (0, 0, 0, 0)))
+            continue
+        crop = f.crop(box)
+        w = max(1, round(crop.width * scale))
+        h = max(1, round(crop.height * scale))
+        crop = crop.resize((w, h), Image.NEAREST)
+        lift = round((ground - box[3]) * scale) if preserve_lift else 0
+        canvas = Image.new("RGBA", (cell, cell), (0, 0, 0, 0))
+        canvas.alpha_composite(crop, (cell // 2 - w // 2, int(cell * foot_line) - h - lift))
+        out.append(canvas)
+    return out
+
+
+def sprite_sheet(frames: list[Image.Image], bg=(0, 0, 0, 0)) -> Image.Image:
+    """横向拼接为 sprite sheet。"""
+    if not frames:
+        raise ValueError("frames 为空")
+    w, h = frames[0].size
+    sheet = Image.new("RGBA", (w * len(frames), h), bg)
+    for i, f in enumerate(frames):
+        sheet.alpha_composite(f.convert("RGBA"), (i * w, 0))
+    return sheet
+
+
+def save_gif(frames: list[Image.Image], path: str, duration: int = 120) -> None:
+    """导出循环 gif 供预览。"""
+    if not frames:
+        raise ValueError("frames 为空")
+    rgba = [f.convert("RGBA") for f in frames]
+    rgba[0].save(path, save_all=True, append_images=rgba[1:], duration=duration, loop=0, disposal=2)