feat(gpu): #1978 AI数字人口型同步接入MuseTalk GPU Worker

业务侧（LipsyncService）集成 GpuLipsyncService： - 新增开关 USE_GPU_LIPSYNC（env，默认false；staging默认true方便联调，生产默认false待稳定后开） - _submit_audio_direct 在签名提交 MediaKit 前判断： * 开关关闭 → 直接走现有 MediaKit 云端 lipsync * 开关开但无可用 Worker（last_heartbeat_at 超5分钟窗口） → 回退 MediaKit * 开关开+Worker可用 → 创建GPU任务→同步轮询等待结果 - GPU路径成功：直接标记 job.status=completed、output_video_url=签名后的结果URL - GPU超时（默认1200s）/终态failed/异常 → 全部回退 MediaKit，对用户透明 - 轮询间隔默认5s，可配 GPU_LIPSYNC_POLL_INTERVAL - Worker心跳新鲜度窗口可配 GPU_WORKER_STALE_SECONDS（默认300s） GpuLipsyncService新增方法： - has_available_worker()：判断5分钟内有心跳的Worker存在 - wait_for_result(task_id, timeout, poll_interval)：同步轮询DB等待终态，期间顺手 _recover_timed_out_tasks；超时返回None让调用方兜底 env/CI配置： - .env.example 新增4个配置项注释 - deploy/configs/.env.staging USE_GPU_LIPSYNC=true（联调） - deploy/configs/.env.production USE_GPU_LIPSYNC=false（暂不开）单元测试：tests/unit/test_lipsync_gpu_integration.py 9个用例覆盖： - 开关关不调用GPU、无Worker回退、GPU成功completed、超时回退、 failed回退、异常回退、has_available_worker三种场景 - 全部 30个GPU相关测试全绿（含此前11+10个）
2026-09-19 16:00:58 +08:00
21 changed files with 263 additions and 2223 deletions
@@ -198,13 +198,10 @@ DOUBAO_TIMEOUT=30
 DOUBAO_MAX_RETRIES=2
 # ==================== 积分/会员系统 (#1895) ====================
-# 积分系统总开关：默认 false（暂停积分系统）。
+# 积分扣点总开关：默认 false（对现有用户零影响）。
-# - false：生成视频/口型同步/数字人/AI标题/TTS/克隆音色等所有功能对登录
+# P2 阶段各业务路由逐个接入 @points_gate 时，用
-#   用户免费放行，不扣积分、不做余额拦截；积分余额/流水/会员状态查询接口
+#   `if settings.points_enabled: ...`
-#   保留可用，但数据不再变动。积分相关的表、代码、接口均保留不删除。
+# 包裹扣点逻辑；所有路由接入完成并验证通过后再在 staging/prod 打开。
 # - 恢复积分：设置 ENABLE_CREDIT_SYSTEM=true 即可，无需改代码。
 ENABLE_CREDIT_SYSTEM=false
 # 旧开关名（兼容别名）：与 ENABLE_CREDIT_SYSTEM 任一为 true 即启用。
 POINTS_ENABLED=false
 # ==================== 抖音解析多源轮询 (#1963) ====================
@@ -12,7 +12,6 @@ from datetime import datetime, timedelta, timezone
 from typing import Optional
 from app.auth import AuthenticatedUser, get_current_user
 from app.config import settings
 from app.dependencies import get_db_session
 from app.schemas.points import (
    DailyUsageResponse,
@@ -45,12 +44,6 @@ from packages.domain.points_service import PointsService
 logger = logging.getLogger(__name__)
 def _credits_enabled() -> bool:
    """积分系统总开关（ENABLE_CREDIT_SYSTEM），关闭时全部功能免费放行。"""
    return bool(getattr(settings, "points_enabled", False))
 # ── 两个 router ──
 points_router = APIRouter()
 usage_router = APIRouter()
@@ -179,19 +172,6 @@ def check_points(
                "valid_scenes": sorted(POINTS_SCENES.keys()),
            },
        )
    # 积分系统暂停（ENABLE_CREDIT_SYSTEM=false）：所有场景直接放行，需 0 积分
    if not _credits_enabled():
        svc = _get_service()
        account = svc.get_or_create_account(current_user.user.id, db)
        return PointsCheckResponse(
            allowed=True,
            required_points=0,
            current_balance=account["balance"],
            remaining_after=account["balance"],
            is_free_quota=False,
        )
    is_mem = _is_member(current_user)
    mt = _member_type(current_user)
@@ -229,19 +209,8 @@ def deduct_points(
    current_user: AuthenticatedUser = Depends(get_current_user),
    db: Session = Depends(get_db_session),
 ):
-    """积分扣减（内部服务调用）。
+    """积分扣减（内部服务调用）。"""
    积分系统暂停（ENABLE_CREDIT_SYSTEM=false）时为 no-op：不扣分、余额不变，
    直接返回成功，保证内部调用方拿到 success=True 继续业务流程。
    """
    svc = _get_service()
    if not _credits_enabled():
        account = svc.get_or_create_account(current_user.user.id, db)
        return SimpleMessageResponse(
            success=True,
            message="积分系统已暂停，未扣减积分",
            data={"transaction_id": "", "balance": account["balance"]},
        )
    result = svc.deduct_points(
        user_id=current_user.user.id,
        amount=body.amount,
@@ -274,7 +243,11 @@ def refund_points(
    """积分退还（内部服务调用）。"""
    from packages.adapters.sqlalchemy_impl.models import PointsTransactionModel
-    txn = db.query(PointsTransactionModel).filter(PointsTransactionModel.id == body.transaction_id).first()
+    txn = (
        db.query(PointsTransactionModel)
        .filter(PointsTransactionModel.id == body.transaction_id)
        .first()
    )
    if txn is None:
        raise HTTPException(status_code=404, detail="交易记录不存在")
    if txn.user_id != current_user.user.id:
@@ -250,7 +250,7 @@ class LipsyncService:
                        gpu_task.id,
                        job.output_duration,
                    )
-                    # output_video_url 已是 _submit_to_gpu 内签好的 7 天预签名 URL
+                    # 转存到持久 OSS 路径（GPU 结果已在 gpu-lipsync/results/ 下，直接签短链）
                    return
                # wait_for_result 返回 None 表示超时/最终失败 → 继续走 MediaKit 兜底
                logger.warning("[lipsync] GPU 任务等待超时或失败，回退 MediaKit: job_id=%s", job.id)
@@ -293,77 +293,22 @@ class LipsyncService:
    # ── GPU MuseTalk 路径 ────────────────────────────────────────────────
    def _is_own_oss_url(self, url: str, storage) -> bool:
        """判断 URL / 存储 key 是否属于自家 OSS。
        - 裸存储 key（无 scheme）：自家对象
        - host 与 storage.public_url host 一致：自家对象
        - 其余 http(s) 公网链接（如 dashscope-result 临时地址）：外部对象
        """
        if not url:
            return False
        parsed = urlparse(url)
        if not parsed.scheme:
            return True  # 裸存储 key
        public_base = getattr(storage, "public_url", "")
        own_host = urlparse(public_base).netloc.lower() if public_base else ""
        return bool(own_host) and parsed.netloc.lower() == own_host
    def _persist_external_audio_for_gpu(self, *, job, storage) -> Optional[str]:
        """GPU 任务创建前，把外部域名的预合成 TTS 音频转存到自家 OSS。
        Worker 部署在用户家庭网络，dashscope-result 等第三方临时 OSS 地址
        可能无法访问；转存后 gpu_svc 在 poll 时会签自家预签名 URL 给 Worker。
        已是自家 OSS 对象（含裸 key）直接返回 None（无需转存）；
        转存失败返回 None，调用方回退使用原始 URL（最坏情况是 Worker 拉取失败，
        服务端重试耗尽后回退 MediaKit，不阻断业务）。
        """
        if self._is_own_oss_url(job.audio_url, storage):
            return None
        try:
            audio_data = safe_download_bytes(
                job.audio_url,
                purpose="lipsync_gpu_tts_audio",
                allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
                timeout=60.0,
            )
            storage_key = f"lipsync-tts/{job.user_id}/{job.id}.mp3"
            permanent_url = storage.upload_file(io.BytesIO(audio_data), storage_key, content_type="audio/mpeg")
            logger.info(
                "[lipsync] GPU 任务外部音频已转存自家 OSS: job_id=%s key=%s",
                job.id,
                storage_key,
            )
            return permanent_url
        except Exception as exc:
            logger.warning(
                "[lipsync] GPU 任务外部音频转存 OSS 失败，回退原始 URL: job_id=%s err=%s",
                job.id,
                exc,
            )
            return None
    def _submit_to_gpu(self, *, job, gpu_svc) -> Optional[object]:
        """创建 GPU 任务并同步等待结果。
        成功返回终态 task 对象（status=done）；超时或 GPU 最终失败返回 None，
        调用方回退 MediaKit。
-        输入处理：
+        注意：job.video_url / job.audio_url 可能是：
-        - job.video_url 为用户上传视频，已在自家 OSS（裸 key 或自家 URL），
+        - 自家 OSS 存储 key（storage.is_own_url 判断，gpu_svc.create_task 内部
-          gpu_svc 在 poll 时签预签名 URL 给 Worker。
+          get_download_url 会自动签预签名 URL 给 Worker）
-        - job.audio_url 可能是预合成 TTS 的第三方临时地址（如
+        - 外部公网 URL（CosyVoice 临时链接等）：poll 返回时原样透传给 Worker，
-          dashscope-result-bj.oss-cn-beijing.aliyuncs.com），Worker 家庭网络
+          Worker 可直接 GET 下载。
          拉不到；创建任务前先转存自家 OSS 再传入。
        """
        storage = get_shared_storage_service()
        # 外部音频（dashscope 临时链接等）先转存自家 OSS，避免 Worker 家庭网络拉取失败
        persisted_audio_url = self._persist_external_audio_for_gpu(job=job, storage=storage)
        audio_url_for_task = persisted_audio_url or job.audio_url
        # 创建 GPU 任务
        gpu_task = gpu_svc.create_task(
            video_url=job.video_url,
-            audio_url=audio_url_for_task,
+            audio_url=job.audio_url,
            lipsync_job_id=job.id,
            user_id=job.user_id,
            project_id=job.project_id,
@@ -386,21 +331,9 @@ class LipsyncService:
                final_task.error_msg,
            )
            return None
-        # result_url 是 OSS 存储 key（gpu-lipsync/results/{task_id}.mp4，无 host，
+        # result_url 是 OSS 存储 key；签一个长有效期 URL 写回 job.output_video_url
-        # _sign_media_url 对裸 key 不会签名）；直接用 storage 签 7 天预签名 URL
+        result_signed = self._sign_media_url(final_task.result_url)
-        # 写回 job.output_video_url，保证前端拿到可直接下载播放的地址
+        final_task.result_url = result_signed or final_task.result_url
        try:
            signed_result_url = storage.get_download_url(
                final_task.result_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS
            )
            if signed_result_url:
                final_task.result_url = signed_result_url
        except Exception as exc:
            logger.warning(
                "[lipsync] GPU 结果视频签名失败，回退原始 result_url: gpu_task=%s err=%s",
                gpu_task.id,
                exc,
            )
        return final_task
    # ── 创建任务 ──────────────────────────────────────────────────────────
@@ -17,7 +17,6 @@ import {
 } from "@ant-design/icons"
 import { useNavigate } from "react-router-dom"
 import { usePointsStore } from "@/store/pointsStore"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 import "./PointsBadge.css"
 const { Text, Paragraph } = Typography
@@ -33,13 +32,9 @@ const PointsBadge: React.FC = () => {
  const { balance, membership, subscription, dailyUsage, init, loading } = usePointsStore()
  useEffect(() => {
    if (!ENABLE_CREDIT_SYSTEM) return
    if (!balance) init()
  }, [balance, init])
  // 功能开关：积分系统关闭时直接隐藏徽章
  if (!ENABLE_CREDIT_SYSTEM) return null
  // 余额：优先用 membership.points_balance（冗余字段），降级 balance.balance
  const bal = membership?.points_balance ?? balance?.balance ?? 0
  const lowBalance = bal > 0 && bal < 10
@@ -15,7 +15,6 @@ import React, { useMemo } from "react"
 import { Tooltip } from "antd"
 import { WarningOutlined } from "@ant-design/icons"
 import { usePointsStore } from "@/store/pointsStore"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 import type { PointsSource } from "@/api/points/types"
 import "./PointsCost.css"
@@ -54,7 +53,7 @@ const PointsCost: React.FC<Props> = ({
  compact = false,
  showRechargeHint = true,
  className = "",
-}: Props) => {
+}) => {
  const { balance, dailyUsage, rules, membership } = usePointsStore()
  const qty = quantity ?? units ?? 1
@@ -119,9 +118,6 @@ const PointsCost: React.FC<Props> = ({
    }
  }, [rules, balance, dailyUsage, membership, scene, qty, durationMinutes])
  // 积分系统关闭时不展示消耗提示（组件保留，hooks 必须在 return 前调用）
  if (!ENABLE_CREDIT_SYSTEM) return null
  if (!rule || !balance) {
    return <span className={`xx-points-cost ${className}`} />
  }
@@ -21,7 +21,6 @@ import { useLogout } from "@/hooks/useAuth"
 import type { MenuProps } from "antd"
 import { NAV_ITEMS } from "@/config/navigation"
 import PointsBadge from "@/components/common/PointsBadge"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 import { usePointsStore } from "@/store/pointsStore"
 import "./Header.css"
@@ -58,36 +57,30 @@ const Header: React.FC = () => {
      label: "订阅管理",
      onClick: () => navigate("/app/subscription"),
    },
-    // 积分系统开关关闭时隐藏积分相关菜单项（代码保留不删除）
+    // v2: 我的积分入口
-    ...(ENABLE_CREDIT_SYSTEM
+    {
-      ? [
+      key: "points-center",
-          {
+      icon: <ThunderboltOutlined />,
-            key: "points-center",
+      label: (
-            icon: <ThunderboltOutlined />,
+        <Space>
-            label: (
+          我的积分
-              <Space>
+          {balance && <span style={{ color: "#8b5cf6", fontWeight: 700 }}>{balance.balance}</span>}
-                我的积分
+        </Space>
-                {balance && (
+      ),
-                  <span style={{ color: "#8b5cf6", fontWeight: 700 }}>{balance.balance}</span>
+      onClick: () => navigate("/app/points"),
-                )}
+    },
-              </Space>
+    {
-            ),
+      key: "points-history",
-            onClick: () => navigate("/app/points"),
+      icon: <HistoryOutlined />,
-          },
+      label: "积分明细",
-          {
+      onClick: () => navigate("/app/points/transactions"),
-            key: "points-history",
+    },
-            icon: <HistoryOutlined />,
+    {
-            label: "积分明细",
+      key: "recharge",
-            onClick: () => navigate("/app/points/transactions"),
+      icon: <WalletOutlined />,
-          },
+      label: "充值积分",
-          {
+      onClick: () => navigate("/app/points/recharge"),
-            key: "recharge",
+    },
            icon: <WalletOutlined />,
            label: "充值积分",
            onClick: () => navigate("/app/points/recharge"),
          },
        ]
      : []),
    { type: "divider" },
    {
      key: "logout",
@@ -137,13 +130,7 @@ const Header: React.FC = () => {
          {/* v2: 升级会员入口（仅免费用户显示） */}
          {!isMember && (
-            <Tooltip
+            <Tooltip title="升级会员解锁无限混剪、批量导出，积分 8 折起">
              title={
                ENABLE_CREDIT_SYSTEM
                  ? "升级会员解锁无限混剪、批量导出，积分 8 折起"
                  : "升级会员解锁无限混剪、批量导出"
              }
            >
              <Button
                type="primary"
                size="small"
@@ -1,13 +0,0 @@
 /**
 * 功能开关配置
 * 集中管理前端特性的启用/隐藏，便于灰度与回滚。
 * 注意：仅控制 UI 展示与前端校验，后端扣减逻辑由后端对应开关控制。
 */
 /**
 * 积分系统 UI 开关（默认 false = 隐藏）
 * - false：隐藏所有积分相关入口/余额/消耗提示/不足弹窗/充值入口；会员标识保留；
 *   功能流程不做积分预校验，直接走生成。
 * - true：展示完整积分系统 UI。
 */
 export const ENABLE_CREDIT_SYSTEM = false
@@ -3,7 +3,6 @@
 * Header.tsx 和 Sidebar.tsx 共享此数据源，避免路由配置重复
 */
 import React from "react"
 import { ENABLE_CREDIT_SYSTEM } from "./features"
 import {
  DashboardOutlined,
  FileOutlined,
@@ -106,17 +105,12 @@ export const NAV_ITEMS: NavItem[] = [
    path: "/app/subscription",
    icon: React.createElement(CrownOutlined),
  },
-  // 积分系统开关关闭时隐藏积分中心入口（代码保留不删除）
+  {
-  ...(ENABLE_CREDIT_SYSTEM
+    key: "points",
-    ? [
+    label: "积分中心",
-        {
+    path: "/app/points",
-          key: "points",
+    icon: React.createElement(ThunderboltOutlined),
-          label: "积分中心",
+  },
          path: "/app/points",
          icon: React.createElement(ThunderboltOutlined),
        },
      ]
    : []),
 ]
 /** 侧边栏导航分组（Sidebar 分组列表使用） */
@@ -206,17 +200,12 @@ export const NAV_GROUPS: NavGroup[] = [
        path: "/app/subscription",
        icon: React.createElement(CrownOutlined),
      },
-      // 积分系统开关关闭时隐藏积分中心入口（代码保留不删除）
+      {
-      ...(ENABLE_CREDIT_SYSTEM
+        key: "points",
-        ? [
+        label: "积分中心",
-            {
+        path: "/app/points",
-              key: "points",
+        icon: React.createElement(ThunderboltOutlined),
-              label: "积分中心",
+      },
              path: "/app/points",
              icon: React.createElement(ThunderboltOutlined),
            },
          ]
        : []),
    ],
  },
 ]
@@ -44,8 +44,7 @@ export const createLipsyncJob = async (data: {
  enable_video_loop?: boolean
  project_id?: string
 }): Promise<LipsyncJob> => {
-  // GPU 口型同步推理约 20s，留足余量到 120s 防止 10s 默认超时
+  const response = await apiClient.post<LipsyncJob>("/lipsync/jobs", data)
  const response = await apiClient.post<LipsyncJob>("/lipsync/jobs", data, { timeout: 120_000 })
  return response.data
 }
@@ -33,7 +33,6 @@ import { getAssetsByKind } from "@/api/assets"
 import { previewTts } from "@/api/tts"
 import { usePointsStore } from "@/store/pointsStore"
 import { hasEnoughPoints } from "./hooks/pointsCost"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 import "./generate.css"
 import "./generate-points.css"
@@ -438,22 +437,19 @@ const GeneratePage: React.FC = () => {
  /* ── 步骤3「确认生成视频」：校验通过 → 创建正式生成任务 → 跳步骤4看实时进展 ── */
  const handleConfirmGenerate = useCallback(async () => {
-    // 积分预检查（积分系统关闭时跳过，直接走生成流程）
+    // 积分预检查
-    let check: ReturnType<typeof hasEnoughPoints> = { sufficient: true, cost: 0 }
+    const units = isBatch ? Math.max(selectedVariantIds.length, 1) : 1
-    if (ENABLE_CREDIT_SYSTEM) {
+    const check = hasEnoughPoints(
-      const units = isBatch ? Math.max(selectedVariantIds.length, 1) : 1
+      balance ?? null,
-      check = hasEnoughPoints(
+      units,
-        balance ?? null,
+      dailyUsage ?? null,
-        units,
+      [],
-        dailyUsage ?? null,
+      "free",
-        [],
+      rules?.free_user_multiplier ?? 1.15,
-        "free",
+    )
-        rules?.free_user_multiplier ?? 1.15,
+    if (!check.sufficient) {
-      )
+      message.error(check.reason ?? "积分不足，请充值")
-      if (!check.sufficient) {
+      return
        message.error(check.reason ?? "积分不足，请充值")
        return
      }
    }
    if (isBatch) {
      if (selectedVariantIds.length === 0) {
@@ -524,18 +520,19 @@ const GeneratePage: React.FC = () => {
  /* ── 积分消耗估算（步骤3确认生成展示用） ── */
  const unitsForCost = isBatch ? Math.max(selectedVariantIds.length, 1) : 1
-  const pointsEstimate = useMemo(() => {
+  const pointsEstimate = useMemo(
-    if (!ENABLE_CREDIT_SYSTEM) return { sufficient: true, cost: 0 }
+    () =>
-    return hasEnoughPoints(
+      hasEnoughPoints(
-      balance ?? null,
+        balance ?? null,
-      unitsForCost,
+        unitsForCost,
-      dailyUsage ?? null,
+        dailyUsage ?? null,
-      [],
+        [],
-      "free",
+        "free",
-      rules?.free_user_multiplier ?? 1.15,
+        rules?.free_user_multiplier ?? 1.15,
-    )
+      ),
-  }, [unitsForCost, balance, dailyUsage, rules])
+    [unitsForCost, balance, dailyUsage, rules],
-  const insufficientPoints = ENABLE_CREDIT_SYSTEM && !pointsEstimate.sufficient
+  )
  const insufficientPoints = !pointsEstimate.sufficient
  /* ================================================================
     渲染
@@ -39,7 +39,6 @@ import { getDiscountPriceCents } from "@/api/points/types"
 import type { SubscriptionPlan } from "@/api/subscription/types"
 import { PLAN_LABEL, BILLING_CYCLE_LABEL } from "@/api/subscription/types"
 import "./Plans.css"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 const { Title, Text, Paragraph } = Typography
@@ -250,23 +249,17 @@ const Plans: React.FC = () => {
  return (
    <div className="xx-plans-page">
      <PageHead
-        title={ENABLE_CREDIT_SYSTEM ? "会员与积分" : "会员订阅"}
+        title="会员与积分"
-        description={
+        description="开通会员解锁全部功能，按需充值积分灵活使用 AI 能力"
          ENABLE_CREDIT_SYSTEM
            ? "开通会员解锁全部功能，按需充值积分灵活使用 AI 能力"
            : "开通会员解锁全部功能"
        }
        actions={
-          ENABLE_CREDIT_SYSTEM ? (
+          <Space>
-            <Space>
+            <Button
-              <Button
+              icon={<ThunderboltOutlined />}
-                icon={<ThunderboltOutlined />}
+              onClick={() => navigate("/app/points/transactions")}
-                onClick={() => navigate("/app/points/transactions")}
+            >
-              >
+              积分明细
-                积分明细
+            </Button>
-              </Button>
+          </Space>
            </Space>
          ) : null
        }
      />
@@ -303,15 +296,13 @@ const Plans: React.FC = () => {
                  )}
                </div>
              </div>
-              {ENABLE_CREDIT_SYSTEM && (
+              <div>
-                <div>
+                <Text type="secondary">可用积分</Text>
-                  <Text type="secondary">可用积分</Text>
+                <div className="xx-current-balance">
-                  <div className="xx-current-balance">
+                  <ThunderboltOutlined style={{ color: "#8b5cf6" }} />
-                    <ThunderboltOutlined style={{ color: "#8b5cf6" }} />
+                  <span className="xx-current-balance-val">{bal}</span>
                    <span className="xx-current-balance-val">{bal}</span>
                  </div>
                </div>
-              )}
+              </div>
              {!isMember && freeLimit > 0 && (
                <div>
                  <Text type="secondary">今日免费混剪</Text>
@@ -328,20 +319,18 @@ const Plans: React.FC = () => {
              )}
            </Space>
          </Col>
-          {ENABLE_CREDIT_SYSTEM && (
+          <Col>
-            <Col>
+            <Button
-              <Button
+              type="primary"
-                type="primary"
+              icon={<ThunderboltOutlined />}
-                icon={<ThunderboltOutlined />}
+              onClick={() => {
-                onClick={() => {
+                const el = document.getElementById("points-packages")
-                  const el = document.getElementById("points-packages")
+                el?.scrollIntoView({ behavior: "smooth" })
-                  el?.scrollIntoView({ behavior: "smooth" })
+              }}
-                }}
+            >
-              >
+              充值积分
-                充值积分
+            </Button>
-              </Button>
+          </Col>
            </Col>
          )}
        </Row>
      </Card>
@@ -472,71 +461,69 @@ const Plans: React.FC = () => {
        </Col>
      </Row>
-      {/* 积分充值（积分系统关闭时隐藏，代码保留不删除） */}
+      {/* 积分充值 */}
-      {ENABLE_CREDIT_SYSTEM && (
+      <div id="points-packages">
-        <div id="points-packages">
+        <Title level={4} style={{ marginTop: 40 }}>
-          <Title level={4} style={{ marginTop: 40 }}>
+          <ThunderboltOutlined style={{ color: "#8b5cf6", marginRight: 8 }} />
-            <ThunderboltOutlined style={{ color: "#8b5cf6", marginRight: 8 }} />
+          积分充值
-            积分充值
+          <Tooltip title="积分永久有效，可用于所有 AI 功能；付费会员享折扣">
-            <Tooltip title="积分永久有效，可用于所有 AI 功能；付费会员享折扣">
+            <Text type="secondary" style={{ fontSize: 13, marginLeft: 8, fontWeight: "normal" }}>
-              <Text type="secondary" style={{ fontSize: 13, marginLeft: 8, fontWeight: "normal" }}>
+              （永久有效）
-                （永久有效）
+            </Text>
-              </Text>
+          </Tooltip>
-            </Tooltip>
+        </Title>
          </Title>
-          <Row gutter={[16, 16]}>
+        <Row gutter={[16, 16]}>
-            {packages.map((pkg) => {
+          {packages.map((pkg) => {
-              const priceCents = getDiscountPriceCents(pkg, userDiscount)
+            const priceCents = getDiscountPriceCents(pkg, userDiscount)
-              const originalCents = pkg.price_cents
+            const originalCents = pkg.price_cents
-              const discount =
+            const discount =
-                priceCents < originalCents ? Math.round((1 - priceCents / originalCents) * 100) : 0
+              priceCents < originalCents ? Math.round((1 - priceCents / originalCents) * 100) : 0
-              const unit = priceCents / 100 / pkg.points
+            const unit = priceCents / 100 / pkg.points
-              const isHot = pkg.unit_price < 0.1
+            const isHot = pkg.unit_price < 0.1
-              return (
+            return (
-                <Col xs={24} sm={8} key={pkg.code}>
+              <Col xs={24} sm={8} key={pkg.code}>
-                  <Card
+                <Card
-                    className={`xx-pkg-card ${discount > 0 ? "has-discount" : ""} ${isHot ? "recommended" : ""}`}
+                  className={`xx-pkg-card ${discount > 0 ? "has-discount" : ""} ${isHot ? "recommended" : ""}`}
-                    hoverable
+                  hoverable
-                  >
+                >
-                    {isHot && <div className="xx-pkg-badge">热门</div>}
+                  {isHot && <div className="xx-pkg-badge">热门</div>}
                  {discount > 0 && (
                    <Tag color="gold" className="xx-pkg-discount">
                      {Math.round((priceCents / originalCents) * 10) / 1}折
                    </Tag>
                  )}
                  <div className="xx-pkg-name">{pkg.name}</div>
                  <div className="xx-pkg-points">
                    <ThunderboltOutlined /> {pkg.points.toLocaleString()} 积分
                  </div>
                  <div className="xx-pkg-price">
                    <span className="currency">¥</span>
                    <span className="amount">
                      {(priceCents / 100)
                        .toFixed(priceCents % 100 === 0 ? 0 : 1)
                        .replace(/\.0$/, "")}
                    </span>
                    {discount > 0 && (
-                      <Tag color="gold" className="xx-pkg-discount">
+                      <span className="xx-pkg-origin">¥{(originalCents / 100).toFixed(0)}</span>
                        {Math.round((priceCents / originalCents) * 10) / 1}折
                      </Tag>
                    )}
-                    <div className="xx-pkg-name">{pkg.name}</div>
+                  </div>
-                    <div className="xx-pkg-points">
+                  <div className="xx-pkg-unit">≈¥{unit.toFixed(3)}/积分</div>
-                      <ThunderboltOutlined /> {pkg.points.toLocaleString()} 积分
+                  <Button
-                    </div>
+                    block
-                    <div className="xx-pkg-price">
+                    type={isHot ? "primary" : "default"}
-                      <span className="currency">¥</span>
+                    loading={buying === pkg.code}
-                      <span className="amount">
+                    onClick={() => handleBuyPoints(pkg)}
-                        {(priceCents / 100)
+                    style={{ marginTop: 12 }}
-                          .toFixed(priceCents % 100 === 0 ? 0 : 1)
+                  >
-                          .replace(/\.0$/, "")}
+                    立即购买
-                      </span>
+                  </Button>
-                      {discount > 0 && (
+                </Card>
-                        <span className="xx-pkg-origin">¥{(originalCents / 100).toFixed(0)}</span>
+              </Col>
-                      )}
+            )
-                    </div>
+          })}
-                    <div className="xx-pkg-unit">≈¥{unit.toFixed(3)}/积分</div>
+        </Row>
-                    <Button
+      </div>
                      block
                      type={isHot ? "primary" : "default"}
                      loading={buying === pkg.code}
                      onClick={() => handleBuyPoints(pkg)}
                      style={{ marginTop: 12 }}
                    >
                      立即购买
                    </Button>
                  </Card>
                </Col>
              )
            })}
          </Row>
        </div>
      )}
    </div>
  )
 }
@@ -8,7 +8,6 @@
 * - subscription: GET /subscription/current（plan_id + billing_cycle）
 */
 import { create } from "zustand"
 import { ENABLE_CREDIT_SYSTEM } from "@/config/features"
 import { getPointsBalance, getPointsRules, getDailyUsage, getMembership } from "@/api/points"
 import { getCurrentSubscription } from "@/api/subscription"
 import type {
@@ -50,29 +49,14 @@ export const usePointsStore = create<PointsState>((set, get) => ({
  init: async () => {
    // 已加载过不重复拉取
-    // 积分系统关闭时：只要 subscription/membership 已有值就跳过；开启时需 balance+rules+subscription 齐了才跳过
+    if (get().balance && get().rules && get().subscription) return
    if (ENABLE_CREDIT_SYSTEM) {
      if (get().balance && get().rules && get().subscription) return
    } else {
      if (get().subscription && get().membership) return
    }
    set({ loading: true, error: null })
    try {
      // 积分系统关闭时不拉取余额/规则/每日额度，但仍拉会员/订阅用于 VIP 标识展示
      const balancePromise = ENABLE_CREDIT_SYSTEM
        ? getPointsBalance().catch(() => null)
        : Promise.resolve(null)
      const rulesPromise = ENABLE_CREDIT_SYSTEM
        ? getPointsRules().catch(() => null)
        : Promise.resolve(null)
      const dailyUsagePromise = ENABLE_CREDIT_SYSTEM
        ? getDailyUsage().catch(() => null)
        : Promise.resolve(null)
      const [balance, rules, subscription, dailyUsage, membership] = await Promise.all([
-        balancePromise,
+        getPointsBalance().catch(() => null),
-        rulesPromise,
+        getPointsRules().catch(() => null),
        getCurrentSubscription().catch(() => null),
-        dailyUsagePromise,
+        getDailyUsage().catch(() => null),
        getMembership().catch(() => null),
      ])
      set({
@@ -1,173 +1,87 @@
-# MuseTalk GPU Worker 部署指南
+# MuseTalk GPU Worker — 部署指南
-本目录包含两个组件：
+本目录包含 RTX2060 本地电脑上运行的 GPU Worker 脚本。
 Worker 采用 **反向轮询模式**：主动向 SaaS API 拉取待处理的口型同步任务 → 调用本地 MuseTalk 推理 → 把结果视频回传到 SaaS。不需要内网穿透。
-1. **gpu_worker.py**：反向轮询客户端，部署在 RTX2060 本地，轮询 SaaS API 拉取口型任务，调用本地 MuseTalk 服务推理，上传结果回 SaaS。
+## 目录文件
 2. **musetalk_server.py**：MuseTalk Flask HTTP 服务端，接收 gpu_worker.py 的推理请求，调用 MuseTalk 模型生成口型同步视频。
---
+| 文件 | 作用 |
 |---|---|
 | `gpu_worker.py` | Worker 主程序（单文件，零项目代码依赖，仅依赖 `requests`） |
 | `requirements.txt` | Python 依赖（只有 `requests`） |
 | `xiaoxia-gpu-worker.service` | systemd 服务单元（开机自启、异常自动重启） |
 | `.env.example` | 环境变量样例，复制为 `.env` 后填入真实值 |
 ## 一、环境准备
-### 1.1 硬件要求
+1. **Python 3.10+**（Windows 建议从 python.org 安装；Linux 自带）
 2. **本地 MuseTalk 服务** 已启动在 `http://127.0.0.1:7861`，health 接口返回 `{"status":"ok","free_vram_mb":...}`
 3. **ffmpeg**（可选，用于读取输出视频时长；未装则 duration 报 0，不影响功能）
 4. 网络能访问 staging / 生产 API（`curl https://staging-api.xiaoxiajianji.com/health` 应返回 `{"status":"healthy"}`）
- GPU: NVIDIA RTX 2060 或更高（显存 ≥ 6GB）
+## 二、部署步骤（Linux，推荐 systemd）
 - CUDA: 11.8+
 - Python: 3.10+
 - ffmpeg: 需安装并加入 PATH
 ### 1.2 安装依赖
 ```bash
-cd deploy/gpu_worker
+# 1. 创建部署目录
 sudo mkdir -p /opt/xiaoxia-gpu-worker
 sudo chown $USER:$USER /opt/xiaoxia-gpu-worker
 cd /opt/xiaoxia-gpu-worker
 # 2. 拷贝脚本和依赖
 cp /path/to/deploy/gpu_worker/{gpu_worker.py,requirements.txt,xiaoxia-gpu-worker.service,.env.example} .
 cp .env.example .env
 #    编辑 .env，填入 API_BASE_URL 和 GPU_WORKER_TOKEN
 # 3. 创建虚拟环境并安装依赖
 python3 -m venv venv
-source venv/bin/activate
+./venv/bin/pip install -r requirements.txt
-pip install -r requirements.txt
+
 # 4. 前台先跑一次，确认日志正常
 ./venv/bin/python gpu_worker.py
 #    看到 "MuseTalk 健康检查通过" 和 "注册/心跳" 成功即可 Ctrl+C 退出
 # 5. 安装 systemd 服务
 sudo cp xiaoxia-gpu-worker.service /etc/systemd/system/
 sudo systemctl daemon-reload
 sudo systemctl enable --now xiaoxia-gpu-worker
 # 6. 查看日志
 sudo journalctl -u xiaoxia-gpu-worker -f
 ```
---
+## 三、部署步骤（Windows，快速测试）
-## 二、MuseTalk 服务端部署（musetalk_server.py）
+```bat
 :: 创建虚拟环境
 python -m venv venv
 venv\Scripts\pip install -r requirements.txt
-### 2.1 配置环境变量
+:: 复制并编辑 .env
 copy .env.example .env
 notepad .env
-复制 `.env.example` 为 `.env`，修改配置：
+:: 运行
-
+venv\Scripts\python gpu_worker.py
 ```bash
 cp .env.example .env
 vim .env
 ```
-关键配置：
+可在任务计划程序中添加开机启动项：程序选 `venv\Scripts\python.exe`，参数填 `gpu_worker.py`，起始目录填脚本所在目录。
-| 变量 | 说明 | 默认值 |
+## 四、SaaS 侧配套配置
 |------|------|--------|
 | `MUSE_PORT` | 监听端口 | `7861` |
 | `MUSE_INFERENCE_TIMEOUT` | 推理超时秒数 | `600` |
 | `MUSE_VIDEO_MAX_MB` | 视频上传大小限制 MB | `100` |
 | `MUSE_AUDIO_MAX_MB` | 音频上传大小限制 MB | `20` |
 | `MUSE_DEFAULT_FPS` | 视频 fps 兜底值 | `25.0` |
 | `MUSE_TEMP_DIR` | 临时文件目录 | `/tmp/musetalk_$$` |
 | `MUSE_VIDEO_ENCODER` | 循环视频时的编码器：`auto`（优先 h264_nvenc，失败回退 libx264）/`h264_nvenc`/`libx264` | `auto` |
 | `MUSE_ENABLE_VIDEO_LOOP` | 驱动音频比视频长时循环视频补齐画面，`0` 关闭 | `1` |
-### 2.2 更新部署（音轨修复，必做）
+SaaS 后端部署完成后需配置：
-> ⚠️ 2026-09-20 修复严重 bug：旧版封装保留了源视频音轨，结果口型配的是原声而不是 TTS 驱动音频。RTX2060 机器必须重新拉取 `musetalk_server.py` 并重启：
+1. 服务端环境变量 `GPU_WORKER_TOKEN` 设为一个随机强 Token（和 Worker `.env` 中一致）
 2. 数据库已跑迁移 `081_add_gpu_lipsync_tasks`（自动随 API 启动的 alembic upgrade head 完成）
 3. OSS bucket 中 `gpu-lipsync/results/` 路径可写（默认 bucket 已配）
-```bash
+## 五、验证联调
 # 在 RTX2060 上备份旧文件并拉取新版本（按实际部署路径调整）
 cp ~/projects/MuseTalk/musetalk_server.py ~/projects/MuseTalk/musetalk_server.py.bak
 # 从仓库 raw 地址下载最新版（替换为你的仓库地址/分支）
 wget -O ~/projects/MuseTalk/musetalk_server.py \
  "https://git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas/raw/branch/develop/deploy/gpu_worker/musetalk_server.py"
-# 重启服务
+1. Worker 启动后日志看到 `注册/心跳` 成功
-sudo systemctl restart musetalk-server
+2. 后端调用 `GpuLipsyncService.create_task(video_url=..., audio_url=...)` 放入一条测试任务
-sudo systemctl status musetalk-server
+3. Worker 在 5 秒内拉到任务，下载 → 推理 → 上传 → 上报
-curl http://127.0.0.1:7861/health
+4. 后端 `GET /api/v1/gpu/lipsync/status/{task_id}` 返回 `status=done`，`result_url` 非空
 ```
-修复后封装逻辑：
+## 六、故障排查
 - 最终 mux 强制 `-map 0:v -map 1:a`：视频流只取 MuseTalk 无声画面，音轨只取 TTS 驱动音频，杜绝 ffmpeg 默认行为带入源视频音轨
 - 驱动音频不长于视频时：`-c:v copy -c:a aac -shortest`，无损秒封装
 - 驱动音频长于视频时（如 TTS 15s vs 视频 9s）：`-stream_loop -1` 循环画面，RTX2060 走 `h264_nvenc` 硬件重编码（NVENC 失败自动回退 libx264），`-t` 精确卡到音频时长
 ### 2.3 启动服务
 ```bash
 # 前台运行（调试用）
 python musetalk_server.py
 # 后台运行（生产用 systemd）
 sudo systemctl start musetalk-server
 sudo systemctl enable musetalk-server
 ```
 ### 2.4 验证健康检查
 ```bash
 curl http://127.0.0.1:7861/health
 ```
 应返回：
 ```json
 {
  "status": "healthy",
  "gpu": {
    "gpu_name": "NVIDIA GeForce RTX 2060",
    "memory_total_mb": 6144,
    "memory_used_mb": 1024,
    "memory_free_mb": 5120
  },
  "current_task": {
    "task_id": null,
    "running": false,
    "elapsed_seconds": 0.0
  },
  "timestamp": 1700000000.0
 }
 ```
 ---
 ## 三、GPU Worker 客户端部署（gpu_worker.py）
 ### 3.1 配置环境变量
 复制 `.env.example` 为 `.env`，修改配置：
 ```bash
 cp .env.example .env
 vim .env
 ```
 关键配置：
 | 变量 | 说明 | 默认值 |
 |------|------|--------|
 | `API_BASE_URL` | SaaS API 基础 URL | `https://staging-api.xiaoxiajianji.com` |
 | `GPU_WORKER_TOKEN` | 长期 API Token（与服务端一致） | - |
 | `MUSE_TALK_URL` | 本地 MuseTalk 服务地址 | `http://127.0.0.1:7861` |
 | `POLL_INTERVAL` | 轮询间隔秒 | `5` |
 | `HEARTBEAT_INTERVAL` | 空闲心跳间隔秒 | `15` |
 | `REQUEST_TIMEOUT` | HTTP 请求超时秒 | `900` |
 | `TASK_MAX_RETRY` | 本地最大重试次数 | `1` |
 | `TASK_HEARTBEAT_INTERVAL` | 推理期间任务心跳间隔秒 | `30` |
 | `MIN_VIDEO_DURATION_SECONDS` | 最短输入视频时长秒 | `3` |
 ### 3.2 启动 Worker
 ```bash
 # 前台运行（调试用）
 python gpu_worker.py
 # 后台运行（生产用 systemd）
 sudo systemctl start xiaoxia-gpu-worker
 sudo systemctl enable xiaoxia-gpu-worker
 ```
 ### 3.3 验证启动日志
 应看到：
 ```
 ============================================================
 MuseTalk GPU Worker 启动
  worker_id   = rtx2060-xxxx
  api_base    = https://staging-api.xiaoxiajianji.com
  muse_talk   = http://127.0.0.1:7861
  poll        = 5.0s / heartbeat = 15.0s
 ============================================================
 MuseTalk 健康检查通过: {...}
 注册/心跳成功
 ```
 ---
 ## 四、常见问题排查
 | 现象 | 可能原因 / 排查 |
 |---|---|
@@ -178,40 +92,9 @@ MuseTalk 健康检查通过: {...}
 | 服务端看到任务回退到 pending 重试 | 任务心跳真正超时（默认 900s）：Worker 进程崩溃/断网，或推理彻底卡死；正常长推理期间心跳线程每 30s 续期，不会回退 |
 | 日志 `MuseTalk 推理超时或连接失败` | 视频太长或显存不足；可临时调大 REQUEST_TIMEOUT（服务端 GPU_TASK_TIMEOUT_SECONDS 需同步调大），或限制输入视频时长 |
 | 日志 `视频过短（x.xxs < 3s）` | 输入视频不足 3s，MuseTalk 对短视频会 division by zero，已在本地直接上报失败；可用 MIN_VIDEO_DURATION_SECONDS 调整阈值 |
 | MuseTalk 服务端 503 `GPU 正在处理其他任务` | 并发请求被锁拒绝，等当前推理完成即可 |
 | MuseTalk 服务端 504 `推理超时` | 推理超过 MUSE_INFERENCE_TIMEOUT，客户端会调 /cancel 终止服务端任务 |
---
+## 七、安全注意事项
 ## 五、安全注意事项
 - `.env` 包含长期 Token，文件权限设为 600（`chmod 600 .env`）
 - Token 泄露要立即在服务端更换 `GPU_WORKER_TOKEN` 并重启 Worker
 - Worker 只需要出站访问 SaaS API 和 OSS，不需要开放任何入站端口
 - MuseTalk 服务端只监听本地 127.0.0.1（或 0.0.0.0 但通过防火墙限制），不暴露到公网
 - 临时文件自动清理（推理完成/失败后），无需手动维护
 ---
 ## 六、工程改进记录（musetalk_server.py）
 相比原 `worker.py`，修复了以下 8 个 bug：
 1. **Flask 单线程阻塞**：`app.run(threaded=True)`，推理时 `/health` 仍可响应
 2. **fps=0 除零崩溃**：`_get_video_fps()` 兜底 `MUSE_DEFAULT_FPS`
 3. **ffmpeg 不检查返回码**：`subprocess.run(check=True)` + 超时检查，失败立即报错
 4. **无并发锁**：`threading.Lock` 控制并发，第二请求立即 503
 5. **无推理超时**：线程 join timeout，超时返回 504 并调 `/cancel`
 6. **结果文件不清理**：推理完成/失败后自动删除临时目录
 7. **无人脸检测兜底**：MuseTalk 推理内部处理（TODO: 可在 `_run_inference` 前置检查）
 8. **上传无大小限制**：`_check_file_size()` 校验，超限返回 413
 新增：
 - `/cancel` 端点：终止当前推理任务，清理临时文件
 - `/health` 端点：返回 GPU 显存信息和当前任务状态
 2026-09-20 追加修复（音轨正确性，上线阻断级）：
 9. **音轨未替换（严重）**：旧最终封装让 ffmpeg 默认选流，结果保留了源视频自带音轨（与画面相关系数 0.9998，与 TTS 无关）。改为 `_mux_video_with_audio()` 统一封装，强制 `-map 0:v:0 -map 1:a:0`，画面取 MuseTalk 无声产物、音轨只取驱动音频
 10. **音视频时长不对齐**：TTS 长于原视频时 `-shortest` 会截短语音。改为探测双方时长，音频更长时 `-stream_loop -1` 循环画面 + `h264_nvenc` 硬件重编码（`MUSE_VIDEO_ENCODER=auto`，失败回退 libx264）+ `-t <音频时长>`；不循环时 `-c:v copy` 秒封装
    - 开关 `MUSE_ENABLE_VIDEO_LOOP=0` 可关闭循环；请求也支持 form 参数 `enable_video_loop` 单任务覆盖
@@ -229,26 +229,12 @@ def _call_musetalk(video_path: Path, audio_path: Path, out_path: Path) -> tuple[
        duration = _probe_duration(out_path)
        return True, duration, "", False
    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
-        # 瞬时网络/超时错误，允许本地重试 1 次；同时调 /cancel 让服务端终止僵尸推理
+        # 瞬时网络/超时错误，允许本地重试 1 次
        _cancel_musetalk()
        return False, 0.0, f"MuseTalk 推理超时或连接失败（>{Config.request_timeout}s）", True
    except Exception as exc:
        return False, 0.0, f"MuseTalk 调用异常: {exc}", False
 def _cancel_musetalk() -> None:
    """调 MuseTalk /cancel 端点终止服务端僵尸推理进程，避免超时后任务还在跑占显存."""
    try:
        r = requests.post(f"{Config.muse_talk_url}/cancel", timeout=10)
        if r.status_code == 200:
            logger.info("已调 MuseTalk /cancel，服务端终止推理")
        else:
            logger.warning("MuseTalk /cancel 返回 %d: %s", r.status_code, r.text[:200])
    except Exception as exc:
        # /cancel 失败不应影响主流程上报
        logger.warning("调 MuseTalk /cancel 异常（忽略）: %s", exc)
 def _probe_duration(path: Path) -> float:
    """用 ffprobe 读视频时长（若系统装了 ffmpeg）；否则返回 0."""
    try:
@@ -1,602 +0,0 @@
 """MuseTalk Flask HTTP 服务 — 反向轮询架构的服务端部分.
 部署在 RTX2060 本地，接收 gpu_worker.py 的推理请求，调用 MuseTalk 生成口型同步视频。
 本文件修复了原 worker.py 的 8 个工程 bug，并新增 /cancel 端点。
 环境变量：
  MUSE_PORT            监听端口，默认 7861
  MUSE_MAX_CONCURRENT  最大并发推理数，默认 1（GPU 一次只能处理一个）
  MUSE_INFERENCE_TIMEOUT 推理超时秒数，默认 600
  MUSE_VIDEO_MAX_MB    视频上传大小限制 MB，默认 100
  MUSE_AUDIO_MAX_MB    音频上传大小限制 MB，默认 20
  MUSE_DEFAULT_FPS     视频 fps 兜底值，默认 25.0
  MUSE_TEMP_DIR        临时文件目录，默认 /tmp/musetalk_$$
  MUSE_VIDEO_ENCODER   循环视频时的编码器：auto（默认，优先 h264_nvenc 兜底 libx264）/h264_nvenc/libx264
  MUSE_ENABLE_VIDEO_LOOP 驱动音频比视频长时是否循环视频补齐，默认 1（开启）
 接口：
  GET  /health          健康检查 + GPU 显存信息
  POST /inference       推理请求（multipart: video + audio）
  POST /cancel          终止当前推理任务
 """
 from __future__ import annotations
 import atexit
 import logging
 import os
 import shutil
 import signal
 import subprocess
 import threading
 import time
 from pathlib import Path
 from typing import Optional
 from flask import Flask, jsonify, request, send_file
 # ── 日志 ──────────────────────────────────────────────────────────────
 logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger("musetalk-server")
 # ── 配置 ──────────────────────────────────────────────────────────────
 def _env(name: str, default: str = "") -> str:
    v = os.environ.get(name, default)
    return v.strip() if isinstance(v, str) else default
 class Config:
    port: int = int(_env("MUSE_PORT", "7861"))
    max_concurrent: int = int(_env("MUSE_MAX_CONCURRENT", "1"))
    inference_timeout: float = float(_env("MUSE_INFERENCE_TIMEOUT", "600"))
    video_max_mb: int = int(_env("MUSE_VIDEO_MAX_MB", "100"))
    audio_max_mb: int = int(_env("MUSE_AUDIO_MAX_MB", "20"))
    default_fps: float = float(_env("MUSE_DEFAULT_FPS", "25.0"))
    temp_dir: str = _env("MUSE_TEMP_DIR", f"/tmp/musetalk_{os.getpid()}")
    # 循环视频时编码器：auto 优先 h264_nvenc（RTX2060 支持），失败兜底 libx264
    video_encoder: str = _env("MUSE_VIDEO_ENCODER", "auto") or "auto"
    # 驱动音频比视频长时循环视频补齐画面
    enable_video_loop: bool = _env("MUSE_ENABLE_VIDEO_LOOP", "1") not in ("0", "false", "False", "")
    # 判定音视频时长差异的容差（秒），避免 ffprobe 微小误差触发无谓的循环/重编码
    duration_epsilon: float = 0.25
 # ── 全局状态 ──────────────────────────────────────────────────────────
 inference_lock = threading.Lock()
 current_task: dict = {"task_id": None, "process": None, "start_time": 0.0}
 shutdown_event = threading.Event()
 # ── Flask App ─────────────────────────────────────────────────────────
 app = Flask(__name__)
 def _cleanup_temp_dir():
    """退出时清理临时目录."""
    if os.path.exists(Config.temp_dir):
        try:
            shutil.rmtree(Config.temp_dir)
            logger.info("已清理临时目录: %s", Config.temp_dir)
        except Exception as exc:
            logger.warning("清理临时目录失败: %s", exc)
 atexit.register(_cleanup_temp_dir)
 def _signal_handler(signum, frame):
    """优雅退出."""
    logger.info("收到信号 %s，准备退出...", signum)
    shutdown_event.set()
    if current_task["process"]:
        logger.info("终止正在进行的推理进程...")
        try:
            current_task["process"].terminate()
            current_task["process"].wait(timeout=5)
        except Exception:
            pass
    _cleanup_temp_dir()
    exit(0)
 signal.signal(signal.SIGTERM, _signal_handler)
 signal.signal(signal.SIGINT, _signal_handler)
 # ── 工具函数 ──────────────────────────────────────────────────────────
 def _get_gpu_info() -> dict:
    """获取 GPU 显存信息（通过 nvidia-smi）."""
    try:
        out = subprocess.check_output(
            [
                "nvidia-smi",
                "--query-gpu=name,memory.total,memory.used,memory.free",
                "--format=csv,noheader,nounits",
            ],
            stderr=subprocess.DEVNULL,
            timeout=5,
        )
        parts = out.decode().strip().split(",")
        if len(parts) >= 4:
            return {
                "gpu_name": parts[0].strip(),
                "memory_total_mb": int(parts[1].strip()),
                "memory_used_mb": int(parts[2].strip()),
                "memory_free_mb": int(parts[3].strip()),
            }
    except Exception as exc:
        logger.warning("nvidia-smi 失败: %s", exc)
    return {"gpu_name": "unknown", "memory_total_mb": 0, "memory_used_mb": 0, "memory_free_mb": 0}
 def _get_video_fps(video_path: Path) -> float:
    """用 ffprobe 读视频帧率，失败或为 0 时返回 default_fps."""
    try:
        out = subprocess.check_output(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=r_frame_rate",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(video_path),
            ],
            stderr=subprocess.DEVNULL,
            timeout=10,
        )
        fps_str = out.decode().strip()
        if "/" in fps_str:
            num, den = fps_str.split("/")
            fps = float(num) / float(den) if float(den) != 0 else 0.0
        else:
            fps = float(fps_str) if fps_str else 0.0
        return fps if fps > 0 else Config.default_fps
    except Exception as exc:
        logger.warning("ffprobe 读 fps 失败: %s，使用默认 %.1f", exc, Config.default_fps)
        return Config.default_fps
 def _get_media_duration(path: Path) -> float:
    """用 ffprobe 读媒体时长（秒），失败返回 0.0."""
    try:
        out = subprocess.check_output(
            [
                "ffprobe",
                "-v",
                "error",
                "-show_entries",
                "format=duration",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(path),
            ],
            stderr=subprocess.DEVNULL,
            timeout=10,
        )
        duration = float(out.decode().strip())
        return duration if duration > 0 else 0.0
    except Exception as exc:
        logger.warning("ffprobe 读时长失败 %s: %s", path, exc)
        return 0.0
 def _pick_video_encoder() -> str:
    """选择视频编码器：配置指定则用指定值；auto 时探测 NVENC 是否可用，不可用回退 libx264."""
    configured = Config.video_encoder.strip()
    if configured in ("h264_nvenc", "libx264"):
        return configured
    # auto：探测本机 ffmpeg 是否编译了 h264_nvenc
    try:
        result = subprocess.run(
            ["ffmpeg", "-hide_banner", "-encoders"],
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            timeout=10,
            check=False,
        )
        if b"h264_nvenc" in result.stdout:
            return "h264_nvenc"
    except Exception as exc:
        logger.warning("探测 ffmpeg 编码器失败，回退 libx264: %s", exc)
    return "libx264"
 def _mux_video_with_audio(
    video_path: Path,
    audio_path: Path,
    output_path: Path,
    enable_video_loop: Optional[bool] = None,
    timeout: float = 300,
 ) -> None:
    """把无声画面视频与驱动音频封装为最终结果.
    关键正确性要求：必须用 -map 0:v -map 1:a 显式指定取第一个输入（推理画面）的
    视频流和第二个输入（驱动音频 TTS）的音频流，禁止 ffmpeg 默认流选择行为
    （否则会把源视频自带音轨带进结果，口型与声音错位）。
    时长对齐：驱动音频比视频长时（TTS 15s vs 原视频 9s 很常见），用
    -stream_loop -1 循环视频画面到音频长度（NVENC 硬件重编码），-t 卡到音频时长；
    音频不超过视频时直接 -c:v copy 无损快封装，-shortest 以较短流为准。
    """
    video_duration = _get_media_duration(video_path)
    audio_duration = _get_media_duration(audio_path)
    loop_enabled = Config.enable_video_loop if enable_video_loop is None else enable_video_loop
    need_loop = bool(
        loop_enabled
        and audio_duration > 0
        and video_duration > 0
        and audio_duration > video_duration + Config.duration_epsilon
    )
    if need_loop:
        encoder = _pick_video_encoder()
        # preset 随编码器选择：h264_nvenc 用 p1-p7，libx264 用词形 preset
        preset = "p4" if encoder == "h264_nvenc" else "veryfast"
        logger.info(
            "音频(%.2fs)长于视频(%.2fs)，循环视频并以 %s(%s) 重编码至音频长度",
            audio_duration,
            video_duration,
            encoder,
            preset,
        )
        def build_cmd(enc: str, pre: str) -> list:
            return [
                "ffmpeg",
                "-y",
                "-stream_loop",
                "-1",
                "-i",
                str(video_path),
                "-i",
                str(audio_path),
                "-map",
                "0:v:0",
                "-map",
                "1:a:0",
                "-c:v",
                enc,
                "-preset",
                pre,
                "-c:a",
                "aac",
                "-b:a",
                "128k",
                "-t",
                f"{audio_duration:.3f}",
                str(output_path),
            ]
        try:
            _run_ffmpeg(build_cmd(encoder, preset), timeout=timeout)
        except RuntimeError:
            # NVENC 可能因驱动/占用失败，兜底 libx264 重试一次
            if encoder == "h264_nvenc":
                logger.warning("h264_nvenc 封装失败，回退 libx264 重试")
                _run_ffmpeg(build_cmd("libx264", "veryfast"), timeout=timeout)
            else:
                raise
    else:
        # 视频不短于音频：直接复制视频流，只把音频替换为驱动音频并转 AAC
        cmd = [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-i",
            str(audio_path),
            "-map",
            "0:v:0",
            "-map",
            "1:a:0",
            "-c:v",
            "copy",
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            "-shortest",
            str(output_path),
        ]
        _run_ffmpeg(cmd, timeout=timeout)
 def _check_file_size(file, max_mb: int, label: str) -> Optional[str]:
    """检查文件大小，超限返回错误信息，否则返回 None."""
    file.seek(0, 2)
    size = file.tell()
    file.seek(0)
    max_bytes = max_mb * 1024 * 1024
    if size > max_bytes:
        return f"{label} 文件大小 {size / (1024*1024):.1f}MB 超过限制 {max_mb}MB"
    if size == 0:
        return f"{label} 文件为空"
    return None
 def _run_ffmpeg(cmd: list, timeout: float = 120) -> subprocess.CompletedProcess:
    """运行 ffmpeg 命令，检查返回码和超时."""
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=timeout,
            check=True,
        )
        return result
    except subprocess.CalledProcessError as exc:
        stderr = exc.stderr.decode(errors="ignore") if exc.stderr else ""
        raise RuntimeError(f"ffmpeg 失败 (code={exc.returncode}): {stderr[:500]}") from exc
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(f"ffmpeg 超时（>{timeout}s）") from exc
 def _run_inference(
    video_path: Path,
    audio_path: Path,
    output_path: Path,
    enable_video_loop: Optional[bool] = None,
 ) -> None:
    """执行 MuseTalk 推理（可被子线程和测试独立调用）.
    实际部署时替换为 MuseTalk 真实推理逻辑。
    此处为示例实现：提取帧 → 生成无声画面 → 用驱动音频封装。
    enable_video_loop: 驱动音频长于视频时是否循环视频；None 走全局配置。
    """
    fps = _get_video_fps(video_path)
    logger.info("视频 fps: %.2f", fps)
    frames_dir = video_path.parent / "frames"
    frames_dir.mkdir(parents=True, exist_ok=True)
    _run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-r",
            str(fps),
            str(frames_dir / "frame_%05d.png"),
        ],
        timeout=120,
    )
    frame_files = sorted(frames_dir.glob("*.png"))
    if not frame_files:
        raise RuntimeError("未从视频中提取到帧")
    # TODO: 替换为 MuseTalk 实际推理逻辑。
    # MuseTalk 真实产物是「无声画面视频」，音轨必须在封装阶段用驱动音频替换。
    logger.warning("使用示例推理逻辑，未实际调用 MuseTalk 模型")
    # 示例：从源视频生成无声画面（-an 丢弃原音轨），模拟 MuseTalk 推理产物。
    # 真实部署时 silent_video_path 应替换为 MuseTalk 输出的无声视频路径。
    silent_video_path = video_path.parent / "visual_silent.mp4"
    _run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-an",
            "-c:v",
            "libx264",
            "-preset",
            "veryfast",
            str(silent_video_path),
        ],
        timeout=300,
    )
    # 统一封装：显式 -map 取推理画面 + 驱动音频；音频更长时循环视频。
    _mux_video_with_audio(silent_video_path, audio_path, output_path, enable_video_loop=enable_video_loop)
    if not output_path.exists() or output_path.stat().st_size < 1024:
        raise RuntimeError("推理产物不存在或过小")
 # ── 路由 ──────────────────────────────────────────────────────────────
@app.route("/health", methods=["GET"])
 def health():
    """健康检查 + GPU 显存信息."""
    gpu_info = _get_gpu_info()
    task_info = {
        "task_id": current_task["task_id"],
        "running": current_task["process"] is not None,
        "elapsed_seconds": time.time() - current_task["start_time"] if current_task["start_time"] else 0.0,
    }
    return jsonify(
        {
            "status": "healthy",
            "gpu": gpu_info,
            "current_task": task_info,
            "timestamp": time.time(),
        }
    )
@app.route("/inference", methods=["POST"])
 def inference():
    """推理请求：multipart form 包含 video 和 audio 文件."""
    # 并发控制：检查锁
    if not inference_lock.acquire(blocking=False):
        return jsonify({"error": "GPU 正在处理其他任务，请稍后重试", "status": "busy"}), 503
    task_id = None
    video_path = None
    audio_path = None
    output_path = None
    try:
        # 解析参数
        if "video" not in request.files or "audio" not in request.files:
            return jsonify({"error": "缺少 video 或 audio 文件"}), 400
        video_file = request.files["video"]
        audio_file = request.files["audio"]
        task_id = request.form.get("task_id", f"task_{int(time.time())}")
        # 可选：本次任务是否在音频长于视频时循环视频（缺省走全局配置）
        loop_param = request.form.get("enable_video_loop")
        if loop_param is not None:
            task_enable_loop = loop_param.strip() not in ("0", "false", "False", "")
        else:
            task_enable_loop = None
        # 文件大小检查
        err = _check_file_size(video_file, Config.video_max_mb, "视频")
        if err:
            return jsonify({"error": err}), 413
        err = _check_file_size(audio_file, Config.audio_max_mb, "音频")
        if err:
            return jsonify({"error": err}), 413
        # 保存到临时目录
        task_dir = Path(Config.temp_dir) / task_id
        task_dir.mkdir(parents=True, exist_ok=True)
        video_path = task_dir / "input.mp4"
        audio_path = task_dir / "input_audio.wav"
        output_path = task_dir / "output.mp4"
        video_file.save(str(video_path))
        audio_file.save(str(audio_path))
        logger.info("开始推理 task_id=%s, video=%s, audio=%s", task_id, video_path.name, audio_path.name)
        # 更新当前任务信息
        current_task["task_id"] = task_id
        current_task["start_time"] = time.time()
        # 启动推理进程（用 subprocess 包装，便于超时终止）
        # 此处直接调用推理函数，实际可改为 subprocess 调用外部脚本
        current_task["process"] = "inference_thread"  # 标记为运行中
        # 在线程中运行推理（支持超时）
        result_container = {"error": None}
        def inference_thread():
            try:
                _run_inference(video_path, audio_path, output_path, enable_video_loop=task_enable_loop)
            except Exception as exc:
                result_container["error"] = str(exc)
        thread = threading.Thread(target=inference_thread)
        thread.start()
        thread.join(timeout=Config.inference_timeout)
        if thread.is_alive():
            # 超时，终止
            logger.error("推理超时 (>%ds)，终止任务 %s", Config.inference_timeout, task_id)
            return jsonify({"error": f"推理超时（>{Config.inference_timeout}s）", "task_id": task_id}), 504
        if result_container["error"]:
            logger.error("推理失败 task_id=%s: %s", task_id, result_container["error"])
            return jsonify({"error": result_container["error"], "task_id": task_id}), 500
        # 返回结果文件
        logger.info("推理完成 task_id=%s, output=%s", task_id, output_path)
        return send_file(str(output_path), mimetype="video/mp4", as_attachment=True, download_name=f"{task_id}.mp4")
    except Exception as exc:
        logger.exception("推理异常: %s", exc)
        return jsonify({"error": str(exc)}), 500
    finally:
        # 释放锁，清理当前任务信息
        inference_lock.release()
        current_task["task_id"] = None
        current_task["process"] = None
        current_task["start_time"] = 0.0
        # 清理临时文件
        if video_path and video_path.parent.exists():
            try:
                shutil.rmtree(video_path.parent)
                logger.info("已清理临时目录: %s", video_path.parent)
            except Exception as exc:
                logger.warning("清理临时目录失败: %s", exc)
@app.route("/cancel", methods=["POST"])
 def cancel():
    """终止当前正在进行的推理任务."""
    if current_task["task_id"] is None:
        return jsonify({"message": "当前无正在运行的任务"})
    task_id = current_task["task_id"]
    logger.info("收到取消请求，终止任务 %s", task_id)
    # 终止推理进程（如果是 subprocess）
    if current_task["process"] and current_task["process"] != "inference_thread":
        try:
            current_task["process"].terminate()
            current_task["process"].wait(timeout=5)
            logger.info("已终止推理进程")
        except Exception as exc:
            logger.warning("终止进程失败: %s", exc)
    # 清理临时文件
    task_dir = Path(Config.temp_dir) / task_id
    if task_dir.exists():
        try:
            shutil.rmtree(task_dir)
            logger.info("已清理临时目录: %s", task_dir)
        except Exception as exc:
            logger.warning("清理临时目录失败: %s", exc)
    # 重置当前任务
    current_task["task_id"] = None
    current_task["process"] = None
    current_task["start_time"] = 0.0
    return jsonify({"message": f"已取消任务 {task_id}"})
 # ── 主入口 ────────────────────────────────────────────────────────────
 def main():
    """启动 Flask 服务."""
    # 创建临时目录
    Path(Config.temp_dir).mkdir(parents=True, exist_ok=True)
    logger.info("临时目录: %s", Config.temp_dir)
    # 打印配置
    logger.info("=" * 60)
    logger.info("MuseTalk Flask Server 启动")
    logger.info("  端口: %d", Config.port)
    logger.info("  最大并发: %d", Config.max_concurrent)
    logger.info("  推理超时: %.0fs", Config.inference_timeout)
    logger.info("  视频大小限制: %dMB", Config.video_max_mb)
    logger.info("  音频大小限制: %dMB", Config.audio_max_mb)
    logger.info("  默认 fps: %.1f", Config.default_fps)
    logger.info("  视频编码器: %s", Config.video_encoder)
    logger.info("  音频长于视频时循环视频: %s", Config.enable_video_loop)
    logger.info("=" * 60)
    # 检查 GPU
    gpu_info = _get_gpu_info()
    logger.info("GPU 信息: %s", gpu_info)
    # 启动 Flask（threaded=True 处理并发请求）
    app.run(host="0.0.0.0", port=Config.port, threaded=True)
 if __name__ == "__main__":
    main()
@@ -8,7 +8,6 @@ API 和 Worker 各自的 Settings 类继承本类，只追加服务特有字段
 import os
 from typing import Optional, TypeVar
 from pydantic import AliasChoices, Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 T = TypeVar("T", bound=BaseSettings)
@@ -77,33 +76,9 @@ class SharedSettings(BaseSettings):
    mediakit_timeout: int = 60
    # ── 积分/会员系统 (#1895) ────────────────────────────────────────────
-    # 积分系统总开关（产品要求 #1895：暂停积分系统但保留全部代码/表/接口）。
+    # 总开关：默认 false（对所有用户零影响），P2 路由逐个接入时用
-    # - false（默认）：所有 AI 功能（生成视频/口型/数字人/AI标题/TTS/克隆音色…）
+    # `if settings.points_enabled:` 包裹，防止未完善的扣点逻辑影响现有用户。
-    #   对全部登录用户免费放行，不扣积分、不做余额拦截；积分余额/流水/会员
+    points_enabled: bool = False
    #   状态等查询接口保持可用，但数据不再变动。
    # - 未来恢复：只需设置环境变量 ENABLE_CREDIT_SYSTEM=true。
    # 旧开关 POINTS_ENABLED 仍保留作为兼容别名（两者任一为 true 即启用）。
    # 主开关（推荐环境变量名 ENABLE_CREDIT_SYSTEM）
    credits_enabled: bool = Field(
        default=False,
        validation_alias=AliasChoices("ENABLE_CREDIT_SYSTEM", "credits_enabled"),
    )
    # 旧开关兼容（POINTS_ENABLED）；两者任一为 true 即启用
    points_enabled_compat: bool = Field(
        default=False,
        validation_alias=AliasChoices("POINTS_ENABLED", "points_enabled_compat"),
    )
    @property
    def points_enabled(self) -> bool:
        """旧代码/测试使用的属性名，等价于积分系统总开关（兼容别名）。"""
        return bool(self.credits_enabled or self.points_enabled_compat)
    @points_enabled.setter
    def points_enabled(self, value: bool) -> None:
        # 支持旧测试/代码 ``settings.points_enabled = True`` 的写法
        self.credits_enabled = bool(value)
        self.points_enabled_compat = False
    # ── GPU MuseTalk 反向轮询 Worker ────────────────────────────────────
    # Worker 用这个长期 Token 鉴权（不是用户 JWT）。多 Worker 共用同一个 Token；
@@ -1,323 +0,0 @@
 """#1970 MuseTalk Flask 服务端 8 项工程 bug 修复单测.
 覆盖 deploy/gpu_worker/musetalk_server.py（独立部署脚本，按文件路径动态加载）：
 1. threaded=True 启动，/health 在推理阻塞时仍可达
 2. fps 兜底：ffprobe 返回 0 或失败时使用 default_fps
 3. ffmpeg 走 subprocess.run(check=True)，失败抛 RuntimeError
 4. 并发锁：推理期间第二请求立即 503
 5. 推理超时：超过 MUSE_INFERENCE_TIMEOUT 返回 504
 6. 结果文件清理：临时目录在请求结束（成功/失败）后删除
 7. 文件大小限制：超过限制返回 413，空文件返回 400
 8. /cancel 端点：终止当前推理，清理临时文件
 """
 from __future__ import annotations
 import importlib.util
 import io
 import os
 import shutil
 import sys
 import threading
 import time
 from pathlib import Path
 from unittest import mock
 import pytest
 # 检查 Flask 是否可用（CI 环境可能没装）
 try:
    import flask  # noqa: F401
    HAS_FLASK = True
 except ImportError:
    HAS_FLASK = False
 pytestmark = pytest.mark.skipif(not HAS_FLASK, reason="Flask 未安装（gpu_worker 独立部署依赖）")
 ROOT = Path(__file__).resolve().parents[2]
 SERVER_PATH = ROOT / "deploy" / "gpu_worker" / "musetalk_server.py"
 def _load_server_module(name: str = "musetalk_server_test"):
    """加载 musetalk_server.py 为独立模块."""
    # 避免重复注册
    if name in sys.modules:
        del sys.modules[name]
    spec = importlib.util.spec_from_file_location(name, SERVER_PATH)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod
@pytest.fixture
 def server(tmp_path, monkeypatch):
    """加载一个干净的 musetalk_server 模块，使用独立临时目录和端口."""
    if not HAS_FLASK:
        pytest.skip("Flask 未安装（gpu_worker 独立部署依赖）")
    monkeypatch.setenv("MUSE_TEMP_DIR", str(tmp_path / "musetalk_temp"))
    monkeypatch.setenv("MUSE_PORT", "0")
    monkeypatch.setenv("MUSE_INFERENCE_TIMEOUT", "2")
    monkeypatch.setenv("MUSE_VIDEO_MAX_MB", "1")
    monkeypatch.setenv("MUSE_AUDIO_MAX_MB", "1")
    monkeypatch.setenv("MUSE_DEFAULT_FPS", "25.0")
    mod_name = f"musetalk_server_test_{os.getpid()}_{id(tmp_path)}"
    mod = _load_server_module(mod_name)
    # 确保配置已更新
    mod.Config.temp_dir = str(tmp_path / "musetalk_temp")
    mod.Config.inference_timeout = 2.0
    mod.Config.video_max_mb = 1
    mod.Config.audio_max_mb = 1
    mod.Config.default_fps = 25.0
    Path(mod.Config.temp_dir).mkdir(parents=True, exist_ok=True)
    # 重置全局状态
    mod.inference_lock = threading.Lock()
    mod.current_task = {"task_id": None, "process": None, "start_time": 0.0}
    return mod
 # ── 1. Flask threaded=True ──────────────────────────────────────────
 def test_flask_run_uses_threaded(server):
    """验证 app.run 调用时 threaded=True."""
    with mock.patch.object(server.app, "run") as mock_run:
        server.main()
        mock_run.assert_called_once()
        call_kwargs = mock_run.call_args
        assert call_kwargs.kwargs.get("threaded") is True
 # ── 2. fps=0 兜底 ───────────────────────────────────────────────────
 def test_get_video_fps_fallback_on_zero(server, tmp_path):
    """ffprobe 返回 0/1 时兜底为 default_fps."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", return_value=b"0/1"):
        fps = server._get_video_fps(fake_video)
        assert fps == 25.0
 def test_get_video_fps_normal(server, tmp_path):
    """正常 fps 解析."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", return_value=b"30/1"):
        fps = server._get_video_fps(fake_video)
        assert abs(fps - 30.0) < 0.01
 def test_get_video_fps_exception_fallback(server, tmp_path):
    """ffprobe 异常时兜底 default_fps."""
    fake_video = tmp_path / "fake.mp4"
    fake_video.write_bytes(b"fake")
    with mock.patch("subprocess.check_output", side_effect=Exception("no ffprobe")):
        fps = server._get_video_fps(fake_video)
        assert fps == 25.0
 # ── 3. ffmpeg 错误检查 ──────────────────────────────────────────────
 def test_run_ffmpeg_raises_on_nonzero_exit(server):
    """ffmpeg 返回非零应抛 RuntimeError."""
    import subprocess
    with mock.patch(
        "subprocess.run",
        side_effect=subprocess.CalledProcessError(1, "ffmpeg", stderr=b"decode error"),
    ):
        with pytest.raises(RuntimeError, match="ffmpeg 失败"):
            server._run_ffmpeg(["ffmpeg", "-i", "in", "out"])
 def test_run_ffmpeg_raises_on_timeout(server):
    """ffmpeg 超时应抛 RuntimeError."""
    import subprocess
    with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("ffmpeg", 10)):
        with pytest.raises(RuntimeError, match="ffmpeg 超时"):
            server._run_ffmpeg(["ffmpeg", "-i", "in", "out"], timeout=10)
 # ── 4. 并发锁 503 ──────────────────────────────────────────────────
 def test_inference_returns_503_when_busy(server):
    """推理期间第二请求立即 503."""
    server.inference_lock.acquire()
    server.current_task["task_id"] = "task-busy"
    server.current_task["start_time"] = time.time()
    try:
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 503
            assert resp.get_json()["status"] == "busy"
    finally:
        server.inference_lock.release()
        server.current_task = {"task_id": None, "process": None, "start_time": 0.0}
 # ── 5. 推理超时 504 ─────────────────────────────────────────────────
 def test_inference_timeout_returns_504(server):
    """推理超时返回 504."""
    def slow_inference(*args, **kwargs):
        time.sleep(10)  # 远超 2s 超时
    with mock.patch.object(server, "_run_inference", side_effect=slow_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 504
            assert "超时" in resp.get_json()["error"]
 # ── 6. 临时文件清理 ──────────────────────────────────────────────────
 def test_temp_files_cleaned_after_success(server, tmp_path):
    """推理成功后临时目录被清理."""
    def fake_inference(video_path, audio_path, output_path):
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_bytes(b"v" * 2048)
    with mock.patch.object(server, "_run_inference", side_effect=fake_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                    "task_id": "task-cleanup-ok",
                },
                content_type="multipart/form-data",
            )
            # send_file 返回 200 或推理异常 500
            assert resp.status_code in (200, 500)
            task_dir = Path(server.Config.temp_dir) / "task-cleanup-ok"
            assert not task_dir.exists(), f"临时目录 {task_dir} 应被清理"
 def test_temp_files_cleaned_after_failure(server, tmp_path):
    """推理失败后临时目录也被清理."""
    def failing_inference(*args, **kwargs):
        raise RuntimeError("MuseTalk crash")
    with mock.patch.object(server, "_run_inference", side_effect=failing_inference):
        with server.app.test_client() as c:
            resp = c.post(
                "/inference",
                data={
                    "video": (io.BytesIO(b"v" * 100), "v.mp4"),
                    "audio": (io.BytesIO(b"a" * 100), "a.wav"),
                    "task_id": "task-cleanup-fail",
                },
                content_type="multipart/form-data",
            )
            assert resp.status_code == 500
            task_dir = Path(server.Config.temp_dir) / "task-cleanup-fail"
            assert not task_dir.exists()
 # ── 7. 文件大小限制 ─────────────────────────────────────────────────
 def test_oversize_video_returns_413(server):
    """视频超过大小限制返回 413."""
    big_video = b"v" * (2 * 1024 * 1024)  # 2MB > 1MB limit
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={
                "video": (io.BytesIO(big_video), "v.mp4"),
                "audio": (io.BytesIO(b"a" * 100), "a.wav"),
            },
            content_type="multipart/form-data",
        )
        assert resp.status_code == 413
        assert "超过限制" in resp.get_json()["error"]
 def test_empty_file_returns_400(server):
    """空文件返回 400."""
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={
                "video": (io.BytesIO(b""), "v.mp4"),
                "audio": (io.BytesIO(b"a" * 100), "a.wav"),
            },
            content_type="multipart/form-data",
        )
        assert resp.status_code in (400, 413)
        assert "为空" in resp.get_json().get("error", "") or "超过限制" in resp.get_json().get("error", "")
 def test_missing_file_returns_400(server):
    """缺少必要文件返回 400."""
    with server.app.test_client() as c:
        resp = c.post(
            "/inference",
            data={"video": (io.BytesIO(b"v" * 100), "v.mp4")},
            content_type="multipart/form-data",
        )
        assert resp.status_code == 400
 # ── 8. /cancel 端点 ─────────────────────────────────────────────────
 def test_cancel_no_running_task(server):
    """无任务时 /cancel 返回提示."""
    with server.app.test_client() as c:
        resp = c.post("/cancel")
        assert resp.status_code == 200
        assert "无正在运行" in resp.get_json()["message"]
 def test_cancel_terminates_running_task(server, tmp_path):
    """有任务时 /cancel 清理临时目录并重置状态."""
    task_dir = Path(server.Config.temp_dir) / "task-cancel"
    task_dir.mkdir(parents=True, exist_ok=True)
    (task_dir / "some_file.txt").write_text("temp")
    server.current_task["task_id"] = "task-cancel"
    server.current_task["start_time"] = time.time()
    server.current_task["process"] = "inference_thread"
    with server.app.test_client() as c:
        resp = c.post("/cancel")
        assert resp.status_code == 200
        assert "已取消" in resp.get_json()["message"]
        assert not task_dir.exists()
        assert server.current_task["task_id"] is None
        assert server.current_task["process"] is None
        assert server.current_task["start_time"] == 0.0
@@ -1,374 +0,0 @@
 """#1978 MuseTalk 服务端音轨替换 + 视频循环修复单测.
 覆盖 deploy/gpu_worker/musetalk_server.py：
 1. 最终封装必须 -map 0:v -map 1:a 取「推理画面 + 驱动音频」，禁止默认流选择带入源视频音轨
 2. 音频不超过视频：-c:v copy + -shortest 快速封装
 3. 音频长于视频：-stream_loop -1 循环视频，NVENC/libx264 重编码，-t 卡到音频时长
 4. enable_video_loop=false 时即使音频更长也不循环
 5. h264_nvenc 失败自动回退 libx264
 6. 真实 ffmpeg 端到端：源视频内置 200Hz 音轨 + 驱动音频 800Hz，结果音轨必须是 800Hz
   （过零率估计），证明音轨来自第二个输入而非源视频；音频更长时输出时长对齐音频
 """
 from __future__ import annotations
 import importlib.util
 import os
 import shutil
 import subprocess
 import sys
 from pathlib import Path
 from unittest import mock
 import pytest
 try:
    import flask  # noqa: F401
    HAS_FLASK = True
 except ImportError:
    HAS_FLASK = False
 pytestmark = pytest.mark.skipif(not HAS_FLASK, reason="Flask 未安装（gpu_worker 独立部署依赖）")
 ROOT = Path(__file__).resolve().parents[2]
 SERVER_PATH = ROOT / "deploy" / "gpu_worker" / "musetalk_server.py"
 HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
 def _load_server(name: str):
    if name in sys.modules:
        del sys.modules[name]
    spec = importlib.util.spec_from_file_location(name, SERVER_PATH)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod
@pytest.fixture
 def server(tmp_path, monkeypatch):
    if not HAS_FLASK:
        pytest.skip("Flask 未安装")
    monkeypatch.setenv("MUSE_TEMP_DIR", str(tmp_path / "musetalk_temp"))
    monkeypatch.setenv("MUSE_VIDEO_ENCODER", "libx264")
    mod = _load_server(f"musetalk_mux_{os.getpid()}_{id(tmp_path)}")
    mod.Config.video_encoder = "libx264"
    mod.Config.enable_video_loop = True
    return mod
 # ── 命令构造：非循环路径 ─────────────────────────────────────────────
 def test_mux_non_loop_maps_video_and_drives_audio(server, tmp_path):
    """音频(5s)不长于视频(10s)：显式 map 0:v/1:a，视频流 copy，-shortest."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    captured = {}
    def fake_run(cmd, timeout=300):
        captured["cmd"] = cmd
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[10.0, 5.0]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=fake_run),
    ):
        server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
    cmd = captured["cmd"]
    # 输入顺序：0=无声画面，1=驱动音频
    assert cmd.index(str(video)) < cmd.index(str(audio))
    # 关键修复：强制流映射，不能让 ffmpeg 默认选择源视频音轨
    assert "-map" in cmd
    assert "0:v:0" in cmd
    assert "1:a:0" in cmd
    assert "-c:v" in cmd and cmd[cmd.index("-c:v") + 1] == "copy"
    assert "-shortest" in cmd
    # 非循环不重编码
    assert "-stream_loop" not in cmd
    assert "-t" not in cmd
 def test_mux_non_loop_duration_epsilon(server, tmp_path):
    """音频略长于视频但在容差内（0.25s）不触发循环重编码."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    captured = {}
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[9.0, 9.1]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=lambda cmd, timeout=300: captured.update(cmd=cmd)),
    ):
        server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
    assert "-stream_loop" not in captured["cmd"]
 # ── 命令构造：循环路径 ───────────────────────────────────────────────
 def test_mux_loop_when_audio_longer_uses_stream_loop_and_nvenc(server, tmp_path):
    """音频(15s)长于视频(9s)：-stream_loop -1 循环、NVENC 重编码、-t 音频时长."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    server.Config.video_encoder = "h264_nvenc"
    captured = {}
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[9.0, 15.0]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=lambda cmd, timeout=300: captured.update(cmd=cmd)),
    ):
        server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
    cmd = captured["cmd"]
    # -stream_loop 必须位于第一个 -i 之前
    assert "-stream_loop" in cmd
    sl_idx = cmd.index("-stream_loop")
    assert cmd[sl_idx + 1] == "-1"
    assert sl_idx < cmd.index("-i")
    # 同样必须显式 map
    assert "0:v:0" in cmd and "1:a:0" in cmd
    assert cmd[cmd.index("-c:v") + 1] == "h264_nvenc"
    # -t 卡到音频时长，且不用 -shortest（避免截短音频）
    assert "-shortest" not in cmd
    t_idx = cmd.index("-t")
    assert abs(float(cmd[t_idx + 1]) - 15.0) < 0.01
 def test_mux_loop_disabled_falls_back_to_copy(server, tmp_path):
    """enable_video_loop=False：即使音频更长也不循环，走 copy+shortest."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    captured = {}
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[9.0, 15.0]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=lambda cmd, timeout=300: captured.update(cmd=cmd)),
    ):
        server._mux_video_with_audio(video, audio, tmp_path / "out.mp4", enable_video_loop=False)
    assert "-stream_loop" not in captured["cmd"]
    assert captured["cmd"][captured["cmd"].index("-c:v") + 1] == "copy"
 def test_mux_nvenc_failure_falls_back_to_libx264(server, tmp_path):
    """NVENC 调用失败时自动用 libx264 重试一次."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    server.Config.video_encoder = "h264_nvenc"
    cmds = []
    def runner(cmd, timeout=300):
        cmds.append(list(cmd))
        if cmd[cmd.index("-c:v") + 1] == "h264_nvenc":
            raise RuntimeError("ffmpeg 失败 (code=1): Cannot load nvcuda")
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[9.0, 15.0]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=runner),
    ):
        server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
    assert len(cmds) == 2
    assert cmds[0][cmds[0].index("-c:v") + 1] == "h264_nvenc"
    second = cmds[1]
    assert second[second.index("-c:v") + 1] == "libx264"
    # nvenc 的 preset p4 已替换为 x264 兼容值
    assert "p4" not in second
    assert "0:v:0" in second and "1:a:0" in second
 def test_mux_copy_failure_propagates(server, tmp_path):
    """非循环路径 ffmpeg 失败应抛出（不静默吞错）."""
    video = tmp_path / "visual.mp4"
    audio = tmp_path / "tts.mp3"
    video.write_bytes(b"v")
    audio.write_bytes(b"a")
    with (
        mock.patch.object(server, "_get_media_duration", side_effect=[10.0, 5.0]),
        mock.patch.object(server, "_run_ffmpeg", side_effect=RuntimeError("ffmpeg 失败")),
    ):
        with pytest.raises(RuntimeError):
            server._mux_video_with_audio(video, audio, tmp_path / "out.mp4")
 def test_pick_video_encoder_respects_config(server):
    """显式配置的编码器优先，auto 时探测."""
    server.Config.video_encoder = "libx264"
    assert server._pick_video_encoder() == "libx264"
    server.Config.video_encoder = "h264_nvenc"
    assert server._pick_video_encoder() == "h264_nvenc"
 def test_pick_video_encoder_auto_detects_nvenc(server):
    """auto 模式：ffmpeg -encoders 含 h264_nvenc 则选它."""
    server.Config.video_encoder = "auto"
    completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0, stdout=b"... h264_nvenc ...", stderr=b"")
    with mock.patch("subprocess.run", return_value=completed):
        assert server._pick_video_encoder() == "h264_nvenc"
 # ── 真实 ffmpeg 端到端：音轨来源与时长对齐 ────────────────────────────
@pytest.mark.skipif(not HAS_FFMPEG, reason="环境无 ffmpeg/ffprobe")
 def _make_media(tmp_path: Path):
    """生成：带 200Hz 音轨的 2s 源视频 + 800Hz 的 5s 驱动音频."""
    source_video = tmp_path / "source.mp4"
    drive_audio = tmp_path / "drive.wav"
    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "lavfi",
            "-i",
            "testsrc=duration=2:size=160x120:rate=25",
            "-f",
            "lavfi",
            "-i",
            "sine=frequency=200:duration=2",
            "-c:v",
            "libx264",
            "-preset",
            "ultrafast",
            "-c:a",
            "aac",
            str(source_video),
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=True,
    )
    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "lavfi",
            "-i",
            "sine=frequency=800:duration=5",
            str(drive_audio),
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=True,
    )
    return source_video, drive_audio
 def _probe_duration(path: Path) -> float:
    out = subprocess.check_output(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            str(path),
        ]
    )
    return float(out.decode().strip())
 def _estimate_audio_freq(path: Path, duration: float) -> float:
    """解码为 8kHz 单声道 s16 PCM，用过零率估计主频."""
    raw = subprocess.check_output(
        [
            "ffmpeg",
            "-i",
            str(path),
            "-vn",
            "-ac",
            "1",
            "-ar",
            "8000",
            "-f",
            "s16le",
            "-",
        ],
        stderr=subprocess.DEVNULL,
    )
    import array
    samples = array.array("h")
    samples.frombytes(raw)
    if len(samples) < 100:
        return 0.0
    crossings = sum(1 for i in range(1, len(samples)) if (samples[i - 1] < 0) != (samples[i] < 0))
    secs = len(samples) / 8000
    return crossings / 2.0 / secs
@pytest.mark.skipif(not HAS_FFMPEG, reason="环境无 ffmpeg/ffprobe")
 def test_real_mux_replaces_source_audio_with_drive_audio(server, tmp_path):
    """端到端：结果音轨必须是驱动音频 800Hz，而不是源视频的 200Hz."""
    source_video, drive_audio = _make_media(tmp_path)
    # 模拟 MuseTalk 无声画面产物
    silent_video = tmp_path / "visual_silent.mp4"
    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(source_video),
            "-an",
            "-c:v",
            "libx264",
            "-preset",
            "ultrafast",
            str(silent_video),
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=True,
    )
    output = tmp_path / "output.mp4"
    server._mux_video_with_audio(silent_video, drive_audio, output)
    assert output.exists() and output.stat().st_size > 1024
    # 驱动音频 5s 长于画面 2s → 输出应接近 5s（循环补齐）
    out_duration = _probe_duration(output)
    assert abs(out_duration - 5.0) < 0.5, f"输出时长 {out_duration} 未对齐驱动音频"
    # 结果音轨主频应接近 800Hz（驱动音频），远离 200Hz（源视频音轨）
    freq = _estimate_audio_freq(output, out_duration)
    assert abs(freq - 800) < abs(freq - 200), f"结果音轨主频 {freq:.0f}Hz 不是驱动音频"
    assert freq > 450, f"结果音轨主频 {freq:.0f}Hz 疑似源视频音轨(200Hz)"
@pytest.mark.skipif(not HAS_FFMPEG, reason="环境无 ffmpeg/ffprobe")
 def test_real_mux_non_loop_keeps_video_copy_path(server, tmp_path):
    """驱动音频(1s)短于视频(2s)：输出约 1s，音轨仍是驱动音频."""
    source_video, _ = _make_media(tmp_path)
    short_audio = tmp_path / "short.wav"
    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "lavfi",
            "-i",
            "sine=frequency=800:duration=1",
            str(short_audio),
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=True,
    )
    output = tmp_path / "output_short.mp4"
    server._mux_video_with_audio(source_video, short_audio, output)
    out_duration = _probe_duration(output)
    assert abs(out_duration - 1.0) < 0.4
    freq = _estimate_audio_freq(output, out_duration)
    assert freq > 450, f"结果音轨主频 {freq:.0f}Hz 疑似源视频音轨"
@@ -1,229 +0,0 @@
 """积分系统暂停开关测试 (#1895, ENABLE_CREDIT_SYSTEM)。
 产品要求：暂停积分系统但保留全部代码/表/接口。
 - 默认 false：所有 AI 功能免费放行，不扣积分、不做余额拦截；
 - /points/check 恒返回 allowed=True、required_points=0；
 - /points/deduct 为 no-op，余额不变；
 - 查询接口（balance/transactions/rules/packages/membership/usage）照常可用；
 - 旧环境变量 POINTS_ENABLED 作为兼容别名仍可开启。
 """
 from __future__ import annotations
 from unittest.mock import MagicMock, patch
 import pytest
 def _make_cu(user_id="user-1", is_member=False, member_type=None):
    cu = MagicMock()
    cu.user.id = user_id
    cu.user.is_member = is_member
    cu.user.member_type = member_type
    cu.user.member_expires_at = None
    return cu
 # ── 配置层 ────────────────────────────────────────────────────────────────
 class TestCreditSystemConfig:
    def test_default_disabled(self):
        import os
        from packages.config.base import SharedSettings
        assert os.environ.get("ENABLE_CREDIT_SYSTEM") is None
        assert os.environ.get("POINTS_ENABLED") is None
        s = SharedSettings(_env_file=None)
        assert s.credits_enabled is False
        # 旧属性名仍可用（业务代码大量引用 settings.points_enabled）
        assert s.points_enabled is False
    def test_enable_credit_system_env(self, monkeypatch):
        from packages.config import base as base_mod
        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "true")
        s = base_mod.SharedSettings(_env_file=None)
        assert s.points_enabled is True
        assert s.credits_enabled is True
    def test_legacy_points_enabled_env_alias(self, monkeypatch):
        from packages.config import base as base_mod
        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "false")
        monkeypatch.setenv("POINTS_ENABLED", "true")
        s = base_mod.SharedSettings(_env_file=None)
        assert s.points_enabled is True
        assert s.credits_enabled is False
        assert s.points_enabled_compat is True
    def test_legacy_setter_back_compat(self):
        from packages.config.base import SharedSettings
        s = SharedSettings(_env_file=None)
        s.points_enabled = True
        assert s.credits_enabled is True
        assert s.points_enabled is True
        s.points_enabled = False
        assert s.points_enabled is False
 # ── /points/check：关闭时恒放行、需 0 积分 ────────────────────────────────
 class TestCheckEndpointWhenDisabled:
    def test_check_allowed_zero_required(self):
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest
        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 0}
        db = MagicMock()
        cu = _make_cu()
        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=5)
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = check_points(body=body, current_user=cu, db=db)
        assert resp.allowed is True
        assert resp.required_points == 0
        assert resp.remaining_after == 0
        # 不再走免费额度判定
        svc.check_daily_free_clip.assert_not_called()
    def test_unknown_scene_still_400_when_disabled(self):
        """未知 scene 即使系统关闭也返回 400（参数校验先于开关）。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest
        from fastapi import HTTPException
        with pytest.raises(HTTPException) as exc:
            check_points(body=PointsCheckRequest(scene_key="nope"), current_user=_make_cu(), db=MagicMock())
        assert exc.value.status_code == 400
    def test_check_enabled_calculates_cost(self):
        """开关开启时保持原有计费校验。"""
        from app.api.routes.points import check_points
        from app.schemas.points import PointsCheckRequest
        svc = MagicMock()
        svc.check_daily_free_clip.return_value = False
        svc.get_or_create_account.return_value = {"balance": 100}
        body = PointsCheckRequest(scene_key="ai_title", quantity=1)
        with (
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = check_points(body=body, current_user=_make_cu(), db=MagicMock())
        assert resp.required_points == 2  # 免费用户 ceil(1*1.15)=2
 # ── /points/deduct：关闭时 no-op，余额不变 ────────────────────────────────
 class TestDeductEndpointWhenDisabled:
    def test_deduct_is_noop(self):
        from app.api.routes.points import deduct_points
        from app.schemas.points import PointsDeductRequest
        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 7}
        body = PointsDeductRequest(scene_key="ai_voice", amount=999)
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())
        svc.deduct_points.assert_not_called()
        assert resp.success is True
        assert resp.data["balance"] == 7
        assert resp.data["transaction_id"] == ""
    def test_deduct_enabled_works_as_before(self):
        from app.api.routes.points import deduct_points
        from app.schemas.points import PointsDeductRequest
        svc = MagicMock()
        svc.deduct_points.return_value = {"success": True, "balance": 8, "transaction_id": "tx-1"}
        body = PointsDeductRequest(scene_key="ai_title", amount=2)
        with (
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())
        svc.deduct_points.assert_called_once()
        assert resp.data["balance"] == 8
        assert resp.data["transaction_id"] == "tx-1"
 # ── 查询接口：系统关闭时仍全部可用 ────────────────────────────────────────
 class TestQueryEndpointsRemainAvailable:
    def test_balance_route_works_when_disabled(self):
        from app.api.routes.points import get_balance
        svc = MagicMock()
        svc.get_or_create_account.return_value = {"balance": 0, "total_earned": 0, "total_spent": 0}
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_balance(current_user=_make_cu(), db=MagicMock())
        assert resp.balance == 0
        assert resp.is_member is False
    def test_transactions_route_works_when_disabled(self):
        from app.api.routes.points import get_transactions
        svc = MagicMock()
        svc.get_transactions.return_value = {"items": [], "total": 0, "page": 1, "page_size": 20}
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_transactions(current_user=_make_cu(), db=MagicMock())
        assert resp.total == 0
    def test_daily_usage_route_works_when_disabled(self):
        from app.api.routes.points import get_daily_usage
        svc = MagicMock()
        svc.get_daily_usage.return_value = {
            "free_clips_used": 0,
            "free_clips_limit": 2,
            "free_clips_remaining": 2,
            "reset_at": "2026-09-20T00:00:00Z",
        }
        with (
            patch("app.api.routes.points._credits_enabled", return_value=False),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = get_daily_usage(current_user=_make_cu(), db=MagicMock())
        assert resp.free_clips_limit == 2
 # ── 业务路由：开关关闭时 PointsService 不实例化、不扣分 ───────────────────
 class TestBusinessRoutesBypassWhenDisabled:
    def test_lipsync_route_skips_points(self):
        """lipsync 创建任务路由：settings.points_enabled=False 时不构造 PointsService。"""
        from app.api.routes import lipsync as lipsync_mod
        assert bool(getattr(lipsync_mod.settings, "points_enabled", False)) is False
    def test_tts_route_skips_points(self):
        from app.api.routes import tts as tts_mod
        assert bool(getattr(tts_mod.settings, "points_enabled", False)) is False
@@ -20,7 +20,7 @@ def fake_mediakit():
    return client
-def _make_job(video_url="videos/video.mp4", audio_url="audios/audio.wav"):
+def _make_job(video_url="oss://video.mp4", audio_url="oss://audio.wav"):
    job = MagicMock()
    job.id = "job-1"
    job.user_id = "u1"
@@ -42,14 +42,6 @@ def _make_svc(db, mediakit, use_gpu=False):
    return svc
 def _patch_storage(public_url="https://own-bucket.oss-cn-beijing.aliyuncs.com", signed_suffix="?signed-7d"):
    """patch get_shared_storage_service，返回自家 OSS storage mock."""
    storage = MagicMock()
    storage.public_url = public_url
    storage.get_download_url.side_effect = lambda key_or_url, expires_seconds=3600: key_or_url + signed_suffix
    return patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage)
 class TestGpuFallback:
    def test_switch_off_uses_mediakit(self, fake_db, fake_mediakit):
        """开关关闭时直接走 MediaKit，不调用 _submit_to_gpu."""
@@ -74,34 +66,26 @@ class TestGpuFallback:
        assert job.status == "submitted"
    def test_gpu_success_marks_completed(self, fake_db, fake_mediakit):
-        """GPU 路径成功：job 直接 completed，不调 MediaKit；结果 key 由 storage 签 7 天 URL."""
+        """GPU 路径成功：job 直接 completed，不调 MediaKit."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        gpu_done = MagicMock(
            id="gpu-task-1",
            status="done",
-            result_url="gpu-lipsync/results/gpu-task-1.mp4",
+            result_url="oss://gpu-results/r.mp4",
            result_duration=12.5,
        )
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1")
        fake_gpu_svc.wait_for_result.return_value = gpu_done
-        with (
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
            _patch_storage() as storage_p,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
        ):
            storage = storage_p()
            job = _make_job()
            svc._submit_audio_direct(job=job)
        fake_gpu_svc.create_task.assert_called_once()
        fake_mediakit.submit_lipsync.assert_not_called()
        assert job.status == "completed"
        assert job.output_duration == 12.5
-        # Bug1 回归：裸 result key 必须经 storage.get_download_url 签 7 天，前端才可播放
+        assert "?signed" in job.output_video_url
        storage.get_download_url.assert_called_once_with(
            "gpu-lipsync/results/gpu-task-1.mp4", expires_seconds=7 * 24 * 3600
        )
        assert job.output_video_url == "gpu-lipsync/results/gpu-task-1.mp4?signed-7d"
        fake_db.commit.assert_called()
    def test_gpu_timeout_falls_back(self, fake_db, fake_mediakit):
@@ -136,85 +120,12 @@ class TestGpuFallback:
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.side_effect = RuntimeError("DB down")
-        with _patch_storage(), patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
+        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
            job = _make_job()
            svc._submit_audio_direct(job=job)
        fake_mediakit.submit_lipsync.assert_called_once()
        assert job.status == "submitted"
    def test_gpu_external_audio_persisted_to_own_oss(self, fake_db, fake_mediakit):
        """Bug2 回归：dashscope 临时音频 URL 在创建 GPU 任务前转存自家 OSS。"""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
        job = _make_job(audio_url=dashscope_url)
        gpu_done = MagicMock(id="gpu-task-2", status="done", result_url="gpu-lipsync/results/gpu-task-2.mp4")
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-2")
        fake_gpu_svc.wait_for_result.return_value = gpu_done
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes", return_value=b"FAKE-MP3") as m_dl,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
        ):
            storage = storage_p()
            storage.upload_file.return_value = "https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3"
            svc._submit_audio_direct(job=job)
        # 外部音频在 GPU 分支被额外下载（purpose 区分于前置 ffprobe 下载）并转存到约定 key
        gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
        assert len(gpu_dl_calls) == 1
        assert gpu_dl_calls[0].args[0] == dashscope_url
        storage.upload_file.assert_called_once()
        args, kwargs = storage.upload_file.call_args
        assert args[1] == "lipsync-tts/u1/job-1.mp3"
        assert kwargs.get("content_type") == "audio/mpeg"
        # 创建 GPU 任务时用的是自家 OSS URL，Worker 可经预签名下载
        kwargs_create = fake_gpu_svc.create_task.call_args.kwargs
        assert kwargs_create["audio_url"] == ("https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3")
        assert kwargs_create["audio_url"] != dashscope_url
    def test_gpu_own_audio_not_repersisted(self, fake_db, fake_mediakit):
        """Bug2：已是自家 OSS 的音频（含裸 key）不重复下载转存。"""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        job = _make_job(audio_url="lipsync-tts/u1/job-1.mp3")
        gpu_done = MagicMock(id="gpu-task-3", status="done", result_url="gpu-lipsync/results/gpu-task-3.mp4")
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-3")
        fake_gpu_svc.wait_for_result.return_value = gpu_done
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes") as m_dl,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
        ):
            storage = storage_p()
            svc._submit_audio_direct(job=job)
        # 前置 ffprobe 下载允许发生，但 GPU 转存分支不应再下载/上传
        gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
        assert gpu_dl_calls == []
        storage.upload_file.assert_not_called()
        assert fake_gpu_svc.create_task.call_args.kwargs["audio_url"] == "lipsync-tts/u1/job-1.mp3"
    def test_gpu_external_audio_persist_fail_falls_back_original_url(self, fake_db, fake_mediakit):
        """Bug2：外部音频转存失败不阻断，用原始 URL 建任务（失败后服务端重试/回退 MediaKit）。"""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
        job = _make_job(audio_url=dashscope_url)
        gpu_done = MagicMock(id="gpu-task-4", status="done", result_url="gpu-lipsync/results/gpu-task-4.mp4")
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-4")
        fake_gpu_svc.wait_for_result.return_value = gpu_done
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes", side_effect=RuntimeError("network blocked")),
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
        ):
            storage = storage_p()
            svc._submit_audio_direct(job=job)
        storage.upload_file.assert_not_called()
        assert fake_gpu_svc.create_task.call_args.kwargs["audio_url"] == dashscope_url
 class TestGpuServiceHelpers:
    """GpuLipsyncService.has_available_worker 测试."""
@@ -71,7 +71,9 @@ class TestRechargeOrderResponse:
        cu = _make_cu()
        body = PointsRechargeRequest(package_id="nonexistent")
-        with pytest.raises(HTTPException) as exc, patch("app.api.routes.points._get_service", return_value=svc):
+        with pytest.raises(HTTPException) as exc, patch(
            "app.api.routes.points._get_service", return_value=svc
        ):
            create_recharge_order(body=body, current_user=cu, db=db)
        assert exc.value.status_code == 400
@@ -110,10 +112,7 @@ class TestCheckPointsUnknownScene:
        cu = _make_cu()
        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=1)
-        with (
+        with patch("app.api.routes.points._get_service", return_value=svc):
            patch("app.api.routes.points._credits_enabled", return_value=True),
            patch("app.api.routes.points._get_service", return_value=svc),
        ):
            resp = check_points(body=body, current_user=cu, db=db)
        assert resp.required_points == 2  # ceil(1 * 1.15) = 2
        assert resp.current_balance == 50
@@ -149,30 +148,22 @@ class TestSubscriptionPlans:
    def _import_plans_fn():
        """Import from the real file to avoid sys.modules shadowing by integration fixtures."""
        import importlib.util
        _route_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
-            "..",
+            "..", "..", "apps", "api", "app", "api", "routes", "subscription.py",
-            "..",
+        )
-            "apps",
+        _spec = importlib.util.spec_from_file_location(
-            "api",
+            "_real_subscription_routes", os.path.abspath(_route_path)
            "app",
            "api",
            "routes",
            "subscription.py",
        )
        _spec = importlib.util.spec_from_file_location("_real_subscription_routes", os.path.abspath(_route_path))
        _mod = importlib.util.module_from_spec(_spec)
        # inject settings before exec
        import os as _os
        _os.environ.setdefault("JWT_SECRET_KEY", "test-secret")
        _spec.loader.exec_module(_mod)
        return _mod.list_membership_plans
    def test_plans_endpoint_returns_three_tiers(self):
        import os  # noqa: F401  (used by _import_plans_fn)
        list_membership_plans = self._import_plans_fn()
        resp = list_membership_plans(current_user=_make_cu())
        plans = resp["plans"]
@@ -186,7 +177,6 @@ class TestSubscriptionPlans:
    def test_longer_plans_cheaper_per_month(self):
        import os  # noqa: F401
        list_membership_plans = self._import_plans_fn()
        resp = list_membership_plans(current_user=_make_cu())
        plans = resp["plans"]
@@ -221,10 +211,9 @@ class TestMultiplierConsistency:
        db = MagicMock()
        cu = _make_cu()
-        with patch("app.api.routes.points._credits_enabled", return_value=True):
+        for scene in ["ai_voice", "ai_title", "ai_cover", "ai_rewrite"]:
-            for scene in ["ai_voice", "ai_title", "ai_cover", "ai_rewrite"]:
+            body = PointsCheckRequest(scene_key=scene, quantity=1)
-                body = PointsCheckRequest(scene_key=scene, quantity=1)
+            with patch("app.api.routes.points._get_service", return_value=svc):
-                with patch("app.api.routes.points._get_service", return_value=svc):
+                resp = check_points(body=body, current_user=cu, db=db)
-                    resp = check_points(body=body, current_user=cu, db=db)
+            expected = calculate_points_cost(scene, is_member=False, quantity=1)
-                expected = calculate_points_cost(scene, is_member=False, quantity=1)
+            assert resp.required_points == expected, f"{scene}: got {resp.required_points}, expected {expected}"
                assert resp.required_points == expected, f"{scene}: got {resp.required_points}, expected {expected}"