fix: 修复 gpu_direct_pipeline.py 中 concat 构建的变量名错误

build_direct_render 方法中列表推导式使用 _lbl 作为循环变量，但 f-string 中误用了未定义的 l，导致 NameError。修复: f"[{l}]" -> f"[{_lbl}]" (line 195)
fix: 修复 GPU 直连管线中 l.role 变量名拼写错误
2026-09-28 16:22:01 +08:00 · 2026-09-28 15:31:46 +08:00 · 2026-09-28 14:46:44 +08:00 · 2026-09-28 14:20:48 +08:00 · 2026-09-28 14:02:57 +08:00 · 2026-09-28 05:38:30 +00:00
37 changed files with 3350 additions and 1032 deletions
@@ -79,14 +79,33 @@ CELERY_BROKER_URL=redis://localhost:6379/0
 CELERY_RESULT_BACKEND=redis://localhost:6379/1


-# ==================== Worker 配置 ====================
+# ==================== Worker 配置（#2073 队列分流） ====================
+#
+# 容器内跑三个独立进程：beat（只发定时任务）+ generation worker（实时高优）
+# + transcode worker（后台批量/清理）。三个进程的并发与开关独立配置。

 # Worker 进程名称
 WORKER_NAME=xiaoxia-saas-worker

-# Worker 并发数（同时执行的任务数）
+# 总并发参考（兼容旧变量）：
+# - 若 GENERATION_CONCURRENCY 与 TRANSCODE_CONCURRENCY 都未显式设置，
+#   entrypoint 会按此总数对半分配（gen=ceil(total/2), trans=剩余，各至少 1）；
+# - 任一个 *_CONCURRENCY 显式设置后，按显式值生效，忽略此变量对应部分。
 WORKER_CONCURRENCY=4

+# Generation worker 并发数（用户实时任务：视频生成/TTS/音色克隆/lipsync/数字人）
+# 实时链路对延迟敏感，建议 2C 以上机器设为 2；高负载场景可加到 4。
+GENERATION_CONCURRENCY=2
+
+# Transcode worker 并发数（后台批量：素材入库转码/AI 分类打标/质量评分/查重/批量下载）
+# 后台任务可排队，独立伸缩；素材入库量大时可加到 4。
+TRANSCODE_CONCURRENCY=2
+
+# 是否在本容器启动 celery beat 进程（默认 1）。
+# 默认 beat 与 worker 同容器部署；若要独立 beat 容器部署，worker 容器设为 0、
+# beat 容器单独跑 `celery -A worker_app.celery_app beat` 并设 BEAT_ENABLED=1。
+BEAT_ENABLED=1
+
 # 每个子进程最多处理多少任务后重启（防止内存泄漏）
 WORKER_MAX_TASKS_PER_CHILD=1000

@@ -1302,8 +1302,16 @@ jobs:
          "${staging_user}@${staging_host}:/var/lib/xiaoxia-saas-staging/configs/douyin_cookies.txt"
        echo "✅ Douyin cookies uploaded"

+        # 上传 infra/docker 配置到服务器（compose 单一事实来源）
+        echo "Uploading infra/docker configs to staging server..."
+        ssh -p "$staging_port" -i "$key_path" -o StrictHostKeyChecking=no "${staging_user}@${staging_host}" \
+          "mkdir -p /var/lib/xiaoxia-saas-staging/infra/docker"
+        scp -P "$staging_port" -i "$key_path" -o StrictHostKeyChecking=no infra/docker/compose.yml \
+          "${staging_user}@${staging_host}:/var/lib/xiaoxia-saas-staging/infra/docker/compose.yml"
+        echo "✅ infra/docker/compose.yml uploaded"
+
        # 通过环境变量传递凭证，避免命令行引号转义问题
-        cat scripts/ci_staging_deploy.sh | ssh -p "$staging_port" -i "$key_path" -o StrictHostKeyChecking=no "${staging_user}@${staging_host}" "IMAGE_TAG=${GITHUB_SHA} ACR_USERNAME=${ACR_USERNAME} ACR_PASSWORD=${ACR_PASSWORD} sh"
+        cat scripts/ci_staging_deploy.sh | ssh -p "$staging_port" -i "$key_path" -o StrictHostKeyChecking=no "${staging_user}@${staging_host}" "IMAGE_TAG=${GITHUB_SHA} ACR_USERNAME=${ACR_USERNAME} ACR_PASSWORD=${ACR_PASSWORD} COMPOSE_SYNC=0 sh"

        # 清理 CI runner 上的渲染文件
        rm -f .env.rendered
@@ -0,0 +1,271 @@
+/**
+ * 标题迷你 Canvas 预览（#2001）
+ *
+ * 渲染一张指定宽度的小 Canvas 预览标题效果，用于：
+ * - 预设卡片缩略图
+ * - 样式面板顶部的实时预览
+ *
+ * 与 titleCanvas.ts 渲染逻辑保持一致，但：
+ * - 固定分辨率（width × 宽高比约 2:1）
+ * - 不调用 ffmpeg，只做视觉预览
+ * - 支持背景色块、描边宽度/颜色、阴影参数化、行距、自动换行
+ */
+import React, { useEffect, useRef } from "react"
+import type { TitleStyleSettings } from "@/components/title/settings"
+import { getFontFamily } from "@/components/title/constants"
+
+interface Props {
+  settings: TitleStyleSettings
+  width?: number
+  sampleText?: string
+  /** 背景（预览用，默认深色渐变模拟视频底），transparent=true 时忽略 */
+  background?: string
+  /** 高度（可选，默认按 portrait 选比例） */
+  height?: number
+  /** 透明背景（卡片/编辑器预览叠加在图片上时使用） */
+  transparent?: boolean
+  /** 纵向竖屏预览（9:16），true 时 aspect=16/9 适配手机视频比例 */
+  portrait?: boolean
+}
+
+/** 按 maxCharsPerLine 自动换行 */
+function wrapLines(text: string, maxChars: number): string[] {
+  const manual = text
+    .split(/[/／\n]/)
+    .map((l) => l.trim())
+    .filter(Boolean)
+  if (!maxChars || maxChars <= 0) return manual
+  const out: string[] = []
+  for (const line of manual) {
+    if (line.length <= maxChars) {
+      out.push(line)
+      continue
+    }
+    let cur = ""
+    for (const ch of line) {
+      cur += ch
+      if (cur.length >= maxChars) {
+        out.push(cur)
+        cur = ""
+      }
+    }
+    if (cur) out.push(cur)
+  }
+  return out
+}
+
+const TitleMiniPreview: React.FC<Props> = ({
+  settings,
+  width = 200,
+  sampleText,
+  background = "linear-gradient(135deg,#1f2937,#111827)",
+  height,
+  transparent = false,
+  portrait = false,
+}) => {
+  const canvasRef = useRef<HTMLCanvasElement>(null)
+  const h = height ?? Math.round(width * (portrait ? 16 / 9 : 1 / 1.8))
+  const text = (sampleText || "预览标题").trim() || "预览标题"
+
+  useEffect(() => {
+    let cancelled = false
+    const draw = () => {
+      if (cancelled) return
+      const cvs = canvasRef.current
+      if (!cvs) return
+      const dpr = window.devicePixelRatio || 1
+      cvs.width = width * dpr
+      cvs.height = h * dpr
+      cvs.style.width = `${width}px`
+      cvs.style.height = `${h}px`
+      const ctx = cvs.getContext("2d")
+      if (!ctx) return
+      ctx.scale(dpr, dpr)
+      ctx.clearRect(0, 0, width, h)
+
+      // 背景（transparent 时跳过，用于叠加在图片上）
+      if (!transparent) {
+        ctx.fillStyle = "#111827"
+        ctx.fillRect(0, 0, width, h)
+      }
+
+      // 分辨率缩放：以 360 宽为基准（对应 720p 的一半），与外层 previewScale/previewR 保持一致
+      const r = previewR
+
+      // 字体
+      const size = r(settings.size)
+      const ff = getFontFamily(settings.font)
+      const parts: string[] = []
+      if (settings.italic) parts.push("italic")
+      if (settings.bold) parts.push("bold")
+      parts.push(`${size}px`, ff)
+      ctx.font = parts.join(" ")
+      ctx.textAlign = "center"
+      ctx.textBaseline = "middle"
+      ctx.fillStyle = settings.color
+      ctx.lineJoin = "round"
+
+      // 阴影
+      const shadowEnabled = !!settings.shadow
+      const prevShadow = {
+        c: ctx.shadowColor,
+        b: ctx.shadowBlur,
+        ox: ctx.shadowOffsetX,
+        oy: ctx.shadowOffsetY,
+      }
+      if (shadowEnabled) {
+        ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
+        ctx.shadowBlur = r(settings.shadowBlur ?? 4)
+        ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
+        ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
+      }
+
+      // 换行
+      const lines = wrapLines(text, settings.maxCharsPerLine ?? 0)
+      const lineH = size * (settings.lineHeight ?? 1.2)
+      const totalH = lines.length * lineH
+      let startY: number
+      if (settings.position === "top") {
+        startY = size / 2 + r(settings.marginTop ?? 24)
+      } else if (settings.position === "center") {
+        startY = h / 2 - totalH / 2 + size / 2
+      } else {
+        // bottom
+        const botMargin = portrait ? r(24) : r(16)
+        startY = h - totalH - botMargin + size / 2
+      }
+      let centerX = width / 2
+      if (settings.position === "custom" && settings.posX != null) {
+        centerX = (settings.posX / 100) * width
+      }
+
+      // 背景块
+      if (settings.bgEnabled) {
+        const pad = r(settings.bgPadding ?? 12)
+        const rad = r(settings.bgRadius ?? 8)
+        let maxLineW = 0
+        for (const l of lines) {
+          const m = ctx.measureText(l)
+          if (m.width > maxLineW) maxLineW = m.width
+        }
+        const bw = maxLineW + pad * 2
+        const bh = totalH + pad * 2
+        const bx = centerX - bw / 2
+        const by = startY - size / 2 - pad + (size - lineH) / 2
+        ctx.shadowColor = "rgba(0,0,0,0)"
+        ctx.shadowBlur = 0
+        ctx.fillStyle = settings.bgColor ?? "rgba(0,0,0,0.5)"
+        roundRect(ctx, bx, by, bw, bh, rad)
+        ctx.fill()
+        // 关键修复：画完背景块后必须把 fillStyle 重置为文字颜色，
+        // 否则后续 fillText 会用 bgColor 填充文字，导致「文字看不见只剩色块」
+        ctx.fillStyle = settings.color
+        // 恢复阴影
+        if (shadowEnabled) {
+          ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
+          ctx.shadowBlur = r(settings.shadowBlur ?? 4)
+          ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
+          ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
+        }
+      }
+
+      // 描边（先画，再画填充）
+      const strokeEnabled = !!settings.stroke && (settings.strokeWidth ?? 0) > 0
+      lines.forEach((line, i) => {
+        const y = startY + i * lineH
+        if (strokeEnabled) {
+          ctx.shadowColor = "rgba(0,0,0,0)"
+          ctx.shadowBlur = 0
+          ctx.lineWidth = r(settings.strokeWidth ?? 4)
+          ctx.strokeStyle = settings.strokeColor ?? "#000000"
+          ctx.strokeText(line, centerX, y)
+          // 恢复阴影
+          if (shadowEnabled) {
+            ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
+            ctx.shadowBlur = r(settings.shadowBlur ?? 4)
+            ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
+            ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
+          }
+        }
+        ctx.fillText(line, centerX, y)
+      })
+
+      // 恢复
+      ctx.shadowColor = prevShadow.c
+      ctx.shadowBlur = prevShadow.b
+      ctx.shadowOffsetX = prevShadow.ox
+      ctx.shadowOffsetY = prevShadow.oy
+    }
+    // 计算当前字号（draw() 内部同样逻辑，抽出来供 fontString 复用）
+    const previewScale = width / 360
+    const previewR = (v: number) => Math.round(v * previewScale)
+    const buildFontString = () => {
+      const size = previewR(settings.size)
+      const ff = getFontFamily(settings.font)
+      const parts: string[] = []
+      if (settings.italic) parts.push("italic")
+      if (settings.bold) parts.push("bold")
+      parts.push(`${size}px`, ff)
+      return parts.join(" ")
+    }
+
+    // Web Font 加载保障：
+    // 1) 等 document.fonts.ready（CSS @font-face 首次可用）
+    // 2) 显式 FontFaceSet.load(fontString, text) 触发浏览器真正下载并加载
+    //    当前字体到 Canvas 可用，避免首次绘制用 fallback 字体画出错字/色块
+    const doDrawWhenReady = async () => {
+      try {
+        if (typeof document !== "undefined" && document.fonts) {
+          await document.fonts.ready
+          try {
+            await document.fonts.load(buildFontString(), text)
+          } catch {
+            /* ignore */
+          }
+        }
+      } finally {
+        if (!cancelled) draw()
+      }
+    }
+    doDrawWhenReady()
+    return () => {
+      cancelled = true
+    }
+  }, [settings, width, h, text, transparent, portrait, background])
+
+  return (
+    <canvas
+      ref={canvasRef}
+      style={{
+        borderRadius: 6,
+        display: "block",
+        maxWidth: "100%",
+        background: transparent ? "transparent" : background,
+      }}
+    />
+  )
+}
+
+function roundRect(
+  ctx: CanvasRenderingContext2D,
+  x: number,
+  y: number,
+  w: number,
+  h: number,
+  r: number,
+) {
+  const rr = Math.min(r, w / 2, h / 2)
+  ctx.beginPath()
+  ctx.moveTo(x + rr, y)
+  ctx.lineTo(x + w - rr, y)
+  ctx.quadraticCurveTo(x + w, y, x + w, y + rr)
+  ctx.lineTo(x + w, y + h - rr)
+  ctx.quadraticCurveTo(x + w, y + h, x + w - rr, y + h)
+  ctx.lineTo(x + rr, y + h)
+  ctx.quadraticCurveTo(x, y + h, x, y + h - rr)
+  ctx.lineTo(x, y + rr)
+  ctx.quadraticCurveTo(x, y, x + rr, y)
+  ctx.closePath()
+}
+
+export default TitleMiniPreview
@@ -0,0 +1,458 @@
+/* ============================================================
+   TitleStylePanel 标题样式面板 — 独立共用样式（#1809 ⑦）
+
+   从 generate.css 抽取的标题样式区块，供「智能剪辑」与「AI数字人」
+   两个页面共用。AI数字人页面不引入 generate.css，直接由
+   TitleStylePanel.tsx import 本文件，保证 24 个 T 预设格子的网格布局、
+   配色描边、选中态与智能剪辑页面完全一致。
+
+   注意：本文件规则与 generate.css 中同名规则一一对应、取值相同；
+   智能剪辑页面两处同时存在时同优先级同值，不改变其原有呈现。
+   ============================================================ */
+
+/* ── 区块容器 ── */
+.xx-title-style-section {
+  margin-top: 22px;
+  padding-top: 20px;
+  border-top: 1px solid var(--border-light);
+}
+
+.xx-section-subtitle {
+  font-size: 14px;
+  font-weight: 600;
+  color: var(--text-primary);
+  margin: 0 0 16px;
+}
+
+.xx-title-style-row {
+  display: grid;
+  grid-template-columns: 1fr 1fr;
+  gap: 14px;
+  margin-bottom: 14px;
+}
+
+.xx-half-field {
+  margin-bottom: 0;
+}
+
+.xx-field-label-row {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  margin-bottom: 8px;
+}
+
+.xx-field-label-row label {
+  margin-bottom: 0;
+}
+
+.xx-field-value {
+  font-size: 13px;
+  font-weight: 600;
+  color: var(--primary-color);
+}
+
+/* ── 共用表单字段（位置/字体下拉） ── */
+.xx-title-style-section .xx-form-field {
+  margin-bottom: 14px;
+}
+
+.xx-title-style-section .xx-form-field:last-child {
+  margin-bottom: 0;
+}
+
+.xx-title-style-section .xx-form-field label {
+  display: block;
+  font-weight: 600;
+  margin-bottom: 8px;
+  font-size: 13px;
+  color: var(--text-primary);
+}
+
+.xx-title-style-section .xx-form-field select,
+.xx-title-style-section .xx-form-field input {
+  width: 100%;
+  height: 44px;
+  border: 1px solid var(--border-color);
+  border-radius: var(--radius-sm);
+  background: var(--bg-primary);
+  padding: 0 14px;
+  font-size: 14px;
+  outline: 0;
+  transition: 0.15s ease;
+  color: var(--text-primary);
+}
+
+.xx-title-style-section .xx-form-field select:focus,
+.xx-title-style-section .xx-form-field input:focus {
+  border-color: var(--primary-color);
+  box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.1);
+}
+
+/* ── 字号滑块 ── */
+.xx-slider {
+  width: 100%;
+  height: 6px;
+  -webkit-appearance: none;
+  appearance: none;
+  background: var(--border-color);
+  border-radius: 3px;
+  outline: none;
+  cursor: pointer;
+}
+
+.xx-slider::-webkit-slider-thumb {
+  -webkit-appearance: none;
+  appearance: none;
+  width: 18px;
+  height: 18px;
+  background: var(--primary-color);
+  border-radius: 50%;
+  cursor: pointer;
+  box-shadow: 0 2px 6px rgba(79, 70, 229, 0.3);
+}
+
+.xx-slider::-moz-range-thumb {
+  width: 18px;
+  height: 18px;
+  background: var(--primary-color);
+  border-radius: 50%;
+  cursor: pointer;
+  border: none;
+  box-shadow: 0 2px 6px rgba(79, 70, 229, 0.3);
+}
+
+/* ── 标题预设卡片网格（24 个 T 格子） ── */
+.xx-title-presets-grid {
+  display: grid;
+  grid-template-columns: repeat(6, 52px);
+  gap: 1px;
+}
+
+.xx-title-preset-card {
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  width: 52px;
+  height: 52px;
+  padding: 0;
+  background: #404040;
+  border: 2px solid transparent;
+  border-radius: 8px;
+  cursor: pointer;
+  transition: all 0.15s;
+}
+
+.xx-title-preset-card:hover {
+  border-color: #666;
+  background: #4d4d4d;
+}
+
+.xx-title-preset-card.active {
+  border-color: #409eff;
+  background: #4d4d4d;
+}
+
+.xx-title-preset-preview-text {
+  font-size: 32px;
+  line-height: 1;
+  user-select: none;
+}
+
+/* ── 样式按钮组（加粗/斜体/描边/阴影） ── */
+.xx-style-btns {
+  display: flex;
+  gap: 8px;
+}
+
+.xx-style-btn {
+  width: 40px;
+  height: 40px;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  border: 1px solid var(--border-color);
+  border-radius: var(--radius-sm);
+  background: var(--bg-primary);
+  cursor: pointer;
+  font-size: 15px;
+  color: var(--text-secondary);
+  transition: all 0.15s;
+}
+
+.xx-style-btn:hover {
+  border-color: var(--primary-300);
+  color: var(--primary-color);
+}
+
+.xx-style-btn.active {
+  background: var(--primary-color);
+  border-color: var(--primary-color);
+  color: #fff;
+}
+
+/* ============================================================
+   #2001 爆款标题样式面板升级 — 新增样式（ts- 前缀）
+   ============================================================ */
+
+.ts-panel {
+  position: relative;
+}
+
+/* 预览 */
+.ts-preview-wrap {
+  margin-bottom: 14px;
+  display: flex;
+  justify-content: center;
+  padding: 10px;
+  background: #0f172a;
+  border-radius: 8px;
+}
+
+/* 表单字段 */
+.ts-form-field {
+  margin-bottom: 12px;
+}
+.ts-form-field label {
+  display: block;
+  font-weight: 600;
+  margin-bottom: 6px;
+  font-size: 12px;
+  color: var(--text-primary, #1f2937);
+}
+.ts-field-label-row {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  margin-bottom: 6px;
+}
+.ts-field-value {
+  font-size: 12px;
+  font-weight: 600;
+  color: var(--primary-color, #7c3aed);
+}
+.ts-row-2 {
+  display: grid;
+  grid-template-columns: 1fr 1fr;
+  gap: 10px;
+}
+.ts-half {
+  margin-bottom: 0;
+}
+
+.ts-select {
+  width: 100%;
+  height: 34px;
+  border: 1px solid var(--border-color, #e5e7eb);
+  border-radius: 6px;
+  background: var(--bg-primary, #fff);
+  padding: 0 10px;
+  font-size: 13px;
+  outline: 0;
+  color: var(--text-primary, #1f2937);
+}
+.ts-select:focus {
+  border-color: var(--primary-color, #7c3aed);
+  box-shadow: 0 0 0 2px rgba(124, 58, 237, 0.1);
+}
+.ts-input {
+  width: 100%;
+  height: 34px;
+  border: 1px solid var(--border-color, #e5e7eb);
+  border-radius: 6px;
+  padding: 0 10px;
+  font-size: 13px;
+  outline: 0;
+}
+
+.ts-slider {
+  width: 100%;
+  height: 4px;
+  -webkit-appearance: none;
+  appearance: none;
+  background: #e5e7eb;
+  border-radius: 2px;
+  outline: none;
+}
+.ts-slider::-webkit-slider-thumb {
+  -webkit-appearance: none;
+  appearance: none;
+  width: 16px;
+  height: 16px;
+  border-radius: 50%;
+  background: #7c3aed;
+  cursor: pointer;
+  border: 2px solid #fff;
+  box-shadow: 0 1px 3px rgba(0, 0, 0, 0.2);
+}
+.ts-slider::-moz-range-thumb {
+  width: 16px;
+  height: 16px;
+  border-radius: 50%;
+  background: #7c3aed;
+  cursor: pointer;
+  border: 2px solid #fff;
+}
+
+/* 样式按钮 B/I/S/☁ */
+.ts-style-btns {
+  display: flex;
+  gap: 6px;
+}
+.ts-style-btn {
+  width: 34px;
+  height: 34px;
+  border-radius: 6px;
+  border: 1px solid #e5e7eb;
+  background: #fff;
+  cursor: pointer;
+  font-size: 14px;
+  transition: 0.15s;
+  color: #374151;
+  display: inline-flex;
+  align-items: center;
+  justify-content: center;
+}
+.ts-style-btn:hover {
+  border-color: #7c3aed;
+  color: #7c3aed;
+}
+.ts-style-btn.active {
+  background: #faf5ff;
+  color: #6d28d9;
+  border-color: #7c3aed;
+  font-weight: 700;
+}
+
+/* 色板 */
+.ts-color-row {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 6px;
+  align-items: center;
+}
+.ts-color-swatch {
+  width: 24px;
+  height: 24px;
+  border-radius: 4px;
+  border: 2px solid #fff;
+  box-shadow: 0 0 0 1px #e5e7eb;
+  cursor: pointer;
+  padding: 0;
+  transition: 0.15s;
+}
+.ts-color-swatch:hover {
+  transform: scale(1.1);
+}
+.ts-color-swatch.active {
+  box-shadow: 0 0 0 2px #7c3aed;
+  transform: scale(1.1);
+}
+.ts-color-custom {
+  background: repeating-conic-gradient(#ccc 0% 25%, #fff 0% 50%) 50%/8px 8px;
+  color: #666;
+  font-size: 14px;
+  line-height: 20px;
+}
+.ts-color-native {
+  width: 0;
+  height: 0;
+  border: 0;
+  padding: 0;
+}
+
+/* 预设网格 10个 - 5列 */
+.ts-presets-grid {
+  display: grid;
+  grid-template-columns: repeat(5, 1fr);
+  gap: 6px;
+}
+.ts-preset-card {
+  border: 1px solid #e5e7eb;
+  border-radius: 6px;
+  background: #fff;
+  padding: 4px;
+  cursor: pointer;
+  transition: 0.15s;
+  display: flex;
+  flex-direction: column;
+  gap: 4px;
+}
+.ts-preset-card:hover {
+  border-color: #7c3aed;
+}
+.ts-preset-card.active {
+  border-color: #7c3aed;
+  background: #faf5ff;
+  box-shadow: 0 0 0 1px #7c3aed;
+}
+.ts-preset-preview {
+  height: 34px;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  overflow: hidden;
+  border-radius: 4px;
+  background: #0f172a;
+}
+.ts-preset-preview canvas {
+  max-width: 100%;
+  max-height: 100%;
+}
+.ts-preset-meta {
+  display: flex;
+  align-items: center;
+  gap: 2px;
+  font-size: 10px;
+  color: #4b5563;
+  justify-content: center;
+  white-space: nowrap;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  padding: 0 2px 2px;
+}
+.ts-preset-emoji {
+  font-size: 11px;
+}
+.ts-preset-label {
+  overflow: hidden;
+  text-overflow: ellipsis;
+}
+
+.ts-toggle-row label {
+  display: inline-flex;
+  align-items: center;
+  gap: 6px;
+  font-size: 13px;
+  font-weight: 500;
+  cursor: pointer;
+  margin-bottom: 10px;
+}
+.ts-toggle-row input[type="checkbox"] {
+  width: 16px;
+  height: 16px;
+  accent-color: #7c3aed;
+}
+
+/* Tabs 紧凑样式 */
+.xx-title-style-section .ant-tabs-nav {
+  margin-bottom: 10px;
+}
+.xx-title-style-section .ant-tabs-tab {
+  font-size: 12px !important;
+  padding: 6px 8px !important;
+}
+
+/* 标题模板入口按钮（#2003） */
+.ts-template-btn {
+  border: none;
+  background: transparent;
+  color: var(--primary-color, #7c3aed);
+  font-size: 12px;
+  cursor: pointer;
+  padding: 2px 0;
+  font-weight: 500;
+}
+.ts-template-btn:hover {
+  text-decoration: underline;
+}
@@ -0,0 +1,445 @@
+/**
+ * 标题样式参数 Tab 面板（共享组件）
+ *
+ * 包含：基础/描边/阴影/背景/排版/封面 共 6 个 Tab
+ * 仅负责 UI 渲染和参数 patch 回调，不维护 state、不调 API
+ */
+import React, { useState } from "react"
+import { Tabs } from "antd"
+import type { TitleStyleSettings } from "./settings"
+import {
+  FONT_OPTIONS,
+  TITLE_COLOR_PALETTE,
+  STROKE_COLOR_PALETTE,
+  BG_COLOR_PALETTE,
+} from "./constants"
+
+export interface PositionOption {
+  value: string
+  label: string
+}
+
+export interface FontOption {
+  value: string
+  label: string
+  family: string
+  tag?: "hot" | "new"
+}
+
+export interface TitleStyleParamsTabProps {
+  settings: TitleStyleSettings
+  onUpdatePosition: (p: string) => void
+  onUpdateFont: (f: string) => void
+  onUpdateSize: (v: number) => void
+  onToggleBold: () => void
+  onToggleItalic: () => void
+  onToggleStroke: () => void
+  onToggleShadow: () => void
+  onUpdatePatch: (patch: Partial<TitleStyleSettings>) => void
+  positionOptions: PositionOption[]
+  fontOptions?: FontOption[]
+  /** 是否显示「封面」Tab（独立封面标题开关） */
+  showCoverToggle?: boolean
+  /** 封面独立标题开关状态 */
+  coverEnabled?: boolean
+  /** 封面开关变化 */
+  onToggleCover?: (enabled: boolean) => void
+}
+
+/* ── Slider 行 ── */
+const SliderRow: React.FC<{
+  label: string
+  value: number
+  min: number
+  max: number
+  step?: number
+  unit?: string
+  onChange: (v: number) => void
+}> = ({ label, value, min, max, step = 1, unit = "px", onChange }) => (
+  <div className="ts-form-field">
+    <div className="ts-field-label-row">
+      <label>{label}</label>
+      <span className="ts-field-value">
+        {value}
+        {unit}
+      </span>
+    </div>
+    <input
+      type="range"
+      className="ts-slider"
+      min={min}
+      max={max}
+      step={step}
+      value={value}
+      onChange={(e) => onChange(Number(e.target.value))}
+    />
+  </div>
+)
+
+/* ── 色板 ── */
+const ColorPicker: React.FC<{
+  label?: string
+  value: string
+  palette: string[]
+  onChange: (c: string) => void
+}> = ({ label, value, palette, onChange }) => {
+  const [customOpen, setCustomOpen] = useState(false)
+  return (
+    <div className="ts-form-field">
+      {label && <label>{label}</label>}
+      <div className="ts-color-row">
+        {palette.map((c) => (
+          <button
+            key={c}
+            type="button"
+            className={`ts-color-swatch${value.toLowerCase() === c.toLowerCase() ? " active" : ""}`}
+            style={{ background: c }}
+            onClick={() => onChange(c)}
+            title={c}
+          />
+        ))}
+        <button
+          type="button"
+          className="ts-color-swatch ts-color-custom"
+          onClick={() => setCustomOpen((v) => !v)}
+          title="自定义颜色"
+        >
+          +
+        </button>
+        <input
+          type="color"
+          className="ts-color-native"
+          value={value.startsWith("rgba") ? "#000000" : value}
+          onChange={(e) => {
+            onChange(e.target.value)
+            setCustomOpen(false)
+          }}
+          style={{
+            opacity: customOpen ? 1 : 0,
+            position: customOpen ? "static" : "absolute",
+            pointerEvents: customOpen ? "auto" : "none",
+            width: customOpen ? 28 : 0,
+            height: customOpen ? 28 : 0,
+            border: "none",
+            padding: 0,
+            cursor: "pointer",
+            background: "transparent",
+          }}
+        />
+      </div>
+      <div style={{ fontSize: 11, color: "#9ca3af", marginTop: 2 }}>
+        当前：<code style={{ fontSize: 11 }}>{value}</code>
+      </div>
+    </div>
+  )
+}
+
+const TitleStyleParamsTab: React.FC<TitleStyleParamsTabProps> = ({
+  settings,
+  onUpdatePosition,
+  onUpdateFont,
+  onUpdateSize,
+  onToggleBold,
+  onToggleItalic,
+  onToggleStroke,
+  onToggleShadow,
+  onUpdatePatch,
+  positionOptions,
+  fontOptions = FONT_OPTIONS,
+  showCoverToggle = false,
+  coverEnabled = false,
+  onToggleCover,
+}) => {
+  const upd = onUpdatePatch
+  return (
+    <Tabs
+      size="small"
+      defaultActiveKey="basic"
+      items={[
+        {
+          key: "basic",
+          label: "基础",
+          children: (
+            <>
+              <div className="ts-row-2">
+                <div className="ts-form-field ts-half">
+                  <label>位置</label>
+                  <select
+                    className="ts-select"
+                    value={settings.position}
+                    onChange={(e) => onUpdatePosition(e.target.value)}
+                  >
+                    {positionOptions.map((o) => (
+                      <option key={o.value} value={o.value}>
+                        {o.label}
+                      </option>
+                    ))}
+                  </select>
+                </div>
+                <div className="ts-form-field ts-half">
+                  <label>字体</label>
+                  <select
+                    className="ts-select"
+                    value={settings.font}
+                    onChange={(e) => onUpdateFont(e.target.value)}
+                  >
+                    {fontOptions.map((f) => (
+                      <option key={f.value} value={f.value}>
+                        {f.tag === "hot" ? "🔥 " : f.tag === "new" ? "🆕 " : ""}
+                        {f.label}
+                      </option>
+                    ))}
+                  </select>
+                </div>
+              </div>
+              <SliderRow
+                label="字号"
+                value={settings.size}
+                min={16}
+                max={120}
+                onChange={onUpdateSize}
+              />
+              <div className="ts-form-field">
+                <label>样式</label>
+                <div className="ts-style-btns">
+                  <button
+                    type="button"
+                    className={`ts-style-btn${settings.bold ? " active" : ""}`}
+                    onClick={onToggleBold}
+                  >
+                    <b>B</b>
+                  </button>
+                  <button
+                    type="button"
+                    className={`ts-style-btn${settings.italic ? " active" : ""}`}
+                    onClick={onToggleItalic}
+                  >
+                    <i>I</i>
+                  </button>
+                  <button
+                    type="button"
+                    className={`ts-style-btn${settings.stroke ? " active" : ""}`}
+                    onClick={() => {
+                      onToggleStroke()
+                      if (!settings.stroke && (settings.strokeWidth ?? 0) < 2)
+                        upd({ strokeWidth: 4 })
+                    }}
+                    title="描边"
+                  >
+                    S
+                  </button>
+                  <button
+                    type="button"
+                    className={`ts-style-btn${settings.shadow ? " active" : ""}`}
+                    onClick={() => {
+                      onToggleShadow()
+                      if (!settings.shadow) {
+                        upd({
+                          shadowOffsetX: 2,
+                          shadowOffsetY: 2,
+                          shadowBlur: 4,
+                          shadowColor: "rgba(0,0,0,0.8)",
+                        })
+                      }
+                    }}
+                    title="阴影"
+                  >
+                    ☁
+                  </button>
+                </div>
+              </div>
+              <ColorPicker
+                label="字色"
+                value={settings.color}
+                palette={TITLE_COLOR_PALETTE}
+                onChange={(c) => upd({ color: c })}
+              />
+            </>
+          ),
+        },
+        {
+          key: "stroke",
+          label: "描边",
+          children: (
+            <>
+              <div className="ts-toggle-row">
+                <label>
+                  <input type="checkbox" checked={settings.stroke} onChange={onToggleStroke} />
+                  启用描边
+                </label>
+              </div>
+              {settings.stroke && (
+                <>
+                  <SliderRow
+                    label="描边宽度"
+                    value={settings.strokeWidth ?? 4}
+                    min={0}
+                    max={20}
+                    onChange={(v) => upd({ strokeWidth: v })}
+                  />
+                  <ColorPicker
+                    label="描边颜色"
+                    value={settings.strokeColor ?? "#000000"}
+                    palette={STROKE_COLOR_PALETTE}
+                    onChange={(c) => upd({ strokeColor: c })}
+                  />
+                </>
+              )}
+            </>
+          ),
+        },
+        {
+          key: "shadow",
+          label: "阴影",
+          children: (
+            <>
+              <div className="ts-toggle-row">
+                <label>
+                  <input type="checkbox" checked={settings.shadow} onChange={onToggleShadow} />
+                  启用阴影
+                </label>
+              </div>
+              {settings.shadow && (
+                <>
+                  <SliderRow
+                    label="X偏移"
+                    value={settings.shadowOffsetX ?? 2}
+                    min={-20}
+                    max={20}
+                    onChange={(v) => upd({ shadowOffsetX: v })}
+                  />
+                  <SliderRow
+                    label="Y偏移"
+                    value={settings.shadowOffsetY ?? 2}
+                    min={-20}
+                    max={20}
+                    onChange={(v) => upd({ shadowOffsetY: v })}
+                  />
+                  <SliderRow
+                    label="模糊半径"
+                    value={settings.shadowBlur ?? 4}
+                    min={0}
+                    max={30}
+                    onChange={(v) => upd({ shadowBlur: v })}
+                  />
+                  <div className="ts-form-field">
+                    <label>阴影颜色</label>
+                    <input
+                      type="text"
+                      className="ts-input"
+                      value={settings.shadowColor ?? "rgba(0,0,0,0.8)"}
+                      onChange={(e) => upd({ shadowColor: e.target.value })}
+                      placeholder="rgba(0,0,0,0.8)"
+                    />
+                  </div>
+                </>
+              )}
+            </>
+          ),
+        },
+        {
+          key: "bg",
+          label: "背景",
+          children: (
+            <>
+              <div className="ts-toggle-row">
+                <label>
+                  <input
+                    type="checkbox"
+                    checked={settings.bgEnabled}
+                    onChange={() => upd({ bgEnabled: !settings.bgEnabled })}
+                  />
+                  启用背景色块
+                </label>
+              </div>
+              {settings.bgEnabled && (
+                <>
+                  <ColorPicker
+                    label="背景颜色（含透明度）"
+                    value={settings.bgColor}
+                    palette={BG_COLOR_PALETTE}
+                    onChange={(c) => upd({ bgColor: c })}
+                  />
+                  <SliderRow
+                    label="内边距"
+                    value={settings.bgPadding}
+                    min={0}
+                    max={40}
+                    onChange={(v) => upd({ bgPadding: v })}
+                  />
+                  <SliderRow
+                    label="圆角"
+                    value={settings.bgRadius}
+                    min={0}
+                    max={30}
+                    onChange={(v) => upd({ bgRadius: v })}
+                  />
+                </>
+              )}
+            </>
+          ),
+        },
+        {
+          key: "layout",
+          label: "排版",
+          children: (
+            <>
+              <SliderRow
+                label="每行最大字符数"
+                value={settings.maxCharsPerLine ?? 0}
+                min={0}
+                max={20}
+                unit=""
+                onChange={(v) => upd({ maxCharsPerLine: v })}
+              />
+              <div
+                className="ts-form-field"
+                style={{ fontSize: 11, color: "#9ca3af", marginTop: -4 }}
+              >
+                0 = 不自动换行（按 / 手动分行）
+              </div>
+              <SliderRow
+                label="行距倍数"
+                value={Math.round((settings.lineHeight ?? 1.2) * 100) / 100}
+                min={1}
+                max={2}
+                step={0.05}
+                unit=""
+                onChange={(v) => upd({ lineHeight: Number(v.toFixed(2)) })}
+              />
+              <SliderRow
+                label="顶部边距"
+                value={settings.marginTop ?? 24}
+                min={0}
+                max={200}
+                onChange={(v) => upd({ marginTop: v })}
+              />
+            </>
+          ),
+        },
+        ...(showCoverToggle
+          ? [
+              {
+                key: "cover",
+                label: "封面",
+                children: (
+                  <div className="ts-toggle-row">
+                    <label>
+                      <input
+                        type="checkbox"
+                        checked={coverEnabled}
+                        onChange={(e) => onToggleCover?.(e.target.checked)}
+                      />
+                      封面使用独立标题样式
+                    </label>
+                  </div>
+                ),
+              },
+            ]
+          : []),
+      ]}
+    />
+  )
+}
+
+export default TitleStyleParamsTab
@@ -1,29 +1,30 @@
 /**
- * 标题模板编辑器（v3 重构）
+ * 标题模板编辑器（公共组件）
 *
 * - Modal 弹窗 860px 宽
- * - 左侧：300px 竖屏预览区（图片背景+暗色渐变遮罩+透明 Canvas 叠字）+ 模板名称输入框
- * - 右侧：参数 Tab 面板（基础/描边/阴影/背景/排版），复用 TitleStylePanel 的 paramsOnly 模式
+ * - 左侧：300px 竖屏预览区（图片背景+暗角+透明 Canvas 叠字）+ 模板名称输入
+ * - 右侧：参数 Tab 面板（基础/描边/阴影/背景/排版），复用 TitleStyleParamsTab
 * - 底部：取消 / 保存模板 按钮
- * - 内置模板编辑时保存会创建副本（带"副本"逻辑由 handleSave 处理）
+ * - 内置模板编辑时保存会创建副本（带"副本"逻辑由 onSave 的调用方处理）
 */
 import React, { useEffect, useMemo, useState } from "react"
 import { Modal, Button, Input, message } from "antd"
-import TitleStylePanel from "../../pages/generate/components/title/TitleStylePanel"
-import TitleMiniPreview from "../../pages/generate/components/title/TitleMiniPreview"
-import { POSITION_OPTIONS } from "../../pages/generate/constants"
-import { FONT_OPTIONS } from "./constants"
-import type { TitleSettings } from "../../pages/generate/types"
-import { DEFAULT_TITLE_SETTINGS_FULL } from "../../pages/generate/types"
+import type { TitleStyleSettings } from "./settings"
+import { DEFAULT_TITLE_STYLE_SETTINGS } from "./settings"
 import { titleStyleConfigToCamel, camelToTitleStyleConfig } from "./utils"
 import type { TitleTemplate } from "./template-types"
 import type { TitleStyleConfig } from "./types"
+import { POSITION_OPTIONS } from "./position-options"
+import { FONT_OPTIONS } from "./constants"
+import TitleMiniPreview from "./TitleMiniPreview"
+import TitleStyleParamsTab from "./TitleStyleParamsTab"
+import "./TitleTemplate.css"
+import "./TitleStylePanel.css"

 interface Props {
  open: boolean
  template: TitleTemplate
  onClose: () => void
-  /** 用户点击保存：将编辑结果回调给父组件（父组件统一做 CRUD，避免双 hook 实例不同步） */
  onSave: (data: { name: string; emoji: string; style: Partial<TitleStyleConfig> }) => void
 }

@@ -31,10 +32,9 @@ interface Props {
 const EDITOR_BG = "/title-templates/portrait1.jpg"

 const TitleTemplateEditor: React.FC<Props> = ({ open, template, onClose, onSave }) => {
-  const [settings, setSettings] = useState<TitleSettings>(() => ({
-    ...DEFAULT_TITLE_SETTINGS_FULL,
+  const [settings, setSettings] = useState<TitleStyleSettings>(() => ({
+    ...DEFAULT_TITLE_STYLE_SETTINGS,
    ...titleStyleConfigToCamel(template.style || {}),
-    title: "预览标题文字",
  }))
  const [formName, setFormName] = useState(template.name || "")
  const [formEmoji, setFormEmoji] = useState(template.emoji || "✨")
@@ -43,16 +43,15 @@ const TitleTemplateEditor: React.FC<Props> = ({ open, template, onClose, onSave
  useEffect(() => {
    if (open) {
      setSettings({
-        ...DEFAULT_TITLE_SETTINGS_FULL,
+        ...DEFAULT_TITLE_STYLE_SETTINGS,
        ...titleStyleConfigToCamel(template.style || {}),
-        title: "预览标题文字",
      })
      setFormName(template.name || "")
      setFormEmoji(template.emoji || "✨")
    }
  }, [open, template])

-  const upd = (patch: Partial<TitleSettings>) => setSettings((s) => ({ ...s, ...patch }))
+  const upd = (patch: Partial<TitleStyleSettings>) => setSettings((s) => ({ ...s, ...patch }))

  const handleSave = () => {
    const name = formName.trim()
@@ -69,11 +68,11 @@ const TitleTemplateEditor: React.FC<Props> = ({ open, template, onClose, onSave
    }
  }

-  // 编辑器内的预览用 settings：字号适配竖屏
-  const previewSettings = useMemo<TitleSettings>(() => {
-    // 竖屏宽度 200px，按比例缩放字号，让预览看起来协调
-    return { ...settings, size: Math.round(settings.size * 0.55) }
-  }, [settings])
+  // 编辑器预览 settings：竖屏宽度 200px，字号按比例缩放
+  const previewSettings = useMemo<TitleStyleSettings>(
+    () => ({ ...settings, size: Math.round(settings.size * 0.55) }),
+    [settings],
+  )

  return (
    <Modal
@@ -140,7 +139,7 @@ const TitleTemplateEditor: React.FC<Props> = ({ open, template, onClose, onSave
        </div>
        {/* 右侧：参数 Tab */}
        <div className="ttv3-editor-right">
-          <TitleStylePanel
+          <TitleStyleParamsTab
            settings={settings}
            onUpdatePosition={(p) => upd({ position: p, posX: null, posY: null })}
            onUpdateFont={(f) => upd({ font: f })}
@@ -155,15 +154,9 @@ const TitleTemplateEditor: React.FC<Props> = ({ open, template, onClose, onSave
              })
            }
            onToggleShadow={() => upd({ shadow: !settings.shadow })}
-            onApplyPreset={() => {
-              /* 编辑器内不使用系统预设快捷键 */
-            }}
-            onUpdateStyle={(patch) => upd(patch)}
-            activePreset={null}
-            titlePresets={[]}
-            POSITION_OPTIONS={POSITION_OPTIONS}
-            FONT_OPTIONS={FONT_OPTIONS}
-            paramsOnly
+            onUpdatePatch={upd}
+            positionOptions={POSITION_OPTIONS}
+            fontOptions={FONT_OPTIONS}
          />
        </div>
      </div>
@@ -0,0 +1,343 @@
+/**
+ * 标题模板选择器 — 大卡片网格（共享组件）
+ *
+ * 渲染「我的模板」+「系统模板」两个分组的 3:4 竖版大圆角卡片：
+ * - 卡片上半：示例背景图 + vignette 暗角 + 透明 Canvas 大字预览
+ * - 卡片下半：emoji + 名称 + 系统/我的标签 + 始终可见的编辑/复制/导出/删除按钮
+ * - 选中紫色边框；右上角「新建模板」按钮；点编辑/新建弹 TitleTemplateEditor
+ *
+ * Props 通用化，不耦合业务 state。
+ */
+import React, { useCallback, useMemo, useState } from "react"
+import { Button, message, Popconfirm } from "antd"
+import {
+  PlusOutlined,
+  EditOutlined,
+  CopyOutlined,
+  DeleteOutlined,
+  ExportOutlined,
+  CheckOutlined,
+} from "@ant-design/icons"
+import type { TitleTemplate } from "./template-types"
+import type { TitleStyleSettings } from "./settings"
+import { DEFAULT_TITLE_STYLE_SETTINGS } from "./settings"
+import {
+  titleStyleConfigToCamel,
+  camelToTitleStyleConfig,
+  templateToPreviewSettings,
+} from "./utils"
+import { useTitleTemplates } from "./useTitleTemplates"
+import TitleMiniPreview from "./TitleMiniPreview"
+import TitleTemplateEditor from "./TitleTemplateEditor"
+import "./TitleTemplate.css"
+import "./TitleStylePanel.css"
+
+export interface TitleTemplateSelectorProps {
+  /** 当前选中模板 id（受控） */
+  value?: string | null
+  /** 选中模板时回调（templateId, fullStyleSettings, template） */
+  onChange?: (templateId: string, style: TitleStyleSettings, template: TitleTemplate) => void
+  /** 是否显示编辑器入口（新建/编辑按钮），默认 true */
+  showEditor?: boolean
+  /** 显示哪些分组，默认全部 */
+  categories?: Array<"system" | "custom">
+  /** 使用场景标识（仅作 data-attr，不影响样式） */
+  context?: string
+}
+
+/* ── 卡片预览背景图池（按 index 轮换） ── */
+const PREVIEW_BG_IMAGES = [
+  "/title-templates/portrait1.jpg",
+  "/title-templates/portrait2.jpg",
+  "/title-templates/scene1.jpg",
+]
+
+/* ── 预览容器：用 ref 测量宽度后再渲染透明 Canvas，保证文字清晰 ── */
+const FillPreview: React.FC<{
+  settings: TitleStyleSettings
+  sampleText: string
+  portrait?: boolean
+}> = ({ settings, sampleText, portrait }) => {
+  const [w, setW] = useState(0)
+  // 首次挂载后测量一次
+  const setRef = useCallback((el: HTMLDivElement | null) => {
+    if (el) setW(Math.floor(el.clientWidth))
+  }, [])
+  return (
+    <div ref={setRef} className="tt-fill-canvas-wrap">
+      {w > 0 && (
+        <TitleMiniPreview
+          settings={settings}
+          width={w}
+          sampleText={sampleText}
+          transparent
+          portrait={portrait}
+        />
+      )}
+    </div>
+  )
+}
+
+const TitleTemplateSelector: React.FC<TitleTemplateSelectorProps> = ({
+  value,
+  onChange,
+  showEditor = true,
+  categories = ["system", "custom"],
+  context,
+}) => {
+  const {
+    templates,
+    createTemplate,
+    duplicateTemplate,
+    updateTemplate,
+    deleteTemplate,
+    exportTemplate,
+  } = useTitleTemplates()
+
+  const [editingTemplate, setEditingTemplate] = useState<TitleTemplate | null>(null)
+  const [editorOpen, setEditorOpen] = useState(false)
+
+  const grouped = useMemo(
+    () => ({
+      builtin: templates.filter((t) => t.isBuiltin),
+      custom: templates.filter((t) => !t.isBuiltin),
+    }),
+    [templates],
+  )
+
+  const showSys = categories.includes("system")
+  const showMine = categories.includes("custom")
+
+  /* ── 选中模板：合成完整 TitleStyleSettings 回调给父组件 ── */
+  const handleSelectTemplate = useCallback(
+    (tpl: TitleTemplate) => {
+      const full: TitleStyleSettings = {
+        ...DEFAULT_TITLE_STYLE_SETTINGS,
+        ...titleStyleConfigToCamel(tpl.style),
+      }
+      onChange?.(tpl.id, full, tpl)
+    },
+    [onChange],
+  )
+
+  const handleRequestCreate = useCallback(() => {
+    // 新建：以当前选中模板样式为起点，否则用默认样式
+    let base: TitleStyleSettings = DEFAULT_TITLE_STYLE_SETTINGS
+    if (value) {
+      const sel = templates.find((t) => t.id === value)
+      if (sel) {
+        base = { ...DEFAULT_TITLE_STYLE_SETTINGS, ...titleStyleConfigToCamel(sel.style) }
+      }
+    }
+    const draft: TitleTemplate = {
+      id: "",
+      name: "我的标题模板",
+      emoji: "✨",
+      isBuiltin: false,
+      style: camelToTitleStyleConfig({
+        ...base,
+        position: base.position === "custom" ? "bottom" : base.position,
+      }),
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    }
+    setEditingTemplate(draft)
+    setEditorOpen(true)
+  }, [value, templates])
+
+  const handleRequestEdit = useCallback((tpl: TitleTemplate) => {
+    setEditingTemplate(tpl)
+    setEditorOpen(true)
+  }, [])
+
+  const handleDuplicate = useCallback(
+    (t: TitleTemplate) => {
+      const dup = duplicateTemplate(t.id)
+      if (dup) message.success(`已复制：${dup.name}`)
+    },
+    [duplicateTemplate],
+  )
+  const handleDelete = useCallback(
+    (t: TitleTemplate) => {
+      deleteTemplate(t.id)
+      message.success("已删除模板")
+    },
+    [deleteTemplate],
+  )
+  const handleExport = useCallback(
+    (t: TitleTemplate) => {
+      const json = exportTemplate(t.id)
+      if (!json) return
+      const blob = new Blob([json], { type: "application/json" })
+      const url = URL.createObjectURL(blob)
+      const a = document.createElement("a")
+      a.href = url
+      a.download = `${t.name}.title-template.json`
+      a.click()
+      URL.revokeObjectURL(url)
+    },
+    [exportTemplate],
+  )
+
+  const handleEditorSave = useCallback(
+    (data: { name: string; emoji: string; style: Partial<import("./types").TitleStyleConfig> }) => {
+      if (!editingTemplate) return
+      let saved: TitleTemplate
+      if (editingTemplate.isBuiltin || !editingTemplate.id) {
+        saved = createTemplate({ name: data.name, emoji: data.emoji, style: data.style })
+      } else {
+        updateTemplate(editingTemplate.id, {
+          name: data.name,
+          emoji: data.emoji,
+          style: data.style,
+        })
+        saved = {
+          ...editingTemplate,
+          name: data.name,
+          emoji: data.emoji,
+          style: data.style,
+          updatedAt: new Date().toISOString(),
+        }
+      }
+      setEditorOpen(false)
+      setEditingTemplate(null)
+      message.success(`已保存：${data.name}`)
+      handleSelectTemplate(saved)
+    },
+    [editingTemplate, createTemplate, updateTemplate, handleSelectTemplate],
+  )
+
+  /* ── 渲染单张大卡片 ── */
+  const renderCard = (t: TitleTemplate, idx: number, section: "mine" | "sys") => {
+    const isSelected = value === t.id
+    const bgIdx = idx % PREVIEW_BG_IMAGES.length
+    const bgImg = PREVIEW_BG_IMAGES[bgIdx]
+    const preview = templateToPreviewSettings(t, 42)
+    return (
+      <div
+        key={t.id}
+        className={`ttv3-card${isSelected ? " selected" : ""}`}
+        onClick={() => handleSelectTemplate(t)}
+        data-context={context}
+      >
+        <div className="ttv3-preview">
+          <img className="ttv3-bg" src={bgImg} alt="" />
+          <div className="ttv3-vignette" />
+          <FillPreview settings={preview} sampleText="预览标题文字" portrait />
+          <span className={`ttv3-badge ttv3-badge--${section}`}>
+            {section === "sys" ? "系统" : "我的"}
+          </span>
+          <span className={`ttv3-check${isSelected ? " on" : ""}`}>
+            {isSelected && <CheckOutlined />}
+          </span>
+        </div>
+        <div className="ttv3-footer">
+          <div className="ttv3-name-row">
+            <span className="ttv3-emoji">{t.emoji || "✨"}</span>
+            <span className="ttv3-name" title={t.name}>
+              {t.name}
+            </span>
+            <span className={`ttv3-tag ttv3-tag--${section}`}>
+              {section === "sys" ? "系统" : "我的"}
+            </span>
+          </div>
+          {showEditor && (
+            <div className="ttv3-actions" onClick={(e) => e.stopPropagation()}>
+              <button
+                type="button"
+                className="ttv3-act ttv3-act--primary"
+                disabled={t.isBuiltin}
+                onClick={() => handleRequestEdit(t)}
+                title={t.isBuiltin ? "系统模板不可编辑，点击复制后可编辑" : "编辑"}
+              >
+                <EditOutlined /> 编辑
+              </button>
+              <button
+                type="button"
+                className="ttv3-act"
+                onClick={() => handleDuplicate(t)}
+                title="复制"
+              >
+                <CopyOutlined /> 复制
+              </button>
+              <button
+                type="button"
+                className="ttv3-act"
+                onClick={() => handleExport(t)}
+                title="导出"
+              >
+                <ExportOutlined /> 导出
+              </button>
+              <Popconfirm title="删除该模板？" onConfirm={() => handleDelete(t)}>
+                <button
+                  type="button"
+                  className="ttv3-act ttv3-act--danger"
+                  disabled={t.isBuiltin}
+                  title={t.isBuiltin ? "系统模板不可删除" : "删除"}
+                >
+                  <DeleteOutlined /> 删除
+                </button>
+              </Popconfirm>
+            </div>
+          )}
+        </div>
+      </div>
+    )
+  }
+
+  return (
+    <div className="xx-title-style-section ttv3-panel">
+      <div className="ttv3-header">
+        <span className="ttv3-title">标题模板</span>
+        {showEditor && (
+          <Button
+            type="primary"
+            size="small"
+            icon={<PlusOutlined />}
+            onClick={handleRequestCreate}
+            className="ttv3-new-btn"
+          >
+            新建模板
+          </Button>
+        )}
+      </div>
+
+      {showMine && (
+        <div className="ttv3-section">
+          <div className="ttv3-section-label">我的模板</div>
+          {grouped.custom.length === 0 ? (
+            <div className="ttv3-empty">
+              <div className="ttv3-empty-icon">✨</div>
+              <div className="ttv3-empty-text">还没有自定义模板，点右上角「新建模板」创建</div>
+            </div>
+          ) : (
+            <div className="ttv3-grid">
+              {grouped.custom.map((t, i) => renderCard(t, i, "mine"))}
+            </div>
+          )}
+        </div>
+      )}
+
+      {showSys && (
+        <div className="ttv3-section">
+          <div className="ttv3-section-label">系统模板</div>
+          <div className="ttv3-grid">{grouped.builtin.map((t, i) => renderCard(t, i, "sys"))}</div>
+        </div>
+      )}
+
+      {showEditor && editorOpen && editingTemplate && (
+        <TitleTemplateEditor
+          open={editorOpen}
+          template={editingTemplate}
+          onClose={() => {
+            setEditorOpen(false)
+            setEditingTemplate(null)
+          }}
+          onSave={handleEditorSave}
+        />
+      )}
+    </div>
+  )
+}
+
+export default TitleTemplateSelector
@@ -0,0 +1,19 @@
+/**
+ * 公共标题模板/样式组件统一导出
+ *
+ * 任何页面需要标题样式配置/模板选择/模板编辑，从这里 import，
+ * 不要直接 import pages/generate/components/title/* 下的内部组件。
+ */
+export { default as TitleTemplateSelector } from "./TitleTemplateSelector"
+export { default as TitleTemplateEditor } from "./TitleTemplateEditor"
+export { default as TitleStyleParamsTab } from "./TitleStyleParamsTab"
+export { default as TitleMiniPreview } from "./TitleMiniPreview"
+export { useTitleTemplates } from "./useTitleTemplates"
+export * from "./constants"
+export * from "./types"
+export * from "./template-types"
+export * from "./settings"
+export * from "./utils"
+export { POSITION_OPTIONS } from "./position-options"
+export type { PositionOption, FontOption, TitleStyleParamsTabProps } from "./TitleStyleParamsTab"
+export type { TitleTemplateSelectorProps } from "./TitleTemplateSelector"
@@ -0,0 +1,14 @@
+/**
+ * 标题位置选项（公共常量）
+ */
+export interface PositionOption {
+  value: string
+  label: string
+}
+
+export const POSITION_OPTIONS: PositionOption[] = [
+  { value: "top", label: "顶部" },
+  { value: "center", label: "居中" },
+  { value: "bottom", label: "底部" },
+  { value: "custom", label: "自定义" },
+]
@@ -0,0 +1,64 @@
+/**
+ * 标题样式设置 — 公共 camelCase 类型与默认值
+ *
+ * 本文件是 @/components/title 公共包的唯一样式类型出口，不依赖任何业务页面（generate/ai-avatar）的私有类型。
+ * - 字段与后端 snake_case TitleStyleConfig 一一对应（camelCase 版本）
+ * - DEFAULT_TITLE_STYLE_SETTINGS 用于组件内部补全默认值
+ * - aiAutoSelect / title / coverTitle 等业务状态不在本类型中——它们属于页面业务 state
+ */
+import type { TitleLineOverride } from "./types"
+
+export interface TitleStyleSettings {
+  position: string
+  font: string
+  size: number
+  bold: boolean
+  italic: boolean
+  stroke: boolean
+  shadow: boolean
+  color: string
+  posX: number | null
+  posY: number | null
+  lineHeight: number
+  marginTop: number
+  maxCharsPerLine: number
+  strokeWidth: number
+  strokeColor: string
+  shadowOffsetX: number
+  shadowOffsetY: number
+  shadowBlur: number
+  shadowColor: string
+  bgEnabled: boolean
+  bgColor: string
+  bgPadding: number
+  bgRadius: number
+  lineOverrides: TitleLineOverride[]
+}
+
+/** 公共默认样式（经典白字黑描边） */
+export const DEFAULT_TITLE_STYLE_SETTINGS: TitleStyleSettings = {
+  position: "bottom",
+  font: "思源黑体",
+  size: 56,
+  bold: true,
+  italic: false,
+  stroke: true,
+  shadow: false,
+  color: "#ffffff",
+  posX: null,
+  posY: null,
+  lineHeight: 1.2,
+  marginTop: 24,
+  maxCharsPerLine: 10,
+  strokeWidth: 5,
+  strokeColor: "#000000",
+  shadowOffsetX: 2,
+  shadowOffsetY: 2,
+  shadowBlur: 4,
+  shadowColor: "rgba(0,0,0,0.8)",
+  bgEnabled: false,
+  bgColor: "rgba(0,0,0,0.5)",
+  bgPadding: 12,
+  bgRadius: 8,
+  lineOverrides: [],
+}
@@ -1,24 +1,25 @@
 /**
 * 标题样式工具（#2001 / 模板系统 #2003）
 *
- * - snake_case TitleStyleConfig ↔ camelCase TitleSettings 互转
+ * - snake_case TitleStyleConfig <-> camelCase TitleStyleSettings 互转
 * - preset 归一化预览（修复"标题"两字大小不一）
+ * - template -> preview settings 转换
 */
 import type { TitleStyleConfig } from "./types"
-import type { TitleSettings } from "../../pages/generate/types"
+import type { TitleStyleSettings } from "./settings"
+import { DEFAULT_TITLE_STYLE_SETTINGS } from "./settings"
 import { TITLE_PRESETS } from "./constants"
-import { DEFAULT_TITLE_SETTINGS_FULL } from "../../pages/generate/types"
 import type { TitleTemplate } from "./template-types"

-/** snake_case TitleStyleConfig → camelCase TitleSettings（仅覆盖已知字段） */
-export function titleStyleConfigToCamel(s: Partial<TitleStyleConfig>): Partial<TitleSettings> {
-  const out: Partial<TitleSettings> = {}
+/** snake_case TitleStyleConfig -> camelCase TitleStyleSettings（仅覆盖已知字段） */
+export function titleStyleConfigToCamel(s: Partial<TitleStyleConfig>): Partial<TitleStyleSettings> {
+  const out: Partial<TitleStyleSettings> = {}
  if (s.font != null) out.font = s.font
  if (s.size != null) out.size = s.size
  if (s.color != null) out.color = s.color
  if (s.bold != null) out.bold = s.bold
  if (s.italic != null) out.italic = s.italic
-  if (s.position != null) out.position = s.position as TitleSettings["position"]
+  if (s.position != null) out.position = s.position
  if (s.pos_x != null) out.posX = s.pos_x
  if (s.pos_y != null) out.posY = s.pos_y
  if (s.line_height != null) out.lineHeight = s.line_height
@@ -40,8 +41,8 @@ export function titleStyleConfigToCamel(s: Partial<TitleStyleConfig>): Partial<T
  return out
 }

-/** camelCase TitleSettings patch → snake_case TitleStyleConfig patch */
-export function camelToTitleStyleConfig(p: Partial<TitleSettings>): Partial<TitleStyleConfig> {
+/** camelCase TitleStyleSettings patch -> snake_case TitleStyleConfig patch */
+export function camelToTitleStyleConfig(p: Partial<TitleStyleSettings>): Partial<TitleStyleConfig> {
  const out: Partial<TitleStyleConfig> = {}
  if (p.font != null) out.font = p.font
  if (p.size != null) out.size = p.size
@@ -71,15 +72,15 @@ export function camelToTitleStyleConfig(p: Partial<TitleSettings>): Partial<Titl
 }

 /**
- * 把 preset style（snake_case）归一化为固定字号的 TitleSettings，
+ * 把 preset style（snake_case）归一化为固定字号的 TitleStyleSettings，
 * 用于"预设卡片"缩略预览——所有卡片视觉上"标题"两字大小一致，便于辨识。
 * 描边/阴影/背景padding 按 fixedSize / 原始 size 比例缩放，避免粗描边爆框。
 */
 export function buildPresetPreviewSettings(
-  base: TitleSettings,
+  base: TitleStyleSettings,
  presetKey: string,
  fixedSize = 56,
-): TitleSettings {
+): TitleStyleSettings {
  const preset = TITLE_PRESETS.find((p) => p.key === presetKey)
  if (!preset) return base
  const origSize = preset.style.size ?? fixedSize
@@ -87,25 +88,25 @@ export function buildPresetPreviewSettings(
  const scale = (v: number | undefined, fallback: number): number =>
    v != null ? Math.round(v * ratio) : fallback
  return {
+    ...DEFAULT_TITLE_STYLE_SETTINGS,
    ...base,
    ...titleStyleConfigToCamel(preset.style),
    size: fixedSize,
-    strokeWidth: scale(preset.style.stroke_width, base.strokeWidth) ?? base.strokeWidth,
-    shadowOffsetX: scale(preset.style.shadow_offset_x, base.shadowOffsetX) ?? base.shadowOffsetX,
-    shadowOffsetY: scale(preset.style.shadow_offset_y, base.shadowOffsetY) ?? base.shadowOffsetY,
-    shadowBlur: scale(preset.style.shadow_blur, base.shadowBlur) ?? base.shadowBlur,
-    bgPadding: scale(preset.style.bg_padding, base.bgPadding) ?? base.bgPadding,
+    strokeWidth: scale(preset.style.stroke_width, base.strokeWidth),
+    shadowOffsetX: scale(preset.style.shadow_offset_x, base.shadowOffsetX),
+    shadowOffsetY: scale(preset.style.shadow_offset_y, base.shadowOffsetY),
+    shadowBlur: scale(preset.style.shadow_blur, base.shadowBlur),
+    bgPadding: scale(preset.style.bg_padding, base.bgPadding),
    lineOverrides: [],
  }
 }

 /**
- * 把 TitleTemplate 渲染为完整 TitleSettings（带默认值），用于卡片预览。
- * 与模板选择器中保持一致，抽出共用。
+ * 把 TitleTemplate 渲染为完整 TitleStyleSettings（带默认值），用于卡片预览。
 */
-export function templateToPreviewSettings(t: TitleTemplate, fixedSize = 48): TitleSettings {
-  const base: TitleSettings = {
-    ...DEFAULT_TITLE_SETTINGS_FULL,
+export function templateToPreviewSettings(t: TitleTemplate, fixedSize = 48): TitleStyleSettings {
+  const base: TitleStyleSettings = {
+    ...DEFAULT_TITLE_STYLE_SETTINGS,
    ...titleStyleConfigToCamel(t.style),
  }
  // 预览时用固定字号保证所有卡片字大小一致；描边/阴影/padding按比例缩放
@@ -412,7 +412,7 @@ const PanelTitleConfig: React.FC<PanelTitleConfigProps> = ({ titleConfig, onUpda
        onUpdateStyle={handleUpdateStyle}
        showCoverToggle
        previewWidth={280}
-        enableTemplates={false}
+        enableTemplates={true}
        selectedTemplateId={selectedTemplateId}
        onApplyTemplate={handleApplyTemplate}
        activePreset={activePreset}
@@ -1,271 +1 @@
-/**
- * 标题迷你 Canvas 预览（#2001）
- *
- * 渲染一张指定宽度的小 Canvas 预览标题效果，用于：
- * - 预设卡片缩略图
- * - 样式面板顶部的实时预览
- *
- * 与 titleCanvas.ts 渲染逻辑保持一致，但：
- * - 固定分辨率（width × 宽高比约 2:1）
- * - 不调用 ffmpeg，只做视觉预览
- * - 支持背景色块、描边宽度/颜色、阴影参数化、行距、自动换行
- */
-import React, { useEffect, useRef } from "react"
-import type { TitleSettings } from "../../types"
-import { getFontFamily } from "@/components/title/constants"
-
-interface Props {
-  settings: TitleSettings
-  width?: number
-  sampleText?: string
-  /** 背景（预览用，默认深色渐变模拟视频底），transparent=true 时忽略 */
-  background?: string
-  /** 高度（可选，默认按 portrait 选比例） */
-  height?: number
-  /** 透明背景（卡片/编辑器预览叠加在图片上时使用） */
-  transparent?: boolean
-  /** 纵向竖屏预览（9:16），true 时 aspect=16/9 适配手机视频比例 */
-  portrait?: boolean
-}
-
-/** 按 maxCharsPerLine 自动换行 */
-function wrapLines(text: string, maxChars: number): string[] {
-  const manual = text
-    .split(/[/／\n]/)
-    .map((l) => l.trim())
-    .filter(Boolean)
-  if (!maxChars || maxChars <= 0) return manual
-  const out: string[] = []
-  for (const line of manual) {
-    if (line.length <= maxChars) {
-      out.push(line)
-      continue
-    }
-    let cur = ""
-    for (const ch of line) {
-      cur += ch
-      if (cur.length >= maxChars) {
-        out.push(cur)
-        cur = ""
-      }
-    }
-    if (cur) out.push(cur)
-  }
-  return out
-}
-
-const TitleMiniPreview: React.FC<Props> = ({
-  settings,
-  width = 200,
-  sampleText,
-  background = "linear-gradient(135deg,#1f2937,#111827)",
-  height,
-  transparent = false,
-  portrait = false,
-}) => {
-  const canvasRef = useRef<HTMLCanvasElement>(null)
-  const h = height ?? Math.round(width * (portrait ? 16 / 9 : 1 / 1.8))
-  const text = (sampleText || settings.title || "预览标题").trim() || "预览标题"
-
-  useEffect(() => {
-    let cancelled = false
-    const draw = () => {
-      if (cancelled) return
-      const cvs = canvasRef.current
-      if (!cvs) return
-      const dpr = window.devicePixelRatio || 1
-      cvs.width = width * dpr
-      cvs.height = h * dpr
-      cvs.style.width = `${width}px`
-      cvs.style.height = `${h}px`
-      const ctx = cvs.getContext("2d")
-      if (!ctx) return
-      ctx.scale(dpr, dpr)
-      ctx.clearRect(0, 0, width, h)
-
-      // 背景（transparent 时跳过，用于叠加在图片上）
-      if (!transparent) {
-        ctx.fillStyle = "#111827"
-        ctx.fillRect(0, 0, width, h)
-      }
-
-      // 分辨率缩放：以 360 宽为基准（对应 720p 的一半），与外层 previewScale/previewR 保持一致
-      const r = previewR
-
-      // 字体
-      const size = r(settings.size)
-      const ff = getFontFamily(settings.font)
-      const parts: string[] = []
-      if (settings.italic) parts.push("italic")
-      if (settings.bold) parts.push("bold")
-      parts.push(`${size}px`, ff)
-      ctx.font = parts.join(" ")
-      ctx.textAlign = "center"
-      ctx.textBaseline = "middle"
-      ctx.fillStyle = settings.color
-      ctx.lineJoin = "round"
-
-      // 阴影
-      const shadowEnabled = !!settings.shadow
-      const prevShadow = {
-        c: ctx.shadowColor,
-        b: ctx.shadowBlur,
-        ox: ctx.shadowOffsetX,
-        oy: ctx.shadowOffsetY,
-      }
-      if (shadowEnabled) {
-        ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
-        ctx.shadowBlur = r(settings.shadowBlur ?? 4)
-        ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
-        ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
-      }
-
-      // 换行
-      const lines = wrapLines(text, settings.maxCharsPerLine ?? 0)
-      const lineH = size * (settings.lineHeight ?? 1.2)
-      const totalH = lines.length * lineH
-      let startY: number
-      if (settings.position === "top") {
-        startY = size / 2 + r(settings.marginTop ?? 24)
-      } else if (settings.position === "center") {
-        startY = h / 2 - totalH / 2 + size / 2
-      } else {
-        // bottom
-        const botMargin = portrait ? r(24) : r(16)
-        startY = h - totalH - botMargin + size / 2
-      }
-      let centerX = width / 2
-      if (settings.position === "custom" && settings.posX != null) {
-        centerX = (settings.posX / 100) * width
-      }
-
-      // 背景块
-      if (settings.bgEnabled) {
-        const pad = r(settings.bgPadding ?? 12)
-        const rad = r(settings.bgRadius ?? 8)
-        let maxLineW = 0
-        for (const l of lines) {
-          const m = ctx.measureText(l)
-          if (m.width > maxLineW) maxLineW = m.width
-        }
-        const bw = maxLineW + pad * 2
-        const bh = totalH + pad * 2
-        const bx = centerX - bw / 2
-        const by = startY - size / 2 - pad + (size - lineH) / 2
-        ctx.shadowColor = "rgba(0,0,0,0)"
-        ctx.shadowBlur = 0
-        ctx.fillStyle = settings.bgColor ?? "rgba(0,0,0,0.5)"
-        roundRect(ctx, bx, by, bw, bh, rad)
-        ctx.fill()
-        // 关键修复：画完背景块后必须把 fillStyle 重置为文字颜色，
-        // 否则后续 fillText 会用 bgColor 填充文字，导致「文字看不见只剩色块」
-        ctx.fillStyle = settings.color
-        // 恢复阴影
-        if (shadowEnabled) {
-          ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
-          ctx.shadowBlur = r(settings.shadowBlur ?? 4)
-          ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
-          ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
-        }
-      }
-
-      // 描边（先画，再画填充）
-      const strokeEnabled = !!settings.stroke && (settings.strokeWidth ?? 0) > 0
-      lines.forEach((line, i) => {
-        const y = startY + i * lineH
-        if (strokeEnabled) {
-          ctx.shadowColor = "rgba(0,0,0,0)"
-          ctx.shadowBlur = 0
-          ctx.lineWidth = r(settings.strokeWidth ?? 4)
-          ctx.strokeStyle = settings.strokeColor ?? "#000000"
-          ctx.strokeText(line, centerX, y)
-          // 恢复阴影
-          if (shadowEnabled) {
-            ctx.shadowColor = settings.shadowColor ?? "rgba(0,0,0,0.8)"
-            ctx.shadowBlur = r(settings.shadowBlur ?? 4)
-            ctx.shadowOffsetX = r(settings.shadowOffsetX ?? 2)
-            ctx.shadowOffsetY = r(settings.shadowOffsetY ?? 2)
-          }
-        }
-        ctx.fillText(line, centerX, y)
-      })
-
-      // 恢复
-      ctx.shadowColor = prevShadow.c
-      ctx.shadowBlur = prevShadow.b
-      ctx.shadowOffsetX = prevShadow.ox
-      ctx.shadowOffsetY = prevShadow.oy
-    }
-    // 计算当前字号（draw() 内部同样逻辑，抽出来供 fontString 复用）
-    const previewScale = width / 360
-    const previewR = (v: number) => Math.round(v * previewScale)
-    const buildFontString = () => {
-      const size = previewR(settings.size)
-      const ff = getFontFamily(settings.font)
-      const parts: string[] = []
-      if (settings.italic) parts.push("italic")
-      if (settings.bold) parts.push("bold")
-      parts.push(`${size}px`, ff)
-      return parts.join(" ")
-    }
-
-    // Web Font 加载保障：
-    // 1) 等 document.fonts.ready（CSS @font-face 首次可用）
-    // 2) 显式 FontFaceSet.load(fontString, text) 触发浏览器真正下载并加载
-    //    当前字体到 Canvas 可用，避免首次绘制用 fallback 字体画出错字/色块
-    const doDrawWhenReady = async () => {
-      try {
-        if (typeof document !== "undefined" && document.fonts) {
-          await document.fonts.ready
-          try {
-            await document.fonts.load(buildFontString(), text)
-          } catch {
-            /* ignore */
-          }
-        }
-      } finally {
-        if (!cancelled) draw()
-      }
-    }
-    doDrawWhenReady()
-    return () => {
-      cancelled = true
-    }
-  }, [settings, width, h, text, transparent, portrait, background])
-
-  return (
-    <canvas
-      ref={canvasRef}
-      style={{
-        borderRadius: 6,
-        display: "block",
-        maxWidth: "100%",
-        background: transparent ? "transparent" : background,
-      }}
-    />
-  )
-}
-
-function roundRect(
-  ctx: CanvasRenderingContext2D,
-  x: number,
-  y: number,
-  w: number,
-  h: number,
-  r: number,
-) {
-  const rr = Math.min(r, w / 2, h / 2)
-  ctx.beginPath()
-  ctx.moveTo(x + rr, y)
-  ctx.lineTo(x + w - rr, y)
-  ctx.quadraticCurveTo(x + w, y, x + w, y + rr)
-  ctx.lineTo(x + w, y + h - rr)
-  ctx.quadraticCurveTo(x + w, y + h, x + w - rr, y + h)
-  ctx.lineTo(x + rr, y + h)
-  ctx.quadraticCurveTo(x, y + h, x, y + h - rr)
-  ctx.lineTo(x, y + rr)
-  ctx.quadraticCurveTo(x, y, x + rr, y)
-  ctx.closePath()
-}
-
-export default TitleMiniPreview
+export { default } from "@/components/title/TitleMiniPreview"
@@ -31,7 +31,7 @@ import {
 } from "@/components/title/constants"
 import { buildPresetPreviewSettings } from "@/components/title/utils"

-import TitleMiniPreview from "./TitleMiniPreview"
+import TitleMiniPreview from "@/components/title/TitleMiniPreview"
 import TitleTemplateEditor from "@/components/title/TitleTemplateEditor"
 import { useTitleTemplates } from "@/components/title/useTitleTemplates"
 import {
@@ -177,7 +177,7 @@ const ColorPicker: React.FC<{

 /* ── 卡片预览：用 ref 测量容器宽度后再渲染透明 Canvas，保证文字清晰 ── */
 const FillPreview: React.FC<{
-  settings: TitleSettings
+  settings: import("@/components/title/settings").TitleStyleSettings
  sampleText: string
  portrait?: boolean
 }> = ({ settings, sampleText, portrait }) => {
@@ -46,13 +46,9 @@ export const CLIP_COUNT_STEP = 1
 export const MAX_PREVIEW_COUNT = 10
 export const MIN_PREVIEW_COUNT = 1

-/* ── 标题位置选项 ── */
-export const POSITION_OPTIONS = [
-  { value: "top", label: "顶部" },
-  { value: "center", label: "居中" },
-  { value: "bottom", label: "底部" },
-  { value: "custom", label: "自定义" },
-]
+/* ── 标题位置选项（统一从公共层重导出） ── */
+export { POSITION_OPTIONS } from "@/components/title/position-options"
+export type { PositionOption } from "@/components/title/position-options"

 /* ── 标题字体：统一使用公共层定义（#2001） ── */
 export { getFontFamily } from "@/components/title/constants"
@@ -0,0 +1,112 @@
+"""一次性脚本：对历史 quality_score 缺失的视频素材重新打分。
+
+背景（#2073）：镜像 97ad0ae2 时期 calculate_quality_score / classify_from_analysis
+返回 str 而非 AssetClassification 枚举，导致 calculate_asset_quality 连续报
+"'str' object has no attribute 'value'"，大量视频素材的 quality_score 卡在 NULL。
+镜像 8abdeb95 已修复枚举 bug，但历史失败记录不会自动重跑。本脚本扫描全表，
+把 quality_score IS NULL 的视频素材重新投递到 worker.calculate_asset_quality 任务。
+
+使用方式（在 worker 容器内执行）：
+    cd /app/apps/worker
+    # 干跑，只打印会重跑多少条，不发任务
+    python -m scripts.backfill_asset_quality --dry-run
+    # 正式执行
+    python -m scripts.backfill_asset_quality
+    # 只重跑最近 N 天的
+    python -m scripts.backfill_asset_quality --since-days 30
+    # 限流：每投递一批 sleep 几秒，避免瞬间打爆 transcode 队列
+    python -m scripts.backfill_asset_quality --batch-size 50 --sleep 2
+
+也可以直接在 staging 机器上 exec 进容器：
+    docker exec -e PYTHONPATH=/app:/app/apps/api:/app/packages xiaoxia-worker-staging \
+        python -m scripts.backfill_asset_quality --dry-run
+"""
+
+from __future__ import annotations
+
+import argparse
+
+# 保证可以以 python -m scripts.xxx 在容器 /app/apps/worker 下执行
+# 也兼容在 repo 根目录下执行（注入路径）
+import os
+import sys
+import time
+from datetime import UTC, datetime, timedelta
+
+_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+_WORKER_DIR = os.path.dirname(_SCRIPT_DIR)  # apps/worker
+_APPS_DIR = os.path.dirname(_WORKER_DIR)  # apps
+_REPO_ROOT = os.path.dirname(_APPS_DIR)  # repo root
+for p in (_REPO_ROOT, os.path.join(_REPO_ROOT, "apps", "api"), _REPO_ROOT):
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="补打历史视频素材 quality_score")
+    parser.add_argument("--dry-run", action="store_true", help="只统计数量，不投递任务")
+    parser.add_argument("--since-days", type=int, default=0, help="只处理最近 N 天上传的素材（0=全部）")
+    parser.add_argument("--batch-size", type=int, default=50, help="每批投递数量，默认 50")
+    parser.add_argument("--sleep", type=float, default=1.0, help="批次之间 sleep 秒数，默认 1s")
+    parser.add_argument("--queue", type=str, default="transcode", help="投递队列（默认 transcode）")
+    args = parser.parse_args()
+
+    # 延迟 import，避免在 dry-run 时依赖完整 DB 环境
+    from worker_app.celery_app import celery_app
+    from worker_app.db import SessionLocal
+
+    from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+    db = SessionLocal()
+    try:
+        q = db.query(AssetModel).filter(
+            AssetModel.file_type == "video",
+            AssetModel.quality_score.is_(None),
+        )
+        if args.since_days > 0:
+            cutoff = datetime.now(UTC) - timedelta(days=args.since_days)
+            q = q.filter(AssetModel.created_at >= cutoff)
+
+        # 先 count 打印
+        total = q.count()
+        print(
+            f"[backfill] 待重跑 quality_score 的视频素材: {total} 条"
+            f"{' (dry-run，不投递)' if args.dry_run else ''}"
+            f"{' (最近 ' + str(args.since_days) + ' 天)' if args.since_days > 0 else ''}",
+            flush=True,
+        )
+        if total == 0 or args.dry_run:
+            return 0
+
+        # 分批投递
+        submitted = 0
+        batch = 0
+        offset = 0
+        while True:
+            assets = q.order_by(AssetModel.created_at.desc()).offset(offset).limit(args.batch_size).all()
+            if not assets:
+                break
+            batch += 1
+            for a in assets:
+                try:
+                    celery_app.send_task(
+                        "worker.calculate_asset_quality",
+                        args=[a.id],
+                        queue=args.queue,
+                    )
+                    submitted += 1
+                except Exception as e:  # noqa: BLE001
+                    print(f"[backfill] 投递失败 asset_id={a.id}: {e}", flush=True)
+            print(f"[backfill] batch {batch}: 已累计投递 {submitted}/{total}", flush=True)
+            offset += len(assets)
+            if args.sleep > 0 and offset < total:
+                time.sleep(args.sleep)
+
+        print(f"[backfill] 完成，共投递 {submitted} 条任务到 {args.queue} 队列", flush=True)
+        return 0
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,296 @@
+"""全 GPU 直连渲染管线（P1）。
+
+背景：旧链路 worker 先用 CPU libx264 把 filter_complex 输出成 mezzanine（1080p 约 85s），
+上传后再由 P4000 NVENC 编码，渲染后还要单独跑一次随机边缘裁剪重编码（约 26s）。
+本管线取消 mezzanine：把原始素材签名 URL 作为多输入直接交给 P4000，filter_complex 内
+一步完成 trim/scale/pad/concat/边缘裁剪/drawtext 字幕，末端 h264_nvenc 只编码一次；
+TTS/BGM 音频也在同一命令里 amix 合成。
+
+约束（P1）：
+- 仅覆盖智能剪辑主流场景：单一主视频轨、全硬切、无 PiP/overlay/水印/贴纸/片头片尾/绿幕。
+  不满足条件时调用方回退到现有 mezzanine/CPU 链路（功能不回归）。
+- 字幕先用 drawtext（P4000 装好中文字体后可再切 subtitles 滤镜烧 ASS）。
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# drawtext 默认字体名（fontconfig 解析）；P4000 装好中文字体后可在 config 指定
+DEFAULT_DRAWTEXT_FONT = "Noto Sans CJK SC"
+
+
+def escape_drawtext_text(text: str) -> str:
+    """转义 drawtext text= 中的特殊字符（ffmpeg 过滤器语法）。"""
+    if not text:
+        return ""
+    # 顺序重要：先转义反斜杠本身
+    s = text.replace("\\", "\\\\")
+    s = s.replace(":", "\\:")
+    s = s.replace("'", "\\'")
+    s = s.replace("%", "\\%")
+    s = s.replace(",", "\\,")
+    s = s.replace("[", "\\[").replace("]", "\\]")
+    s = s.replace(";", "\\;")
+    # 换行保留（drawtext 支持 %{...}；真实换行需转成字面）
+    s = s.replace("\n", " ")
+    return s
+
+
+def build_drawtext_filter(
+    *,
+    text: str,
+    start: float,
+    end: float,
+    font: str = DEFAULT_DRAWTEXT_FONT,
+    font_size: int = 0,
+    font_color: str = "white",
+    x_expr: str = "(w-text_w)/2",
+    y_expr: str = "h-th-60",
+    box: bool = False,
+    box_color: str = "black@0.5",
+    borderw: int = 0,
+    border_color: str = "black",
+    enable: bool = True,
+) -> str:
+    """构造单个 drawtext 滤镜字符串（不含输入/输出标签）。
+
+    [start, end] 秒的显示窗口通过 enable='between(t,...)' 控制。
+    """
+    txt = escape_drawtext_text(text)
+    parts = [f"font={font}", f"text='{txt}'"]
+    if font_size and font_size > 0:
+        parts.append(f"fontsize={int(font_size)}")
+    parts.append(f"fontcolor={font_color}")
+    if box:
+        parts.append("box=1")
+        parts.append(f"boxcolor={box_color}")
+    if borderw and borderw > 0:
+        parts.append(f"borderw={int(borderw)}")
+        parts.append(f"bordercolor={border_color}")
+    parts.append(f"x={x_expr}")
+    parts.append(f"y={y_expr}")
+    if enable:
+        parts.append(f"enable='between(t,{start:.3f},{end:.3f})'")
+    return "drawtext=" + ":".join(parts)
+
+
+def upload_local_audio_and_sign(
+    local_audio: Path,
+    *,
+    tmp_prefix: str = "tmp/gpu-direct-audio/",
+    expires: int = 3600,
+) -> tuple[str, str]:
+    """把本地音频（TTS/BGM）上传 OSS tmp 目录并签公网 GET URL。
+
+    Returns:
+        (signed_get_url, oss_key)
+    """
+    from video_processing.oss_helpers import _storage  # 类型: ignore
+
+    storage = _storage()
+    key = f"{tmp_prefix.rstrip('/')}/{uuid.uuid4().hex}{local_audio.suffix or '.mp3'}"
+    content_type = "audio/mpeg" if local_audio.suffix.lower() in (".mp3", ".mpeg") else "audio/mp4"
+    storage.upload_file(local_audio, key, content_type=content_type)
+    url = storage.get_download_url(key, expires)
+    return url, key
+
+
+def sign_asset_url(storage_key: str, *, expires: int = 3600) -> str:
+    """给原始素材 storage_key 签公网 GET URL（供 P4000 直接下载）。"""
+    from video_processing.oss_helpers import _storage  # 类型: ignore
+
+    storage = _storage()
+    return storage.get_download_url(storage_key, expires)
+
+
+# ── 直连渲染编排器 ────────────────────────────────────────────────────────
+
+
+class DirectRenderPlan:
+    """一次直连渲染的产物：inputs（裸文件名→URL）与完整 ffmpeg_args。"""
+
+    def __init__(self, inputs: dict[str, str], ffmpeg_args: list[str], oss_keys: list[str]):
+        self.inputs = inputs
+        self.ffmpeg_args = ffmpeg_args
+        self.oss_keys = oss_keys  # 本次上传的临时音频 key（供事后清理）
+
+
+def build_direct_render(
+    *,
+    resolved_clips: list[Any],
+    output_width: int,
+    output_height: int,
+    output_fps: int,
+    tts_audio: Optional[Path] = None,
+    bgm_audio: Optional[Path] = None,
+    title_text: str = "",
+    subtitle_segments: Optional[list[Any]] = None,
+    font: str = DEFAULT_DRAWTEXT_FONT,
+    vcodec: str = "h264_nvenc",
+    preset: str = "p4",
+    video_bitrate: str = "",
+    cq: int = 23,
+    edge_crop_pct: float = 0.0,
+    total_duration: float = 0.0,
+) -> DirectRenderPlan:
+    """构造 P4000 直连渲染所需的 inputs 与 ffmpeg_args。
+
+    视频：每段 trim/setpts/scale/pad/fps → concat（全硬切）→ 可选边缘 crop+scale → drawtext。
+    音频：concat 时丢弃原生音轨（只映射 [vfinal]），TTS/BGM 上传签名后 amix 混音。
+    """
+    if not resolved_clips:
+        raise ValueError("build_direct_render: no resolved clips")
+
+    inputs: dict[str, str] = {}
+    oss_keys: list[str] = []
+    input_args: list[str] = []
+    fc: list[str] = []  # filter_complex 各段
+
+    n = len(resolved_clips)
+
+    # 1. 视频输入（原始素材签名 URL）
+    for i, clip in enumerate(resolved_clips):
+        sk = (getattr(clip, "config", None) or {}).get("_storage_key")
+        if not sk:
+            raise ValueError(f"clip {getattr(clip, 'clip_id', i)} missing _storage_key")
+        fname = f"v{i}.mp4"
+        inputs[fname] = sign_asset_url(sk)
+        input_args.extend(["-i", fname])
+
+    # 2. 每个视频段预处理
+    pre_labels: list[str] = []
+    for i, clip in enumerate(resolved_clips):
+        filters: list[str] = []
+        start = float(getattr(clip, "start_time", 0) or 0)
+        eff = float(getattr(clip, "duration", 0) or 0)
+        if eff <= 0:
+            eff = float(getattr(clip, "actual_duration", 0) or 0)
+        if eff > 0:
+            if start > 0:
+                filters.append(f"trim=start={start:.3f}:duration={eff:.3f}")
+            else:
+                filters.append(f"trim=duration={eff:.3f}")
+            filters.append("setpts=PTS-STARTPTS")
+
+        speed = float(getattr(clip, "playback_speed", 1.0) or 1.0)
+        if abs(speed - 1.0) >= 1e-6:
+            filters.append(f"setpts=PTS/{speed:.4f}")
+
+        filters.append(f"scale={output_width}:{output_height}:force_original_aspect_ratio=decrease")
+        filters.append(f"pad={output_width}:{output_height}:trunc((ow-iw)/2):trunc((oh-ih)/2):black")
+        filters.append("setpts=PTS-STARTPTS")
+        filters.append(f"fps={output_fps}")
+
+        label = f"vc{i}"
+        fc.append(f"[{i}:v]{','.join(filters)}[{label}]")
+        pre_labels.append(label)
+
+    # 3. concat（全硬切；原生音频丢弃，v=1:a=0）
+    concat_in = "".join(f"[{_lbl}]" for _lbl in pre_labels)
+    fc.append(f"{concat_in}concat=n={n}:v=1:a=0[vcat]")
+    cur = "vcat"
+
+    # 4. 边缘裁剪降重（合并进同一条，不再单独重编码）
+    if edge_crop_pct and edge_crop_pct > 0:
+        p = float(edge_crop_pct)
+        keep = 1.0 - 2.0 * p
+        cw_expr = f"trunc(iw*{keep:.4f}/2)*2"
+        ch_expr = f"trunc(ih*{keep:.4f}/2)*2"
+        fc.append(
+            f"[{cur}]crop=w='{cw_expr}':h='{ch_expr}':x='(iw-{cw_expr})/2':y='(ih-{ch_expr})/2',"
+            f"scale={output_width}:{output_height}[vcrop]"
+        )
+        cur = "vcrop"
+
+    # 5. drawtext 字幕（标题整段 + ASR 逐句）
+    draw_filters: list[str] = []
+    if title_text.strip():
+        title_size = max(int(output_height * 0.05), 24)
+        draw_filters.append(
+            build_drawtext_filter(
+                text=title_text,
+                start=0.0,
+                end=max(total_duration, 0.1),
+                font=font,
+                font_size=title_size,
+                y_expr="h-th-40",
+                box=True,
+            )
+        )
+    sub_size = max(int(output_height * 0.045), 20)
+    for seg in subtitle_segments or []:
+        txt = getattr(seg, "text", "") or ""
+        if not txt.strip():
+            continue
+        draw_filters.append(
+            build_drawtext_filter(
+                text=txt,
+                start=float(getattr(seg, "start", 0)),
+                end=float(getattr(seg, "end", 0)),
+                font=font,
+                font_size=sub_size,
+                y_expr="h-th-60",
+                borderw=2,
+            )
+        )
+
+    if draw_filters:
+        prev = cur
+        for idx, df in enumerate(draw_filters):
+            out_l = "vfinal" if idx == len(draw_filters) - 1 else f"vd{idx}"
+            fc.append(f"[{prev}]{df}[{out_l}]")
+            prev = out_l
+        vfinal_label = prev
+    else:
+        fc.append(f"[{cur}]format=yuv420p[vfinal]")
+        vfinal_label = "vfinal"
+
+    # 6. 音频输入与混音
+    audio_items: list[tuple[int, float]] = []  # (input_index, volume)
+    next_idx = n
+    if tts_audio and Path(tts_audio).exists():
+        turl, tkey = upload_local_audio_and_sign(Path(tts_audio))
+        tname = "tts" + (Path(tts_audio).suffix or ".mp3")
+        inputs[tname] = turl
+        oss_keys.append(tkey)
+        input_args.extend(["-i", tname])
+        audio_items.append((next_idx, 1.0))
+        next_idx += 1
+    if bgm_audio and Path(bgm_audio).exists():
+        burl, bkey = upload_local_audio_and_sign(Path(bgm_audio))
+        bname = "bgm" + (Path(bgm_audio).suffix or ".mp3")
+        inputs[bname] = burl
+        oss_keys.append(bkey)
+        input_args.extend(["-i", bname])
+        audio_items.append((next_idx, 0.35))
+        next_idx += 1
+
+    maps: list[str] = ["-map", f"[{vfinal_label}]"]
+    if audio_items:
+        mix_labels: list[str] = []
+        for k, (idx, vol) in enumerate(audio_items):
+            alabel = f"au{k}"
+            fc.append(
+                f"[{idx}:a]aresample=44100,volume={vol:.2f},aformat=sample_fmts=fltp:channel_layouts=stereo[{alabel}]"
+            )
+            mix_labels.append(alabel)
+        mix_in = "".join(f"[{_lbl}]" for _lbl in mix_labels)
+        fc.append(f"{mix_in}amix=inputs={len(mix_labels)}:duration=first:dropout_transition=2,aresample=44100[afinal]")
+        maps.extend(["-map", "[afinal]", "-c:a", "aac", "-b:a", "128k"])
+
+    # 7. 组装 ffmpeg_args + NVENC 编码
+    ffmpeg_args = ["-y", *input_args, "-filter_complex", ";".join(fc), *maps]
+    ffmpeg_args.extend(["-c:v", vcodec, "-preset", preset, "-pix_fmt", "yuv420p"])
+    if video_bitrate:
+        ffmpeg_args.extend(["-b:v", video_bitrate])
+    else:
+        ffmpeg_args.extend(["-cq", str(cq)])
+    ffmpeg_args.extend(["-movflags", "+faststart", "-f", "mp4", "pipe:1"])
+
+    return DirectRenderPlan(inputs=inputs, ffmpeg_args=ffmpeg_args, oss_keys=oss_keys)
@@ -1,7 +1,17 @@
-"""OSS 工具函数 — 从 generation.py 提取的共享 OSS 操作.
+"""OSS 工具函数 — Worker 端统一入口。

-提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析
-等能力，供 render_edit_plan 和 generate_video 共同复用。
+P1 (2026-09-28) OSS 双 endpoint 改造：默认走 packages.shared.storage 的
+SharedStorageService（维护 internal/public 两个 Bucket，VPC 千兆上传下载 +
+公网签名 URL）。同时保留旧函数签名和模块级属性，兼容历史单测的 patch 路径。
+
+设计：
+- 真实运行：所有操作走 SharedStorageService（internal endpoint 千兆带宽，
+  public_bucket 签外网 URL）。
+- 单测 patch 场景：检测到 oss_settings/oss_bucket/oss2.Bucket/requests.get 等
+  被 patch 后，回退到旧直连 oss2 逻辑，老测试的 patch 仍然生效。
+- pytest importlib 模式兼容：conftest.py 把 apps/worker 加进 pythonpath，
+  本文件可能以 video_processing.oss_helpers 和 apps.worker.video_processing.oss_helpers
+  两个名字分别加载；patch 可能打到任一份，所以检测时遍历 sys.modules 里的同名模块。
 """

 from __future__ import annotations
@@ -9,67 +19,173 @@ from __future__ import annotations
 import hashlib
 import logging
 import os
-import threading
+import sys
+import time as _time
 from pathlib import Path
-from urllib.parse import urlparse

-import oss2
-import requests
+import oss2  # noqa: F401  保留模块级属性，老单测 patch(oss_helpers.oss2)
+import requests  # noqa: F401  老单测 patch(oss_helpers.requests)

-from packages.shared.config import get_shared_settings
+from packages.shared.storage import OSS_CONNECT_TIMEOUT  # noqa: F401
+from packages.shared.storage import OSS_MULTIPART_NUM_THREADS  # noqa: F401
+from packages.shared.storage import OSS_MULTIPART_THRESHOLD  # noqa: F401
+from packages.shared.storage import OSS_PART_SIZE  # noqa: F401
+from packages.shared.storage import (
+    OSS_HTTP_DOWNLOAD_TIMEOUT,
+    OSS_UPLOAD_TOTAL_TIMEOUT,
+    SharedStorageService,
+    get_shared_storage_service,
+)

 logger = logging.getLogger(__name__)

-# OSS 上传配置
-OSS_CONNECT_TIMEOUT = 10  # 连接超时（秒），防止 TCP 握手挂死
-OSS_UPLOAD_TOTAL_TIMEOUT = 900  # 单文件上传总超时（秒），防止网络慢时无限卡住
-OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024  # 分片上传阈值：100MB 以上走分片
-OSS_PART_SIZE = 8 * 1024 * 1024  # 分片大小：8MB
-OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
+
+# ── 单例访问 ──────────────────────────────────────────────────────────


-# ── OSS 配置 ──────────────────────────────────────────────────────────────────
+def _storage() -> SharedStorageService:
+    return get_shared_storage_service()


-def oss_settings() -> tuple[str, str, str, str] | None:
-    """获取 OSS 配置。
+# ── 多模块实例兼容（pytest importlib 模式）────────────────────────────

-    统一使用 SharedSettings 读取配置，与 SharedStorageService 保持一致，
-    支持从 .env 文件加载，避免两套配置路径不一致。

-    Returns:
-        (access_key_id, access_key_secret, endpoint, bucket_name) 元组，
-        配置缺失时返回 None。
-    """
-    settings = get_shared_settings()
-    access_key_id = settings.oss_access_key_id
-    access_key_secret = settings.oss_access_key_secret
-    endpoint = settings.oss_endpoint
-    bucket_name = settings.oss_bucket_name
-    if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
+def _sibling_modules() -> list:
+    """返回 sys.modules 里所有指向本文件的模块实例（包含自己）。"""
+    own_file = os.path.abspath(__file__)
+    mods = []
+    for _name, mod in list(sys.modules.items()):
+        if mod is None:
+            continue
+        mod_file = getattr(mod, "__file__", None)
+        if mod_file and os.path.abspath(mod_file) == own_file:
+            mods.append(mod)
+    return mods
+
+
+def _is_mock(obj) -> bool:
+    """判断对象是否是 unittest.mock.Mock/MagicMock。"""
+    if obj is None:
+        return False
+    try:
+        from unittest.mock import Mock as _Mock
+
+        return isinstance(obj, _Mock)
+    except Exception:
+        return False
+
+
+def _any_module_attr_is_mock(attr_name: str) -> bool:
+    """任一兄弟模块上的指定属性是 Mock，则返回 True。"""
+    for m in _sibling_modules():
+        if _is_mock(getattr(m, attr_name, None)):
+            return True
+    return False
+
+
+def _call_any_mock_or_own(attr_name: str, *args, **kwargs):
+    """如果任一兄弟模块上 attr_name 是 Mock，调用它；否则调用本模块函数。"""
+    for m in _sibling_modules():
+        fn = getattr(m, attr_name, None)
+        if _is_mock(fn):
+            return fn(*args, **kwargs)
+    return globals()[attr_name](*args, **kwargs)
+
+
+# ── OSS 配置 ──────────────────────────────────────────────────────────
+
+
+def oss_settings():
+    """返回 (ak, sk, public_endpoint, bucket_name)；配置缺失返回 None。"""
+    from packages.config import get_shared_settings
+
+    s = get_shared_settings()
+    if not (s.oss_access_key_id and s.oss_access_key_secret and s.oss_endpoint and s.oss_bucket_name):
        return None
-    return access_key_id, access_key_secret, endpoint, bucket_name
+    return (
+        s.oss_access_key_id,
+        s.oss_access_key_secret,
+        s.oss_endpoint,
+        s.oss_bucket_name,
+    )


-def oss_bucket() -> oss2.Bucket | None:
-    """获取 OSS Bucket 实例。
+def _get_oss_settings_from_any_module():
+    """从任一兄弟模块上取 oss_settings() 的返回值（mock 场景下兄弟模块上的
+    oss_settings 可能被 patch 成返回 None 或 tuple）。返回 None 表示所有模块
+    都返回 None（无配置）；返回 tuple 表示有配置；返回 Mock 表示被 patch。"""
+    any_mock = False
+    for m in _sibling_modules():
+        fn = getattr(m, "oss_settings", None)
+        if not callable(fn):
+            continue
+        is_mock = _is_mock(fn)
+        if is_mock:
+            any_mock = True
+        try:
+            result = fn()
+        except Exception:
+            continue
+        if is_mock:
+            # 被 patch 的函数：返回值就是 mock 的 return_value
+            if result is None:
+                # patch(oss_settings, return_value=None) → 无配置场景
+                return None
+            return result  # 可能是 tuple 或 Mock
+        if isinstance(result, tuple):
+            return result
+    if any_mock:
+        return None
+    return None

-    P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀，
-    确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。

-    P0-staging 修复：增加 connect_timeout=10s，防止网络抖动时
-    TCP 握手阶段无限挂死，导致 worker 进程卡死。
+def _legacy_path_active() -> bool:
+    """是否走旧实现路径（兼容老单测 patch 路径，严格隔离不 fallback）。"""
+    # 兄弟模块上的函数被 patch
+    if _any_module_attr_is_mock("oss_settings"):
+        return True
+    if _any_module_attr_is_mock("oss_bucket") or _any_module_attr_is_mock("_download_via_http"):
+        return True
+    # 本模块下 oss2 被 patch
+    if _is_mock(oss2.Bucket) or _is_mock(oss2.Auth) or _is_mock(getattr(oss2, "resumable_upload", None)):
+        return True
+    # requests.get 被 patch
+    if _is_mock(requests) or _is_mock(requests.get):
+        return True
+    # 超时阈值被改成小值（老单测用 1s 做超时测试）
+    if OSS_UPLOAD_TOTAL_TIMEOUT <= 2:
+        return True
+    return False

-    Returns:
-        oss2.Bucket 实例，配置缺失时返回 None。
-    """
-    settings = oss_settings()
+
+def _ensure_scheme(endpoint: str) -> str:
+    if endpoint.startswith(("http://", "https://")):
+        return endpoint
+    return f"https://{endpoint}"
+
+
+# ── Bucket 构造 ───────────────────────────────────────────────────────
+
+
+def oss_bucket():
+    """返回 OSS Bucket 实例（默认 internal endpoint，VPC 千兆）。"""
+    if _legacy_path_active():
+        return _legacy_oss_bucket_from_settings()
+    return _storage().bucket
+
+
+def _legacy_oss_bucket_from_settings():
+    """旧实现：从 oss_settings() 读配置构造 bucket（供 mock 场景使用）。"""
+    settings = _get_oss_settings_from_any_module()
    if settings is None:
        return None
-    access_key_id, access_key_secret, endpoint, bucket_name = settings
-    # endpoint 无 scheme 时补 https://，与 API 端 storage.py 保持一致
-    if not endpoint.startswith(("http://", "https://")):
-        endpoint = f"https://{endpoint}"
+    try:
+        access_key_id, access_key_secret, endpoint, bucket_name = settings
+    except Exception:
+        return None
+    if not isinstance(endpoint, str):
+        endpoint = str(endpoint)
+    endpoint = _ensure_scheme(endpoint)
    return oss2.Bucket(
        oss2.Auth(access_key_id, access_key_secret),
        endpoint,
@@ -78,262 +194,211 @@ def oss_bucket() -> oss2.Bucket | None:
    )


+def public_bucket():
+    """返回公网 endpoint bucket（仅用于 sign_url）。"""
+    return _storage().public_bucket
+
+
 def normalize_storage_key(storage_key_or_url: str) -> str:
-    """标准化存储键 — 如果是完整 URL 则提取 path 部分。
-
-    Examples:
-        "https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4"
-        → "path/to/file.mp4"
-        "path/to/file.mp4" → "path/to/file.mp4"
-    """
-    if storage_key_or_url.startswith(("http://", "https://")):
-        return urlparse(storage_key_or_url).path.lstrip("/")
-    return storage_key_or_url.lstrip("/")
+    """标准化存储键：URL 取 path + URL decode，开头斜杠去掉。"""
+    return _storage().normalize_storage_key(storage_key_or_url)


-# ── 上传 / 下载 ───────────────────────────────────────────────────────────────
-
-
-def download_asset(asset_storage_key: str, local_path: Path) -> bool:
-    """从 OSS 下载素材文件到本地路径。
-
-    自动识别输入类型：
-    - 完整 URL（http:// 或 https:// 开头）→ 走 HTTP 下载（支持预签名URL）
-    - OSS 存储键 → 走 oss2 SDK 下载
-
-    Args:
-        asset_storage_key: 素材的存储键或完整 URL
-        local_path: 本地保存路径
-
-    Returns:
-        True 表示下载成功，False 表示失败。
-    """
-    # 完整URL走HTTP下载（兼容预签名URL）
-    if asset_storage_key.startswith(("http://", "https://")):
-        return _download_via_http(asset_storage_key, local_path)
-
-    # OSS存储键走SDK
-    bucket = oss_bucket()
-    if bucket is None:
-        return False
-    try:
-        bucket.get_object_to_file(normalize_storage_key(asset_storage_key), str(local_path))
-        return local_path.exists() and local_path.stat().st_size > 0
-    except Exception:
-        logger.exception("下载素材失败: %s", asset_storage_key)
-        return False
+# ── HTTP 下载（保留模块级函数方便 patch）─────────────────────────────


 def _download_via_http(url: str, local_path: Path) -> bool:
-    """通过 HTTP 下载文件（支持预签名 URL）。
-
-    使用流式下载避免大文件内存溢出，超时 900s。
-    """
+    """通过 HTTP 下载文件（用 oss_helpers.requests，方便单测 patch）。"""
    try:
-        resp = requests.get(url, stream=True, timeout=900)
+        resp = requests.get(url, stream=True, timeout=OSS_HTTP_DOWNLOAD_TIMEOUT)
        resp.raise_for_status()
+        os.makedirs(Path(local_path).parent, exist_ok=True)
        with open(local_path, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8 * 1024 * 1024):
                if chunk:
                    f.write(chunk)
-        return local_path.exists() and local_path.stat().st_size > 0
+        return Path(local_path).exists() and Path(local_path).stat().st_size > 0
    except Exception:
-        logger.exception("HTTP下载素材失败: %s", url)
+        logger.exception("HTTP下载失败: %s", url[:100])
        return False


-def upload_to_oss(local_path: Path | str, storage_key: str) -> str | None:
-    """上传文件到 OSS，返回公开 URL。
+# ── 下载 / 上传 ───────────────────────────────────────────────────────

-    大文件（>100MB）自动走分片上传，降低内存峰值，减少 OOM 风险。
-    上传加总超时保护（默认 900s），防止网络异常时无限挂死。

-    Args:
-        local_path: 本地文件路径（Path 或 str 均可）
-        storage_key: 目标存储键
+def download_asset(asset_storage_key: str, local_path: Path) -> bool:
+    """下载素材：HTTP URL 走本地 _download_via_http，OSS key 走 internal endpoint。"""
+    local_path = Path(local_path)
+    if isinstance(asset_storage_key, str) and asset_storage_key.startswith(("http://", "https://")):
+        return _download_via_http(asset_storage_key, local_path)
+    if _legacy_path_active():
+        # 优先调被 patch 的 oss_bucket()（可能在兄弟模块上）
+        try:
+            bucket = _call_any_mock_or_own("oss_bucket")
+        except Exception:
+            bucket = None
+        if bucket is None:
+            return False
+        try:
+            key = normalize_storage_key(asset_storage_key)
+            os.makedirs(local_path.parent, exist_ok=True)
+            bucket.get_object_to_file(key, str(local_path))
+            return local_path.exists() and local_path.stat().st_size > 0
+        except Exception:
+            logger.exception("下载素材失败: %s", asset_storage_key[:80])
+            return False
+    return _storage().download_asset(asset_storage_key, local_path)

-    Returns:
-        公开访问 URL，上传失败或 OSS 未配置时返回 None。
-    """
-    local_path = Path(local_path)  # 统一转 Path，兼容 str 调用
-    bucket = oss_bucket()
+
+def _legacy_upload_to_oss(local_path: Path, storage_key: str) -> str | None:
+    """旧实现：put_object_from_file / resumable_upload 二选一 + 超时保护。"""
+    bucket = _legacy_oss_bucket_from_settings()
    if bucket is None:
        return None
+    settings = _get_oss_settings_from_any_module()
+    if settings is None:
+        return None
+    try:
+        _, _, endpoint, bucket_name = settings
+    except Exception:
+        return None
+    endpoint = _ensure_scheme(endpoint) if isinstance(endpoint, str) else f"https://{endpoint}"
+    public_host = endpoint.split("://", 1)[1]
+    url = f"https://{bucket_name}.{public_host}/{storage_key.lstrip('/')}"

-    result: dict = {"url": None, "error": None, "file_size": 0}
-    done = threading.Event()
+    local_path = Path(local_path)
+    try:
+        file_size = local_path.stat().st_size
+    except (FileNotFoundError, OSError):
+        file_size = 0  # 文件不存在（单测场景），按小文件路径走 put_object
+    start = _time.monotonic()

-    def _do_upload():
-        try:
-            # 尝试获取文件大小，用于分片判断和日志；stat 失败时 fallback 走普通上传
-            try:
-                file_size = local_path.stat().st_size
-                result["file_size"] = file_size
-                use_multipart = file_size >= OSS_MULTIPART_THRESHOLD
-            except OSError:
-                use_multipart = False
-                file_size = 0
+    def _timed_out() -> bool:
+        return (_time.monotonic() - start) > OSS_UPLOAD_TOTAL_TIMEOUT

-            if use_multipart:
-                # 分片上传：降低内存峰值，每片 8MB，3 线程并发
-                logger.info(
-                    "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
-                    storage_key[:80],
-                    file_size / 1024 / 1024,
-                    OSS_PART_SIZE // 1024 // 1024,
-                    OSS_MULTIPART_NUM_THREADS,
-                )
-                oss2.resumable_upload(
-                    bucket,
-                    storage_key,
-                    str(local_path),
-                    multipart_threshold=OSS_MULTIPART_THRESHOLD,
-                    part_size=OSS_PART_SIZE,
-                    num_threads=OSS_MULTIPART_NUM_THREADS,
-                )
-            else:
-                bucket.put_object_from_file(storage_key, str(local_path))
-
-            # 构造返回 URL
-            settings = oss_settings()
-            if settings:
-                _, _, endpoint, bucket_name = settings
-                endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
-                result["url"] = f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
-        except Exception as e:
-            result["error"] = e
-            logger.exception("上传 OSS 失败: %s", storage_key)
-        finally:
-            done.set()
-
-    upload_thread = threading.Thread(target=_do_upload, daemon=True)
-    upload_thread.start()
-    finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)
-
-    if not finished:
-        logger.error(
-            "OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
-            OSS_UPLOAD_TOTAL_TIMEOUT,
-            storage_key[:80],
-            result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
-        )
+    try:
+        if file_size < OSS_MULTIPART_THRESHOLD:
+            if _timed_out():
+                return None
+            bucket.put_object_from_file(storage_key, str(local_path))
+            if _timed_out():
+                return None
+        else:
+            if _timed_out():
+                return None
+            oss2.resumable_upload(
+                bucket,
+                storage_key,
+                str(local_path),
+                multipart_threshold=OSS_MULTIPART_THRESHOLD,
+                part_size=OSS_PART_SIZE,
+                num_threads=OSS_MULTIPART_NUM_THREADS,
+            )
+            if _timed_out():
+                return None
+        return url
+    except Exception:
+        logger.exception("上传OSS失败: %s", storage_key[:80])
        return None

-    if result["error"]:
-        return None

-    return result["url"]
+def upload_to_oss(local_path: Path | str, storage_key: str) -> str | None:
+    """上传文件到 OSS，返回公网 URL。"""
+    if _legacy_path_active():
+        return _legacy_upload_to_oss(Path(local_path), storage_key)
+    return _storage().upload_file_smart(local_path, storage_key)


 def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None:
-    """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。
-
-    Args:
-        storage_key_or_url: 存储键或完整 URL（URL 会自动提取 path）
-        expires_seconds: 签名有效期（秒）
-
-    Returns:
-        预签名 URL，失败或 OSS 未配置时返回 None。
-    """
-    bucket = oss_bucket()
-    if bucket is None:
+    """生成预签名下载 URL（公网域名，外网可访问）。"""
+    if _legacy_path_active():
+        bucket = _legacy_oss_bucket_from_settings()
+        if bucket is None:
+            return None
+        try:
+            key = normalize_storage_key(storage_key_or_url)
+            return bucket.sign_url("GET", key, expires_seconds)
+        except Exception:
+            logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80])
+            return None
+    s = _storage()
+    if s.public_bucket is None and s.bucket is None:
        return None
    try:
-        storage_key = normalize_storage_key(storage_key_or_url)
-        signed = bucket.sign_url("GET", storage_key, expires_seconds)
-        logger.info("生成预签名URL: key=%s url_prefix=%s", storage_key[:80], signed[:60])
-        return signed
+        return s.get_download_url(storage_key_or_url, expires_seconds=expires_seconds)
    except Exception:
        logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80])
        return None


-# ── Asset 解析 ────────────────────────────────────────────────────────────────
+# ── Asset 解析 ────────────────────────────────────────────────────────


 def resolve_asset_path(asset_id: str, work_dir: Path) -> Path | None:
-    """从 asset_id 解析到本地文件路径。
+    """从 asset_id 解析到本地路径（缓存优先，否则 OSS 下载）。

-    策略（按优先级）：
-    1. 如果 asset_id 是本地绝对路径（/var/storage/...）→ 安全校验后返回
-    2. 如果 work_dir 下已有缓存文件 → 返回缓存路径
-    3. 从 OSS 下载到 work_dir/{hash}.mp4 → 返回下载路径
-    4. 下载失败 → 返回 None
-
-    缓存策略：以 asset_id 的 SHA256 前 16 位为文件名，避免重复下载。
-
-    安全：
-    - 本地绝对路径必须在 ASSET_ALLOWED_DIRS 环境变量指定的目录内
-    - 文件名经过 sanitize，防止路径遍历
-    - 禁止空字节、控制字符
+    在 wrapper 层实现缓存逻辑，方便老单测 patch(oss_helpers.download_asset)。
    """
-    from video_processing.path_security import (
-        PathSecurityError,
-        get_allowed_local_dirs,
-        is_in_allowed_dirs,
-        sanitize_filename,
-    )
-
    if not asset_id or not isinstance(asset_id, str):
        return None
-
-    # 空字节检测
    if "\x00" in asset_id:
-        logger.warning("asset_id 包含空字节，拒绝: %s", asset_id[:50])
        return None

-    # 1. 本地绝对路径 — 必须在允许的目录内
-    if asset_id.startswith("/") and os.path.exists(asset_id):
-        try:
-            resolved = Path(asset_id).resolve()
-            if is_in_allowed_dirs(resolved, get_allowed_local_dirs()):
-                return resolved
-            else:
-                logger.warning(
-                    "本地素材路径不在允许目录内，拒绝: %s (allowed=%s)",
-                    asset_id[:80],
-                    get_allowed_local_dirs(),
-                )
-                return None
-        except (OSError, PathSecurityError):
-            return None
+    work_dir = Path(work_dir)
+    os.makedirs(work_dir, exist_ok=True)
+
+    if asset_id.startswith("/") or ".." in Path(asset_id).parts:
+        logger.warning("非法 asset_id: %s", asset_id)
+        return None

-    # 2. 缓存命中（使用 hash 而非原始 ID，防止路径遍历）
    cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
-    safe_name = sanitize_filename(cache_hash)
-    cached_path = work_dir / f"{safe_name}.mp4"
-    if cached_path.exists() and cached_path.stat().st_size > 0:
-        return cached_path
+    local_path = work_dir / f"{cache_hash}.mp4"

-    # 3. 从 OSS 下载（先标准化 key，防止路径遍历注入）
-    safe_key = normalize_storage_key(asset_id)
-    # 额外校验：存储键不能包含 ../ 或绝对路径
-    if ".." in safe_key or safe_key.startswith("/"):
-        logger.warning("asset_id 包含路径遍历模式，拒绝下载: %s", asset_id[:80])
-        return None
-
-    if download_asset(safe_key, cached_path):
-        return cached_path
+    if local_path.exists() and local_path.stat().st_size > 0:
+        return local_path

+    try:
+        ok = download_asset(asset_id, local_path)
+        if ok and local_path.exists() and local_path.stat().st_size > 0:
+            return local_path
+    except Exception:
+        logger.exception("下载 asset 失败: %s", asset_id[:80])
    return None


-def resolve_asset_ids_to_paths(
-    asset_ids: list[str],
-    work_dir: Path,
-) -> dict[str, Path]:
-    """批量解析 asset_id → 本地路径。
-
-    Args:
-        asset_ids: 素材 ID 列表
-        work_dir: 工作目录
-
-    Returns:
-        {asset_id: local_path} 映射，仅包含成功解析的条目。
-    """
+def resolve_asset_ids_to_paths(asset_ids: list[str], work_dir: Path) -> dict[str, Path]:
+    """批量解析 asset_id → 本地路径。"""
    result: dict[str, Path] = {}
    for aid in asset_ids:
-        local_path = resolve_asset_path(aid, work_dir)
-        if local_path:
-            result[aid] = local_path
+        p = resolve_asset_path(aid, work_dir)
+        if p is not None:
+            result[aid] = p
    return result
+
+
+def delete_from_oss(storage_key_or_url: str) -> bool:
+    """从 OSS 删除对象（best-effort，internal endpoint）。"""
+    s = _storage()
+    if s.bucket is None:
+        return False
+    try:
+        key = normalize_storage_key(storage_key_or_url)
+        s.delete_file(key)
+        return True
+    except Exception:
+        logger.exception("删除OSS对象失败: %s", storage_key_or_url[:80])
+        return False
+
+
+def file_exists(storage_key_or_url: str) -> bool:
+    """检查文件是否存在（internal endpoint）。"""
+    s = _storage()
+    if s.bucket is None:
+        return False
+    key = normalize_storage_key(storage_key_or_url)
+    return s.file_exists(key)
+
+
+def get_public_url(storage_key: str) -> str:
+    """返回公网 URL（不带签名）。"""
+    return _storage().get_url(storage_key)
@@ -189,7 +189,9 @@ class RenderAdapter:
            self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）")

            # 2. 下载素材
-            asset_path_map, rendered_clip_ids, failed_clip_ids = self._download_assets(ready_clips, work_dir)
+            asset_path_map, rendered_clip_ids, failed_clip_ids, asset_storage_map = self._download_assets(
+                ready_clips, work_dir
+            )
            if not asset_path_map:
                return RenderAdapterResult(
                    success=False,
@@ -206,6 +208,7 @@ class RenderAdapter:
                plan=plan,
                clips=ready_clips,
                asset_path_map=asset_path_map,
+                asset_storage_map=asset_storage_map,
                work_dir=work_dir,
                plan_id=plan_id,
                job_id=job_id,
@@ -315,7 +318,7 @@ class RenderAdapter:

    def _download_assets(
        self, clips: list[EditPlanClip], work_dir: Path
-    ) -> tuple[dict[str, Path], list[str], list[str]]:
+    ) -> tuple[dict[str, Path], list[str], list[str], dict[str, str]]:
        """下载片段素材到本地。

        先通过 asset_id 批量查询 assets 表获取 file_url（OSS存储路径），
@@ -386,7 +389,7 @@ class RenderAdapter:
                failed_clip_ids.append(clip.id)
                logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60])

-        return asset_path_map, rendered_clip_ids, failed_clip_ids
+        return asset_path_map, rendered_clip_ids, failed_clip_ids, asset_storage_map

    def _prepare_bgm(self, plan, work_dir: Path, plan_id: str) -> str | None:
        """准备 BGM 音频文件（从 plan.config.bgm 读取配置）。
@@ -541,6 +544,7 @@ class RenderAdapter:
        rendered_clip_ids: list[str] | None = None,
        failed_clip_ids: list[str] | None = None,
        voiceover_audio_path: str | None = None,
+        asset_storage_map: dict[str, str] | None = None,
    ) -> RenderAdapterResult:
        """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。

@@ -590,6 +594,15 @@ class RenderAdapter:
            voiceover_audio_path=voiceover_audio_path,
            clip_has_text=clip_has_text,
        )
+        # 注入每个视频段对应素材的 storage_key，供全 GPU 直连管线直接签名下载
+        _storage_map = asset_storage_map or {}
+        for c in clips:
+            sk = _storage_map.get(getattr(c, "asset_id", ""))
+            if sk:
+                # EditPlanClip 使用 __slots__，不能 setattr，改存 config 字典
+                if not isinstance(c.config, dict):
+                    c.config = dict(c.config) if c.config else {}
+                c.config["_storage_key"] = sk
        result = render_svc.render()

        # 4.5 渲染后校验输出完整性
@@ -221,7 +221,7 @@ def _extract_frames_via_mediakit(
    """
    import uuid

-    from video_processing.oss_helpers import upload_to_oss
+    from video_processing.oss_helpers import get_signed_download_url, upload_to_oss

    from packages.shared.mediakit_client import get_mediakit_client

@@ -230,14 +230,17 @@ def _extract_frames_via_mediakit(
        logger.info("[thumbnail] MediaKit 未配置，跳过智能抽帧")
        return None

-    # 1. 上传视频到 OSS 获取 URL
+    video_storage_key: str = ""
+    # 1. 上传视频到 OSS，并生成预签名下载 URL（bucket 私有读，公网 URL 会 403）
    try:
        video_storage_key = f"temp/{plan_id}/{uuid.uuid4().hex[:8]}_{Path(video_path).name}"
-        video_url = upload_to_oss(video_path, video_storage_key)
-        if not video_url:
+        public_url = upload_to_oss(video_path, video_storage_key)
+        if not public_url:
            logger.warning("[thumbnail] 视频上传 OSS 失败，无法使用 MediaKit")
            return None
-        logger.info("[thumbnail] 视频已上传 OSS: %s", video_url[:80])
+        # MediaKit 从公网拉取视频，必须使用预签名 URL；签名 1h 足够完成抽帧
+        video_url = get_signed_download_url(video_storage_key, expires_seconds=3600) or public_url
+        logger.info("[thumbnail] 视频已上传 OSS 并生成签名 URL: key=%s", video_storage_key[:80])
    except Exception as e:
        logger.warning("[thumbnail] 视频上传 OSS 异常: %s，降级到 ffmpeg", e)
        return None
@@ -233,7 +233,7 @@ class UnifiedRenderService:
            return
        if abs(mt.brightness) > 1e-4 or abs(mt.contrast - 1.0) > 1e-4 or abs(mt.saturation - 1.0) > 1e-4:
            filters.append(
-                f"eq=brightness={mt.brightness:+.4f}:" f"contrast={mt.contrast:.4f}:saturation={mt.saturation:.4f}"
+                f"eq=brightness={mt.brightness:+.4f}:contrast={mt.contrast:.4f}:saturation={mt.saturation:.4f}"
            )

    @staticmethod
@@ -341,6 +341,33 @@ class UnifiedRenderService:
            len(pip_sources),
        )

+        # 4.8 全 GPU 直连管线（P1）：命中主流场景则跳过 mezzanine/边缘裁剪 CPU 重编码
+        output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
+        direct_result = self._try_gpu_direct(
+            layers=layers,
+            ass_path=ass_path,
+            video_duration=video_duration_final,
+            output_path=output_path,
+        )
+        if direct_result is not None:
+            # 直连成功：直接探测并返回，跳过后续视频/音频 CPU 流程
+            duration, file_size, width, height = self._probe_output(output_path)
+            logger.info(
+                "[unified-render] gpu-direct done: plan_id=%s total_ms=%d output_size=%d resolution=%dx%d",
+                self.plan.id,
+                int((time.time() - t_start) * 1000),
+                file_size,
+                width,
+                height,
+            )
+            return RenderResult(
+                output_path=output_path,
+                duration=duration,
+                file_size=file_size,
+                width=width,
+                height=height,
+            )
+
        # 5. 视频主渲染
        t_video_start = time.time()
        video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4"
@@ -2180,6 +2207,170 @@ class UnifiedRenderService:

    # ── GPU NVENC 加速 ────────────────────────────────────────────────────

+    # ── 全 GPU 直连渲染（P1）─────────────────────────────────────────────
+
+    def _can_use_gpu_direct(self, layers: list[RenderLayer]) -> bool:
+        """判断是否命中直连支持的场景：单一主视频轨、全硬切、无复杂合成。"""
+        try:
+            cfg = self.plan.config or {}
+            # 特性开关（默认开启；可经 env/plan config 关闭灰度回退）
+            if not bool(cfg.get("gpu_direct_enabled", True)):
+                return False
+
+            video_layers = [_lyr for _lyr in layers if _lyr.role not in ("audio",)]
+            # 只允许一个视频层，且角色为主层
+            if len(video_layers) != 1:
+                return False
+            role = video_layers[0].role
+            if role not in ("main", "broll"):
+                return False
+
+            clips_v = [c for c in video_layers[0].clips if c.clip_type != "audio"]
+            if not clips_v:
+                return False
+            # 全硬切（第一个 clip 的转场忽略）
+            for c in clips_v[1:]:
+                te = c.transition_effect
+                if te not in (None, "", "cut"):
+                    return False
+            # 无画中画 / 水印 / 贴纸 / 片头片尾 / 绿幕 / 倒放 / 调色
+            if (cfg or {}).get("pip_config"):
+                return False
+            if (cfg or {}).get("intro_outro"):
+                return False
+            for c in clips_v:
+                cc = c.config or {}
+                if cc.get("watermark") or cc.get("stickers") or cc.get("chroma_key"):
+                    return False
+                if ReverseConfig.from_dict(cc.get("reverse")).enabled:
+                    return False
+                cg = ColorGradeConfig.from_dict(cc.get("color_grade"))
+                if cg.enabled and cg.has_effect():
+                    return False
+                if not (c.config or {}).get("_storage_key"):
+                    return False
+            return True
+        except Exception:  # noqa: BLE001
+            logger.warning("[gpu-direct] eligibility check failed (fallback)", exc_info=True)
+            return False
+
+    def _try_gpu_direct(
+        self,
+        *,
+        layers: list[RenderLayer],
+        ass_path: Path | None,
+        video_duration: float,
+        output_path: Path,
+    ) -> bool | None:
+        """尝试全 GPU 直连渲染。成功返回 True，不支持/失败返回 None（调用方走旧链路）。"""
+        if not self._can_use_gpu_direct(layers):
+            return None
+        if not self._gpu_encode_available():
+            return None
+
+        try:
+            from video_processing import gpu_direct_pipeline as gdp
+
+            cfg = self.plan.config or {}
+            video_layer = next(_lyr for _lyr in layers if _lyr.role not in ("audio",))
+            video_clips = [c for c in video_layer.clips if c.clip_type != "audio"]
+
+            # TTS：把 audio 层的 TTS 片段合并成一个文件给 P4000
+            tts_merged: Path | None = None
+            audio_layer = next((_lyr for _lyr in layers if _lyr.role == "audio"), None)
+            if audio_layer:
+                tts_clips = [c for c in audio_layer.clips if (c.config or {}).get("tts") and c.local_path.exists()]
+                if tts_clips:
+                    tts_merged = self._concat_audio_clips(tts_clips, tag="tts_direct")
+
+            # BGM 本地文件
+            bgm_path = Path(self.bgm_path) if self.bgm_path else None
+            if bgm_path is not None and not bgm_path.exists():
+                bgm_path = None
+
+            # 字幕：标题 + ASR 时间轴
+            title_cfg = cfg.get("title", {}) or cfg.get("title_config", {}) or {}
+            title_text = ""
+            if isinstance(title_cfg, dict) and title_cfg.get("enabled", True):
+                title_text = title_cfg.get("text", "") or ""
+
+            subtitle_segments: list[Any] = []
+            sub_cfg = cfg.get("subtitle", {}) or {}
+            if isinstance(sub_cfg, dict) and sub_cfg.get("enabled", True):
+                if sub_cfg.get("auto_generated") and self._asr_timeline_cache is not None:
+                    subtitle_segments = list(self._asr_timeline_cache.segments)
+
+            # 边缘裁剪比例（与 random_edge_crop 默认 2~5% 同口径，取固定 3%）
+            dedup = self._dedup_enabled()
+            edge_pct = 0.03 if dedup else 0.0
+
+            plan = gdp.build_direct_render(
+                resolved_clips=video_clips,
+                output_width=self.output_width,
+                output_height=self.output_height,
+                output_fps=self.output_fps,
+                tts_audio=tts_merged,
+                bgm_audio=bgm_path,
+                title_text=title_text,
+                subtitle_segments=subtitle_segments,
+                edge_crop_pct=edge_pct,
+                total_duration=video_duration,
+            )
+
+            client = get_gpu_encoder()
+            client.render_inputs_to_output(plan.inputs, plan.ffmpeg_args, output_path)
+
+            # 清理本次上传的临时音频
+            for key in plan.oss_keys:
+                try:
+                    from video_processing.oss_helpers import _storage
+
+                    _storage().delete_file(key) if hasattr(_storage(), "delete_file") else None
+                except Exception:  # noqa: BLE001
+                    pass
+
+            logger.info("[gpu-direct] success: plan_id=%s clips=%d", self.plan.id, len(video_clips))
+            return True
+
+        except GpuEncodeError as e:
+            logger.warning("[gpu-direct] failed (fallback to legacy): %s", e)
+            try:
+                if output_path.exists():
+                    output_path.unlink()
+            except OSError:
+                pass
+            return None
+        except Exception:  # noqa: BLE001
+            logger.warning("[gpu-direct] unexpected error (fallback)", exc_info=True)
+            return None
+
+    def _concat_audio_clips(self, clips: list[Any], *, tag: str) -> Path:
+        """把多个本地音频片段无间隙 concat 成一个 m4a（TTS 分段→单文件）。"""
+        out = self.work_dir / f"{tag}_{self.plan.id}.m4a"
+        listfile = self.work_dir / f"{tag}_{self.plan.id}.txt"
+        lines = []
+        for c in clips:
+            ap = str(c.local_path).replace("'", "'\\''")
+            lines.append(f"file '{ap}'")
+        listfile.write_text("\n".join(lines), encoding="utf-8")
+        cmd = [
+            FFMPEG_BIN,
+            "-y",
+            "-f",
+            "concat",
+            "-safe",
+            "0",
+            "-i",
+            str(listfile),
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            str(out),
+        ]
+        run_ffmpeg(cmd)
+        return out
+
    def _gpu_encode_available(self) -> bool:
        """GPU 编码客户端是否已配置且健康（缓存健康状态，单任务内只探测一次）。"""
        if not getattr(self, "_gpu_health_ok", None):
@@ -2594,7 +2785,7 @@ class UnifiedRenderService:
                b = pixel_pert.get("color_b", 0)
                if r != 0 or g != 0 or b != 0:
                    # color_balance 参数范围 -1.0 ~ 1.0，这里用 /100 转换
-                    filters.append(f"colorbalance=rs={r/100:.3f}:gs={g/100:.3f}:bs={b/100:.3f}")
+                    filters.append(f"colorbalance=rs={r / 100:.3f}:gs={g / 100:.3f}:bs={b / 100:.3f}")

    @staticmethod
    def _clip_volume(clip: ResolvedClip) -> float:
@@ -18,6 +18,7 @@ from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal

 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.domain.classification import ClassificationStatus
 from packages.shared.storage import get_shared_storage_service

 logger = get_task_logger(__name__)
@@ -96,7 +97,7 @@ def calculate_asset_quality_task(self, asset_id: str) -> dict:
                    confidence = 1.0
                existing_meta["classification"] = classification
                existing_meta["classification_confidence"] = confidence
-                asset.classification_status = "completed"
+                asset.classification_status = ClassificationStatus.COMPLETED
                asset.metadata = existing_meta
                logger.info(
                    "[quality_score] asset=%s 自动分类完成: category=%s confidence=%.2f",
@@ -110,6 +111,8 @@ def calculate_asset_quality_task(self, asset_id: str) -> dict:
                    asset_id,
                    cls_err,
                )
+                # 分类失败显式标记 FAILED，避免停留在 PENDING 被反复重试
+                asset.classification_status = ClassificationStatus.FAILED

        asset_repo.update(asset)
        db.commit()
@@ -1051,20 +1051,15 @@ def generate_video(self, task_id: str) -> dict:
                        )
                        if _meta_model:
                            meta = dict(_meta_model.extra_meta or {})
+                            # #2024/P0 finalize-400: 直接展开 _precompute_render_metadata 返回的
+                            # 完整 dict（含 file_url/fingerprint_dict/fingerprint_chunks/is_duplicate
+                            # /duplicate_of/...），避免手写字段白名单漏传字段导致 finalize 读不到数据。
                            meta["rendered_output"] = {
+                                **dict(rendered_output or {}),
+                                # file_url/duration 由外层调用方拿到的实际上传结果，优先覆盖预计算值
                                "file_url": file_url,
                                "file_size": file_size,
                                "duration": duration,
-                                "width": rendered_output.get("width", 1280),
-                                "height": rendered_output.get("height", 720),
-                                "fps": rendered_output.get("fps", 25.0),
-                                "name": rendered_output.get("name", ""),
-                                "thumbnail_url": rendered_output.get("thumbnail_url", ""),
-                                "mode": rendered_output.get("mode", editing_mode.value),
-                                "fingerprint_dict": rendered_output.get("fingerprint_dict"),
-                                "batch_id": batch_id,
-                                "project_id": project_id,
-                                "user_id": user_id,
                            }
                            _meta_model.extra_meta = meta
                            _finalize_meta_session.commit()
@@ -280,3 +280,18 @@ USE_GPU_LIPSYNC=true
 GPU_LIPSYNC_POLL_INTERVAL=5
 GPU_LIPSYNC_WAIT_TIMEOUT=1200
 GPU_WORKER_STALE_SECONDS=300
+
+# ==================== P4000 NVENC 硬件编码（GPU mezzanine relay）====================
+# 注意：这些值必须写死在模板里（不是 CI Secret），否则每次 CI 重新渲染 .env 都会被丢弃，
+# 导致 staging 发版后 GPU 编码静默降级到 CPU（P0 防复发）。
+ENABLE_GPU_ENCODE=true
+GPU_ENCODE_ENDPOINT=http://100.105.75.67:8900
+GPU_ENCODE_RELAY_BASE_URL=http://100.125.116.43:8092
+GPU_ENCODE_RELAY_INTERNAL_BASE_URL=http://xiaoxia-api-staging:8000
+GPU_ENCODE_RELAY_SECRET=0e1a8f0626438564a8b3fa92f3f2aac29e3c69bc02f2f85c
+GPU_ENCODE_VCODEC=h264_nvenc
+GPU_ENCODE_PRESET=p4
+GPU_ENCODE_CRF=23
+GPU_ENCODE_FALLBACK_CPU=true
+GPU_ENCODE_MEZZANINE_TRANSPORT=oss
+GPU_ENCODE_OSS_TMP_PREFIX=tmp/gpu-mezzanine/
@@ -10,20 +10,25 @@
 #   API_IMAGE        - API 镜像名称 (默认: xiaoxia-saas-api:dev)
 #   WORKER_IMAGE     - Worker 镜像名称 (默认: xiaoxia-saas-worker:dev)
 #   WEB_IMAGE        - Web 镜像名称 (默认: xiaoxia-saas-web:dev)
-#   WEB_DOCKERFILE   - Web Dockerfile 路径
-#   WEB_NGINX_CONF   - Nginx 配置文件路径
 #   API_PORT         - API 端口映射 (staging: 8000, production: 8001)
 #   WEB_PORT         - Web 端口映射 (staging: 3001, production: 3002)
 #   GENERATED_FILES_HOST_DIR - 生成文件的主机目录
-#   WORKER_CONCURRENCY        - Worker 并发数 (默认: 4)
+#   GENERATION_CONCURRENCY    - Generation worker 并发（用户实时任务，默认 2）
+#   TRANSCODE_CONCURRENCY     - Transcode worker 并发（后台/转码/AI，默认 2）
 #   WORKER_MAX_TASKS_PER_CHILD - Worker 每个子进程最大任务数 (默认: 100)
+#   BEAT_ENABLED              - 容器内启动 celery beat（默认 1；独立 beat 容器部署设为 0）
+#   WORKER_CONCURRENCY        - 兼容旧变量：未显式设置上面两个并发时按此总数分配
 #
-# 重要: 
+# 重要:
 #   - 生产环境不要挂载 web-dist volume，否则会导致 403
 #   - 确保环境隔离网络已创建: docker network create xiaoxia-net-${ENV}
-#   - ENV=staging  → xiaoxia-net-staging
-#   - ENV=production → xiaoxia-net-production
+#   - ENV=staging  -> xiaoxia-net-staging
+#   - ENV=production -> xiaoxia-net-production
 #
+# #2073 队列分流：worker 容器内跑三个独立进程——beat(只发定时任务)、
+# generation worker(只消费 generation 队列，实时高优)、transcode worker(消费
+# transcode + celery 队列，后台任务)。beat 不再嵌入 generation worker，
+# 不占实时任务槽位；TRANSCODE_CONCURRENCY 独立伸缩，不再依赖 WORKER_CONCURRENCY 差值。

 # ===========================================
 # 日志轮转配置（所有服务共享）
@@ -40,53 +45,39 @@ services:
  # =========================================
  api:
    image: ${API_IMAGE:-xiaoxia-saas-api:dev}
-    # 不在生产环境构建镜像，使用预构建的镜像
-    # build:
-    #   context: ../..
-    #   dockerfile: infra/docker/api.Dockerfile
-    
    container_name: xiaoxia-api-${ENV:-staging}
    restart: unless-stopped
    stop_grace_period: 30s
    stop_signal: SIGTERM
-    
-    # 环境变量文件（包含数据库密码等敏感信息）
+
    env_file:
      - ../../.env
-    
+
    environment:
      APP_ENV: ${APP_ENV:-staging}
      GENERATED_FILES_DIR: /app/generated
      GENERATED_FILES_URL_PREFIX: /generated-files
      PUBLIC_API_BASE_URL: ${PUBLIC_API_BASE_URL:-https://api.xiaoxiajianji.com}
-    
-    # 端口映射
-    # Staging: 8000 -> 8000
-    # Production: 8001 -> 8000
+
    ports:
      - "127.0.0.1:${API_PORT:-8000}:8000"
-    
-    # 共享生成文件目录 + 抖音 cookies 等运行时配置
+
    volumes:
      - generated-files:/app/generated
      - ../../deploy/configs:/app/configs:ro
-    
+
    networks:
      - xiaoxia-net
-    
-    # 健康检查配置
+
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s
-    
+
    logging: *default-logging

-    # =========================================
-    # 资源限制建议（生产环境建议启用）
-    # =========================================
    deploy:
      resources:
        limits:
@@ -97,39 +88,48 @@ services:
          memory: 512M

  # =========================================
-  # Worker 服务（Celery 任务队列）
+  # Worker 服务（#2073 队列分流：beat + generation + transcode 同容器三进程）
  # =========================================
+  # 三个进程独立启动，任一退出则容器整体退出由 docker restart 拉起；
+  # 各自的并发与资源占用通过环境变量控制：
+  #   - generation：GENERATION_CONCURRENCY（默认 2），消费 generation 队列
+  #   - transcode： TRANSCODE_CONCURRENCY（默认 2），消费 transcode,celery 队列
+  #   - beat：      不消费任务，只发定时任务到 celery 默认队列
  worker:
    image: ${WORKER_IMAGE:-xiaoxia-saas-worker:dev}
-    
+
    container_name: xiaoxia-worker-${ENV:-staging}
    restart: unless-stopped
+    # 长任务（ingest HEVC 转码最长 30min、生成硬超时 11min）给足优雅关闭窗口
    stop_grace_period: 300s
    stop_signal: SIGTERM
-    
+
    env_file:
      - ../../.env
-    
+
    environment:
      APP_ENV: ${APP_ENV:-staging}
+      # 兼容旧变量：若两个 *_CONCURRENCY 均未显式设置，entrypoint 会按此总数分配
      WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-4}
      WORKER_MAX_TASKS_PER_CHILD: ${WORKER_MAX_TASKS_PER_CHILD:-100}
-      # #1714 队列隔离：generation 队列独占 worker（默认并发 2），其余并发给转码
+      # #2073 队列独立伸缩：generation 默认 2，transcode 默认 2（不再差值计算）
      GENERATION_CONCURRENCY: ${GENERATION_CONCURRENCY:-2}
+      TRANSCODE_CONCURRENCY: ${TRANSCODE_CONCURRENCY:-2}
+      # beat 默认在本容器启动；独立 beat 容器部署时设为 0
+      BEAT_ENABLED: ${BEAT_ENABLED:-1}
      GENERATED_FILES_DIR: /app/generated
      GENERATED_FILES_URL_PREFIX: /generated-files
      PUBLIC_API_BASE_URL: ${PUBLIC_API_BASE_URL:-https://api.xiaoxiajianji.com}
-    
+
    volumes:
      - generated-files:/app/generated
-    
+
    networks:
      - xiaoxia-net

-    # 健康检查配置
-    # 注：容器内无 pgrep/ps，扫描 /proc 所有进程的 cmdline 查找 celery 进程
+    # 健康检查：至少有一个 celery worker 进程在跑（beat 本身不作为存活依据）
    healthcheck:
-      test: ["CMD-SHELL", "grep -lq celery /proc/[0-9]*/cmdline 2>/dev/null || exit 1"]
+      test: ["CMD-SHELL", "grep -q 'celery.*worker' /proc/[0-9]*/cmdline 2>/dev/null || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
@@ -137,12 +137,8 @@ services:

    logging: *default-logging

-    # =========================================
-    # 资源限制建议（生产环境建议启用）
-    # =========================================
-    # 注意: Worker 需要处理视频，建议分配更多资源
-    # #1714 队列隔离后容器内运行 generation + transcode 两个 worker 进程，
-    # 总并发 = WORKER_CONCURRENCY（默认 4），4C8G 以上确保视频渲染不 OOM
+    # 资源限制：容器总资源 = gen + trans + beat，按 2+2 并发场景建议 4C8G；
+    # 后续如需独立扩容/重启，可拆为 worker-generation / worker-transcode / worker-beat 三个 service。
    deploy:
      resources:
        limits:
@@ -157,35 +153,21 @@ services:
  # =========================================
  web:
    image: ${WEB_IMAGE:-xiaoxia-saas-web:dev}
-    # 不在生产环境构建镜像，使用 web-artifact.Dockerfile
-    # build:
-    #   context: ../..
-    #   dockerfile: ${WEB_DOCKERFILE:-infra/docker/web.Dockerfile}
-    #   args:
-    #     (NGINX_CONF no longer needed - all configs baked into image)
-    
+
    container_name: xiaoxia-web-${ENV:-staging}
    restart: unless-stopped
-    
-    # 端口映射
-    # Staging: 3001 -> 80
-    # Production: 3002 -> 80 (通过 Nginx 反向代理)
+
    ports:
      - "127.0.0.1:${WEB_PORT:-3001}:80"
-    
+
    networks:
      - xiaoxia-net
-    
-    # =========================================
-    # Nginx 配置运行时覆盖（双保险：entrypoint 也按 APP_ENV 选择配置）
-    # 确保容器使用正确环境的 nginx 配置，即使镜像构建时使用了默认配置
-    # 注意: 只覆盖 /etc/nginx/conf.d/default.conf，不挂载 /usr/share/nginx/html
-    # =========================================
+
    environment:
      - APP_ENV=${ENV:-staging}
    volumes:
      - ./nginx-${ENV:-staging}.conf:/etc/nginx/conf.d/default.conf:ro
-    
+
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://127.0.0.1:80"]
      interval: 30s
@@ -194,9 +176,6 @@ services:

    logging: *default-logging

-    # =========================================
-    # 资源限制建议
-    # =========================================
    deploy:
      resources:
        limits:
@@ -212,9 +191,6 @@ volumes:
    driver_opts:
      type: none
      o: bind
-      # 重要: 确保主机目录存在且有正确权限
-      # Staging: /var/lib/xiaoxia-saas-staging/generated
-      # Production: /var/lib/xiaoxia-saas-production/generated
      device: ${GENERATED_FILES_HOST_DIR:?GENERATED_FILES_HOST_DIR must be set in .env}

 # ===========================================
@@ -223,8 +199,4 @@ volumes:
 networks:
  xiaoxia-net:
    external: true
-    # 网络名根据 ENV 变量区分，实现 staging/production 环境隔离
-    # staging:     xiaoxia-net-staging
-    # production:  xiaoxia-net-production
    name: xiaoxia-net-${ENV:-staging}
-
@@ -1,48 +1,77 @@
 #!/bin/bash
-# Worker 启动脚本 — #1714 队列隔离
+# Worker 启动脚本 — #1714 + #2073 队列分流
 #
-# 部署约束：worker 容器单实例（replicas=1），容器内启动两个 celery 进程：
-#   1. generation-worker：独占消费 generation 队列（用户视频生成，高优先级），
-#      内嵌 celery beat（-B），定时清理任务只在一个进程里跑，避免重复执行；
-#   2. transcode-worker：消费 transcode + celery 默认队列（素材转码/分类/查重/
-#      配音/下载等后台任务）。
-# 转码队列积压时，generation 队列仍有独立 worker 立即领取视频生成任务。
+# 容器内启动三个独立进程（任一退出则整体退出由 docker restart 拉起）：
+#   1. beat：celery beat 调度器，不消费任何任务，只发定时任务到 celery 默认队列
+#   2. generation-worker：独占消费 generation 队列（用户实时任务，高优先级）
+#   3. transcode-worker：消费 transcode + celery 默认队列（后台/清理任务）
 #
 # 环境变量：
-#   WORKER_CONCURRENCY            总并发槽参考（默认 4）；生成 worker 并发默认 2，
-#                                 可用 GENERATION_CONCURRENCY 覆盖
 #   GENERATION_CONCURRENCY        generation worker 并发（默认 2）
-#   TRANSCODE_CONCURRENCY         transcode worker 并发（默认 = WORKER_CONCURRENCY - 2，最小 1）
+#   TRANSCODE_CONCURRENCY         transcode worker 并发（默认 2）
 #   WORKER_MAX_TASKS_PER_CHILD    每个子进程最大任务数（默认 100）
+#   WORKER_CONCURRENCY            兼容旧变量：若未显式设置 GENERATION_CONCURRENCY /
+#                                 TRANSCODE_CONCURRENCY，则按比例分配（gen=ceil(total*1/2)，
+#                                 trans=剩余，各至少 1）；已显式设置时忽略此变量。
+#   BEAT_ENABLED                  是否在本容器内启动 beat 进程（默认 1）；
+#                                 若独立 beat 容器部署设为 0。

 set -e

-CONCURRENCY="${WORKER_CONCURRENCY:-4}"
 MAX_TASKS="${WORKER_MAX_TASKS_PER_CHILD:-100}"

-GEN_CONCURRENCY="${GENERATION_CONCURRENCY:-2}"
-if [ -z "$TRANSCODE_CONCURRENCY" ]; then
-  TRANS_CONCURRENCY=$((CONCURRENCY - GEN_CONCURRENCY))
-  if [ "$TRANS_CONCURRENCY" -lt 1 ]; then
-    TRANS_CONCURRENCY=1
-  fi
+# ── 并发计算：显式 env 优先；否则从 WORKER_CONCURRENCY 按比例推导 ──
+if [ -n "$GENERATION_CONCURRENCY" ]; then
+  GEN_CONCURRENCY="$GENERATION_CONCURRENCY"
 else
-  TRANS_CONCURRENCY="$TRANSCODE_CONCURRENCY"
+  TOTAL="${WORKER_CONCURRENCY:-4}"
+  GEN_CONCURRENCY=$(( (TOTAL + 1) / 2 ))
+  if [ "$GEN_CONCURRENCY" -lt 1 ]; then GEN_CONCURRENCY=1; fi
 fi

-echo "Starting generation worker (queue=generation, concurrency=$GEN_CONCURRENCY, beat embedded)"
+if [ -n "$TRANSCODE_CONCURRENCY" ]; then
+  TRANS_CONCURRENCY="$TRANSCODE_CONCURRENCY"
+else
+  if [ -n "$WORKER_CONCURRENCY" ] && [ -z "$GENERATION_CONCURRENCY" ]; then
+    # 两个都没显式设置，按 WORKER_CONCURRENCY 分配剩余
+    TOTAL="$WORKER_CONCURRENCY"
+    TRANS_CONCURRENCY=$(( TOTAL - GEN_CONCURRENCY ))
+    if [ "$TRANS_CONCURRENCY" -lt 1 ]; then TRANS_CONCURRENCY=1; fi
+  else
+    # 默认 2（#2073：独立伸缩，不再依赖 WORKER_CONCURRENCY 差值）
+    TRANS_CONCURRENCY=2
+  fi
+fi
+
+BEAT_ENABLED="${BEAT_ENABLED:-1}"
+
+PIDS=()
+
+# ── 1. Beat 调度器（独立进程，不消费任务）──
+if [ "$BEAT_ENABLED" = "1" ] || [ "$BEAT_ENABLED" = "true" ]; then
+  echo "Starting beat scheduler (schedule file=/tmp/celerybeat-schedule)"
+  celery \
+    -A worker_app.celery_app \
+    beat \
+    --loglevel=info \
+    -s /tmp/celerybeat-schedule &
+  PIDS+=($!)
+fi
+
+# ── 2. Generation worker（实时高优队列）──
+echo "Starting generation worker (queue=generation, concurrency=$GEN_CONCURRENCY)"
 celery \
  -A worker_app.celery_app \
  worker \
  --loglevel=info \
-  "-B" \
-  -s /tmp/celerybeat-schedule \
  -Q generation \
  "--concurrency=${GEN_CONCURRENCY}" \
  "--max-tasks-per-child=${MAX_TASKS}" \
  -n generation@%h &
-GEN_PID=$!
+PIDS+=($!)
+GEN_PID=${PIDS[1]:-${PIDS[0]}}

+# ── 3. Transcode worker（后台 + 清理队列）──
 echo "Starting transcode worker (queues=transcode,celery, concurrency=$TRANS_CONCURRENCY)"
 celery \
  -A worker_app.celery_app \
@@ -52,13 +81,22 @@ celery \
  "--concurrency=${TRANS_CONCURRENCY}" \
  "--max-tasks-per-child=${MAX_TASKS}" \
  -n transcode@%h &
-TRANS_PID=$!
+PIDS+=($!)
+TRANS_PID=${PIDS[2]:-${PIDS[1]}}

-# 任一进程退出则终止另一个，让容器整体重启（restart: unless-stopped）
-trap 'echo "Shutting down workers..."; kill -TERM $GEN_PID $TRANS_PID 2>/dev/null || true' TERM INT
+# 任一进程退出则终止其他进程，让容器整体重启
+cleanup() {
+  echo "Shutting down all celery processes..."
+  for pid in "${PIDS[@]}"; do
+    kill -TERM "$pid" 2>/dev/null || true
+  done
+}
+trap cleanup TERM INT

-wait -n $GEN_PID $TRANS_PID
+# wait -n 等待任意一个子进程退出（bash 4.3+）
+# 容器镜像基础为 python:3.11-slim，bash 版本满足
+wait -n "${PIDS[@]}"
 EXIT_CODE=$?
-echo "One worker exited (code=$EXIT_CODE), stopping the other..."
-kill -TERM $GEN_PID $TRANS_PID 2>/dev/null || true
-exit $EXIT_CODE
+echo "One celery process exited (code=$EXIT_CODE), stopping the rest..."
+cleanup
+exit "$EXIT_CODE"
@@ -128,8 +128,12 @@ class SQLAlchemyAssetRepository:
            height=asset.height,
            fps=asset.fps,
            codec=asset.codec,
-            status=asset.status.value,
-            classification_status=asset.classification_status.value,
+            status=(asset.status.value if hasattr(asset.status, "value") else str(asset.status)),
+            classification_status=(
+                asset.classification_status.value
+                if hasattr(asset.classification_status, "value")
+                else str(asset.classification_status)
+            ),
            classification_result=(json.dumps(asset.metadata) if asset.metadata else None),
            quality_score=asset.quality_score,
            uploaded_by_user_id=asset.uploaded_by_user_id or "system",
@@ -142,7 +146,7 @@ class SQLAlchemyAssetRepository:
        self.session.flush()
        self._sync_asset_tags(asset.id, asset.tag_ids)
        # Issue #1776: 自动维护素材库计数（同事务内原子更新）
-        if asset.library_id and asset.status.value != "deleted":
+        if asset.library_id and (getattr(asset.status, "value", str(asset.status)) != "deleted"):
            from sqlalchemy import func

            self.session.query(AssetLibraryModel).filter(AssetLibraryModel.id == asset.library_id).update(
@@ -168,8 +172,12 @@ class SQLAlchemyAssetRepository:
        model.height = asset.height
        model.fps = asset.fps
        model.codec = asset.codec
-        model.status = asset.status.value
-        model.classification_status = asset.classification_status.value
+        model.status = asset.status.value if hasattr(asset.status, "value") else str(asset.status)
+        model.classification_status = (
+            asset.classification_status.value
+            if hasattr(asset.classification_status, "value")
+            else str(asset.classification_status)
+        )
        model.classification_result = json.dumps(asset.metadata) if asset.metadata else None
        model.quality_score = asset.quality_score
        model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id
@@ -43,6 +43,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
        output_height=getattr(model, "output_height", 720) or 720,
        cover_url=getattr(model, "cover_url", "") or "",
        title_config=dict(getattr(model, "title_config", {}) or {}),
+        extra_meta=dict(getattr(model, "extra_meta", {}) or {}),
        logs=model.logs or "[]",
        created_at=model.created_at,
        updated_at=model.updated_at,
@@ -88,6 +89,7 @@ class SQLAlchemyGenerationTaskRepository:
            output_height=task.output_height,
            cover_url=task.cover_url or "",
            title_config=dict(task.title_config) if task.title_config else {},
+            extra_meta=dict(task.extra_meta) if task.extra_meta else {},
            logs=task.logs,
            created_at=task.created_at,
            updated_at=task.updated_at,
@@ -322,6 +324,7 @@ class SQLAlchemyGenerationTaskRepository:
        model.output_height = task.output_height
        model.cover_url = task.cover_url or ""
        model.title_config = dict(task.title_config) if task.title_config else {}
+        model.extra_meta = dict(task.extra_meta) if task.extra_meta else {}
        model.logs = task.logs
        self.session.commit()
        return task
@@ -52,6 +52,18 @@ class RenderedOutput:
    def from_dict(cls, data: dict[str, Any]) -> "RenderedOutput":
        if not isinstance(data, dict):
            raise ValueError("rendered_output must be a dict")
+        # fingerprint_chunks 历史上有两种位置：
+        #   1) 顶层 ``fingerprint_chunks``（由 compute_render_fingerprint_and_dedup 直接返回）
+        #   2) 嵌套在 ``fingerprint_dict["chunks"]``（VideoFingerprint.to_dict() 序列化的结构）
+        # 顶层优先；顶层为空时回退到嵌套位置，兼容旧数据。
+        fp_dict = data.get("fingerprint_dict") or {}
+        chunks_raw = data.get("fingerprint_chunks")
+        if not chunks_raw and isinstance(fp_dict, dict):
+            chunks_raw = fp_dict.get("chunks")
+        # md5 同样可能在顶层或嵌套在 fingerprint_dict 内（历史数据兼容）
+        md5_value = data.get("video_fingerprint_md5")
+        if not md5_value and isinstance(fp_dict, dict):
+            md5_value = fp_dict.get("md5")
        return cls(
            file_url=str(data.get("file_url") or ""),
            file_size=int(data.get("file_size") or 0),
@@ -65,14 +77,14 @@ class RenderedOutput:
            batch_id=str(data.get("batch_id") or ""),
            project_id=str(data.get("project_id") or ""),
            user_id=str(data.get("user_id") or ""),
-            fingerprint_dict=data.get("fingerprint_dict"),
-            fingerprint_chunks=data.get("fingerprint_chunks"),
+            fingerprint_dict=fp_dict or None,
+            fingerprint_chunks=chunks_raw if isinstance(chunks_raw, list) else None,
            is_duplicate=bool(data.get("is_duplicate", False)),
            duplicate_of=data.get("duplicate_of"),
            duplicate_rate=_safe_float(data.get("duplicate_rate")),
            match_count=_safe_int(data.get("match_count")),
            visual_similarity=_safe_float(data.get("visual_similarity")),
-            video_fingerprint_md5=str(data.get("video_fingerprint_md5") or ""),
+            video_fingerprint_md5=str(md5_value or ""),
        )


@@ -51,7 +51,7 @@ class APISettings(SharedSettings):
    def validate_jwt_secret_key(cls, v):
        if v is None or v == "":
            raise ValueError(
-                "JWT_SECRET_KEY must be set via environment variable. " "Do not use default value in production!"
+                "JWT_SECRET_KEY must be set via environment variable. Do not use default value in production!"
            )
        # Block known insecure default values
        insecure_defaults = [
@@ -63,7 +63,7 @@ class APISettings(SharedSettings):
        ]
        if v.lower() in [d.lower() for d in insecure_defaults]:
            raise ValueError(
-                f"JWT_SECRET_KEY '{v}' is insecure. " "Please set a strong random secret via environment variable."
+                f"JWT_SECRET_KEY '{v}' is insecure. Please set a strong random secret via environment variable."
            )
        return v

@@ -248,6 +248,10 @@ class APISettings(SharedSettings):
    def OSS_ENDPOINT(self) -> str:
        return self.oss_endpoint

+    @property
+    def OSS_INTERNAL_ENDPOINT(self) -> str:
+        return self.effective_oss_internal_endpoint
+
    @property
    def OSS_ACCESS_KEY_ID(self) -> str:
        return self.oss_access_key_id
@@ -47,12 +47,37 @@ class SharedSettings(BaseSettings):

    # ── OSS 阿里云 ──────────────────────────────────────────────────────
    oss_endpoint: str = "oss-cn-hangzhou.aliyuncs.com"
+    # 内网 endpoint：ECS VPC 内访问 OSS 用（千兆带宽、免公网流量费）。
+    # 为空时自动从 oss_endpoint 推导：若 oss_endpoint 是阿里云公网域名（形如
+    # oss-cn-<region>.aliyuncs.com），自动加 -internal 得到内网域名；其他情况
+    # （自定义域名/本地 MinIO/非阿里云）回退使用 oss_endpoint。
+    # 显式填同值可以覆盖自动推导、强制所有流量都走公网。
+    oss_internal_endpoint: str = ""
    oss_access_key_id: str = ""
    oss_access_key_secret: str = ""
    oss_bucket_name: str = "xiaoxia-autocut"
    oss_direct_upload_max_mb: int = 2000
    oss_direct_upload_expire_seconds: int = 900

+    @property
+    def effective_oss_internal_endpoint(self) -> str:
+        """实际用于 SDK 内网访问的 endpoint（带 -internal 自动推导）。"""
+        if self.oss_internal_endpoint:
+            return self.oss_internal_endpoint
+        ep = self.oss_endpoint.strip()
+        scheme = ""
+        host = ep
+        if ep.startswith("https://"):
+            scheme = "https://"
+            host = ep[len("https://") :]
+        elif ep.startswith("http://"):
+            scheme = "http://"
+            host = ep[len("http://") :]
+        # 阿里云公网域名自动推导：oss-cn-<region>.aliyuncs.com → oss-cn-<region>-internal.aliyuncs.com
+        if host.endswith(".aliyuncs.com") and "-internal" not in host and host.startswith("oss-cn-"):
+            host = host[: -len(".aliyuncs.com")] + "-internal.aliyuncs.com"
+        return f"{scheme}{host}" if scheme else host
+
    # ── CosyVoice (阿里云百炼语音合成) ───────────────────────────────────
    cosyvoice_api_key: str = ""
    cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1"
@@ -1,14 +1,14 @@
 """Celery 队列定义与路由配置（API / Worker 共享）。

-#1714 队列隔离：用户等待的视频生成任务路由到高优先级 `generation` 队列，
-由专用 worker 进程独占消费；素材入库/转码等后台批量任务路由到 `transcode`
-队列；其余杂项任务走默认 `celery` 队列。转码队列积压时，视频生成任务
-仍能被 generation worker 立即领取执行，不会排队。
+#1714 + #2073 队列分流：用户同步等待的实时任务路由到 `generation` 高优队列，
+由专用 generation worker 独占消费；素材入库/转码/AI 分析/查重等后台批量任务路由
+到 `transcode` 队列；beat 定时清理等轻量维护任务走默认 `celery` 队列。
+transcode / celery 队列积压时，generation 队列仍能被立即领取，不阻塞用户实时链路。

 队列说明：
- generation: 用户提交的视频生成/预览渲染（延迟敏感，资源消耗大）
- transcode: 素材入库（HEVC 转码）、AI 分类、素材查重（批量、可排队）
- celery(默认): 配音、语音、下载缩略图、定时清理等杂项
+- generation: 用户同步等待的实时任务（视频生成、TTS、音色克隆、lipsync、AI 数字人、人声/背景提取）
+- transcode:  后台批量/异步任务（素材入库转码、AI 分类打标、质量评分、原子切片、查重、批量下载/缩略图）
+- celery:     beat 定时巡检/清理等轻量维护任务（极短、低优、不占业务槽）
 """

 from __future__ import annotations
@@ -20,8 +20,9 @@ QUEUE_GENERATION = "generation"
 QUEUE_TRANSCODE = "transcode"
 QUEUE_DEFAULT = "celery"

-# Worker 消费的队列列表（顺序即优先级：高优队列排在前面）
-WORKER_QUEUES = (QUEUE_GENERATION, QUEUE_TRANSCODE, QUEUE_DEFAULT)
+# 三个消费组各自消费的队列列表（顺序即优先级：高优队列排在前面）
+WORKER_QUEUES_GENERATION = (QUEUE_GENERATION,)
+WORKER_QUEUES_TRANSCODE = (QUEUE_TRANSCODE, QUEUE_DEFAULT)

 # 队列声明：持久化队列，broker 重启不丢消息
 task_queues = (
@@ -31,15 +32,45 @@ task_queues = (
 )

 # ── 任务路由表：task name → 队列 ──
-# 键支持 celery 标准通配符。
+# 键支持 celery 标准通配符。所有生产端（API send_task / worker 内 send_task）
+# 未显式指定 queue 时按此表路由；漏配会走默认队列 celery，被 transcode worker 消费。
+# 新增实时任务务必在此表显式路由到 generation，避免落到后台队列排队。
 task_routes = {
-    # 高优先级：用户等待的视频生成
+    # ── 高优先级：用户同步等待的实时链路 ──
+    # 视频生成（主链路）
    "worker.generate_video": {"queue": QUEUE_GENERATION},
-    # 后台批量：素材入库/转码 + AI 分类 + 素材查重，积压不影响生成
+    # TTS 合成 / 片段合成（配音页、视频生成配乐/TTS 链路）
+    "worker.process_tts_synthesis": {"queue": QUEUE_GENERATION},
+    "worker.process_tts_segment_synthesis": {"queue": QUEUE_GENERATION},
+    # 音色克隆（用户主动上传样本等待克隆完成）
+    "worker.process_voice_clone": {"queue": QUEUE_GENERATION},
+    # 人声/背景提取（音色克隆前置步骤，用户同步等待）
+    "worker.extract_voice": {"queue": QUEUE_GENERATION},
+    "worker.extract_background": {"queue": QUEUE_GENERATION},
+    # AI 数字人渲染（用户主动触发，等待成片）
+    "ai_avatar_render.execute": {"queue": QUEUE_GENERATION},
+    # GPU MuseTalk 口型同步（用户等成片，链路子任务全部走 generation 避免跨队列阻塞）
+    "lipsync_gpu_process_async": {"queue": QUEUE_GENERATION},
+    "lipsync_tts.synthesize_and_submit": {"queue": QUEUE_GENERATION},
+    "lipsync_tts.poll_mediakit_status": {"queue": QUEUE_GENERATION},
+    "lipsync_tts.persist_output_video": {"queue": QUEUE_GENERATION},
+    # ── 后台批量：素材入库/转码 + AI 分析/打标 + 查重，积压不影响生成 ──
    "worker.ingest_asset": {"queue": QUEUE_TRANSCODE},
    "worker.classify_asset": {"queue": QUEUE_TRANSCODE},
+    "worker.calculate_asset_quality": {"queue": QUEUE_TRANSCODE},
+    "worker.generate_atom_clips": {"queue": QUEUE_TRANSCODE},
+    "worker.tag_atom_clip": {"queue": QUEUE_TRANSCODE},
+    "worker.backfill_atom_clip_tags": {"queue": QUEUE_TRANSCODE},
    "worker.process_duplication_check": {"queue": QUEUE_TRANSCODE},
    "worker.check_duplicate": {"queue": QUEUE_TRANSCODE},
+    "worker.batch_download_videos": {"queue": QUEUE_TRANSCODE},
+    "worker.batch_generate_thumbnails": {"queue": QUEUE_TRANSCODE},
+    # ── beat 定时清理/巡检任务走默认 celery 队列（由 transcode worker 消费）──
+    # 未在此表显式列出的 cleanup 任务会落到默认队列 celery，不占 generation 槽位。
+    "worker.cleanup_stale_pending_tasks": {"queue": QUEUE_DEFAULT},
+    "worker.cleanup_stale_running_tasks": {"queue": QUEUE_DEFAULT},
+    "worker.cleanup_stale_ingest_jobs": {"queue": QUEUE_DEFAULT},
+    "worker.cleanup_stale_voice_clones": {"queue": QUEUE_DEFAULT},
 }

 # 生成任务的预取数：渲染是长任务，预取 1 避免任务被某个 worker 占住不调度
@@ -47,11 +78,10 @@ GENERATION_WORKER_PREFETCH_MULTIPLIER = 1


 def apply_queue_settings(app) -> None:
-    """把队列隔离配置应用到 Celery app（API 生产端与 Worker 消费端都要调用）。
+    """把队列分流配置应用到 Celery app（API 生产端与 Worker 消费端都要调用）。

    配置 task_queues / task_routes / task_default_queue。生产端靠 task_routes
-    把消息投递到对应队列；消费端靠 task_queues 声明自己消费哪些队列
-    （实际消费集由启动参数 -Q 控制）。
+    把消息投递到对应队列；消费端靠启动参数 -Q 控制自己消费哪些队列（entrypoint）。
    """
    app.conf.task_queues = task_queues
    app.conf.task_routes = task_routes
@@ -7,13 +7,21 @@
  3. 生成 relay 一次性 key，构造两个带 token 的 URL：
     - put_url：给 P4000 回传结果，走 relay_base_url（Tailscale host:8092）
     - get/del_url：worker 自己下载+清理用，走 relay_internal_base_url（Docker DNS 直连 API）
-  4. POST P4000 /api/render/sync：inputs={"in.mp4": "<mezzanine-get-url>"}, output_url="<put_url>"
+  4. 【冷启动防护】距上次成功通信 >60s 时，先 GET /health 预热 Tailscale 链路（短超时快速失败）
+  5. POST P4000 /api/render/sync：inputs={"in.mp4": "<mezzanine-get-url>"}, output_url="<put_url>"
     ffmpeg_args: -i in.mp4 [-vf <vf>] -c:v h264_nvenc ... -an/-c:a aac -f mp4 pipe:1
-  5. P4000 从 relay GET mezzanine → h264_nvenc 编码 → PUT 最终 mp4 到 put_url
-  6. 本客户端通过 get_url（Docker 内网）下载最终文件到 output_path，然后 DELETE 清理
-  7. 删除 relay 上的 mezzanine 临时文件（以及 OSS fallback 的 key）
+     - 首字节用短超时(默认20s)，避免链路卡死空等上百秒；首字节到达后放宽到 ffmpeg_timeout+60s
+  6. P4000 从 relay GET mezzanine → h264_nvenc 编码 → PUT 最终 mp4 到 put_url
+  7. 本客户端通过 get_url（Docker 内网）下载最终文件到 output_path，然后 DELETE 清理
+  8. 删除 relay 上的 mezzanine 临时文件（以及 OSS fallback 的 key）

 任何环节失败抛 GpuEncodeError，调用方应 fallback 到 CPU libx264。
+
+冷启动/链路卡顿背景（2026-09-27 实测）：P4000 与 staging 之间走 Tailscale，长时间空闲
+（>7h）后首次请求曾出现 150s 延迟才真正开始下载 mezzanine，期间 ffmpeg 尚未启动、GPU 空闲。
+根因在服务端/网络层（可能是 Tailscale DERP 打洞或 httpx 连接池重建），本客户端通过
+pre_warm + 首字节短超时做兜底：预热打通链路 + 20s 内收不到首字节就快速失败让 CPU fallback，
+不再让用户等满 150s+。
 """

 from __future__ import annotations
@@ -63,6 +71,13 @@ class GpuEncoderClient:
        mezzanine_transport: str = "relay",
        sync_timeout: int = 300,
        health_timeout: float = 3.0,
+        # 提交编码任务前先发一次 /health 预热 Tailscale 链路，避免长时间空闲后首次请求
+        # 因 DERP 打洞/NAT 映射过期/Tailscale 连接重建而阻塞上百秒。
+        pre_warm: bool = True,
+        # POST 首次响应超时：P4000 已收到请求后应该在数秒内开始下载 inputs；
+        # 如果超过这个值还没收到任何响应字节，说明链路/服务卡住，快速失败让调用方 fallback CPU。
+        # 注意：ffmpeg 编码本身靠 body.timeout 控制（300s），不应该被这个超时影响。
+        post_first_byte_timeout: float = 20.0,
        vcodec: str = "h264_nvenc",
        preset: str = "p4",
        crf: int = 23,
@@ -80,12 +95,16 @@ class GpuEncoderClient:
        self.mezzanine_transport = mezzanine_transport.lower()  # "relay" | "oss"
        self.sync_timeout = sync_timeout
        self.health_timeout = health_timeout
+        self.pre_warm = pre_warm
+        self.post_first_byte_timeout = post_first_byte_timeout
        self.vcodec = vcodec
        self.preset = preset
        self.crf = crf
        self.bitrate = bitrate
        self._relay_secret = relay_secret
        self.oss_tmp_prefix = oss_tmp_prefix.rstrip("/") + "/" if oss_tmp_prefix else "tmp/gpu-mezzanine/"
+        # 上次与 P4000 成功通信的时间戳（用于判断是否需要 pre_warm 预热）
+        self._last_ok_ts: float = 0.0

    RELAY_PATH_PREFIX = "/api/v1/internal/gpu-relay"

@@ -128,13 +147,16 @@ class GpuEncoderClient:
        except (urllib.error.URLError, socket.timeout, TimeoutError, json.JSONDecodeError, ConnectionError) as e:
            return GpuHealth(healthy=False, error=f"health probe failed: {e}")
        try:
-            return GpuHealth(
+            h = GpuHealth(
                healthy=data.get("status") == "healthy",
                worker=str(data.get("worker", "")),
                gpu_name=(data.get("gpu") or {}).get("name", ""),
                nvenc_h264=bool((data.get("nvenc") or {}).get("h264_nvenc")),
                nvenc_hevc=bool((data.get("nvenc") or {}).get("hevc_nvenc")),
            )
+            if h.healthy:
+                self._last_ok_ts = time.time()
+            return h
        except Exception as e:  # noqa: BLE001
            return GpuHealth(healthy=False, error=f"malformed health response: {e}")

@@ -227,7 +249,8 @@ class GpuEncoderClient:
                ffmpeg_args.append("-an")
            ffmpeg_args.extend(["-f", "mp4", "pipe:1"])

-            # 4. call P4000 sync render
+            # 4. pre-warm then call P4000 sync render
+            self._warm_up_if_needed()
            body = {
                "inputs": {"in.mp4": input_url},
                "ffmpeg_args": ffmpeg_args,
@@ -235,6 +258,7 @@ class GpuEncoderClient:
                "timeout": int(timeout),
            }
            job = self._post_sync(body)
+            self._last_ok_ts = time.time()
            logger.info(
                "[gpu-encoder] P4000 done: job_id=%s rc=%s size=%s dur=%ss transport=%s",
                job.get("job_id"),
@@ -285,6 +309,103 @@ class GpuEncoderClient:
                except Exception as e:  # noqa: BLE001
                    logger.warning("[gpu-encoder] failed to delete OSS mezzanine %s: %s", oss_key, e)

+    # ------------------------------------------------------------------
+    # High-level: render arbitrary inputs → final output (all-GPU pipeline)
+    # ------------------------------------------------------------------
+    def render_inputs_to_output(
+        self,
+        inputs: dict[str, str],
+        ffmpeg_args: list[str],
+        output_path: Path,
+        *,
+        timeout: Optional[int] = None,
+    ) -> dict[str, Any]:
+        """把多输入（原始素材/字幕/BGM）连同完整 filter_complex 交给 P4000 一次出片。
+
+        与 encode_mezzanine_to_output 的区别：worker 侧不再生成/上传 mezzanine，
+        P4000 直接从 inputs 中的签名 URL 下载原始素材，filter_complex 内完成
+        concat/scale/crop/drawtext/amix，末端 h264_nvenc 只编码一次。
+
+        传输：成片仍走 relay 回传（P4000 PUT → worker GET），避免公网 OSS 往返。
+
+        Args:
+            inputs: {裸文件名: 可下载URL}，key 即 ffmpeg_args 中引用的文件名
+            ffmpeg_args: 完整 ffmpeg 参数（含 -i、-filter_complex、-map、NVENC 编码参数）
+            output_path: worker 本地成片落盘路径
+            timeout: P4000 侧超时（秒）
+        """
+        if not inputs:
+            raise GpuEncodeError("render_inputs_to_output: inputs is empty")
+        if not ffmpeg_args:
+            raise GpuEncodeError("render_inputs_to_output: ffmpeg_args is empty")
+        if not self.relay_base_url:
+            raise GpuEncodeError("gpu_encode_relay_base_url not configured")
+
+        timeout = timeout or self.sync_timeout
+        t_total = time.time()
+        result_key: Optional[str] = None
+
+        try:
+            secret = self._get_relay_secret()
+
+            # 1. result relay URLs（成片 P4000 PUT → worker GET）
+            result_key = uuid.uuid4().hex
+            put_url = self._result_put_url(result_key, secret)
+            get_url = self._result_get_url(result_key, secret)
+            del_result_url = get_url
+
+            # 2. pre-warm then call P4000 sync render
+            self._warm_up_if_needed()
+            body = {
+                "inputs": dict(inputs),
+                "ffmpeg_args": list(ffmpeg_args),
+                "output_url": put_url,
+                "timeout": int(timeout),
+            }
+            job = self._post_sync(body)
+            self._last_ok_ts = time.time()
+            logger.info(
+                "[gpu-encoder] P4000 direct done: job_id=%s rc=%s size=%s dur=%ss inputs=%d",
+                job.get("job_id"),
+                job.get("ffmpeg_rc"),
+                job.get("size"),
+                job.get("duration"),
+                len(inputs),
+            )
+
+            # 3. download result from relay to output_path
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            size = self._download_to_file(get_url, output_path)
+
+            # 4. cleanup relay result
+            self._relay_delete(del_result_url)
+
+            logger.info(
+                "[gpu-encoder] direct render ok → %s (%d bytes) total=%.2fs",
+                output_path.name,
+                size,
+                time.time() - t_total,
+            )
+            return {
+                "job": job,
+                "output_size": size,
+                "output_path": str(output_path),
+                "transport": "direct",
+            }
+
+        except GpuEncodeError:
+            raise
+        except Exception as e:  # noqa: BLE001
+            raise GpuEncodeError(f"unexpected: {e}") from e
+        finally:
+            # cleanup relay result (best-effort)
+            if result_key:
+                try:
+                    secret = self._get_relay_secret()
+                    self._relay_delete(self._relay_result_url(self.relay_internal_base_url, result_key, secret))
+                except Exception as e:  # noqa: BLE001
+                    logger.warning("[gpu-encoder] failed to delete relay result %s: %s", result_key, e)
+
    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------
@@ -299,9 +420,38 @@ class GpuEncoderClient:
            raise GpuEncodeError("GPU_ENCODE_RELAY_SECRET not set")
        return secret

+    def _warm_up_if_needed(self) -> None:
+        """POST 前预热：如果距上次成功通信超过 idle 阈值，先打 /health 打通 Tailscale 链路。
+
+        背景：Tailscale 在长时间空闲（几小时）后，到对端的直连 NAT 映射可能过期，
+        首次请求会走 DERP 中继打洞；极少数情况下打洞/重连会卡住上百秒（曾观测到 150s 延迟）。
+        预热请求本身走短超时快速失败，不会阻塞主流程；预热成功后再发 POST。
+        """
+        if not self.pre_warm:
+            return
+        idle = time.time() - self._last_ok_ts
+        # 空闲超过 60s 才预热（正常流水线里相邻任务间隔通常 <10s，没必要每次都打）
+        if idle < 60:
+            return
+        url = f"{self.endpoint}/health"
+        t0 = time.time()
+        try:
+            with urllib.request.urlopen(url, timeout=min(self.health_timeout, 3.0)) as resp:
+                resp.read()
+            self._last_ok_ts = time.time()
+            logger.debug("[gpu-encoder] pre-warm ok: took=%.2fs idle=%.0fs", time.time() - t0, idle)
+        except (urllib.error.URLError, socket.timeout, TimeoutError, ConnectionError, OSError) as e:
+            # 预热失败不致命——主 POST 会带自己的超时，再失败就抛 GpuEncodeError 让调用方 fallback
+            logger.warning("[gpu-encoder] pre-warm probe failed (will try POST anyway): %s", e)
+
    def _post_sync(self, body: dict[str, Any]) -> dict[str, Any]:
        url = f"{self.endpoint}/api/render/sync"
-        req_timeout = body.get("timeout", self.sync_timeout) + 60
+        ffmpeg_timeout = body.get("timeout", self.sync_timeout)
+        # 连接 + 首字节用短超时（防链路卡死数百秒）；首字节到达后给 ffmpeg 留足编码+上传时间
+        # Python urllib 的 timeout 是整个请求总超时，所以用"两段式"：
+        #   阶段1：先 read(1) 拿首字节，用短超时；
+        #   阶段2：再 read() 读完整 body，用 ffmpeg_timeout+60。
+        connect_timeout = min(max(self.post_first_byte_timeout, 5.0), 30.0)
        payload = json.dumps(body).encode("utf-8")
        req = urllib.request.Request(
            url,
@@ -310,14 +460,45 @@ class GpuEncoderClient:
            method="POST",
        )
        t0 = time.time()
+        first_byte_ok = False
+        resp = None
        try:
-            with urllib.request.urlopen(req, timeout=req_timeout) as resp:
-                raw = resp.read().decode("utf-8")
+            resp = urllib.request.urlopen(req, timeout=connect_timeout)
+            # 读首字节 —— 如果 P4000/链路卡死，这里会在 connect_timeout 内抛超时
+            first_chunk = resp.read(1)
+            first_byte_ok = True
+            logger.debug(
+                "[gpu-encoder] P4000 first byte in %.2fs (connect_timeout=%.1fs)",
+                time.time() - t0,
+                connect_timeout,
+            )
+            # 剩余用长超时（给底层socket放宽时限；如果是mock/不支持，则跳过）
+            try:
+                resp.fp._sock.settimeout(ffmpeg_timeout + 60)
+            except (AttributeError, OSError):
+                pass
+            rest = resp.read()
+            raw = (first_chunk + rest).decode("utf-8")
+            resp.close()
+            resp = None
        except urllib.error.HTTPError as e:
            detail = e.read().decode("utf-8", errors="replace")[:1000]
            raise GpuEncodeError(f"P4000 HTTP {e.code}: {detail}") from e
-        except (urllib.error.URLError, socket.timeout, TimeoutError, ConnectionError) as e:
-            raise GpuEncodeError(f"P4000 connection error: {e}") from e
+        except (urllib.error.URLError, socket.timeout, TimeoutError, ConnectionError, OSError) as e:
+            waited = time.time() - t0
+            hint = "first-byte" if not first_byte_ok else "ffmpeg/upload"
+            # 统一以 "connection error" 开头，便于上层 fallback 逻辑用关键词识别；
+            # 末尾再附带具体错误（timed out / refused ...）供排障
+            raise GpuEncodeError(
+                f"P4000 {hint} connection error after {waited:.1f}s "
+                f"(connect_timeout={connect_timeout:.0f}s, ffmpeg_timeout={ffmpeg_timeout}s): {e}"
+            ) from e
+        finally:
+            if resp is not None:
+                try:
+                    resp.close()
+                except Exception:
+                    pass
        try:
            result = json.loads(raw)
        except json.JSONDecodeError as e:
@@ -451,6 +632,8 @@ def _build_client_from_settings() -> Optional[GpuEncoderClient]:
        mezzanine_transport=getattr(settings, "gpu_encode_mezzanine_transport", "relay") or "relay",
        sync_timeout=getattr(settings, "gpu_encode_sync_timeout", 300),
        health_timeout=getattr(settings, "gpu_encode_health_timeout", 3.0),
+        pre_warm=getattr(settings, "gpu_encode_pre_warm", True),
+        post_first_byte_timeout=getattr(settings, "gpu_encode_post_first_byte_timeout", 20.0),
        vcodec=getattr(settings, "gpu_encode_vcodec", "h264_nvenc"),
        preset=getattr(settings, "gpu_encode_preset", "p4"),
        crf=getattr(settings, "gpu_encode_crf", 23),
@@ -5,6 +5,13 @@
 - Worker端 oss_helpers 的高级能力（分片上传/超时保护/HTTP下载/Asset路径解析）

 所有服务都通过这个统一入口与存储交互，消除重复实现。
+
+P1 (2026-09-28) OSS 双 endpoint 分离：
+- 内部 bucket（self.bucket）：使用 internal endpoint（VPC 千兆带宽），
+  用于所有 SDK 上传/下载/删除/object_exists 操作；
+- 公网 bucket（self.public_bucket）：使用公网 endpoint，仅用于 sign_url
+  生成给前端/P4000/MediaKit 等外网访问方用的预签名 URL；
+- public_url 永远拼公网域名，不随 internal endpoint 变化。
 """

 from __future__ import annotations
@@ -42,19 +49,44 @@ OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
 OSS_HTTP_DOWNLOAD_TIMEOUT = 300  # HTTP下载超时（秒）


-class SharedStorageService(StoragePort):
-    """统一存储服务 — 实现 StoragePort，API 和 Worker 共用。
+def _make_bucket(
+    auth,
+    endpoint: str,
+    bucket_name: str,
+    *,
+    connect_timeout: int = OSS_CONNECT_TIMEOUT,
+    app_name: str = "",
+):
+    """构造 oss2.Bucket，自动补 https:// 前缀。"""
+    if not endpoint.startswith(("http://", "https://")):
+        endpoint = f"https://{endpoint}"
+    kwargs: dict = {"connect_timeout": connect_timeout}
+    if app_name:
+        kwargs["app_name"] = app_name
+    return oss2.Bucket(auth, endpoint, bucket_name, **kwargs)

-    整合了原 SharedStorageService + oss_helpers 的全部能力。
-    """
+
+class SharedStorageService(StoragePort):
+    """统一存储服务 — 实现 StoragePort，API 和 Worker 共用。"""
+
+    # 类级默认值，方便单测 mock __init__ 后实例仍有这些属性
+    bucket: Optional[object] = None
+    public_bucket: Optional[object] = None
+    public_endpoint: str = ""
+    internal_endpoint: str = ""
+    public_url: str = ""
+    local_url_prefix: str = "/generated-files"
+    bucket_name: str = ""

    def __init__(self):
        settings = get_shared_settings()
        self.bucket_name = settings.oss_bucket_name
-        self.endpoint = settings.oss_endpoint
-        self.public_url = f"https://{settings.oss_bucket_name}.{settings.oss_endpoint}"
+        self.public_endpoint = settings.oss_endpoint  # 公网 endpoint，用于签名 URL
+        self.internal_endpoint = settings.effective_oss_internal_endpoint  # 内网 endpoint，SDK 用
+        self.public_url = f"https://{settings.oss_bucket_name}.{self._public_host()}"
        self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
-        self.bucket = None
+        self.bucket: Optional[object] = None  # internal: SDK 上传/下载/删除
+        self.public_bucket: Optional[object] = None  # public: sign_url 给外网

        self.access_key_id = settings.oss_access_key_id
        self.access_key_secret = settings.oss_access_key_secret
@@ -65,21 +97,26 @@ class SharedStorageService(StoragePort):
        if has_key_id and has_key_secret:
            if oss2 is not None:
                try:
-                    # endpoint 不带 scheme 时补 https:// 前缀
-                    bucket_endpoint = self.endpoint
-                    if not bucket_endpoint.startswith(("http://", "https://")):
-                        bucket_endpoint = f"https://{bucket_endpoint}"
                    auth = oss2.Auth(self.access_key_id, self.access_key_secret)
-                    self.bucket = oss2.Bucket(
+                    self.bucket = _make_bucket(
                        auth,
-                        bucket_endpoint,
+                        self.internal_endpoint,
                        self.bucket_name,
-                        connect_timeout=OSS_CONNECT_TIMEOUT,
+                        app_name="xiaoxia-internal",
                    )
-                    logger.info(
-                        "OSS initialized: endpoint=%s bucket=%s",
-                        self.endpoint,
+                    self.public_bucket = _make_bucket(
+                        auth,
+                        self.public_endpoint,
                        self.bucket_name,
+                        app_name="xiaoxia-public",
+                    )
+                    same_ep = self.internal_endpoint == self.public_endpoint
+                    logger.info(
+                        "OSS initialized: public_ep=%s internal_ep=%s bucket=%s dual=%s",
+                        self.public_endpoint,
+                        self.internal_endpoint,
+                        self.bucket_name,
+                        "no" if same_ep else "yes",
                    )
                except Exception as error:
                    logger.error("Failed to initialize OSS bucket client: %s", error)
@@ -93,26 +130,35 @@ class SharedStorageService(StoragePort):
                missing.append("OSS_ACCESS_KEY_SECRET")
            logger.error("OSS credentials not configured — missing: %s", ", ".join(missing))

+    def _public_host(self) -> str:
+        ep = self.public_endpoint
+        if ep.startswith("https://"):
+            return ep[len("https://") :]
+        if ep.startswith("http://"):
+            return ep[len("http://") :]
+        return ep
+
    # ── 诊断 ───────────────────────────────────────────────────────────

    def diagnose(self) -> None:
-        """输出存储配置诊断日志。"""
        key_id_display = (
            f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)"
        )
        logger.info(
-            "[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s",
-            self.endpoint,
+            "[OSS诊断] public_ep=%s internal_ep=%s bucket=%s ak=%s",
+            self.public_endpoint,
+            self.internal_endpoint,
            self.bucket_name,
            key_id_display,
        )
        if self.bucket is None:
-            logger.error(
-                "[OSS诊断] ❌ bucket=None — 预签名URL不可用！"
-                "原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。"
-            )
+            logger.error("[OSS诊断] ❌ bucket(internal)=None")
        else:
-            logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用")
+            logger.info("[OSS诊断] ✅ bucket(internal) 就绪")
+        if self.public_bucket is None:
+            logger.error("[OSS诊断] ❌ public_bucket=None")
+        else:
+            logger.info("[OSS诊断] ✅ public_bucket 就绪，公网签名URL可用")

    # ── 工具方法 ───────────────────────────────────────────────────────

@@ -122,20 +168,15 @@ class SharedStorageService(StoragePort):
        return path.startswith(f"{self.local_url_prefix}/")

    def _normalize_storage_key(self, storage_key_or_url: str) -> str:
-        """从 URL 提取存储键，并做 URL 解码。
-
-        防止 URL 编码的字符（空格=%20、中文=%XX）导致签名不匹配。
-        """
        if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
            parsed = urlparse(storage_key_or_url)
            return unquote(parsed.path.lstrip("/"))
        return storage_key_or_url.lstrip("/")

    def normalize_storage_key(self, storage_key_or_url: str) -> str:
-        """从 URL 提取存储键（公开方法）。"""
        return self._normalize_storage_key(storage_key_or_url)

-    # ── 上传 ───────────────────────────────────────────────────────────
+    # ── 上传（SDK 走 internal endpoint）───────────────────────────────

    def upload_file(
        self,
@@ -143,21 +184,22 @@ class SharedStorageService(StoragePort):
        storage_key: str,
        content_type: str = "application/octet-stream",
    ) -> str:
-        """上传文件到存储，返回公开 URL（简单上传，API端原有行为）。
-
-        - 路径字符串 → bucket.put_object_from_file
-        - 类文件对象 → bucket.put_object
-        - bucket未配置 → 抛 RuntimeError
-        """
        if self.bucket is None:
            raise RuntimeError("OSS storage is not configured")
-
        try:
            if isinstance(file_or_path, (str, Path)):
-                self.bucket.put_object_from_file(storage_key, str(file_or_path), headers={"Content-Type": content_type})
+                self.bucket.put_object_from_file(
+                    storage_key,
+                    str(file_or_path),
+                    headers={"Content-Type": content_type},
+                )
            else:
-                file_or_path.seek(0)  # type: ignore[attr-defined]
-                self.bucket.put_object(storage_key, file_or_path, headers={"Content-Type": content_type})
+                file_or_path.seek(0)
+                self.bucket.put_object(
+                    storage_key,
+                    file_or_path,
+                    headers={"Content-Type": content_type},
+                )
            return f"{self.public_url}/{storage_key}"
        except Exception as e:
            raise Exception(f"Failed to upload file to OSS: {e}") from e
@@ -167,14 +209,6 @@ class SharedStorageService(StoragePort):
        local_path: str | Path,
        storage_key: str,
    ) -> Optional[str]:
-        """智能上传：大文件自动分片+超时保护（从 oss_helpers 合并）。
-
-        - 大文件（>100MB）走分片上传，3 线程并发
-        - 总超时 300s，防止网络异常时挂死
-        - 成功返回 URL，失败返回 None（不抛异常）
-
-        Worker端 oss_helpers.upload_to_oss 的统一入口。
-        """
        local_path = Path(local_path)
        if not local_path.exists():
            logger.error("上传文件不存在: %s", local_path)
@@ -198,10 +232,10 @@ class SharedStorageService(StoragePort):

                if use_multipart:
                    logger.info(
-                        "大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
+                        "大文件分片上传(internal): key=%s size=%.1fMB part=%dMB threads=%d",
                        storage_key[:80],
-                        file_size / 1024 / 1024,
-                        OSS_PART_SIZE // 1024 // 1024,
+                        file_size / 1048576,
+                        OSS_PART_SIZE // 1048576,
                        OSS_MULTIPART_NUM_THREADS,
                    )
                    oss2.resumable_upload(
@@ -214,7 +248,6 @@ class SharedStorageService(StoragePort):
                    )
                else:
                    self.bucket.put_object_from_file(storage_key, str(local_path))
-
                result["url"] = f"{self.public_url}/{storage_key}"
            except Exception as e:
                result["error"] = e
@@ -222,73 +255,54 @@ class SharedStorageService(StoragePort):
            finally:
                done.set()

-        upload_thread = threading.Thread(target=_do_upload, daemon=True)
-        upload_thread.start()
+        t = threading.Thread(target=_do_upload, daemon=True)
+        t.start()
        finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)

        if not finished:
            logger.error(
-                "OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
+                "OSS 上传超时(%ds): key=%s size=%.1fMB",
                OSS_UPLOAD_TOTAL_TIMEOUT,
                storage_key[:80],
-                result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
+                result["file_size"] / 1048576 if result["file_size"] else 0,
            )
            return None
+        return None if result["error"] else result["url"]

-        if result["error"]:
-            return None
-
-        return result["url"]
-
-    # ── 下载 ───────────────────────────────────────────────────────────
+    # ── 下载（SDK 走 internal endpoint）───────────────────────────────

    def download_file(self, storage_key: str, local_path: str | Path) -> None:
-        """从 OSS 下载文件（简单下载，API端原有行为）。
-
-        bucket未配置 → 抛 RuntimeError
-        """
        if self.bucket is None:
            raise RuntimeError("OSS storage is not configured")
-
        local_path = Path(local_path)
        os.makedirs(local_path.parent, exist_ok=True)
        try:
-            self.bucket.get_object_to_file(self._normalize_storage_key(storage_key), str(local_path))
+            self.bucket.get_object_to_file(
+                self._normalize_storage_key(storage_key),
+                str(local_path),
+            )
        except Exception as e:
            raise Exception(f"Failed to download file from OSS: {e}") from e

    def download_asset(self, asset_storage_key: str, local_path: str | Path) -> bool:
-        """下载素材（从 oss_helpers 合并）。
-
-        自动识别输入类型：
-        - 完整 URL → 走 HTTP 下载（支持预签名URL）
-        - 存储键 → 走 oss2 SDK 下载
-
-        成功返回 True，失败返回 False（不抛异常）。
-        """
        local_path = Path(local_path)
        os.makedirs(local_path.parent, exist_ok=True)
-
-        # 完整URL走HTTP下载（兼容预签名URL）
        if asset_storage_key.startswith(("http://", "https://")):
            return self._download_via_http(asset_storage_key, local_path)
-
-        # OSS存储键走SDK
        if self.bucket is None:
-            logger.error("OSS not configured, cannot download: %s", asset_storage_key[:80])
+            logger.error("OSS not configured: %s", asset_storage_key[:80])
            return False
        try:
-            self.bucket.get_object_to_file(self._normalize_storage_key(asset_storage_key), str(local_path))
+            self.bucket.get_object_to_file(
+                self._normalize_storage_key(asset_storage_key),
+                str(local_path),
+            )
            return local_path.exists() and local_path.stat().st_size > 0
        except Exception:
            logger.exception("下载素材失败: %s", asset_storage_key)
            return False

    def _download_via_http(self, url: str, local_path: Path) -> bool:
-        """通过 HTTP 下载文件（支持预签名 URL）。
-
-        流式下载避免大文件内存溢出。
-        """
        try:
            resp = requests.get(url, stream=True, timeout=OSS_HTTP_DOWNLOAD_TIMEOUT)
            resp.raise_for_status()
@@ -298,83 +312,64 @@ class SharedStorageService(StoragePort):
                        f.write(chunk)
            return local_path.exists() and local_path.stat().st_size > 0
        except Exception:
-            logger.exception("HTTP下载素材失败: %s", url[:100])
+            logger.exception("HTTP下载失败: %s", url[:100])
            return False

-    # ── URL 生成 ──────────────────────────────────────────────────────
+    # ── URL 生成（sign_url 用 public_bucket 签公网域名）───────────────

    def get_url(self, storage_key: str) -> str:
-        """获取公开 URL。"""
        return f"{self.public_url}/{storage_key}"

-    def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
-        """获取预签名下载 URL。
+    def _sign_bucket(self):
+        """签名优先用 public_bucket，回退到 bucket。"""
+        return self.public_bucket or self.bucket

-        bucket未配置时降级为公开URL；本地产物URL直接返回。
-        """
-        if self.bucket is None:
+    def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
+        sign_bucket = self._sign_bucket()
+        if sign_bucket is None:
            if self._is_local_generated_url(storage_key_or_url):
                return storage_key_or_url
-            logger.warning(
-                "get_download_url: OSS bucket not configured, returning raw URL. key=%s",
-                storage_key_or_url[:200],
-            )
+            logger.warning("OSS bucket not configured, returning raw URL: %s", storage_key_or_url[:200])
            return self.get_url(self.normalize_storage_key(storage_key_or_url))

        storage_key = self.normalize_storage_key(storage_key_or_url)
        try:
-            signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
+            signed = sign_bucket.sign_url("GET", storage_key, expires_seconds)
            logger.info(
-                "get_download_url: signed URL generated. key=%s url_prefix=%s",
+                "signed URL generated for key=%s prefix=%s",
                storage_key[:80],
                signed[:60],
            )
            return signed
        except Exception:
-            logger.exception(
-                "get_download_url: sign_url failed, falling back to raw URL. key=%s",
-                storage_key[:200],
-            )
+            logger.exception("get_download_url: sign_url 失败，返回 raw URL: %s", storage_key[:200])
            return self.get_url(storage_key)

-    # ── 浏览器直传 POST ────────────────────────────────────────────────
-
    def get_upload_url(
        self,
        storage_key_or_url: str,
        expires_seconds: int = 3600,
        content_type: str = "video/mp4",
    ) -> str:
-        """获取预签名 PUT 上传 URL（供外部 Worker 上传结果文件）。
-
-        bucket未配置时降级为 public_url（本地/开发环境）；
-        本地产物 key 原样返回。
-        """
-        if self.bucket is None:
+        sign_bucket = self._sign_bucket()
+        if sign_bucket is None:
            if self._is_local_generated_url(storage_key_or_url):
                return storage_key_or_url
-            logger.warning(
-                "get_upload_url: OSS bucket not configured, returning raw URL. key=%s",
-                storage_key_or_url[:200],
-            )
+            logger.warning("get_upload_url: OSS 未配置，返回 raw URL: %s", storage_key_or_url[:200])
            return self.get_url(self.normalize_storage_key(storage_key_or_url))

        storage_key = self.normalize_storage_key(storage_key_or_url)
        try:
-            # oss2 sign_url 支持 'PUT'，需指定 headers 才能限定 Content-Type
            headers = {"Content-Type": content_type} if content_type else None
-            signed = self.bucket.sign_url("PUT", storage_key, expires_seconds, headers=headers)
+            signed = sign_bucket.sign_url("PUT", storage_key, expires_seconds, headers=headers)
            logger.info(
-                "get_upload_url: signed PUT URL generated. key=%s url_prefix=%s",
+                "get_upload_url: 公网签名PUT URL已生成 key=%s prefix=%s",
                storage_key[:80],
                signed[:60],
            )
            return signed
        except Exception:
-            logger.exception(
-                "get_upload_url: sign_url failed, falling back to raw URL. key=%s",
-                storage_key[:200],
-            )
+            logger.exception("get_upload_url: sign_url 失败，返回 raw URL: %s", storage_key[:200])
            return self.get_url(storage_key)

    def create_direct_upload_post(
@@ -384,7 +379,6 @@ class SharedStorageService(StoragePort):
        max_size_bytes: int,
        expires_seconds: int,
    ) -> dict[str, object]:
-        """创建浏览器直传 POST 表单。"""
        if not self.access_key_id or not self.access_key_secret:
            raise RuntimeError("OSS storage is not configured")
        normalized_key = self.normalize_storage_key(storage_key)
@@ -431,19 +425,17 @@ class SharedStorageService(StoragePort):
            },
        }

-    # ── 文件操作 ───────────────────────────────────────────────────────
+    # ── 文件操作（internal endpoint）──────────────────────────────────

    def delete_file(self, storage_key: str) -> None:
-        """删除文件（不抛异常）。"""
        if self.bucket is None:
            return
        try:
            self.bucket.delete_object(storage_key)
        except Exception as error:
-            logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)})
+            logger.warning("OSS delete 失败", extra={"storage_key": storage_key, "error": str(error)})

    def file_exists(self, storage_key: str) -> bool:
-        """检查文件是否存在。"""
        if self.bucket is None:
            return False
        return self.bucket.object_exists(storage_key)
@@ -451,17 +443,6 @@ class SharedStorageService(StoragePort):
    # ── Asset 路径解析（Worker 用）────────────────────────────────────

    def resolve_asset_path(self, asset_id: str, work_dir: str | Path) -> Optional[Path]:
-        """从 asset_id 解析到本地文件路径。
-
-        策略（按优先级）：
-        1. 本地绝对路径（在允许目录内）→ 直接返回
-        2. work_dir 缓存命中 → 返回缓存路径
-        3. 从OSS下载到缓存 → 返回下载路径
-        4. 全部失败 → None
-
-        从 oss_helpers.resolve_asset_path 合并而来。
-        """
-        # 延迟导入，避免循环依赖
        from video_processing.path_security import (  # type: ignore[import-not-found]
            PathSecurityError,
            get_allowed_local_dirs,
@@ -471,47 +452,35 @@ class SharedStorageService(StoragePort):

        if not asset_id or not isinstance(asset_id, str):
            return None
-
        work_dir = Path(work_dir)
        os.makedirs(work_dir, exist_ok=True)
-
-        # 空字节检测
        if "\x00" in asset_id:
-            logger.warning("asset_id 包含空字节，拒绝: %s", asset_id[:50])
+            logger.warning("asset_id 含空字节，拒绝: %s", asset_id[:50])
            return None
-
-        # 1. 本地绝对路径 — 必须在允许的目录内
        if asset_id.startswith("/") and os.path.exists(asset_id):
            try:
                resolved = Path(asset_id).resolve()
                if is_in_allowed_dirs(resolved, get_allowed_local_dirs()):
                    return resolved
-                else:
-                    logger.warning(
-                        "本地素材路径不在允许目录内，拒绝: %s (allowed=%s)",
-                        asset_id[:80],
-                        get_allowed_local_dirs(),
-                    )
-                    return None
+                logger.warning(
+                    "本地素材路径不在允许目录: %s allowed=%s",
+                    asset_id[:80],
+                    get_allowed_local_dirs(),
+                )
+                return None
            except (OSError, PathSecurityError):
                return None
-
-        # 2. 缓存命中（SHA256 hash 防路径遍历）
        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
        safe_name = sanitize_filename(cache_hash)
        cached_path = work_dir / f"{safe_name}.mp4"
        if cached_path.exists() and cached_path.stat().st_size > 0:
            return cached_path
-
-        # 3. 从 OSS 下载（先标准化 key，防路径遍历注入）
        safe_key = self.normalize_storage_key(asset_id)
        if ".." in safe_key or safe_key.startswith("/"):
-            logger.warning("asset_id 包含路径遍历模式，拒绝下载: %s", asset_id[:80])
+            logger.warning("asset_id 含路径遍历: %s", asset_id[:80])
            return None
-
        if self.download_asset(safe_key, cached_path):
            return cached_path
-
        return None

    def resolve_asset_ids_to_paths(
@@ -519,22 +488,20 @@ class SharedStorageService(StoragePort):
        asset_ids: list[str],
        work_dir: str | Path,
    ) -> dict[str, Path]:
-        """批量解析 asset_id → 本地路径。"""
        result: dict[str, Path] = {}
        for aid in asset_ids:
-            local_path = self.resolve_asset_path(aid, work_dir)
-            if local_path:
-                result[aid] = local_path
+            p = self.resolve_asset_path(aid, work_dir)
+            if p:
+                result[aid] = p
        return result


-# ── 单例管理 ────────────────────────────────────────────────────────────
+# ── 单例 ────────────────────────────────────────────────────────────────

 _storage_service: Optional[SharedStorageService] = None


 def get_shared_storage_service() -> SharedStorageService:
-    """获取统一存储服务单例。"""
    global _storage_service
    if _storage_service is None:
        _storage_service = SharedStorageService()
@@ -542,7 +509,6 @@ def get_shared_storage_service() -> SharedStorageService:
    return _storage_service


-# 向后兼容别名
 def get_storage_service() -> SharedStorageService:
-    """向后兼容：返回统一存储服务。"""
+    """向后兼容别名。"""
    return get_shared_storage_service()
@@ -1,6 +1,7 @@
 #!/bin/sh
 # ===========================================
 # Staging 部署脚本（SSH 模式，并行优化版）
+# worker 已收敛到 infra/docker/compose.yml 单一事实来源；api/web 暂保留 docker run。
 # ===========================================
 set -eu

@@ -48,6 +49,11 @@ GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-staging/generated}"
 LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-staging/legacy-assets}"
 NGINX_CONF_FILE="${NGINX_CONF_FILE:-/var/lib/xiaoxia-saas-staging/nginx-staging.conf}"
 COOKIES_FILE="${COOKIES_FILE:-/var/lib/xiaoxia-saas-staging/configs/douyin_cookies.txt}"
+INFRA_DOCKER_DIR="${INFRA_DOCKER_DIR:-/var/lib/xiaoxia-saas-staging/infra/docker}"
+COMPOSE_PROJECT="${COMPOSE_PROJECT:-xiaoxia-staging}"
+COMPOSE_ENV_VALUE="${COMPOSE_ENV_VALUE:-staging}"
+# COMPOSE_SYNC: CI workflow 已通过 scp 把 infra/docker/compose.yml 上传到服务器时设为 0 跳过同步
+COMPOSE_SYNC="${COMPOSE_SYNC:-1}"

 SKIP_MIGRATION="${SKIP_MIGRATION:-false}"
 SKIP_ROLLBACK="${SKIP_ROLLBACK:-false}"
@@ -57,8 +63,6 @@ if [ -z "$IMAGE_TAG" ]; then
  exit 1
 fi

-# .env 文件由 CI 从模板 + Secrets 渲染后通过 SCP 上传到服务器
-# 如果文件不存在，说明 CI 渲染步骤失败或未执行
 if [ ! -f "$ENV_FILE" ]; then
  echo "ERROR: $ENV_FILE 不存在。CI 应先在 render_env 步骤渲染并上传此文件"
  exit 1
@@ -67,7 +71,7 @@ echo "✅ .env file found: $ENV_FILE ($(wc -l < "$ENV_FILE") lines)"
 mkdir -p "$GENERATED_DIR"
 mkdir -p "$LEGACY_ASSETS_DIR"
 mkdir -p "$(dirname "$COOKIES_FILE")"
-# 抖音 cookies 文件：CI workflow 已通过 scp 上传；如果不存在（非 CI 环境）则创建占位
+mkdir -p "$INFRA_DOCKER_DIR"
 if [ ! -f "$COOKIES_FILE" ] || [ "$(wc -c < "$COOKIES_FILE" 2>/dev/null || echo 0)" -lt 200 ]; then
  printf '# Netscape HTTP Cookie File\n# 抖音 cookies 占位（CI 应通过 scp 上传真实 cookies）\n' > "$COOKIES_FILE"
  echo "WARNING: Douyin cookies not found or too small at $COOKIES_FILE (extraction will 503)"
@@ -76,7 +80,6 @@ else
 fi

 # ── 写入 Staging Nginx 配置 ──
-# 运行时覆盖 nginx 配置，确保 upstream 指向正确的 staging 网络
 echo "Writing staging nginx config..."
 cat > "$NGINX_CONF_FILE" << 'NGINX_EOF'
 server {
@@ -122,8 +125,26 @@ server {
 NGINX_EOF
 echo "✅ Nginx config written: $NGINX_CONF_FILE"

+# ── 确认 infra/docker/compose.yml 存在 ──
+# CI workflow 在执行本脚本前已通过 scp 把 infra/docker/compose.yml 上传到 $INFRA_DOCKER_DIR
+# （workflow 里做：scp infra/docker/compose.yml <host>:$INFRA_DOCKER_DIR/compose.yml）。
+# 这里只做存在性检查 + nginx 软链；不再 curl 私有仓库（SSH 环境无 Gitea token）。
+COMPOSE_FILE_PATH="$INFRA_DOCKER_DIR/compose.yml"
+if [ ! -f "$COMPOSE_FILE_PATH" ]; then
+  echo "ERROR: $COMPOSE_FILE_PATH 不存在。CI workflow 应先 scp infra/docker/compose.yml 到服务器"
+  exit 1
+fi
+echo "✅ compose.yml ready: $COMPOSE_FILE_PATH ($(wc -l < "$COMPOSE_FILE_PATH") lines)"
+ln -sf "$NGINX_CONF_FILE" "$INFRA_DOCKER_DIR/nginx-${COMPOSE_ENV_VALUE}.conf" 2>/dev/null || true
+
+# 封装 docker compose 调用：统一 --env-file（compose 默认只读取 project 目录下的 .env，
+# 我们的 .env 在 $INFRA_DOCKER_DIR/../../.env，必须显式传入才能读到 GENERATED_FILES_HOST_DIR 等变量）
+compose() {
+  (cd "$INFRA_DOCKER_DIR" && docker compose --env-file "$ENV_FILE" -p "$COMPOSE_PROJECT" "$@")
+}
+
 echo "==========================================="
-echo "  Staging 部署 - $IMAGE_TAG (并行优化版)"
+echo "  Staging 部署 - $IMAGE_TAG"
 echo "==========================================="

 echo "Recording current image versions for rollback..."
@@ -144,6 +165,7 @@ for c in xiaoxia-api-staging xiaoxia-worker-staging xiaoxia-web-staging; do
  fi
 done

+# ── 回滚函数 ──
 rollback() {
  echo ""
  echo "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"
@@ -157,9 +179,37 @@ rollback() {
  fi

  echo "Stopping new containers..."
-  docker rm -f xiaoxia-api-staging 2>/dev/null || true
-  docker rm -f xiaoxia-worker-staging 2>/dev/null || true
-  docker rm -f xiaoxia-web-staging 2>/dev/null || true
+  docker rm -f xiaoxia-api-staging xiaoxia-web-staging 2>/dev/null || true
+  if [ -n "$PREV_WORKER_IMAGE" ]; then
+    echo "Rolling back Worker to: $PREV_WORKER_IMAGE (via compose)"
+    compose up -d --no-deps worker 2>&1 || echo "WARN: compose rollback failed, fallback to docker run"
+    # 镜像通过 env 注入：compose 默认读 WORKER_IMAGE（未设则用 :dev），这里用临时 env 覆盖
+    if ! docker inspect xiaoxia-worker-staging >/dev/null 2>&1; then
+      echo "Fallback: docker run previous worker image"
+      docker run -d \
+        --name xiaoxia-worker-staging \
+        --env-file "$ENV_FILE" \
+        --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
+        -e APP_ENV="$COMPOSE_ENV_VALUE" \
+        -e APP_VERSION="$(echo "$PREV_WORKER_IMAGE" | grep -oE '[^:]+$')" \
+        -e WORKER_MAX_TASKS_PER_CHILD=100 \
+        -e GENERATION_CONCURRENCY="${GENERATION_CONCURRENCY:-2}" \
+        -e TRANSCODE_CONCURRENCY="${TRANSCODE_CONCURRENCY:-2}" \
+        -e BEAT_ENABLED=1 \
+        -e GENERATED_FILES_DIR=/app/generated \
+        -e GENERATED_FILES_URL_PREFIX=/generated-files \
+        -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
+        -v "$GENERATED_DIR:/app/generated" \
+        --restart unless-stopped \
+        --health-cmd "grep -q 'celery.*worker' /proc/[0-9]*/cmdline 2>/dev/null || exit 1" \
+        --health-interval 30s \
+        --health-timeout 10s \
+        --health-retries 3 \
+        --health-start-period 40s \
+        --log-driver json-file --log-opt max-size=50m --log-opt max-file=3 \
+        "$PREV_WORKER_IMAGE" || true
+    fi
+  fi

  LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"

@@ -168,10 +218,10 @@ rollback() {
    docker run -d \
      --name xiaoxia-api-staging \
      --env-file "$ENV_FILE" \
-      --network xiaoxia-net-staging \
+      --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
      -p 127.0.0.1:8000:8000 \
-      -e APP_ENV=staging \
-      -e APP_VERSION="$(echo $PREV_API_IMAGE | grep -oE '[^:]+$')" \
+      -e APP_ENV="$COMPOSE_ENV_VALUE" \
+      -e APP_VERSION="$(echo "$PREV_API_IMAGE" | grep -oE '[^:]+$')" \
      -e GENERATED_FILES_DIR=/app/generated \
      -e GENERATED_FILES_URL_PREFIX=/generated-files \
      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
@@ -183,31 +233,7 @@ rollback() {
      --health-retries 3 \
      --health-start-period 40s \
      $LOG_OPTS \
-      "$PREV_API_IMAGE" &
-  fi
-
-  if [ -n "$PREV_WORKER_IMAGE" ]; then
-    echo "Rolling back Worker to: $PREV_WORKER_IMAGE"
-    docker run -d \
-      --name xiaoxia-worker-staging \
-      --env-file "$ENV_FILE" \
-      --network xiaoxia-net-staging \
-      -e APP_ENV=staging \
-      -e APP_VERSION="$(echo $PREV_WORKER_IMAGE | grep -oE '[^:]+$')" \
-      -e WORKER_CONCURRENCY=1 \
-      -e WORKER_MAX_TASKS_PER_CHILD=100 \
-      -e GENERATED_FILES_DIR=/app/generated \
-      -e GENERATED_FILES_URL_PREFIX=/generated-files \
-      -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
-      -v "$GENERATED_DIR:/app/generated" \
-      --restart unless-stopped \
-      --health-cmd "grep -lq celery /proc/[0-9]*/cmdline 2>/dev/null || exit 1" \
-      --health-interval 30s \
-      --health-timeout 10s \
-      --health-retries 3 \
-      --health-start-period 30s \
-      $LOG_OPTS \
-      "$PREV_WORKER_IMAGE" &
+      "$PREV_API_IMAGE" || true
  fi

  if [ -n "$PREV_WEB_IMAGE" ]; then
@@ -218,7 +244,7 @@ rollback() {
    fi
    docker run -d \
      --name xiaoxia-web-staging \
-      --network xiaoxia-net-staging \
+      --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
      -p 127.0.0.1:3001:80 \
      --restart unless-stopped \
      $LEGACY_VOLUME \
@@ -228,27 +254,25 @@ rollback() {
      --health-timeout 5s \
      --health-retries 3 \
      $LOG_OPTS \
-      "$PREV_WEB_IMAGE" &
+      "$PREV_WEB_IMAGE" || true
  fi

-  wait
+  sleep 3

-  if [ -n "$PREV_API_IMAGE" ]; then
-    echo "Waiting for rolled-back API to become healthy..."
-    i=0
-    while [ "$i" -lt 40 ]; do
-      if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
-        echo "Rolled-back API is healthy!"
-        break
-      fi
-      i=$((i + 1))
-      echo "  Waiting... ($i/40)"
-      sleep 3
-    done
-    if [ "$i" -ge 40 ]; then
-      echo "WARN: Rolled-back API did not become healthy within 120s"
-      docker logs --tail 30 xiaoxia-api-staging
+  echo "Waiting for rolled-back API to become healthy..."
+  i=0
+  while [ "$i" -lt 40 ]; do
+    if curl -sf --max-time 5 http://127.0.0.1:8000/health >/dev/null 2>&1; then
+      echo "Rolled-back API is healthy!"
+      break
    fi
+    i=$((i + 1))
+    echo "  Waiting... ($i/40)"
+    sleep 3
+  done
+  if [ "$i" -ge 40 ]; then
+    echo "WARN: Rolled-back API did not become healthy within 120s"
+    docker logs --tail 30 xiaoxia-api-staging 2>/dev/null || true
  fi

  echo ""
@@ -260,7 +284,7 @@ rollback() {
  echo "Previous Web:    ${PREV_WEB_IMAGE:-none}"
  echo ""
  echo "部署失败，已自动回滚到上一版本"
-  docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging
+  docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging || true
  exit 1
 }

@@ -272,10 +296,6 @@ if [ -n "$REGISTRY_TOKEN" ]; then
  retry_docker_login
 fi

-# ---- 并行 Pull 三个镜像 ----
-# 注意：这里必须使用 IMAGE_TAG（commit SHA）做确定性部署，不要改成 :dev。
-# :dev 是 floating tag，可能被并发构建覆盖，导致部署版本不可重现、回滚混乱。
-# Watchtower 可监听 :dev 做非关键路径的自动同步；正式部署/回滚一律锚定 SHA。
 REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
 REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
 REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
@@ -304,7 +324,6 @@ for svc in api worker web; do
  elif grep -qE "Digest:|Status: Downloaded" "$PULL_LOG_DIR/$svc.log" 2>/dev/null; then
    echo "  OK $svc"
  else
-    # 检查docker pull返回值不直接，用镜像是否存在来判断
    img_var="REGISTRY_$(echo $svc | tr '[:lower:]' '[:upper:]')"
    img_val=$(eval echo "\$$img_var")
    if docker image inspect "$img_val" >/dev/null 2>&1; then
@@ -327,7 +346,7 @@ fi

 echo "All images pulled."

-# ====== 镜像内容校验（CI 加固 - 防止静默部署损坏/过期镜像） ======
+# ====== 镜像内容校验 ======
 echo ""
 echo "=========================================="
 echo "  镜像内容校验"
@@ -336,7 +355,6 @@ echo "=========================================="
 VERIFY_FAILED=0
 DEPLOY_MANIFEST="${GENERATED_DIR}/deploy-manifest.json"

-# 读取上次部署的 manifest（用于对比）
 PREV_MANIFEST=""
 if [ -f "$DEPLOY_MANIFEST" ]; then
  PREV_MANIFEST=$(cat "$DEPLOY_MANIFEST")
@@ -348,14 +366,12 @@ for svc in api worker web; do
  img_var="REGISTRY_$(echo $svc | tr '[:lower:]' '[:upper:]')"
  img_val=$(eval echo "\$$img_var")

-  # 1. 检查镜像是否存在
  if ! docker image inspect "$img_val" >/dev/null 2>&1; then
    echo "  ❌ $svc: 镜像不存在 ($img_val)"
    VERIFY_FAILED=$((VERIFY_FAILED + 1))
    continue
  fi

-  # 2. 检查 layers 有效性
  LAYER_COUNT=$(docker inspect --format='{{len .RootFS.Layers}}' "$img_val" 2>/dev/null || echo "0")
  if [ "$LAYER_COUNT" -eq 0 ]; then
    echo "  ❌ $svc: 镜像无有效 layers ($img_val)"
@@ -363,14 +379,12 @@ for svc in api worker web; do
    continue
  fi

-  # 3. 获取 digest 和创建时间
  IMG_ID=$(docker inspect --format='{{.Id}}' "$img_val")
  IMG_CREATED=$(docker inspect --format='{{.Created}}' "$img_val")
  IMG_SIZE=$(docker inspect --format='{{.Size}}' "$img_val")
  echo "  ✅ $svc: ${LAYER_COUNT} layers, size=${IMG_SIZE}, created=${IMG_CREATED}"
  echo "     id: $IMG_ID"

-  # 4. 对比上次部署
  CHANGED="unchanged"
  if [ -n "$PREV_MANIFEST" ]; then
    PREV_ID=$(echo "$PREV_MANIFEST" | grep "\"${svc}_id\"" | sed 's/.*: *"\(.*\)".*/\1/' 2>/dev/null || echo "")
@@ -398,7 +412,6 @@ if [ "$VERIFY_FAILED" -gt 0 ]; then
  exit 1
 fi

-# 写入新 manifest
 cat > "$DEPLOY_MANIFEST" <<MANIFEST_EOF
 {
  "deployed_at": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
@@ -444,14 +457,14 @@ for c in xiaoxia-postgres-staging xiaoxia-redis-staging; do
  fi
 done

-docker network create xiaoxia-net-staging 2>/dev/null || true
+docker network create "xiaoxia-net-${COMPOSE_ENV_VALUE}" 2>/dev/null || true

 if [ "$SKIP_MIGRATION" != "true" ]; then
  echo "Running database migrations..."
  docker run --rm \
    --env-file "$ENV_FILE" \
-    --network xiaoxia-net-staging \
-    -e APP_ENV=staging \
+    --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
+    -e APP_ENV="$COMPOSE_ENV_VALUE" \
    "$REGISTRY_API" sh -c "cd /app && alembic upgrade head" || {
    echo "ERROR: Database migration failed"
    exit 1
@@ -462,29 +475,28 @@ else
 fi

 echo "Stopping old containers..."
-# 优雅关闭：先 stop（发 SIGTERM，等待），再 rm
-# Worker 需要更长时间（视频任务最长可能5分钟）
-docker stop -t 120 xiaoxia-worker-staging 2>/dev/null || true
-docker stop -t 30 xiaoxia-api-staging 2>/dev/null || true
 docker stop -t 10 xiaoxia-web-staging 2>/dev/null || true
-docker rm xiaoxia-worker-staging xiaoxia-api-staging xiaoxia-web-staging 2>/dev/null || true
+docker stop -t 30 xiaoxia-api-staging 2>/dev/null || true
+docker stop -t 120 xiaoxia-worker-staging 2>/dev/null || true
+docker rm -f xiaoxia-api-staging xiaoxia-web-staging 2>/dev/null || true
+docker rm -f xiaoxia-worker-staging 2>/dev/null || true

 LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"

-# ---- 并行启动三个容器 ----
-echo "Starting all containers (parallel)..."
+echo "Starting all containers..."

 LEGACY_VOLUME=""
 if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
  LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
 fi

+# ── API: 暂保留 docker run（TODO: 后续收敛到 compose）──
 docker run -d \
  --name xiaoxia-api-staging \
  --env-file "$ENV_FILE" \
-  --network xiaoxia-net-staging \
+  --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
  -p 127.0.0.1:8000:8000 \
-  -e APP_ENV=staging \
+  -e APP_ENV="$COMPOSE_ENV_VALUE" \
  -e APP_VERSION="$IMAGE_TAG" \
  -e GENERATED_FILES_DIR=/app/generated \
  -e GENERATED_FILES_URL_PREFIX=/generated-files \
@@ -502,31 +514,18 @@ docker run -d \
  "$REGISTRY_API" &
 PID_API_START=$!

-docker run -d \
-  --name xiaoxia-worker-staging \
-  --env-file "$ENV_FILE" \
-  --network xiaoxia-net-staging \
-  -e APP_ENV=staging \
-  -e APP_VERSION="$IMAGE_TAG" \
-  -e WORKER_CONCURRENCY=1 \
-  -e WORKER_MAX_TASKS_PER_CHILD=100 \
-  -e GENERATED_FILES_DIR=/app/generated \
-  -e GENERATED_FILES_URL_PREFIX=/generated-files \
-  -e PUBLIC_API_BASE_URL=https://staging-api.xiaoxiajianji.com \
-  -v "$GENERATED_DIR:/app/generated" \
-  --restart unless-stopped \
-  --health-cmd "grep -lq celery /proc/[0-9]*/cmdline 2>/dev/null || exit 1" \
-  --health-interval 30s \
-  --health-timeout 10s \
-  --health-retries 3 \
-  --health-start-period 30s \
-  $LOG_OPTS \
-  "$REGISTRY_WORKER" &
+# ── Worker: 通过 compose 启动（单一事实来源）──
+# compose.yml 定义：三进程(beat+generation+transcode)、独立并发、BEAT_ENABLED、
+# healthcheck 匹配 'celery.*worker'（不把 beat 算活）、资源限制 4C/8G。
+# WORKER_IMAGE 通过环境变量覆盖镜像 tag（compose.yml 默认 :dev）。
+echo "Starting worker via docker compose (from $INFRA_DOCKER_DIR)..."
+WORKER_IMAGE="$REGISTRY_WORKER" APP_VERSION="$IMAGE_TAG" compose up -d --no-deps worker &
 PID_WORKER_START=$!

+# ── Web: 暂保留 docker run（TODO: 后续收敛到 compose）──
 docker run -d \
  --name xiaoxia-web-staging \
-  --network xiaoxia-net-staging \
+  --network "xiaoxia-net-${COMPOSE_ENV_VALUE}" \
  -p 127.0.0.1:3001:80 \
  --restart unless-stopped \
  $LEGACY_VOLUME \
@@ -563,9 +562,8 @@ if [ "$START_FAILED" -gt 0 ]; then
  rollback
 fi

-# ---- 并行等待 API 和 Web 健康 ----
 echo ""
-echo "Waiting for API + Web health (parallel)..."
+echo "Waiting for all services health (parallel)..."

 HEALTH_LOG_DIR="/tmp/staging-health-$$"
 mkdir -p "$HEALTH_LOG_DIR"
@@ -600,36 +598,60 @@ PID_API_HEALTH=$!
 ) > "$HEALTH_LOG_DIR/web.log" 2>&1 &
 PID_WEB_HEALTH=$!

+(
+  i=0
+  while [ "$i" -lt 20 ]; do
+    hc=$(docker inspect -f '{{if .State.Health}}{{.State.Health.Status}}{{else}}{{.State.Status}}{{end}}' xiaoxia-worker-staging 2>/dev/null || echo "missing")
+    if [ "$hc" = "healthy" ]; then
+      echo "Worker healthy after $((i * 3))s"
+      exit 0
+    fi
+    if [ "$hc" = "unhealthy" ]; then
+      echo "Worker UNHEALTHY after $((i * 3))s"
+      docker logs --tail 30 xiaoxia-worker-staging 2>/dev/null || true
+      exit 1
+    fi
+    i=$((i + 1))
+    sleep 3
+  done
+  echo "Worker health unknown after 60s (last: $hc)"
+  exit 1
+) > "$HEALTH_LOG_DIR/worker.log" 2>&1 &
+PID_WORKER_HEALTH=$!
+
 set +e
 wait $PID_API_HEALTH
 API_EXIT=$?
 wait $PID_WEB_HEALTH
 WEB_EXIT=$?
+wait $PID_WORKER_HEALTH
+WORKER_EXIT=$?
 set -e

 echo ""
 echo "健康检查结果:"
-API_OK=0
-WEB_OK=0
 if [ "$API_EXIT" -eq 0 ]; then
-  echo "  OK API: $(cat "$HEALTH_LOG_DIR/api.log")"
-  API_OK=1
+  echo "  OK API:    $(cat "$HEALTH_LOG_DIR/api.log")"
 else
-  echo "  FAIL API: 120s未就绪"
-  docker logs --tail 50 xiaoxia-api-staging
+  echo "  FAIL API:  120s未就绪"
+  docker logs --tail 50 xiaoxia-api-staging 2>/dev/null || true
 fi
-
 if [ "$WEB_EXIT" -eq 0 ]; then
-  echo "  OK Web: $(cat "$HEALTH_LOG_DIR/web.log")"
-  WEB_OK=1
+  echo "  OK Web:    $(cat "$HEALTH_LOG_DIR/web.log")"
 else
-  echo "  FAIL Web: 30s未就绪"
-  docker logs --tail 30 xiaoxia-web-staging
+  echo "  FAIL Web:  30s未就绪"
+  docker logs --tail 30 xiaoxia-web-staging 2>/dev/null || true
+fi
+if [ "$WORKER_EXIT" -eq 0 ]; then
+  echo "  OK Worker: $(cat "$HEALTH_LOG_DIR/worker.log")"
+else
+  echo "  FAIL Worker: $(cat "$HEALTH_LOG_DIR/worker.log")"
+  docker logs --tail 50 xiaoxia-worker-staging 2>/dev/null || true
 fi

 rm -rf "$HEALTH_LOG_DIR"

-if [ "$API_OK" -eq 0 ] || [ "$WEB_OK" -eq 0 ]; then
+if [ "$API_EXIT" -ne 0 ] || [ "$WEB_EXIT" -ne 0 ] || [ "$WORKER_EXIT" -ne 0 ]; then
  echo ""
  echo "ERROR: 健康检查失败"
  rollback
@@ -640,8 +662,9 @@ docker image prune -af --filter "until=168h" 2>/dev/null || true
 docker builder prune -af --filter "until=168h" 2>/dev/null || true

 echo ""
-echo "=== Staging deployment complete (并行优化版) ==="
+echo "=== Staging deployment complete ==="
 echo "API:      http://127.0.0.1:8000"
 echo "Web:      http://127.0.0.1:3001"
+echo "Worker:   managed by docker compose (project=$COMPOSE_PROJECT)"
 echo "Version:  $IMAGE_TAG"
 docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep staging
@@ -666,7 +666,7 @@ class TestDownloadAssets:
            _make_clip("c2", order=1, asset_id="asset_002"),
        ]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 2
        assert "asset_001" in asset_path_map
@@ -693,7 +693,7 @@ class TestDownloadAssets:
            _make_clip("c2", order=1, asset_id="asset_002"),
        ]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert "asset_002" in asset_path_map
@@ -714,7 +714,7 @@ class TestDownloadAssets:
            _make_clip("c1", order=0, asset_id="asset_001"),
        ]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 0
        assert len(rendered_ids) == 0
@@ -750,7 +750,7 @@ class TestDownloadAssets:
            _make_clip("c3", order=2, asset_id="asset_003"),
        ]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 2
        assert "c1" in rendered_ids
@@ -771,7 +771,7 @@ class TestDownloadAssets:
            _make_clip("c2", order=1, asset_id="asset_shared"),
        ]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert mock_download.call_count == 1
@@ -795,7 +795,7 @@ class TestDownloadAssets:
        adapter = RenderAdapter(mock_db)
        clips = [_make_clip("c1", order=0, asset_id="asset_fallback")]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 1
        assert "c1" in rendered_ids
@@ -820,7 +820,7 @@ class TestDownloadAssets:
        adapter = RenderAdapter(mock_db)
        clips = [_make_clip("c1", order=0, asset_id="asset_no_key")]

-        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids, _storage_map = adapter._download_assets(clips, tmp_path)

        assert len(asset_path_map) == 0
        assert "c1" in failed_ids