chore(points): worker settle 打印预估/实际 tokens diff 便于公式校准 (#2155)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

.env.example

@@ -212,9 +212,14 @@ COSYVOICE_CLONE_MODEL=voice-enrollment
 
 DOUBAO_API_KEY=your-doubao-api-key
 DOUBAO_MODEL=doubao-seed-1-6-250615
+DOUBAO_FAST_MODEL=doubao-1-5-pro-32k-250115
 DOUBAO_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 DOUBAO_TIMEOUT=30
 DOUBAO_MAX_RETRIES=2
+# 视觉模型：pro 精度高，lite 速度快（viral-video 商品识别默认用 lite 提速）
+DOUBAO_VISION_MODEL=doubao-1-5-vision-pro-250328
+DOUBAO_VISION_LITE_MODEL=doubao-1-5-vision-lite-250315
+DOUBAO_VISION_USE_LITE=true
 
 # ==================== 积分/会员系统 (#1895) ====================
 # 积分系统总开关：默认 false（暂停积分系统）。

alembic/versions/089_viral_video_cols.py

@@ -47,7 +47,15 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    for name in ("copy_result", "video_model", "video_ratio", "voice_source", "voice_id", "generated_copy_text", "storyboard"):
+    for name in (
+        "copy_result",
+        "video_model",
+        "video_ratio",
+        "voice_source",
+        "voice_id",
+        "generated_copy_text",
+        "storyboard",
+    ):
         try:
             op.drop_column("viral_video_jobs", name)
         except Exception:

alembic/versions/090_viral_video_phase_msg.py

@@ -0,0 +1,35 @@
+"""viral video add phase_message column (#2134)
+
+Revision ID: 090_viral_video_phase_msg
+Revises: 089_viral_video_cols
+Create Date: 2026-10-02
+
+#2134 阶段细粒度提示：viral_video 表新增 phase_message 列（中文阶段提示文案）。
+current_stage 列已在之前版本存在，本迁移只补 phase_message。
+幂等 ADD COLUMN IF NOT EXISTS。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "090_viral_video_phase_msg"
+down_revision = "089_viral_video_cols"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # SQLite/PostgreSQL 兼容的幂等添加列
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    cols = {c["name"] for c in inspector.get_columns("viral_video_jobs")}
+    if "phase_message" not in cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column("phase_message", sa.String(length=500), nullable=False, server_default=""),
+        )
+
+
+def downgrade() -> None:
+    op.drop_column("viral_video_jobs", "phase_message")

alembic/versions/091_viral_video_stage.py

@@ -0,0 +1,49 @@
+"""viral video add current_stage column (#2137 follow-up)
+
+Revision ID: 091_viral_video_stage
+Revises: 090_viral_video_phase_msg
+Create Date: 2026-10-02
+
+#2137 follow-up fix: 090 migration missed current_stage column on viral_video_jobs,
+causing UndefinedColumn errors and 500s on all authenticated viral-video endpoints.
+Idempotently add current_stage and double-check phase_message.
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "091_viral_video_stage"
+down_revision = "090_viral_video_phase_msg"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    cols = {c["name"] for c in inspector.get_columns("viral_video_jobs")}
+    if "current_stage" not in cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column(
+                "current_stage",
+                sa.String(length=200),
+                nullable=False,
+                server_default="",
+            ),
+        )
+    if "phase_message" not in cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column(
+                "phase_message",
+                sa.String(length=500),
+                nullable=False,
+                server_default="",
+            ),
+        )
+
+
+def downgrade() -> None:
+    op.drop_column("viral_video_jobs", "current_stage")

alembic/versions/092_viral_video_heartbeat.py

@@ -0,0 +1,42 @@
+"""viral_video_jobs 增加 heartbeat_at 列（worker 心跳，用于僵尸任务超时回收）
+
+Revision ID: 092_viral_video_heartbeat
+Revises: 091_viral_video_stage
+Create Date: 2026-10-02
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "092_viral_video_heartbeat"
+down_revision = "091_viral_video_stage"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    cols = {c["name"] for c in inspector.get_columns("viral_video_jobs")}
+    if "heartbeat_at" not in cols:
+        op.add_column("viral_video_jobs", sa.Column("heartbeat_at", sa.DateTime(), nullable=True))
+        op.execute(
+            "UPDATE viral_video_jobs SET heartbeat_at = updated_at " "WHERE status = 'running' AND heartbeat_at IS NULL"
+        )
+        try:
+            op.create_index("ix_viral_video_jobs_heartbeat_at", "viral_video_jobs", ["heartbeat_at"])
+        except Exception:
+            pass
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    cols = {c["name"] for c in inspector.get_columns("viral_video_jobs")}
+    if "heartbeat_at" in cols:
+        try:
+            op.drop_index("ix_viral_video_jobs_heartbeat_at", table_name="viral_video_jobs")
+        except Exception:
+            pass
+        op.drop_column("viral_video_jobs", "heartbeat_at")

alembic/versions/093_viral_video_pricing_points_float.py

@@ -0,0 +1,87 @@
+"""viral_video 动态积分定价 + 积分字段从 Integer 改为 Float (#2151)
+
+Revision ID: 093
+Revises: 092_viral_video_heartbeat
+Create Date: 2026-10-02
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "093"
+down_revision = "092_viral_video_heartbeat"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # 1) points_accounts 三列 Integer -> Float
+    pa_cols = {c["name"]: c for c in inspector.get_columns("points_accounts")}
+    for col in ("balance", "total_earned", "total_spent"):
+        if col in pa_cols:
+            op.alter_column(
+                "points_accounts",
+                col,
+                existing_type=sa.Integer(),
+                type_=sa.Float(),
+                existing_nullable=False,
+            )
+
+    # 2) points_transactions amount/balance_after Integer -> Float
+    pt_cols = {c["name"]: c for c in inspector.get_columns("points_transactions")}
+    for col in ("amount", "balance_after"):
+        if col in pt_cols:
+            op.alter_column(
+                "points_transactions",
+                col,
+                existing_type=sa.Integer(),
+                type_=sa.Float(),
+                existing_nullable=False,
+            )
+
+    # 3) users.points_balance Integer -> Float
+    user_cols = {c["name"]: c for c in inspector.get_columns("users")}
+    if "points_balance" in user_cols:
+        op.alter_column(
+            "users",
+            "points_balance",
+            existing_type=sa.Integer(),
+            type_=sa.Float(),
+            existing_nullable=False,
+        )
+
+    # 4) viral_video_jobs.credits_cost Integer -> Float
+    vv_cols = {c["name"]: c for c in inspector.get_columns("viral_video_jobs")}
+    if "credits_cost" in vv_cols:
+        op.alter_column(
+            "viral_video_jobs",
+            "credits_cost",
+            existing_type=sa.Integer(),
+            type_=sa.Float(),
+            existing_nullable=False,
+        )
+
+    # 5) viral_video_jobs 新增列
+    if "video_resolution" not in vv_cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column("video_resolution", sa.String(20), nullable=False, server_default="720p"),
+        )
+    if "credits_prepaid" not in vv_cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column("credits_prepaid", sa.Float(), nullable=False, server_default="0"),
+        )
+    if "credits_transaction_id" not in vv_cols:
+        op.add_column(
+            "viral_video_jobs",
+            sa.Column("credits_transaction_id", sa.String(36), nullable=False, server_default=""),
+        )
+
+
+def downgrade() -> None:
+    pass

apps/api/app/api/routes/ai_avatar_render.py

@@ -29,8 +29,6 @@ from app.services.ai_avatar_render_service import (
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
-from packages.middleware.points_gate import points_gate
-
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -44,7 +42,6 @@ def _get_service(db: Session = Depends(get_db_session)) -> AiAvatarRenderService
 
 
 @router.post("", response_model=AiAvatarRenderJobResponse, status_code=201)
-@points_gate("ai_digital_human", per_unit=15)
 def create_render_job(
     body: CreateAiAvatarRenderRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/generation_cover.py

@@ -27,7 +27,6 @@ from packages.adapters.sqlalchemy_impl.generation_task_repository import (
 )
 from packages.application import ListGeneratedVideosByTaskUseCase
 from packages.domain.config_schemas import normalize_plan_config
-from packages.middleware.points_gate import points_gate
 from packages.shared.storage import get_shared_storage_service
 
 from .templates_editor.dependencies import get_draft_plan_id, get_editor_services
@@ -346,7 +345,6 @@ def _is_trusted_media_url(url: str) -> bool:
 
 
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
-@points_gate("ai_cover")
 def generate_cover(
     body: GenerateCoverRequest,
     template_id: str = Query(..., description="模板 ID"),

apps/api/app/api/routes/generation_preview.py

@@ -41,7 +41,6 @@ from packages.application import (
     GetGenerationTaskUseCase,
     ListGeneratedVideosByTaskUseCase,
 )
-from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -270,7 +269,6 @@ def _variant_value(values: list[str], index: int, fallback: str = "") -> str:
 
 
 @router.post("/preview", response_model=BatchPreviewGenerationTaskResponse, status_code=201)
-@points_gate("ai_video", quantity_field="preview_count")
 def create_preview_generation_task(
     request: CreatePreviewGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/generation_tasks.py

@@ -163,7 +163,6 @@ def _infer_expected_categories(script_tags: set[str] | None) -> set[str] | None:
     return matched or None
 
 
-from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -465,7 +464,6 @@ def _resolve_project_and_library(
 
 
 @router.post("/tasks", response_model=BatchGenerationTaskResponse)
-@points_gate("ai_video", quantity_field="count")
 def create_generation_task(
     request: CreateGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/lipsync.py

@@ -12,11 +12,9 @@
 from __future__ import annotations
 
 import logging
-import math
 from datetime import UTC
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.config import settings
 from app.dependencies import (
     get_db_session,
     get_voice_clone_profile_repository,
@@ -32,9 +30,6 @@ from app.services.mediakit_client import MediaKitError
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
-
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -61,37 +56,6 @@ def create_lipsync_job(
     db: Session = Depends(get_db_session),
     svc: LipsyncService = Depends(_get_service),
 ):
-    user_id = current_user.user.id
-
-    # ── 积分扣点（#1895 P2） ──
-    _points_deducted = 0
-    _points_scene = "ai_digital_human"
-    _points_svc = PointsService() if settings.points_enabled else None
-    if _points_svc is not None:
-        # 口型同步：TTS 模式按 script_text 估时长（240字/分钟）；音频直传按 audio_duration（秒→分钟）
-        if body.audio_url and body.audio_duration and body.audio_duration > 0:
-            est_minutes = max(1.0, math.ceil(body.audio_duration / 60.0))
-        elif body.script_text:
-            est_minutes = max(1.0, math.ceil(len(body.script_text) / 240))
-        else:
-            est_minutes = 1.0
-        _points_deducted = calculate_points_cost(
-            _points_scene,
-            is_member=getattr(current_user.user, "is_member", False),
-            duration_minutes=est_minutes,
-            member_type=getattr(current_user.user, "member_type", None),
-        )
-        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
-        if not _deduct_res["success"]:
-            raise HTTPException(
-                status_code=402,
-                detail={
-                    "code": "INSUFFICIENT_POINTS",
-                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
-                    "required": _points_deducted,
-                    "balance": _deduct_res["balance"],
-                },
-            )
     """提交对口型任务.
 
     三种模式：
@@ -101,6 +65,8 @@ def create_lipsync_job(
     - 预合成音频（#1845 新主路径）：传 {video_url, audio_url, audio_duration, sentence_timings}，
       后端同步ffprobe+写入timings+直接提交MediaKit（~2-3s）。
     """
+    user_id = current_user.user.id
+
     try:
         job = svc.create_job(
             user_id=user_id,
@@ -118,18 +84,8 @@ def create_lipsync_job(
             project_id=body.project_id,
         )
     except ValueError as exc:
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"对口型 ValueError 退积分异常: err={refund_err}")
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     except MediaKitError as exc:
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"对口型 MediaKitError 退积分异常: err={refund_err}")
         status_code = 502
         if exc.code in ("VoiceForbidden",):
             status_code = 403
@@ -145,24 +101,11 @@ def create_lipsync_job(
         ) from exc
     except Exception as exc:
         logger.error("创建对口型任务异常: %s", exc, exc_info=True)
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"对口型异常退积分异常: err={refund_err}")
         raise HTTPException(
             status_code=400,
             detail=f"创建对口型任务失败: {exc}",
         ) from exc
 
-    # 创建成功但状态为 failed（同步路径失败已抛异常到上面 except；此处处理 Celery 调度失败等）
-    # 若任务已创建且状态为 failed，退费
-    if _points_deducted > 0 and _points_svc is not None and getattr(job, "status", None) == "failed":
-        try:
-            _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
-        except Exception as refund_err:
-            logger.warning(f"对口型任务失败退积分异常: job_id={job.id}, err={refund_err}")
-
     return job
 
 
@@ -176,37 +119,14 @@ def preview_tts(
     db: Session = Depends(get_db_session),
     svc: LipsyncService = Depends(_get_service),
 ):
-    user_id = current_user.user.id
-
-    # ── 积分扣点（#1895 P2） ──
-    _points_deducted = 0
-    _points_scene = "ai_digital_human"
-    _points_svc = PointsService() if settings.points_enabled else None
-    if _points_svc is not None:
-        est_minutes = max(1.0, math.ceil(len(body.script_text or "") / 240)) if body.script_text else 1.0
-        _points_deducted = calculate_points_cost(
-            _points_scene,
-            is_member=getattr(current_user.user, "is_member", False),
-            duration_minutes=est_minutes,
-            member_type=getattr(current_user.user, "member_type", None),
-        )
-        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
-        if not _deduct_res["success"]:
-            raise HTTPException(
-                status_code=402,
-                detail={
-                    "code": "INSUFFICIENT_POINTS",
-                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
-                    "required": _points_deducted,
-                    "balance": _deduct_res["balance"],
-                },
-            )
     """步骤1「生成配音」同步 TTS 预合成.
 
     同步执行 TTS 合成 → 下载音频 → ffprobe 时长 → 句子时间戳计算，
     不创建 LipsyncJob、不转存 OSS，直接返回 CosyVoice 临时 URL（~24h 有效）。
     耗时约 2-3 秒。
     """
+    user_id = current_user.user.id
+
     try:
         result = svc.preview_tts(
             user_id=user_id,
@@ -218,11 +138,6 @@ def preview_tts(
             emotion=body.emotion,
         )
     except MediaKitError as exc:
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"TTS 预合成 MediaKitError 退积分异常: err={refund_err}")
         status_code = 400
         if exc.code in ("VoiceForbidden",):
             status_code = 403
@@ -237,11 +152,6 @@ def preview_tts(
         ) from exc
     except Exception as exc:
         logger.error("TTS 预合成异常: %s", exc, exc_info=True)
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"TTS 预合成异常退积分异常: err={refund_err}")
         raise HTTPException(
             status_code=400,
             detail=f"TTS 合成失败: {exc}",

apps/api/app/api/routes/points.py

@@ -169,17 +169,7 @@ def check_points(
     current_user: AuthenticatedUser = Depends(get_current_user),
     db: Session = Depends(get_db_session),
 ):
-    """消费前检查余额是否足够。未知 scene_key 返回 400（而非 500）。"""
-    if body.scene_key not in POINTS_SCENES:
-        raise HTTPException(
-            status_code=400,
-            detail={
-                "code": "UNKNOWN_SCENE",
-                "message": f"未知场景: {body.scene_key}",
-                "valid_scenes": sorted(POINTS_SCENES.keys()),
-            },
-        )
-
+    """消费前检查余额是否足够。已下线/未知场景返回 cost=0（免费）。"""
     # 积分系统暂停（ENABLE_CREDIT_SYSTEM=false）：所有场景直接放行，需 0 积分
     if not _credits_enabled():
         svc = _get_service()
@@ -195,13 +185,6 @@ def check_points(
     is_mem = _is_member(current_user)
     mt = _member_type(current_user)
 
-    # 混剪场景先检查免费额度
-    is_free_quota = False
-    if body.scene_key == "ai_video" and not is_mem:
-        svc = _get_service()
-        if svc.check_daily_free_clip(current_user.user.id, db):
-            is_free_quota = True
-
     required = calculate_points_cost(
         body.scene_key,
         is_mem,
@@ -215,11 +198,11 @@ def check_points(
     balance = account["balance"]
 
     return PointsCheckResponse(
-        allowed=is_free_quota or balance >= required,
+        allowed=balance >= required,
         required_points=required,
         current_balance=balance,
         remaining_after=balance - required,
-        is_free_quota=is_free_quota,
+        is_free_quota=False,
     )
 
 

apps/api/app/api/routes/scripts_ai.py

@@ -44,7 +44,6 @@ from app.services.script_asr_service import (
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 
-from packages.middleware.points_gate import points_gate
 from packages.shared.ai_client import get_doubao_client
 
 logger = logging.getLogger(__name__)
@@ -373,7 +372,6 @@ def douyin_diag():
 
 
 @router.post("/extract-from-douyin", response_model=ExtractFromDouyinResponse)
-@points_gate("douyin_extract")
 def extract_from_douyin(
     request: ExtractFromDouyinRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
@@ -497,7 +495,6 @@ def extract_from_douyin(
 
 
 @router.post("/ai-rewrite", response_model=AiRewriteResponse)
-@points_gate("ai_rewrite")
 def ai_rewrite(
     request: AiRewriteRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
@@ -537,7 +534,6 @@ def ai_rewrite(
 
 
 @router.post("/ai-generate-titles", response_model=AiGenerateTitlesResponse)
-@points_gate("ai_title")
 def ai_generate_titles(
     request: AiGenerateTitlesRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/tts.py

@@ -4,14 +4,12 @@ from __future__ import annotations
 
 import json
 import logging
-import math
 import subprocess
 import tempfile
 from pathlib import Path
 from typing import Any, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.config import settings
 from app.core.celery_app import celery_app
 from app.core.storage import get_storage_service
 from app.dependencies import (
@@ -53,8 +51,6 @@ from packages.application.tts_job.use_cases import (
 )
 from packages.application.tts_job.workflow import TTSWorkflowService
 from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
 from packages.domain.voice_presets import list_voices
 from packages.ports.asset_library_repository import AssetLibraryRepository
 from packages.ports.asset_repository import AssetRepository
@@ -144,31 +140,6 @@ def synthesize(
     """
     user_id = authenticated_user.user.id
 
-    # ── 积分扣点（#1895 P2） ──
-    _points_deducted = 0
-    _points_scene = "ai_voice"
-    _points_svc = PointsService() if settings.points_enabled else None
-    if _points_svc is not None:
-        # 中文按 ~240 字/分钟粗估时长，至少按 1 分钟扣 1 分
-        est_minutes = max(1.0, math.ceil(len(request.text) / 240))
-        _points_deducted = calculate_points_cost(
-            _points_scene,
-            is_member=getattr(authenticated_user.user, "is_member", False),
-            duration_minutes=est_minutes,
-            member_type=getattr(authenticated_user.user, "member_type", None),
-        )
-        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
-        if not _deduct_res["success"]:
-            raise HTTPException(
-                status_code=402,
-                detail={
-                    "code": "INSUFFICIENT_POINTS",
-                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
-                    "required": _points_deducted,
-                    "balance": _deduct_res["balance"],
-                },
-            )
-
     # 解析 voice_id：前端可能传克隆音色 profile UUID（而非 CosyVoice voice_id），
     # 与 /tts/preview 保持一致：命中 profile → 校验归属 → 取 CosyVoice voice_id
     actual_voice_id = request.voice_id
@@ -231,7 +202,6 @@ def synthesize(
         cosyvoice_service=cosyvoice_service,
     )
 
-    synthesis_error: Exception | None = None
     try:
         job = workflow.start_synthesis(job.id)
     except Exception as e:
@@ -239,18 +209,10 @@ def synthesize(
         # 但 DB 异常、网络异常等意外错误可能逃逸。
         # 与音色克隆接口保持一致：标记 failed，返回 201，不抛 500。
         logger.error(f"TTS 合成异常: job_id={job.id}, error={e}", exc_info=True)
-        synthesis_error = e
         try:
             job = workflow.process_synthesis_failure(job.id, str(e))
         except Exception as inner_e:
             logger.error(f"标记 TTS job 失败时出错: job_id={job.id}, error={inner_e}")
-    # 合成失败且已扣积分 → 退费
-    if synthesis_error is not None and _points_deducted > 0 and _points_svc is not None:
-        try:
-            _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
-        except Exception as refund_err:
-            logger.warning(f"TTS 合失败退积分异常: job_id={job.id}, err={refund_err}")
-
     # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询
     if job.status.value == "processing":
         # 分段合成任务 vs 普通单段任务
@@ -269,13 +231,6 @@ def synthesize(
                 workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}")
             except Exception as inner_e:
                 logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}")
-            # 调度失败退费
-            if _points_deducted > 0 and _points_svc is not None:
-                try:
-                    _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
-                except Exception as refund_err:
-                    logger.warning(f"Celery 调度失败退积分异常: job_id={job.id}, err={refund_err}")
-
     return TTSSynthesizeResponse(
         job_id=job.id,
         status=job.status,
@@ -610,31 +565,6 @@ def preview_tts(
     用于前端预览配音效果，限制文本长度 200 字以内。
     支持预设音色和克隆音色：克隆音色传的是 profile UUID，需解析为 CosyVoice voice_id。
     """
-    user_id = authenticated_user.user.id
-    # ── 积分扣点（#1895 P2） ──
-    _points_deducted = 0
-    _points_scene = "ai_voice"
-    _points_svc = PointsService() if settings.points_enabled else None
-    if _points_svc is not None:
-        est_minutes = max(1.0, math.ceil(len(request.text) / 240))
-        _points_deducted = calculate_points_cost(
-            _points_scene,
-            is_member=getattr(authenticated_user.user, "is_member", False),
-            duration_minutes=est_minutes,
-            member_type=getattr(authenticated_user.user, "member_type", None),
-        )
-        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
-        if not _deduct_res["success"]:
-            raise HTTPException(
-                status_code=402,
-                detail={
-                    "code": "INSUFFICIENT_POINTS",
-                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
-                    "required": _points_deducted,
-                    "balance": _deduct_res["balance"],
-                },
-            )
-
     # 解析 voice_id：前端可能传 VoiceCloneProfile UUID 或预设音色 ID
     actual_voice_id = request.voice_id
     profile = voice_clone_repo.get(request.voice_id)
@@ -664,12 +594,6 @@ def preview_tts(
             language=getattr(request, "language", "zh-CN"),
         )
     except (CosyVoiceError, ValueError) as e:
-        # 合成失败退费
-        if _points_deducted > 0 and _points_svc is not None:
-            try:
-                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
-            except Exception as refund_err:
-                logger.warning(f"TTS 预览失败退积分异常: {refund_err}")
         if isinstance(e, CosyVoiceError):
             raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"TTS 合成失败: {e}") from e
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e

apps/api/app/api/routes/viral_video.py

@@ -32,7 +32,11 @@ from app.schemas.viral_video import (
     ConfirmCopyRequest,
     ConfirmIntentRequest,
     CreateViralVideoRequest,
+    CreditsFormulaBreakdown,
+    EstimateCreditsRequest,
+    EstimateCreditsResponse,
     GenerateCopyRequest,
+    RetryViralVideoRequest,
     StyleTemplateListResponse,
     StyleTemplateResponse,
     ViralVideoHistoryResponse,
@@ -128,6 +132,8 @@ def _to_response(job) -> ViralVideoJobResponse:
         style_guide=job.style_guide,
         style_template_id=job.style_template_id,
         status=job.status,
+        current_stage=getattr(job, "current_stage", "") or "",
+        phase_message=getattr(job, "phase_message", "") or "",
         image_analysis=getattr(job, "image_analysis", None),
         storyboard=getattr(job, "storyboard", None),
         generated_copy_text=getattr(job, "generated_copy_text", "") or "",
@@ -138,7 +144,9 @@ def _to_response(job) -> ViralVideoJobResponse:
         video_model=getattr(job, "video_model", "") or "",
         intent_result=job.intent_result,
         result_video_url=job.result_video_url,
-        credits_cost=job.credits_cost,
+        video_resolution=getattr(job, "video_resolution", "720p") or "720p",
+        credits_prepaid=float(getattr(job, "credits_prepaid", 0) or 0),
+        credits_cost=float(getattr(job, "credits_cost", 0) or 0),
         error_msg=job.error_msg,
         retry_count=job.retry_count,
         started_at=job.started_at,
@@ -191,6 +199,7 @@ def create_viral_video(
         voice_source=getattr(request, "voice_source", "") or "",
         video_ratio=getattr(request, "video_ratio", "9:16") or "9:16",
         video_model=getattr(request, "video_model", "") or "",
+        video_resolution=getattr(request, "video_resolution", "720p") or "720p",
         copy_result=None,
     )
 
@@ -233,6 +242,7 @@ def analyze_images(
         voice_source=request.voice_source or "",
         video_ratio=request.video_ratio or "9:16",
         video_model=request.video_model or "",
+        video_resolution=getattr(request, "video_resolution", "720p") or "720p",
         duration=request.duration or 15,
     )
     repo.save(job)
@@ -296,6 +306,7 @@ def generate_copy(
     job.voice_source = request.voice_source or job.voice_source
     job.video_ratio = request.video_ratio or job.video_ratio or "9:16"
     job.video_model = request.video_model or job.video_model or ""
+    job.video_resolution = getattr(request, "video_resolution", "") or job.video_resolution or "720p"
 
     job.resume_from_image_analyzed()
     repo.update(job)
@@ -328,6 +339,41 @@ def confirm_copy(
     if job.status != ViralVideoStatus.COPY_GENERATED:
         raise HTTPException(status_code=409, detail=f"任务当前状态 {job.status} 不能确认文案（需 copy_generated）")
 
+    # 积分预扣（已扣过/重试任务跳过）
+    from app.config import settings as _settings
+
+    if _settings.points_enabled:
+        already_paid = (float(getattr(job, "credits_prepaid", 0) or 0) > 0) or (
+            float(getattr(job, "credits_cost", 0) or 0) > 0
+        )
+        if not already_paid:
+            from packages.domain.points_rules import calculate_viral_video_credits, resolve_video_dimensions
+            from packages.domain.points_service import PointsService
+
+            w, h = resolve_video_dimensions(
+                getattr(job, "video_resolution", "720p") or "720p",
+                job.video_ratio or "9:16",
+            )
+            est_credits = calculate_viral_video_credits(
+                int(job.duration or 15), w, h, job.video_model or "seedance-2.5"
+            )
+            svc = PointsService()
+            res = svc.deduct_viral_video(authenticated_user.user.id, est_credits, job.id, session)
+            if not res.get("success"):
+                balance = res.get("balance", 0)
+                raise HTTPException(
+                    status_code=402,
+                    detail={
+                        "code": "INSUFFICIENT_POINTS",
+                        "message": f"积分不足，需要 {est_credits} 积分，当前余额 {balance}",
+                        "required": est_credits,
+                        "balance": balance,
+                    },
+                )
+            job.credits_prepaid = est_credits
+            job.credits_transaction_id = res.get("transaction_id", "") or ""
+            repo.update(job)
+
     job.resume_from_copy_generated(edited_copy=request.edited_copy or None)
     repo.update(job)
 
@@ -342,6 +388,35 @@ def confirm_copy(
     return _to_response(job)
 
 
+@router.post("/estimate-credits", response_model=EstimateCreditsResponse)
+def estimate_credits(
+    request: EstimateCreditsRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+) -> EstimateCreditsResponse:
+    """爆款视频积分预估（纯计算，不扣费、不创建任务）。
+
+    返回 estimated_credits 与 formula_breakdown（tokens / video_cost / fixed_cost /
+    profit_multiplier / model_price / width / height / fps），便于前端展示计费明细。
+    同时兼容前端传 model 或 video_model、resolution 或 video_resolution、ratio 或 video_ratio。
+    """
+    from packages.domain.points_rules import (
+        calculate_viral_video_credits_with_breakdown,
+        resolve_video_dimensions,
+    )
+
+    model = (request.model or "").strip() or "seedance-2.5"
+    resolution = (request.resolution or "").strip() or "720p"
+    ratio = (request.ratio or "").strip() or "9:16"
+    duration = int(request.duration or 15)
+
+    w, h = resolve_video_dimensions(resolution, ratio)
+    credits, bd = calculate_viral_video_credits_with_breakdown(
+        duration, w, h, model,
+    )
+    breakdown = CreditsFormulaBreakdown(**bd)
+    return EstimateCreditsResponse(estimated_credits=credits, formula_breakdown=breakdown)
+
+
 @router.get("/history", response_model=ViralVideoHistoryResponse)
 def list_viral_video_history(
     limit: int = 50,
@@ -395,31 +470,149 @@ def get_viral_video_job(
 @router.post("/{job_id}/retry", response_model=ViralVideoJobResponse)
 def retry_viral_video_job(
     job_id: str,
+    request: RetryViralVideoRequest | None = None,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     session: Session = Depends(get_db_session),
 ) -> ViralVideoJobResponse:
-    """重试失败的爆款视频任务。"""
+    """重试失败的爆款视频任务（也支持对僵尸/超时 running 任务强制重置后重试）。
+
+    可选 body (RetryViralVideoRequest)：若传入新的 duration/video_resolution/video_ratio/
+    video_model，会重新预估积分并与原 credits_prepaid 做差额多退少补（不足抛 402 阻止重试）；
+    不传 body 或参数无变化时，保持原参数、原预扣金额不变，仅重置状态并入队。
+    credits_prepaid 为 0 的老任务首次重试会走预扣流程（与 confirm-copy 一致）。
+    """
+    from datetime import datetime, timezone
+
     repo = _get_job_repo(session)
     job = repo.get(job_id)
     if job is None:
         raise HTTPException(status_code=404, detail="任务不存在")
     if job.user_id != authenticated_user.user.id:
         raise HTTPException(status_code=403, detail="无权操作此任务")
-    if job.status != ViralVideoStatus.FAILED:
-        raise HTTPException(status_code=409, detail="只有失败的任务可以重试")
+
+    # 判定是否为僵尸 running 任务：running 超过 10 分钟且心跳停止超过 2 分钟
+    now = datetime.now(timezone.utc)
+    is_stale_running = False
+    if job.status == ViralVideoStatus.RUNNING and job.started_at is not None:
+        hb = getattr(job, "heartbeat_at", None) or job.updated_at
+        if (now - job.started_at).total_seconds() > 10 * 60 and hb is not None and (now - hb).total_seconds() > 2 * 60:
+            is_stale_running = True
+
+    if job.status != ViralVideoStatus.FAILED and not is_stale_running:
+        raise HTTPException(status_code=409, detail="只有失败或超时的任务可以重试")
+
+    # ── 参数变更检测 + 积分多退少补 ──────────────────────────────────────
+    req = request or RetryViralVideoRequest()
+    new_duration = req.duration
+    new_resolution = (req.video_resolution or "").strip() or None
+    new_ratio = (req.video_ratio or "").strip() or None
+    new_model = (req.video_model or "").strip() or None
+
+    old_duration = int(getattr(job, "duration", 15) or 15)
+    old_resolution = (getattr(job, "video_resolution", "720p") or "720p").strip() or "720p"
+    old_ratio = (getattr(job, "video_ratio", "9:16") or "9:16").strip() or "9:16"
+    old_model = (getattr(job, "video_model", "") or "").strip()
+
+    # 仅当有任意字段传入且值不同才算"参数变更"
+    param_changed = bool(
+        (new_duration is not None and int(new_duration) != old_duration)
+        or (new_resolution is not None and new_resolution != old_resolution)
+        or (new_ratio is not None and new_ratio != old_ratio)
+        or (new_model is not None and new_model != old_model)
+    )
+
+    from app.config import settings as _settings
+
+    need_points_settle = False
+    new_est = 0.0
+    if _settings.points_enabled and param_changed:
+        from packages.domain.points_rules import (
+            calculate_viral_video_credits_with_breakdown,
+            resolve_video_dimensions,
+        )
+
+        eff_dur = int(new_duration if new_duration is not None else old_duration)
+        eff_res = new_resolution if new_resolution is not None else old_resolution
+        eff_ratio = new_ratio if new_ratio is not None else old_ratio
+        eff_model = new_model if new_model is not None else (old_model or "seedance-2.5")
+        w, h = resolve_video_dimensions(eff_res, eff_ratio)
+        new_est, _ = calculate_viral_video_credits_with_breakdown(eff_dur, w, h, eff_model or "seedance-2.5")
+        need_points_settle = True
+
+    # 写入新参数（即使不开 points 也要允许用户重试时改参数）
+    if new_duration is not None:
+        job.duration = max(5, min(30, int(new_duration)))
+    if new_resolution is not None:
+        job.video_resolution = new_resolution
+    if new_ratio is not None:
+        job.video_ratio = new_ratio
+    if new_model is not None:
+        job.video_model = new_model
+
+    if need_points_settle:
+        from packages.domain.points_service import PointsService
+
+        old_prepaid = float(getattr(job, "credits_prepaid", 0) or 0)
+        svc = PointsService()
+        diff = round(new_est - old_prepaid, 2)
+        if abs(diff) >= 0.01:
+            if diff > 0:
+                # 新预扣更多：补扣差额
+                res = svc.deduct_viral_video(authenticated_user.user.id, diff, job.id, session)
+                if not res.get("success"):
+                    balance = res.get("balance", 0)
+                    raise HTTPException(
+                        status_code=402,
+                        detail={
+                            "code": "INSUFFICIENT_POINTS",
+                            "message": f"重试参数变更后需补扣 {diff} 积分，余额不足（当前 {balance}，需 {new_est}）",
+                            "required": new_est,
+                            "balance": balance,
+                            "delta": diff,
+                        },
+                    )
+                job.credits_prepaid = round(old_prepaid + diff, 2)
+                logger.info(
+                    "[爆款视频][retry] 补扣差额 job_id=%s diff=%.2f new_prepaid=%.2f",
+                    job.id, diff, job.credits_prepaid,
+                )
+            else:
+                # 新预扣更少：退还差额
+                refund = round(-diff, 2)
+                txn_id = getattr(job, "credits_transaction_id", "") or ""
+                svc.refund_points(
+                    user_id=authenticated_user.user.id,
+                    amount=refund,
+                    source="viral_video",
+                    db=session,
+                    ref_id=txn_id or job.id,
+                    description="爆款视频重试参数变更退费",
+                )
+                job.credits_prepaid = round(old_prepaid - refund, 2)
+                logger.info(
+                    "[爆款视频][retry] 退还差额 job_id=%s refund=%.2f new_prepaid=%.2f",
+                    job.id, refund, job.credits_prepaid,
+                )
+        # 差额为 0 则不调整
 
     # 重置状态
     job.retry_count += 1
     job.status = ViralVideoStatus.PENDING
-    job.error_msg = ""
+    job.error_msg = "" if not is_stale_running else "任务执行超时，已重置重试"
     job.started_at = None
     job.completed_at = None
+    job.current_stage = ""
+    job.phase_message = ""
+    job.heartbeat_at = None
     repo.update(job)
 
     # 重新入队
     try:
         celery_app.send_task("worker.run_viral_video_pipeline", args=[job.id])
-        logger.info("[爆款视频] 重试入队: job_id=%s retry_count=%d", job.id, job.retry_count)
+        logger.info(
+            "[爆款视频] 重试入队: job_id=%s retry_count=%d stale=%s params_changed=%s",
+            job.id, job.retry_count, is_stale_running, param_changed,
+        )
     except Exception as e:
         logger.error("[爆款视频] 重试入队失败: %s", e, exc_info=True)
         job.mark_failed(f"重试入队失败: {e}")

apps/api/app/schemas/points.py

@@ -13,9 +13,9 @@ from pydantic import BaseModel, Field
 class PointsBalanceResponse(BaseModel):
     """积分余额 + 会员状态"""
 
-    balance: int = Field(..., description="当前积分余额")
-    total_earned: int = Field(..., description="累计获得积分")
-    total_spent: int = Field(..., description="累计消耗积分")
+    balance: float = Field(..., description="当前积分余额")
+    total_earned: float = Field(..., description="累计获得积分")
+    total_spent: float = Field(..., description="累计消耗积分")
     is_member: bool = Field(default=False, description="是否付费会员")
     member_type: Optional[str] = Field(None, description="会员类型: monthly/quarterly/yearly")
     member_expires_at: Optional[datetime] = Field(None, description="会员到期时间")
@@ -30,8 +30,8 @@ class PointsTransactionItem(BaseModel):
     id: str
     type: str = Field(..., description="类型: add/deduct")
     source: str = Field(..., description="来源场景")
-    amount: int
-    balance_after: int
+    amount: float
+    balance_after: float
     description: str = ""
     ref_id: str = ""
     created_at: Optional[str] = None
@@ -99,9 +99,9 @@ class PointsCheckResponse(BaseModel):
     """消费前余额检查响应"""
 
     allowed: bool
-    required_points: int
-    current_balance: int
-    remaining_after: int
+    required_points: float
+    current_balance: float
+    remaining_after: float
     is_free_quota: bool = False
 
 
@@ -112,7 +112,7 @@ class PointsDeductRequest(BaseModel):
     """积分扣减请求"""
 
     scene_key: str
-    amount: int
+    amount: float
     description: Optional[str] = ""
     ref_id: Optional[str] = ""
 
@@ -170,7 +170,7 @@ class MembershipStatusResponse(BaseModel):
     is_member: bool
     member_type: Optional[str] = None
     member_expires_at: Optional[datetime] = None
-    points_balance: int
+    points_balance: float
     max_resolution: str = Field(
         default="1080p",
         description="可用最高分辨率: 720p(free) / 1080p(paid)",

apps/api/app/schemas/viral_video.py

@@ -22,6 +22,7 @@ VALID_STAGES = (
 )
 VALID_VIDEO_RATIOS = ("9:16", "16:9", "1:1", "4:3", "3:4", "21:9")
 VALID_DURATIONS = (5, 10, 15, 20, 25, 30)
+VALID_VIDEO_RESOLUTIONS = ("480p", "720p", "1080p", "普清", "高清", "超清")
 
 
 # -- 编导脚本结构（v1.6） --
@@ -88,6 +89,7 @@ class CreateViralVideoRequest(BaseModel):
     voice_source: str = ""
     video_ratio: str = "9:16"
     video_model: str = ""
+    video_resolution: str = "720p"
 
     @field_validator("fusion_level")
     @classmethod
@@ -117,6 +119,7 @@ class AnalyzeImagesRequest(BaseModel):
     voice_source: str = ""
     video_ratio: str = "9:16"
     video_model: str = ""
+    video_resolution: str = "720p"
     duration: int = Field(default=15, ge=5, le=30)
 
 
@@ -141,6 +144,7 @@ class GenerateCopyRequest(BaseModel):
     voice_source: str = ""
     video_ratio: str = "9:16"
     video_model: str = ""
+    video_resolution: str = "720p"
 
     @field_validator("fusion_level")
     @classmethod
@@ -201,6 +205,10 @@ class ViralVideoJobResponse(BaseModel):
     style_guide: dict | None = None
     style_template_id: str = ""
     status: str
+    current_stage: str = (
+        ""  # 细粒度阶段 snake_case（analyzing_images/parsing_intent/generating_script/reviewing/tts_synthesizing/rendering_video/uploading）
+    )
+    phase_message: str = ""  # 中文阶段提示文案（前端轮询/SSE 直接展示）
     image_analysis: dict | None = None
     # v1.6 编导脚本（推荐前端使用）
     copy_result: dict | None = None
@@ -214,7 +222,9 @@ class ViralVideoJobResponse(BaseModel):
     video_model: str = ""
     intent_result: dict | None = None
     result_video_url: str = ""
-    credits_cost: int = 0
+    video_resolution: str = "720p"
+    credits_prepaid: float = 0.0
+    credits_cost: float = 0.0
     error_msg: str = ""
     retry_count: int = 0
     started_at: datetime | None = None
@@ -246,6 +256,57 @@ class AnalyzeStyleResponse(BaseModel):
     style_guide: dict | None = None
 
 
+# -- 积分预估 --
+
+
+class EstimateCreditsRequest(BaseModel):
+    """爆款视频积分预估请求。
+
+    前端可传 model 或 video_model（兼容老字段）；resolution/ratio/duration 为预估所需参数。
+    """
+
+    model: str = Field(default="", alias="video_model")
+    resolution: str = Field(default="720p", alias="video_resolution")
+    ratio: str = Field(default="9:16", alias="video_ratio")
+    duration: int = Field(default=15, ge=5, le=30)
+
+    model_config = {"populate_by_name": True}
+
+
+class CreditsFormulaBreakdown(BaseModel):
+    """爆款视频积分计费公式明细（前端展示用）。"""
+
+    tokens: float = Field(..., description="估算视频 tokens 数 (duration*width*height*fps/1024)")
+    video_cost: float = Field(..., description="视频生成成本（元）= tokens/1e6 * model_price")
+    fixed_cost: float = Field(..., description="固定成本（元），含 VLM/LLM/TTS/OSS/服务器")
+    profit_multiplier: float = Field(..., description="利润系数（默认 1.3）")
+    model_price: float = Field(..., description="模型单价（元/百万 tokens）")
+    width: int = Field(..., description="视频宽度像素")
+    height: int = Field(..., description="视频高度像素")
+    fps: int = Field(..., description="视频帧率")
+
+
+class EstimateCreditsResponse(BaseModel):
+    """爆款视频积分预估响应。"""
+
+    estimated_credits: float
+    formula_breakdown: CreditsFormulaBreakdown = Field(..., description="计费公式明细")
+
+
+class RetryViralVideoRequest(BaseModel):
+    """重试爆款视频任务的请求体（可选，允许改参数重新预估积分多退少补）。
+
+    不传 body 或字段全缺省：保持原参数、不重新扣点，走默认重置+入队逻辑。
+    传入新的 duration/video_resolution/video_ratio/video_model：重新预估积分，
+    与原 credits_prepaid 比较后多退少补（差额补扣不足抛 402）。
+    """
+
+    duration: int | None = Field(default=None, ge=5, le=30, description="重试时新的视频时长（秒）")
+    video_resolution: str | None = Field(default=None, description="重试时新的分辨率，如 720p/1080p")
+    video_ratio: str | None = Field(default=None, description="重试时新的画幅比，如 9:16/16:9")
+    video_model: str | None = Field(default=None, description="重试时新的视频模型，如 seedance-2.5")
+
+
 # -- WebSocket 事件 Schema --
 
 

apps/api/app/services/narrative_service.py

@@ -11,14 +11,12 @@
    存储路径与元信息约定），返回 asset_id —— 下游仍以 voice_library_id（实为
    audio asset id）消费，渲染链路零改动。
 
-积分扣点与 /tts 合成端点保持一致（ai_voice 场景），失败退费。
 """
 
 from __future__ import annotations
 
 import json
 import logging
-import math
 import subprocess
 import tempfile
 from dataclasses import dataclass
@@ -32,13 +30,10 @@ from packages.application.cosyvoice_service import CosyVoiceService
 from packages.application.tts_job.use_cases import CreateTTSJobUseCase
 from packages.application.tts_job.workflow import TTSWorkflowService
 from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
 from packages.shared.storage import SharedStorageService
 
 logger = logging.getLogger(__name__)
 
-_POINTS_SCENE = "ai_voice"
 _SYNTH_TIMEOUT = 180.0  # 叙事配音在 HTTP 请求内同步等待，长文案分段合成时留出余量
 _CONTENT_TYPE_MAP = {"mp3": "audio/mpeg", "wav": "audio/wav", "pcm": "audio/pcm", "opus": "audio/opus"}
 
@@ -273,24 +268,6 @@ def prepare_narrative_voice(
         voice_clone_repository=voice_clone_repository,
     )
 
-    # 积分扣点（与 /tts 合成端点同口径），失败时在合成失败分支退费
-    points_svc = PointsService() if points_enabled else None
-    points_deducted = 0
-    if points_svc is not None:
-        est_minutes = max(1.0, math.ceil(len(content) / 240))
-        points_deducted = calculate_points_cost(
-            _POINTS_SCENE,
-            is_member=is_member,
-            duration_minutes=est_minutes,
-            member_type=member_type,
-        )
-        deduct_res = points_svc.deduct_points(user_id, points_deducted, _POINTS_SCENE, db)
-        if not deduct_res["success"]:
-            raise NarrativeError(
-                f"积分不足，需要 {points_deducted} 积分，当前余额 {deduct_res['balance']}",
-                status_code=402,
-            )
-
     use_case = CreateTTSJobUseCase(tts_repository)
     job = use_case.execute(
         user_id=user_id,
@@ -311,19 +288,9 @@ def prepare_narrative_voice(
             workflow.process_synthesis_failure(job.id, str(e))
         except Exception:  # noqa: BLE001
             logger.warning("标记叙事 TTS job 失败出错: job_id=%s", job.id, exc_info=True)
-        if points_deducted and points_svc is not None:
-            try:
-                points_svc.refund_points(user_id, points_deducted, _POINTS_SCENE, db, ref_id=job.id)
-            except Exception:  # noqa: BLE001
-                logger.warning("叙事 TTS 失败退积分异常: job_id=%s", job.id, exc_info=True)
         raise NarrativeError(f"配音合成失败：{e}", status_code=502) from e
 
     if not job.is_completed:
-        if points_deducted and points_svc is not None:
-            try:
-                points_svc.refund_points(user_id, points_deducted, _POINTS_SCENE, db, ref_id=job.id)
-            except Exception:  # noqa: BLE001
-                logger.warning("叙事 TTS 未完成退积分异常: job_id=%s", job.id, exc_info=True)
         raise NarrativeError("配音合成未完成，请稍后重试", status_code=504)
 
     asset = _save_tts_job_as_voice_asset(

apps/web/src/api/viral-video/index.ts

@@ -51,6 +51,17 @@ export function analyzeViralStyle(id: string) {
   return apiClient.post<ViralVideoJob>(`/viral-video/${id}/analyze-style`).then((r) => r.data)
 }
 
+/** 动态预估积分消耗（STEP3 参数变化时调用） */
+export function estimateViralVideoCredits(params: {
+  video_model: string
+  resolution: string
+  video_ratio: string
+  duration: number
+}) {
+  return apiClient
+    .post<{ estimated_credits: number }>("/viral-video/estimate-credits", params)
+    .then((r) => r.data)
+}
 /** ── 三步拆分：前端 mock 辅助函数（后端新接口上线后可替换） ── */
 
 /**

apps/web/src/pages/viral-video/components/PresetVoicePickerModal.tsx

@@ -34,83 +34,63 @@ interface Props {
 
 /** 兜底 mock 音色（后端 /api/v1/tts/presets 返回字段不够时使用） */
 const MOCK_VOICES: PresetVoice[] = [
+  // ⚠️ 兜底 mock，仅在 /voices/presets 接口不可达时使用；ID 必须与后端
+  // packages/domain/preset_voices.py PRESET_VOICES 的 voice_id 对齐（v3后缀）
   {
-    id: "long-xiaochun",
+    id: "longxiaochun_v3",
     name: "龙小淳",
     gender: "female",
     category: "女声",
     desc: "知性积极女声，适合语音助手",
   },
   {
-    id: "long-xiaoxia",
+    id: "longxiaoxia_v3",
     name: "龙小夏",
     gender: "female",
     category: "女声",
     desc: "沉稳权威女声，适合新闻播报",
   },
   {
-    id: "long-xiaoyan",
-    name: "龙小颜",
-    gender: "female",
-    category: "女声",
-    desc: "温柔甜美女声，适合情感口播",
-  },
-  {
-    id: "long-xiaotong",
-    name: "龙小彤",
-    gender: "female",
-    category: "女声",
-    desc: "活力少女音，适合短视频带货",
-  },
-  {
-    id: "long-sanshu",
+    id: "longsanshu_v3",
     name: "龙三叔",
     gender: "male",
     category: "男声",
     desc: "沉稳质感男声，适合有声书",
   },
   {
-    id: "long-xiaogang",
-    name: "龙小刚",
-    gender: "male",
-    category: "男声",
-    desc: "阳光活力男声，适合解说",
-  },
-  {
-    id: "long-xiaocheng",
-    name: "龙小诚",
-    gender: "male",
-    category: "男声",
-    desc: "磁性商务男声，适合品牌宣传",
-  },
-  {
-    id: "long-xiaozhi",
-    name: "龙小智",
-    gender: "child",
-    category: "童声",
-    desc: "可爱童声，适合亲子内容",
-  },
-  {
-    id: "long-yue",
+    id: "longyue_v3",
     name: "龙悦",
     gender: "female",
-    category: "情绪",
-    desc: "温柔治愈女声，适合睡前/助眠",
+    category: "女声",
+    desc: "温暖磁性女声，适合广告配音",
   },
   {
-    id: "long-xiaodong",
-    name: "龙晓东",
+    id: "longshu_v3",
+    name: "龙书",
     gender: "male",
-    category: "方言",
-    desc: "东北方言男声，接地气",
+    category: "男声",
+    desc: "沉稳青年男声，适合教育讲解",
   },
-  { id: "long-xiaoling", name: "龙小玲", gender: "female", category: "方言", desc: "粤语女声" },
   {
-    id: "long-xiaoxiao-neural",
-    name: "晓晓",
+    id: "longyingjing_v3",
+    name: "龙应静",
     gender: "female",
     category: "女声",
-    desc: "温柔女声",
+    desc: "低调冷静女声，适合纪录片解说",
+  },
+  {
+    id: "longshuo_v3",
+    name: "龙硕",
+    gender: "male",
+    category: "男声",
+    desc: "博才干练男声，适合科技类内容",
+  },
+  {
+    id: "longtian_v3",
+    name: "龙甜",
+    gender: "female",
+    category: "女声",
+    desc: "活泼女声，适合短视频配音",
   },
 ]
 
@@ -178,23 +158,22 @@ const PresetVoicePickerModal: React.FC<Props> = ({
 
   // 合并真实数据和 mock：如果真实数据 gender/category 缺失，用 mock 兜底
   const allVoices: PresetVoice[] = useMemo(() => {
+    // 真实 API 返回的 voice_id 以 API 为准（如 longxiaochun_v3），前端不做硬编码覆盖
     const realList: PresetVoice[] = (voices || []).map((v) => {
-      // 按 name 模糊匹配 mock 获取补充信息
-      const mockMatch = MOCK_VOICES.find(
-        (m) => v.name?.includes(m.name.slice(1)) || m.name.includes(v.name?.slice(0, 2) || "___"),
-      )
+      // 按 id 精确匹配 mock 获取补充元信息（id 即 voice_id，唯一稳定键）
+      const mockMatch = MOCK_VOICES.find((m) => m.id === v.id)
       return {
         ...v,
         gender: v.gender || mockMatch?.gender,
         category:
           v.category ||
           mockMatch?.category ||
-          (v.gender === "female" ? "女声" : v.gender === "male" ? "男声" : undefined),
+          (v.gender === "female" ? "女声" : v.gender === "male" ? "男声" : "其他"),
         desc: v.desc || mockMatch?.desc,
         sample_audio_url: v.sample_audio_url,
       }
     })
-    // 如果没有真实数据，使用 mock
+    // 如果没有真实数据，使用兜底 mock（接口失败时）
     return realList.length > 0 ? realList : MOCK_VOICES
   }, [voices])
 

packages/adapters/sqlalchemy_impl/models.py

@@ -56,7 +56,7 @@ class UserModel(Base):
     is_member = Column(Boolean, nullable=False, default=False)
     member_type = Column(String(20), nullable=True)
     member_expires_at = Column(DateTime, nullable=True)
-    points_balance = Column(Integer, nullable=False, default=0)
+    points_balance = Column(Float, nullable=False, default=0)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
 
 
@@ -775,9 +775,9 @@ class PointsAccountModel(Base):
 
     id = Column(String(36), primary_key=True)
     user_id = Column(String(36), nullable=False, unique=True, index=True)
-    balance = Column(Integer, nullable=False, default=0)
-    total_earned = Column(Integer, nullable=False, default=0)
-    total_spent = Column(Integer, nullable=False, default=0)
+    balance = Column(Float, nullable=False, default=0)
+    total_earned = Column(Float, nullable=False, default=0)
+    total_spent = Column(Float, nullable=False, default=0)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
 
@@ -792,8 +792,8 @@ class PointsTransactionModel(Base):
     account_id = Column(String(36), nullable=False, index=True)
     type = Column(String(20), nullable=False, index=True)  # earn / spend / refund
     source = Column(String(50), nullable=False, index=True)
-    amount = Column(Integer, nullable=False)
-    balance_after = Column(Integer, nullable=False)
+    amount = Column(Float, nullable=False)
+    balance_after = Column(Float, nullable=False)
     description = Column(String(255), nullable=False, default="")
     ref_id = Column(String(100), nullable=False, default="")
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
@@ -950,6 +950,9 @@ class ViralVideoJobModel(Base):
     video_model = Column(String(100), nullable=False, default="")
     # 结果与状态
     status = Column(String(30), nullable=False, default="pending", index=True)
+    current_stage = Column(String(200), nullable=False, default="")  # 细粒度阶段 snake_case
+    phase_message = Column(String(500), nullable=False, default="")  # 阶段中文提示文案
+    heartbeat_at = Column(DateTime, nullable=True, index=True)  # worker 心跳，用于僵尸任务超时回收
     intent_result = Column(JSON, nullable=True)
     image_analysis = Column(JSON, nullable=True)
     storyboard = Column(JSON, nullable=True)
@@ -958,7 +961,10 @@ class ViralVideoJobModel(Base):
         JSON, nullable=True
     )  # v1.6: 编导脚本结构{overview,scene_and_lighting,shots,hard_constraints,negative_prompts,voiceover_script}
     result_video_url = Column(String(1000), nullable=False, default="")
-    credits_cost = Column(Integer, nullable=False, default=0)
+    credits_cost = Column(Float, nullable=False, default=0)
+    video_resolution = Column(String(20), nullable=False, default="720p")
+    credits_prepaid = Column(Float, nullable=False, default=0.0)
+    credits_transaction_id = Column(String(36), nullable=False, default="")
     error_msg = Column(Text, nullable=False, default="")
     retry_count = Column(Integer, nullable=False, default=0)
     started_at = Column(DateTime(timezone=True), nullable=True)

packages/adapters/sqlalchemy_impl/viral_video_repository.py

@@ -37,13 +37,19 @@ def _to_domain(model: ViralVideoJobModel) -> ViralVideoJob:
         video_ratio=getattr(model, "video_ratio", "9:16") or "9:16",
         video_model=getattr(model, "video_model", "") or "",
         status=ViralVideoStatus(model.status) if model.status else ViralVideoStatus.PENDING,
+        current_stage=getattr(model, "current_stage", "") or "",
+        phase_message=getattr(model, "phase_message", "") or "",
+        heartbeat_at=getattr(model, "heartbeat_at", None),
         intent_result=dict(model.intent_result) if model.intent_result else None,
         image_analysis=dict(model.image_analysis) if getattr(model, "image_analysis", None) else None,
         storyboard=list(model.storyboard) if getattr(model, "storyboard", None) else None,
         generated_copy_text=getattr(model, "generated_copy_text", "") or "",
         copy_result=dict(model.copy_result) if getattr(model, "copy_result", None) else None,
         result_video_url=model.result_video_url or "",
-        credits_cost=model.credits_cost or 0,
+        video_resolution=getattr(model, "video_resolution", "720p") or "720p",
+        credits_prepaid=float(getattr(model, "credits_prepaid", 0) or 0),
+        credits_transaction_id=getattr(model, "credits_transaction_id", "") or "",
+        credits_cost=float(model.credits_cost or 0),
         error_msg=model.error_msg or "",
         retry_count=model.retry_count or 0,
         started_at=model.started_at,
@@ -83,13 +89,19 @@ class SQLAlchemyViralVideoJobRepository:
             video_ratio=job.video_ratio,
             video_model=job.video_model,
             status=job.status,
+            current_stage=job.current_stage or "",
+            phase_message=job.phase_message or "",
+            heartbeat_at=job.heartbeat_at,
             intent_result=job.intent_result,
             image_analysis=job.image_analysis,
             storyboard=job.storyboard,
             generated_copy_text=job.generated_copy_text,
             copy_result=job.copy_result,
             result_video_url=job.result_video_url,
-            credits_cost=job.credits_cost,
+            video_resolution=getattr(job, "video_resolution", "720p") or "720p",
+            credits_prepaid=float(getattr(job, "credits_prepaid", 0) or 0),
+            credits_transaction_id=getattr(job, "credits_transaction_id", "") or "",
+            credits_cost=float(getattr(job, "credits_cost", 0) or 0),
             error_msg=job.error_msg,
             retry_count=job.retry_count,
             started_at=job.started_at,
@@ -106,13 +118,19 @@ class SQLAlchemyViralVideoJobRepository:
         if model is None:
             raise ValueError(f"ViralVideoJob {job.id} not found")
         model.status = job.status
+        model.current_stage = job.current_stage or ""
+        model.phase_message = job.phase_message or ""
+        model.heartbeat_at = job.heartbeat_at
         model.intent_result = job.intent_result
         model.image_analysis = job.image_analysis
         model.storyboard = job.storyboard
         model.generated_copy_text = job.generated_copy_text or ""
         model.copy_result = job.copy_result
         model.result_video_url = job.result_video_url
-        model.credits_cost = job.credits_cost
+        model.video_resolution = getattr(job, "video_resolution", "720p") or "720p"
+        model.credits_prepaid = float(getattr(job, "credits_prepaid", 0) or 0)
+        model.credits_transaction_id = getattr(job, "credits_transaction_id", "") or ""
+        model.credits_cost = float(getattr(job, "credits_cost", 0) or 0)
         model.error_msg = job.error_msg
         model.retry_count = job.retry_count
         model.started_at = job.started_at

packages/config/base.py

@@ -90,11 +90,14 @@ class SharedSettings(BaseSettings):
 
     # ── 豆包大模型（火山引擎方舟） ────────────────────────────────────────
     doubao_api_key: str = ""
-    doubao_model: str = "doubao-seed-1-6-250615"
+    doubao_model: str = "doubao-seed-1-6-250615"  # 推理模型（通用兜底）
+    doubao_fast_model: str = "doubao-1-5-pro-32k-250115"  # 快速结构化输出模型（编导脚本/意图解析/审核）
     doubao_base_url: str = "https://ark.cn-beijing.volces.com/api/v3"
     doubao_timeout: int = 30
     doubao_max_retries: int = 2
-    doubao_vision_model: str = "doubao-1-5-vision-pro-250915"
+    doubao_vision_model: str = "doubao-1-5-vision-pro-250328"  # 高精度视觉（备用）
+    doubao_vision_lite_model: str = "doubao-1-5-vision-lite-250315"  # 快速视觉（商品识别默认，速度优先）
+    doubao_vision_use_lite: bool = True  # viral-video 图片分析默认用 lite 提速
     doubao_embedding_model: str = "doubao-embedding-large-text-240915"
     doubao_video_model: str = "doubao-seedance-2-5-260628"
     doubao_video_timeout: int = 600  # 视频生成轮询总超时（秒）

packages/domain/points_account.py

@@ -9,9 +9,9 @@ from uuid import uuid4
 class PointsAccount:
     id: str
     user_id: str
-    balance: int = 0
-    total_earned: int = 0
-    total_spent: int = 0
+    balance: float = 0.0
+    total_earned: float = 0.0
+    total_spent: float = 0.0
     created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
     updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
 

packages/domain/points_rules.py

@@ -1,32 +1,198 @@
-"""积分消耗规则配置 (#1895)"""
+"""积分消耗规则配置 (#1895)
+
+v1.6.1: 按产品决策，智能混剪/AI数字人/AI配音/抖音解析/改写/标题/封面 全部免费，
+仅保留声音克隆合成（voice_clone_synth）的扣点逻辑；声音克隆训练保持免费。
+爆款视频（viral_video）走动态定价，见本文件 VIRAL_VIDEO_MODEL_PRICES + calculate_viral_video_credits。
+"""
 
 from __future__ import annotations
 
 import math
 
+# ============ 爆款视频动态定价 (#2151) ============
+# key = (model_id, resolution, has_video_input)，单位：元/百万token
+VIRAL_VIDEO_MODEL_PRICES: dict[tuple[str, str, bool], float] = {
+    ("seedance-2.5", "480p", False): 70.0,
+    ("seedance-2.5", "720p", False): 70.0,
+    ("seedance-2.5", "1080p", False): 77.0,
+    ("seedance-2.5", "480p", True): 42.0,
+    ("seedance-2.5", "720p", True): 42.0,
+    ("seedance-2.5", "1080p", True): 46.0,
+    ("seedance-2.0", "480p", False): 46.0,
+    ("seedance-2.0", "720p", False): 46.0,
+    ("seedance-2.0", "1080p", False): 51.0,
+}
+
+# 固定成本（元）：VLM 分析 + LLM 文案 + TTS + OSS + 服务器
+VIRAL_VIDEO_FIXED_COST = 0.15
+# 利润系数
+VIRAL_VIDEO_PROFIT_MULTIPLIER = 1.3
+# Seedance 输出帧率
+VIRAL_VIDEO_FPS = 24
+
+# 分辨率别名映射 -> 标准 key
+_RESOLUTION_ALIASES: dict[str, str] = {
+    "480p": "480p",
+    "普清": "480p",
+    "default": "480p",
+    "low": "480p",
+    "sd": "480p",
+    "720p": "720p",
+    "高清": "720p",
+    "medium": "720p",
+    "hd": "720p",
+    "1080p": "1080p",
+    "超清": "1080p",
+    "high": "1080p",
+    "ultra": "1080p",
+    "全能": "1080p",
+    "fhd": "1080p",
+}
+# 分辨率 -> 短边像素数（p 值代表短边，不是 height）
+_RESOLUTION_SHORT_SIDE: dict[str, int] = {"480p": 480, "720p": 720, "1080p": 1080}
+
+
+def resolve_video_dimensions(resolution: str, ratio: str) -> tuple[int, int]:
+    """把 (resolution, ratio) 解析为 (width, height)。
+
+    resolution 数字代表短边像素数（480p/720p/1080p 等）：
+    - 横屏 16:9：短边是 height，width = short * 16/9
+    - 竖屏 9:16：短边是 width，height = short * 16/9
+    - 方屏 1:1：width = height = short
+    """
+    key = str(resolution or "").strip()
+    key_l = key.lower()
+    res_key = _RESOLUTION_ALIASES.get(key_l) or _RESOLUTION_ALIASES.get(key) or "720p"
+    short = _RESOLUTION_SHORT_SIDE.get(res_key, 720)
+    r = str(ratio or "").strip().lower()
+    if r == "16:9":
+        # 横屏：短边是 height，width 向上取整并对齐偶数
+        w = math.ceil(short * 16 / 9)
+        h = short
+    elif r == "1:1":
+        w, h = short, short
+    else:
+        # 9:16 竖屏（默认）：短边是 width，height 向上取整并对齐偶数
+        w = short
+        h = math.ceil(short * 16 / 9)
+    # 对齐到偶数（视频编码要求）
+    w = w + (w % 2)
+    h = h + (h % 2)
+    return int(w), int(h)
+
+
+def _match_model_prefix(model: str) -> str:
+    """匹配 model 前缀。"""
+    m = (model or "").strip().lower()
+    for prefix in ("seedance-2.5", "seedance-2.0"):
+        if m.startswith(prefix):
+            return prefix
+    return "seedance-2.5"
+
+
+def _infer_resolution_key(width: int, height: int) -> str:
+    """从实际 (width, height) 用短边推断 resolution key。"""
+    short = min(int(width or 720), int(height or 720))
+    if short >= 1000:
+        return "1080p"
+    if short >= 650:
+        return "720p"
+    return "480p"
+
+
+def calculate_viral_video_credits_with_breakdown(
+    duration_seconds: int,
+    width: int,
+    height: int,
+    model: str = "seedance-2.5",
+    has_video_input: bool = False,
+    actual_tokens: int | None = None,
+    fps: int = VIRAL_VIDEO_FPS,
+) -> tuple[float, dict]:
+    """计算爆款视频所需积分（1 积分 = 1 元），并返回计费公式明细。
+
+    公式:
+        tokens = duration * width * height * fps / 1024
+        video_cost = tokens / 1_000_000 * model_token_price
+        total = round((video_cost + fixed_cost) * profit_multiplier, 2)
+    若传入 actual_tokens 则用它替代计算值。
+
+    Returns:
+        (credits, breakdown) 二元组：
+        - credits: 四舍五入保留两位小数的最终积分
+        - breakdown: dict，包含 tokens / video_cost / fixed_cost / profit_multiplier /
+          model_price / width / height / fps 字段，便于前端展示计费明细。
+    """
+    w = max(1, int(width or 1))
+    h = max(1, int(height or 1))
+    effective_fps = int(fps or VIRAL_VIDEO_FPS)
+
+    prefix = _match_model_prefix(model)
+    res_key = _infer_resolution_key(w, h)
+    key = (prefix, res_key, bool(has_video_input))
+    price = VIRAL_VIDEO_MODEL_PRICES.get(key)
+    if price is None:
+        price = VIRAL_VIDEO_MODEL_PRICES.get(("seedance-2.5", res_key, False), 70.0)
+    if actual_tokens is not None and actual_tokens > 0:
+        tokens = float(actual_tokens)
+    else:
+        dur = max(1, int(duration_seconds or 15))
+        tokens = dur * w * h * effective_fps / 1024.0
+
+    video_cost = tokens / 1_000_000.0 * float(price)
+    total = (video_cost + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER
+    credits = round(float(total), 2)
+    breakdown = {
+        "tokens": float(tokens),
+        "video_cost": float(video_cost),
+        "fixed_cost": float(VIRAL_VIDEO_FIXED_COST),
+        "profit_multiplier": float(VIRAL_VIDEO_PROFIT_MULTIPLIER),
+        "model_price": float(price),
+        "width": int(w),
+        "height": int(h),
+        "fps": int(effective_fps),
+    }
+    return credits, breakdown
+
+
+def calculate_viral_video_credits(
+    duration_seconds: int,
+    width: int,
+    height: int,
+    model: str = "seedance-2.5",
+    has_video_input: bool = False,
+    actual_tokens: int | None = None,
+    fps: int = VIRAL_VIDEO_FPS,
+) -> float:
+    """计算爆款视频所需积分（1 积分 = 1 元），仅返回积分值（向后兼容包装器）。
+
+    内部调用 calculate_viral_video_credits_with_breakdown，仅返回 credits 部分，
+    保持旧调用方签名与返回值类型不变。
+
+    公式:
+        tokens = duration * width * height * fps / 1024
+        video_cost = tokens / 1_000_000 * model_token_price
+        total = round((video_cost + fixed_cost) * profit_multiplier, 2)
+    若传入 actual_tokens 则用它替代计算值。
+    """
+    credits, _ = calculate_viral_video_credits_with_breakdown(
+        duration_seconds=duration_seconds,
+        width=width,
+        height=height,
+        model=model,
+        has_video_input=has_video_input,
+        actual_tokens=actual_tokens,
+        fps=fps,
+    )
+    return credits
+
+
 # ============ 场景定义 ============
-# 每个场景: base_points(基础积分), unit(计费单位), name(显示名称)
+# 每个场景: base_points(基础积分), unit(计费单位), name(显示名称), dynamic(是否动态定价)
+# 说明：爆款视频（viral_video）走动态定价（预扣→结算多退少补），因此不使用 @points_gate
+# 装饰器，base_points=0，dynamic=True；前端展示场景列表时仍可看到。
 
 POINTS_SCENES: dict[str, dict] = {
-    "ai_voice": {
-        "base_points": 1,
-        "unit": "分钟",
-        "name": "AI 配音",
-        "description": "AI 配音每分钟消耗 1 积分（免费用户上浮 15%，会员 8~9 折）",
-    },
-    "ai_video": {
-        "base_points": 3,
-        "unit": "条",
-        "name": "智能混剪",
-        "extra_per_30s": 1,
-        "description": "智能混剪每条 3 积分起，视频超过 30 秒后每 30 秒加 1 积分；免费用户每日 2 条免费额度",
-    },
-    "ai_digital_human": {
-        "base_points": 15,
-        "unit": "分钟",
-        "name": "AI 数字人",
-        "description": "AI 数字人每分钟消耗 15 积分",
-    },
     "voice_clone_train": {
         "base_points": 0,
         "unit": "次",
@@ -39,23 +205,16 @@ POINTS_SCENES: dict[str, dict] = {
         "name": "声音克隆合成",
         "description": "克隆音色合成每分钟消耗 1 积分",
     },
-    "douyin_extract": {
-        "base_points": 1,
+    "viral_video": {
+        "base_points": 0,
         "unit": "次",
-        "name": "抖音链接提取",
-        "description": "抖音文案提取每次 1 积分",
+        "name": "爆款视频",
+        "dynamic": True,
+        "description": "爆款视频动态定价（按视频时长/分辨率/模型计算，预扣→结算多退少补）",
     },
-    "ai_rewrite": {"base_points": 1, "unit": "次", "name": "AI 改写文案", "description": "AI 改写文案每次 1 积分"},
-    "ai_title": {
-        "base_points": 1,
-        "unit": "次",
-        "name": "AI 标题生成",
-        "description": "AI 生成标题每次 1 积分（免费用户实际上浮后 2 积分/次）",
-    },
-    "ai_cover": {"base_points": 1, "unit": "张", "name": "AI 封面生成", "description": "AI 封面生成每张 1 积分"},
 }
 
-# 免费用户积分消耗上浮系数
+# 免费用户积分消耗上浮系数（仅对 voice_clone_synth 生效）
 FREE_USER_MULTIPLIER = 1.15
 
 # ============ 积分包定义 ============
@@ -81,9 +240,6 @@ MEMBER_DISCOUNT: dict[str, float] = {
     "yearly": 0.8,
 }
 
-# 每日免费混剪次数（免费用户）
-DAILY_FREE_CLIP_LIMIT = 2
-
 
 def calculate_points_cost(
     scene_key: str,
@@ -91,48 +247,45 @@ def calculate_points_cost(
     quantity: int = 1,
     duration_minutes: float = 0,
     member_type: str | None = None,
-) -> int:
+) -> float:
     """计算指定场景的积分消耗。
 
     Args:
-        scene_key: 场景标识，如 "ai_voice"、"ai_video"
+        scene_key: 场景标识（当前支持 voice_clone_train/voice_clone_synth/viral_video；
+                   viral_video 为动态定价场景，此处返回 0，由业务侧调用
+                   calculate_viral_video_credits 手动计算）
         is_member: 是否付费会员
         quantity: 数量（按次计费场景）
         duration_minutes: 时长分钟数（按时长计费场景）
         member_type: 会员类型 (monthly/quarterly/yearly)，用于折扣
 
     Returns:
-        实际消耗积分（已含免费用户 ×1.15 上浮或会员折扣）
-
-    Raises:
-        ValueError: 未知场景标识
+        实际消耗积分（float；已含免费用户 ×1.15 上浮或会员折扣）；免费/动态/已下线场景统一返回 0。
     """
     scene = POINTS_SCENES.get(scene_key)
     if not scene:
-        raise ValueError(f"Unknown points scene: {scene_key}")
+        # 已下线/未注册的场景统一返回 0（免费），保持向后兼容
+        return 0.0
+
+    # 动态定价场景（如 viral_video）由业务侧手动计算，这里统一返回 0
+    if scene.get("dynamic"):
+        return 0.0
 
     base = scene["base_points"]
     if base == 0:
-        return 0
+        return 0.0
 
-    # —— 计算基础消耗 ——
     unit = scene["unit"]
     if unit == "分钟":
         total_base = base * max(1, math.ceil(duration_minutes))
-    elif unit in ("条", "次", "张"):
+    elif unit in ("次", "张"):
         total_base = base * quantity
-        # 混剪特殊逻辑：视频超过 30s 后每 +30s 额外加 1 积分
-        if scene_key == "ai_video" and duration_minutes > 0.5:
-            extra_segments = math.ceil((duration_minutes * 60 - 30) / 30)
-            if extra_segments > 0:
-                total_base += scene.get("extra_per_30s", 1) * extra_segments
     else:
         total_base = base
 
-    # —— 会员折扣 / 免费用户上浮 ——
     if is_member and member_type and member_type in MEMBER_DISCOUNT:
         total_base = max(1, math.floor(total_base * MEMBER_DISCOUNT[member_type]))
     elif not is_member:
         total_base = math.ceil(total_base * FREE_USER_MULTIPLIER)
 
-    return total_base
+    return float(total_base)

packages/domain/points_service.py

@@ -13,7 +13,6 @@ from typing import Any
 from sqlalchemy.orm import Session
 
 from packages.domain.points_rules import (
-    DAILY_FREE_CLIP_LIMIT,
     POINTS_PACKAGES,
 )
 
@@ -84,7 +83,7 @@ class PointsService:
 
     # ──────────────── 余额检查 ────────────────
 
-    def check_balance(self, user_id: str, amount: int, db: Session) -> dict[str, Any]:
+    def check_balance(self, user_id: str, amount: float, db: Session) -> dict[str, Any]:
         """检查余额是否足够。"""
         account_data = self.get_or_create_account(user_id, db)
         balance = account_data["balance"]
@@ -100,7 +99,7 @@ class PointsService:
     def deduct_points(
         self,
         user_id: str,
-        amount: int,
+        amount: float,
         source: str,
         db: Session,
         description: str = "",
@@ -109,7 +108,7 @@ class PointsService:
         """扣减积分（事务性：SELECT FOR UPDATE → 检查余额 → 扣减 → 流水 → 同步用户表）。
 
         Returns:
-            {"success": True/False, "balance": int, "transaction_id": str|None}
+            {"success": True/False, "balance": float, "transaction_id": str|None}
         """
         PointsAccountModel, PointsTransactionModel, _, _, UserModel = _get_models()
 
@@ -173,7 +172,7 @@ class PointsService:
         except Exception:
             db.rollback()
             logger.exception(
-                "积分扣减失败: user_id=%s, amount=%d, source=%s",
+                "积分扣减失败: user_id=%s, amount=%.2f, source=%s",
                 user_id,
                 amount,
                 source,
@@ -185,7 +184,7 @@ class PointsService:
     def add_points(
         self,
         user_id: str,
-        amount: int,
+        amount: float,
         source: str,
         db: Session,
         description: str = "",
@@ -242,7 +241,7 @@ class PointsService:
         except Exception:
             db.rollback()
             logger.exception(
-                "积分增加失败: user_id=%s, amount=%d, source=%s",
+                "积分增加失败: user_id=%s, amount=%.2f, source=%s",
                 user_id,
                 amount,
                 source,
@@ -254,7 +253,7 @@ class PointsService:
     def refund_points(
         self,
         user_id: str,
-        amount: int,
+        amount: float,
         source: str,
         db: Session,
         ref_id: str = "",
@@ -270,6 +269,92 @@ class PointsService:
             ref_id=ref_id,
         )
 
+    # ──────────────── 爆款视频（viral_video）动态定价 ────────────────
+
+    def deduct_viral_video(self, user_id: str, credits: float, job_id: str, db: Session) -> dict[str, Any]:
+        """爆款视频预扣积分（confirm-copy 阶段）。"""
+        return self.deduct_points(
+            user_id=user_id,
+            amount=float(credits or 0),
+            source="viral_video",
+            db=db,
+            description="爆款视频生成",
+            ref_id=job_id,
+        )
+
+    def settle_viral_video(
+        self,
+        user_id: str,
+        estimated: float,
+        actual: float,
+        txn_id: str,
+        db: Session,
+    ) -> dict[str, Any]:
+        """爆款视频完成后按实际 tokens 结算（多退少补）。
+
+        - actual < estimated: 退差额
+        - actual > estimated: 补扣差额（余额不足时记 warning，不阻塞完成）
+        - |diff| < 0.01: 不动
+        """
+        diff = round(float(actual or 0) - float(estimated or 0), 2)
+        if abs(diff) < 0.01:
+            return {"success": True, "action": "none", "diff": 0.0}
+        if diff < 0:
+            refund = round(-diff, 2)
+            try:
+                res = self.refund_points(
+                    user_id=user_id,
+                    amount=refund,
+                    source="viral_video",
+                    db=db,
+                    ref_id=txn_id,
+                    description="爆款视频结算退费",
+                )
+                return {"success": bool(res.get("success")), "action": "refund", "diff": -refund, "amount": refund}
+            except Exception:
+                logger.exception("[viral_video] 结算退费异常 user_id=%s refund=%.2f", user_id, refund)
+                return {"success": False, "action": "refund", "diff": -refund}
+        else:
+            extra = round(diff, 2)
+            try:
+                res = self.deduct_points(
+                    user_id=user_id,
+                    amount=extra,
+                    source="viral_video",
+                    db=db,
+                    description="爆款视频结算补扣",
+                    ref_id=txn_id,
+                )
+                if not res.get("success"):
+                    logger.warning(
+                        "[viral_video] 结算补扣余额不足 user_id=%s extra=%.2f balance=%s (不阻塞任务完成)",
+                        user_id,
+                        extra,
+                        res.get("balance"),
+                    )
+                return {"success": bool(res.get("success")), "action": "deduct", "diff": extra, "amount": extra}
+            except Exception:
+                logger.exception("[viral_video] 结算补扣异常 user_id=%s extra=%.2f", user_id, extra)
+                return {"success": False, "action": "deduct", "diff": extra}
+
+    def refund_viral_video(self, user_id: str, credits: float, txn_id: str, db: Session) -> dict[str, Any]:
+        """爆款视频失败全额退款。"""
+        amount = float(credits or 0)
+        if amount <= 0:
+            return {"success": True, "action": "none", "amount": 0.0}
+        try:
+            return self.refund_points(
+                user_id=user_id,
+                amount=amount,
+                source="viral_video",
+                db=db,
+                ref_id=txn_id,
+                description="爆款视频失败退款",
+            )
+        except Exception:
+            logger.exception("[viral_video] 失败退款异常 user_id=%s amount=%.2f", user_id, amount)
+            return {"success": False, "action": "refund", "amount": amount}
+
     # ──────────────── 流水查询 ────────────────
 
     def get_transactions(
@@ -324,132 +409,16 @@ class PointsService:
             "page_size": page_size,
         }
 
-    # ──────────────── 每日免费混剪额度 ────────────────
-
-    def _daily_key(self, user_id: str) -> str:
-        """生成 Redis 每日额度 key。格式: daily_usage:{user_id}:{YYYYMMDD}:free_clip"""
-        today = datetime.now(UTC).strftime("%Y%m%d")
-        return f"daily_usage:{user_id}:{today}:free_clip"
-
-    def check_daily_free_clip(self, user_id: str, db: Session) -> bool:
-        """检查今日是否还有免费混剪额度。
-
-        优先查 Redis，Redis 不可用时降级到 DB。
-        """
-        redis_client = _get_redis_client()
-        if redis_client:
-            try:
-                key = self._daily_key(user_id)
-                current = redis_client.get(key)
-                if current is None:
-                    return True
-                return int(current) < DAILY_FREE_CLIP_LIMIT
-            except Exception:
-                logger.warning("Redis 不可用，降级到 DB 查询每日额度")
-
-        # 降级到 DB
-        _, _, _, DailyUsageRecordModel, _ = _get_models()
-        today_start = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
-        record = (
-            db.query(DailyUsageRecordModel)
-            .filter(
-                DailyUsageRecordModel.user_id == user_id,
-                DailyUsageRecordModel.usage_type == "free_clip",
-                DailyUsageRecordModel.usage_date >= today_start,
-            )
-            .first()
-        )
-        if record is None:
-            return True
-        return record.count < DAILY_FREE_CLIP_LIMIT
-
-    def record_daily_free_clip(self, user_id: str, db: Session) -> bool:
-        """记录使用一次免费混剪。
-
-        先 INCR Redis；如果超限回退 Redis。DB 使用 upsert 语义（唯一约束）。
-        """
-        redis_client = _get_redis_client()
-        if redis_client:
-            try:
-                key = self._daily_key(user_id)
-                new_count = redis_client.incr(key)
-                if new_count == 1:
-                    redis_client.expire(key, 48 * 3600)  # TTL 48h
-                if new_count <= DAILY_FREE_CLIP_LIMIT:
-                    return True
-                # 超限，回退 Redis
-                redis_client.decr(key)
-            except Exception:
-                logger.warning("Redis 不可用，降级到 DB 记录每日额度")
-
-        # 降级/兜底到 DB（upsert 语义）
-        _, _, _, DailyUsageRecordModel, _ = _get_models()
-        today_start = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
-
-        record = (
-            db.query(DailyUsageRecordModel)
-            .filter(
-                DailyUsageRecordModel.user_id == user_id,
-                DailyUsageRecordModel.usage_type == "free_clip",
-                DailyUsageRecordModel.usage_date >= today_start,
-            )
-            .first()
-        )
-
-        if record is None:
-            if DAILY_FREE_CLIP_LIMIT <= 0:
-                return False
-            record = DailyUsageRecordModel(
-                id=uuid.uuid4().hex,
-                user_id=user_id,
-                usage_type="free_clip",
-                usage_date=datetime.now(UTC),
-                count=1,
-            )
-            db.add(record)
-        else:
-            if record.count >= DAILY_FREE_CLIP_LIMIT:
-                return False
-            record.count += 1
-
-        db.commit()
-        return True
+    # ──────────────── 每日免费混剪额度（已下线：智能混剪全免费） ────────────────
 
     def get_daily_usage(self, user_id: str, db: Session) -> dict[str, Any]:
-        """查询今日免费额度使用情况。"""
-        redis_client = _get_redis_client()
-        used = 0
-
-        if redis_client:
-            try:
-                key = self._daily_key(user_id)
-                val = redis_client.get(key)
-                used = int(val) if val else 0
-            except Exception:
-                pass
-
-        if used == 0:
-            # 从 DB 查
-            _, _, _, DailyUsageRecordModel, _ = _get_models()
-            today_start = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
-            record = (
-                db.query(DailyUsageRecordModel)
-                .filter(
-                    DailyUsageRecordModel.user_id == user_id,
-                    DailyUsageRecordModel.usage_type == "free_clip",
-                    DailyUsageRecordModel.usage_date >= today_start,
-                )
-                .first()
-            )
-            used = record.count if record else 0
-
+        """查询今日免费额度使用情况（智能混剪已全免费，返回 unlimited）。"""
         now = datetime.now(UTC)
         tomorrow = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
-
         return {
-            "free_clips_used": used,
-            "free_clips_limit": DAILY_FREE_CLIP_LIMIT,
-            "free_clips_remaining": max(0, DAILY_FREE_CLIP_LIMIT - used),
+            "free_clips_used": 0,
+            "free_clips_limit": -1,  # -1 表示 unlimited
+            "free_clips_remaining": -1,
             "reset_at": tomorrow.isoformat(),
         }
 

packages/domain/viral_video.py

@@ -69,8 +69,6 @@ class PromptType(StrEnum):
     STYLE_CONSTRAINT = "style_constraint"
 
 
-CREDITS_VIRAL_VIDEO_COST = 50
-
 STAGE_LABELS = {
     ViralVideoStage.IMAGE_ANALYSIS: "图片分析",
     ViralVideoStage.VIDEO_ANALYSIS: "视频风格分析",
@@ -117,8 +115,14 @@ class ViralVideoJob:
     # 状态
     id: str = field(default_factory=lambda: uuid4().hex)
     status: ViralVideoStatus = ViralVideoStatus.PENDING
+    current_stage: str = ""  # 细粒度阶段（ViralVideoStage.value，snake_case）
+    phase_message: str = ""  # 阶段中文提示文案，前端轮询直接展示
+    heartbeat_at: datetime | None = None  # worker 心跳时间，用于超时僵尸任务检测
     result_video_url: str = ""
-    credits_cost: int = 0
+    video_resolution: str = "720p"
+    credits_prepaid: float = 0.0
+    credits_transaction_id: str = ""
+    credits_cost: float = 0.0
     error_msg: str = ""
     retry_count: int = 0
     started_at: datetime | None = None
@@ -138,9 +142,19 @@ class ViralVideoJob:
         ):
             raise ValueError(f"Cannot transition from {self.status} to running")
         self.status = ViralVideoStatus.RUNNING
+        now = datetime.now(timezone.utc)
         if self.started_at is None:
-            self.started_at = datetime.now(timezone.utc)
-        self.updated_at = datetime.now(timezone.utc)
+            self.started_at = now
+        self.heartbeat_at = now
+        self.updated_at = now
+
+    def touch_heartbeat(self) -> None:
+        """更新心跳时间（worker 在长任务中周期性调用，用于超时检测）。"""
+        now = datetime.now(timezone.utc)
+        if self.started_at is None:
+            self.started_at = now
+        self.heartbeat_at = now
+        self.updated_at = now
 
     def mark_image_analyzed(self) -> None:
         if self.status not in (ViralVideoStatus.PENDING, ViralVideoStatus.RUNNING):

packages/middleware/points_gate.py

@@ -185,19 +185,6 @@ def _execute_with_gate_impl(
     is_member = getattr(user, "is_member", False)
     member_type = getattr(user, "member_type", None)
 
-    if scene_key == "ai_video":
-        from packages.domain.points_service import PointsService
-
-        svc = PointsService()
-        if not is_member:
-            if svc.check_daily_free_clip(user.id, db):
-                svc.record_daily_free_clip(user.id, db)
-                kwargs["_points_deducted"] = 0
-                kwargs["_is_free_quota"] = True
-                if is_async:
-                    return _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
-                return func(*args, **_filter_kwargs_impl(func, kwargs))
-
     if per_unit is not None:
         total_points = per_unit
     else:

packages/shared/ai_client.py

@@ -40,6 +40,8 @@ class DoubaoClient:
         self.timeout: int = settings.doubao_timeout
         self.max_retries: int = settings.doubao_max_retries
         self.vision_model: str = settings.doubao_vision_model
+        self.vision_lite_model: str = settings.doubao_vision_lite_model
+        self.fast_model: str = settings.doubao_fast_model
 
     def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
         """调用豆包文本 Embedding API，返回浮点向量；失败返回 None。"""
@@ -92,6 +94,7 @@ class DoubaoClient:
         messages: list[dict[str, str]],
         temperature: float = 0.7,
         max_tokens: int = 1024,
+        model: str | None = None,
     ) -> Optional[str]:
         """调用 Chat Completion 接口.
 
@@ -112,7 +115,7 @@ class DoubaoClient:
             "Content-Type": "application/json",
         }
         payload: dict[str, Any] = {
-            "model": self.model,
+            "model": model or self.model,
             "messages": messages,
             "temperature": temperature,
             "max_tokens": max_tokens,
@@ -154,11 +157,12 @@ class DoubaoClient:
         max_tokens: int = 2048,
         temperature: float = 0.3,
         timeout: int | None = None,
+        model: str | None = None,
     ) -> Optional[str]:
         """调用豆包视觉理解 API（OpenAI 兼容多模态格式）.
 
         将 images 附加到最后一条 user message 的 content 中，
-        使用 vision_model（默认 doubao-1-5-vision-pro-250915）。
+        使用 vision_model（默认 doubao-1-5-vision-pro-250328）。
 
         Args:
             messages: 对话消息列表。最后一条 user message 会被注入图片内容。
@@ -204,7 +208,7 @@ class DoubaoClient:
             "Content-Type": "application/json",
         }
         payload: dict[str, Any] = {
-            "model": self.vision_model,
+            "model": model or self.vision_model,
             "messages": vision_messages,
             "temperature": temperature,
             "max_tokens": max_tokens,
@@ -257,27 +261,17 @@ class DoubaoClient:
         reference_images: list[str] | None = None,
         reference_audios: list[str] | None = None,
         reference_videos: list[str] | None = None,
-    ) -> str | None:
-        """调用 Seedance 2.5 文生/图生视频（异步任务→轮询→下载），返回本地 MP4 路径；失败返回 None。
+    ) -> dict | None:
+        """调用 Seedance 2.5 生视频（异步任务→轮询→下载）。
 
-        v1.6: 支持多参考图（产品素材）+ 参考音频（TTS口型驱动）+ 参考视频，单次生成最长 30 秒。
+        成功返回 {"video_path": str, "usage": dict | None}，失败返回 None。
+        usage 是 Seedance 返回的计费信息（含 completion_tokens）。
 
-        Args:
-            prompt: 文本提示词（含完整编导脚本：总览+场景光线+逐镜头时间轴+硬约束+负面词）
-            image_url: 首帧参考图 URL（可选，提供则走图生视频首帧模式，ratio 跟随首帧）
-            duration: 视频时长 4~30 秒
-            ratio: 宽高比 16:9/9:16/1:1/4:3/3:4/21:9/adaptive；image_url 存在时自动忽略
-            resolution: 480p/720p/1080p
-            generate_audio: 是否让模型原生合成音效/BGM（v1.6 默认 True，配合 reference_audios 做口型驱动）
-            watermark: 是否加水印
-            output_dir: 下载目录，默认 /tmp
-            model: 指定模型 ID；空则用 settings.doubao_video_model
-            reference_images: 多参考图 URL 列表（产品素材，最多30张；注意 image_url 为首帧单独传）
-            reference_audios: 参考音频 URL 列表（TTS口播，驱动口型，最多10段）
-            reference_videos: 参考视频 URL 列表（风格参考）
-
-        Returns:
-            本地 MP4 文件路径，失败返回 None。
+        【v1.6.1 修复】严格按官方 content 数组协议构造请求：
+        - 所有参考（图/音/视）必须放进 content 数组并带 role 字段，不能放顶层 reference_audios/reference_videos（非官方字段，会被忽略或导致异常）。
+        - 首帧图（first_frame 模式）Seedance 2.5 强制 ratio=adaptive；走 omni_reference（参考生视频）模式时才能指定 9:16/1:1 等具体比例。
+          判定：传了参考音频/视频或 ≥1 张多参考图时，走 omni_reference（首张图 role=reference_image）；纯首帧无参考时走 first_frame（ratio 强制 adaptive）。
+        - 创建任务若因 ratio 报错（HTTP 400），自动回退到 ratio=adaptive 重试一次。
         """
         if not self.is_available:
             return None
@@ -286,18 +280,48 @@ class DoubaoClient:
 
         settings = get_shared_settings()
         poll_interval = getattr(settings, "doubao_video_poll_interval", 10) or 10
-        total_timeout = getattr(settings, "doubao_video_timeout", 900) or 900
+        # 收紧总超时：轮询 8min + 下载 2min = 最长 ~10min，防止出现 20min 卡死
+        total_timeout = getattr(settings, "doubao_video_timeout", 480) or 480
         default_video_model = getattr(settings, "doubao_video_model", None) or "doubao-seedance-2-5-260628"
         video_model = model or default_video_model
 
+        ref_audios = [u for u in (reference_audios or [])[:10] if u and isinstance(u, str)]
+        ref_videos = [u for u in (reference_videos or [])[:3] if u and isinstance(u, str)]
+        ref_imgs = [u for u in (reference_images or [])[:9] if u and isinstance(u, str)]
+
+        # 判断任务模式：有参考音/视/多图 → omni_reference（支持指定 ratio）；纯首帧 → first_frame（ratio=adaptive）
+        has_extra_refs = bool(ref_audios or ref_videos or ref_imgs)
+        is_first_frame_mode = bool(image_url) and not has_extra_refs
+        # 最终 ratio：first_frame 模式强制 adaptive，否则按用户传值（默认 9:16）
+        final_ratio = "adaptive" if is_first_frame_mode else (ratio or "9:16")
+
+        # 构造 content 数组：text + 图 + 音 + 视
         content: list[dict[str, Any]] = [{"type": "text", "text": prompt.strip()}]
         if image_url:
-            content.append({"type": "image_url", "image_url": {"url": image_url}})
-        # v1.6: 多参考图（产品素材）
-        if reference_images:
-            for url in reference_images[:30]:
-                if url and isinstance(url, str):
-                    content.append({"type": "image_url", "image_url": {"url": url}})
+            if has_extra_refs:
+                # omni_reference：首张图作为 reference_image，允许指定 ratio
+                content.append(
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": image_url},
+                        "role": "reference_image",
+                    }
+                )
+            else:
+                # 纯首帧：不带 role，服务端识别为 first_frame（或显式 role=first_frame）
+                content.append(
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": image_url},
+                        "role": "first_frame",
+                    }
+                )
+        for u in ref_imgs:
+            content.append({"type": "image_url", "image_url": {"url": u}, "role": "reference_image"})
+        for u in ref_audios:
+            content.append({"type": "audio_url", "audio_url": {"url": u}, "role": "reference_audio"})
+        for u in ref_videos:
+            content.append({"type": "video_url", "video_url": {"url": u}, "role": "reference_video"})
 
         create_payload: dict[str, Any] = {
             "model": video_model,
@@ -306,18 +330,8 @@ class DoubaoClient:
             "duration": int(duration),
             "resolution": resolution,
             "watermark": bool(watermark),
+            "ratio": final_ratio,
         }
-        # v1.6: 参考音频（TTS 驱动口型）
-        if reference_audios:
-            create_payload["reference_audios"] = [
-                {"url": u, "role": "audio_url"} for u in reference_audios[:10] if u and isinstance(u, str)
-            ]
-        # v1.6: 参考视频（风格参考）
-        if reference_videos:
-            create_payload["reference_videos"] = [{"url": u} for u in reference_videos[:5] if u and isinstance(u, str)]
-        # Bug #2110 / v1.6: ratio=None 时不传（首帧图生视频跟随原图比例）
-        if ratio and not image_url:
-            create_payload["ratio"] = ratio
 
         headers = {
             "Authorization": f"Bearer {self.api_key}",
@@ -325,78 +339,95 @@ class DoubaoClient:
         }
         create_url = f"{self.base_url}/contents/generations/tasks"
         logger.info(
-            "Seedance 创建任务请求: url=%s model=%s duration=%ds ratio=%s gen_audio=%s image_url=%s ref_imgs=%d ref_audios=%d ref_videos=%d",
-            create_url,
+            "Seedance 创建任务: model=%s dur=%ds ratio=%s mode=%s gen_audio=%s img=%d aud=%d vid=%d",
             video_model,
             duration,
-            ratio or "(follow-image)",
+            final_ratio,
+            "first_frame" if is_first_frame_mode else "omni_ref",
             generate_audio,
-            bool(image_url),
-            len(reference_images or []),
-            len(reference_audios or []),
-            len(reference_videos or []),
+            (1 if image_url else 0) + len(ref_imgs),
+            len(ref_audios),
+            len(ref_videos),
         )
 
-        # 1) 创建任务（带重试）
-        task_id: str | None = None
-        last_error: Exception | None = None
-        for attempt in range(self.max_retries + 1):
-            try:
-                resp = httpx.post(create_url, headers=headers, json=create_payload, timeout=self.timeout)
-                # 测试环境下 MagicMock().status_code 是 MagicMock，与 int 比较会抛 TypeError；
-                # 用显式 int() 转换+类型判断，避免误判。
+        def _do_create(payload: dict) -> tuple[str | None, Exception | None, int, str]:
+            """返回 (task_id, last_err, status_code, body_text)。"""
+            last_err: Exception | None = None
+            for attempt in range(self.max_retries + 1):
                 try:
-                    _status = int(resp.status_code)
-                except (TypeError, ValueError):
-                    _status = 200
-                if _status >= 400:
-                    # 把响应体完整打出来（通常含 error.code/message，能直接定位：模型未开通/Key 无权限/模型 ID 错误）
-                    logger.error(
-                        "Seedance 创建任务 HTTP %d: body=%s",
-                        resp.status_code,
-                        (resp.text or "")[:1000],
-                    )
-                resp.raise_for_status()
-                data = resp.json()
-                task_id = data.get("id")
-                if task_id:
-                    break
-                last_error = RuntimeError(f"create task returned no id: {str(data)[:200]}")
-            except Exception as e:
-                last_error = e
-                if attempt < self.max_retries:
-                    wait = 0.5 * (2**attempt)
-                    logger.warning(
-                        "Seedance 创建任务失败，%.1fs 后重试 (%d/%d): %s", wait, attempt + 1, self.max_retries + 1, e
-                    )
-                    time.sleep(wait)
+                    resp = httpx.post(create_url, headers=headers, json=payload, timeout=self.timeout)
+                    sc = int(getattr(resp, "status_code", 0) or 0)
+                    body = (getattr(resp, "text", "") or "")[:1500]
+                    if sc >= 400:
+                        logger.error("Seedance 创建任务 HTTP %d: body=%s", sc, body)
+                        try:
+                            resp.raise_for_status()
+                        except Exception as ee:
+                            last_err = ee
+                            if attempt < self.max_retries:
+                                time.sleep(0.5 * (2**attempt))
+                                continue
+                            return None, last_err, sc, body
+                    data = resp.json()
+                    tid = data.get("id")
+                    if tid:
+                        return tid, None, sc, body
+                    last_err = RuntimeError(f"create ok but no id: {str(data)[:300]}")
+                except Exception as e:
+                    last_err = e
+                    if attempt < self.max_retries:
+                        wait = 0.5 * (2**attempt)
+                        logger.warning(
+                            "Seedance 创建任务失败，%.1fs 后重试 (%d/%d): %s",
+                            wait,
+                            attempt + 1,
+                            self.max_retries + 1,
+                            e,
+                        )
+                        time.sleep(wait)
+            return None, last_err, 0, ""
+
+        # 第一次尝试
+        task_id, last_err, sc, body = _do_create(create_payload)
+
+        # ratio 兜底：HTTP 400 且 body 提到 ratio / adaptive → 回退 adaptive 再试一次
+        if (
+            not task_id
+            and sc == 400
+            and final_ratio != "adaptive"
+            and (
+                "ratio" in (body or "").lower()
+                or "aspect" in (body or "").lower()
+                or "adaptive" in (body or "").lower()
+            )
+        ):
+            logger.warning("Seedance 创建因 ratio 失败，回退 ratio=adaptive 重试")
+            create_payload["ratio"] = "adaptive"
+            task_id, last_err, sc2, body2 = _do_create(create_payload)
+
         if not task_id:
             logger.error(
-                "Seedance 创建任务最终失败: model=%s base_url=%s err=%s  【排查建议】"
-                "1) 确认方舟控制台已开通 Doubao-Seedance-2.5 模型；"
-                "2) DOUBAO_API_KEY 对应的账号有该模型调用权限；"
-                "3) DOUBAO_BASE_URL 必须为 https://ark.cn-beijing.volces.com/api/v3；"
-                "4) 若控制台用「推理接入点」(endpoint)，请把 DOUBAO_VIDEO_MODEL 改为 ep-xxx 接入点 ID。",
+                "Seedance 创建任务最终失败: model=%s base_url=%s err=%s body=%s  【排查】"
+                "1) 方舟控制台已开通 doubao-seedance-2-5-260628；2) API Key 有该模型权限；"
+                "3) DOUBAO_BASE_URL=https://ark.cn-beijing.volces.com/api/v3；4) 参考素材 URL 公网可访问。",
                 video_model,
                 self.base_url,
-                last_error,
+                last_err,
+                (body or "")[:500],
             )
             return None
 
-        logger.info(
-            "Seedance 任务已创建: task_id=%s model=%s duration=%ds gen_audio=%s",
-            task_id,
-            video_model,
-            duration,
-            generate_audio,
-        )
+        logger.info("Seedance 任务已创建: task_id=%s ratio=%s", task_id, create_payload["ratio"])
 
         # 2) 轮询状态
         poll_url = f"{create_url}/{task_id}"
         deadline = time.time() + total_timeout
         video_url: str | None = None
+        usage: dict | None = None
         last_status: str = "queued"
+        poll_count = 0
         while time.time() < deadline:
+            poll_count += 1
             try:
                 resp = httpx.get(poll_url, headers=headers, timeout=self.timeout)
                 try:
@@ -410,49 +441,73 @@ class DoubaoClient:
                 if status == "succeeded":
                     content_obj = data.get("content") or {}
                     video_url = content_obj.get("video_url")
+                    usage = data.get("usage") or content_obj.get("usage") or None
                     if video_url:
+                        logger.info("Seedance 任务成功: task_id=%s polls=%d usage=%s", task_id, poll_count, usage)
                         break
-                    last_error = RuntimeError(f"task succeeded but no video_url: {str(data)[:300]}")
+                    last_err = RuntimeError(f"task succeeded but no video_url: {str(data)[:300]}")
+                    logger.error("Seedance succeeded 但无 video_url: %s", last_err)
                     break
                 if status == "failed":
                     err = data.get("error") or {}
-                    last_error = RuntimeError(f"task failed: {err.get('code','')} {err.get('message','')}")
+                    last_err = RuntimeError(f"task failed: code={err.get('code','')} msg={err.get('message','')}")
+                    logger.error("Seedance 任务失败 task_id=%s: %s", task_id, last_err)
                     break
                 if status in ("expired", "cancelled"):
-                    last_error = RuntimeError(f"task {status}")
+                    last_err = RuntimeError(f"task {status}")
+                    logger.error("Seedance 任务 %s: task_id=%s", status, task_id)
                     break
-                # queued / running: 继续轮询
+                # 每 5 次轮询打一次 info 日志，便于观察进度
+                if poll_count % 5 == 0:
+                    logger.info("Seedance 轮询中: task_id=%s status=%s polls=%d", task_id, status, poll_count)
             except httpx.HTTPStatusError as e:
-                last_error = e
-                logger.warning(
-                    "Seedance 轮询 HTTP %d: body=%s",
-                    e.response.status_code,
-                    (e.response.text or "")[:500],
-                )
+                last_err = e
+                logger.warning("Seedance 轮询 HTTP %d: %s", e.response.status_code, (e.response.text or "")[:300])
             except Exception as e:
-                last_error = e
+                last_err = e
                 logger.debug("Seedance 轮询异常: %s", e)
             time.sleep(poll_interval)
 
         if not video_url:
-            logger.error("Seedance 任务未成功: task_id=%s status=%s err=%s", task_id, last_status, last_error)
+            logger.error(
+                "Seedance 任务未成功: task_id=%s last_status=%s polls=%d err=%s (总等待 %.0fs)",
+                task_id,
+                last_status,
+                poll_count,
+                last_err,
+                total_timeout,
+            )
             return None
 
-        # 3) 下载到本地
+        # 3) 下载到本地（下载超时收紧到 120s）
         try:
             out_dir = output_dir or "/tmp"
             os.makedirs(out_dir, exist_ok=True)
             local_path = f"{out_dir}/seedance_{task_id}_{uuid.uuid4().hex[:8]}.mp4"
-            with httpx.stream("GET", video_url, timeout=300) as r:
+            download_timeout = 120.0
+            logger.info(
+                "Seedance 开始下载: task_id=%s url=%s timeout=%.0fs", task_id, video_url[:120], download_timeout
+            )
+            with httpx.stream("GET", video_url, timeout=download_timeout) as r:
                 r.raise_for_status()
+                downloaded = 0
                 with open(local_path, "wb") as f:
                     for chunk in r.iter_bytes(chunk_size=1024 * 256):
                         if chunk:
                             f.write(chunk)
-            logger.info("Seedance 视频下载完成: %s (%d bytes)", local_path, os.path.getsize(local_path))
-            return local_path
+                            downloaded += len(chunk)
+            size = os.path.getsize(local_path)
+            logger.info("Seedance 视频下载完成: %s size=%d bytes", local_path, size)
+            if size == 0:
+                logger.error("Seedance 下载文件大小为 0")
+                try:
+                    os.remove(local_path)
+                except Exception:
+                    pass
+                return None
+            return {"video_path": local_path, "usage": usage}
         except Exception as e:
-            logger.error("Seedance 视频下载失败: %s", e)
+            logger.error("Seedance 视频下载失败: %s", e, exc_info=True)
             return None
 
 

packages/shared/ai_service.py

@@ -496,16 +496,32 @@ def run_generate_cover(
 # ── 通用 LLM / Vision 调用（#2039 ViralVideoOrchestrator 使用，复用现有豆包客户端）──
 
 
-def call_llm(prompt: str, temperature: float = 0.7) -> object:
-    """调用豆包大模型（文本对话），返回解析后的 JSON（dict/list）或原文字符串；失败返回 None。"""
+def call_llm(
+    prompt: str,
+    temperature: float = 0.7,
+    max_tokens: int = 2048,
+    model: str | None = None,
+    system_prompt: str | None = None,
+) -> object:
+    """调用豆包大模型（文本对话），返回解析后的 JSON（dict/list）或原文字符串；失败返回 None。
+
+    Args:
+        prompt: 用户侧提示。
+        temperature: 采样温度。
+        max_tokens: 输出上限（结构化任务默认 2048，长文案可按需加大）。
+        model: 覆盖默认模型（如 fast_model 提速用），None 走配置默认推理模型。
+        system_prompt: 覆盖默认 system prompt。
+    """
     client = get_doubao_client()
     if not client.is_available:
         return None
+    if system_prompt is None:
+        system_prompt = "你是专业的短视频内容策划助手。需要结构化输出时请严格使用 JSON。"
     messages = [
-        {"role": "system", "content": "你是专业的短视频内容策划助手。需要结构化输出时请严格使用 JSON。"},
+        {"role": "system", "content": system_prompt},
         {"role": "user", "content": prompt},
     ]
-    raw = client.chat_completion(messages, temperature=temperature, max_tokens=4096)
+    raw = client.chat_completion(messages, temperature=temperature, max_tokens=max_tokens, model=model)
     if raw is None:
         return None
     try:
@@ -514,14 +530,26 @@ def call_llm(prompt: str, temperature: float = 0.7) -> object:
         return raw
 
 
-def call_vision(image_url: str, prompt: str) -> object:
+def call_vision(
+    image_url: str,
+    prompt: str,
+    *,
+    model: str | None = None,
+    max_tokens: int = 1024,
+    temperature: float = 0.2,
+    timeout: int = 45,
+    system_prompt: str | None = None,
+) -> object:
     """调用豆包视觉大模型分析图片，返回解析后的 JSON 或原文字符串；失败返回 None。
 
-    Bug #2114 (VLM 牛头不对马嘴根因修复):
-    之前误走 client.chat_completion(用文本模型 doubao-seed-1.6)，多模态 content list 被当成
-    纯文本发给文本模型 → 模型要么看不到图、要么抛 400，静默被 except 吞掉 → 返回 None →
-    _step_image_analysis fallback 到 {"name":"未识别"} → 后续文案/分镜完全没图的信息。
-    现改走 vision_completion，走视觉模型 doubao-1-5-vision-pro-250915。
+    Args:
+        image_url: 可公网访问的图片 URL（直接传给豆包视觉模型，无需本地下载）。
+        prompt: 用户侧文本提示。
+        model: 覆盖默认视觉模型（如 vision_lite_model 提速用），None 走配置默认。
+        max_tokens: 输出上限，商品识别用 800~1200 足够，避免长输出拖慢首 token。
+        temperature: 温度。
+        timeout: 单次请求超时（秒）。
+        system_prompt: 覆盖默认 system prompt（viral-video 商品分析会传专门的详细 prompt）。
     """
     client = get_doubao_client()
     if not client.is_available:
@@ -531,28 +559,33 @@ def call_vision(image_url: str, prompt: str) -> object:
         logger.warning("[call_vision] 空 image_url，跳过视觉分析")
         return None
 
-    system_prompt = (
-        "你是资深电商视觉分析师。请严格基于用户提供的图片观察回答，"
-        "图片里没有的信息不要凭空想象或编造；看不清或无法判断时明确说"
-        "「图片中无法判断」，不要猜测。输出必须是严格 JSON，不要附加 Markdown 或解释文字。"
-    )
+    if system_prompt is None:
+        system_prompt = (
+            "你是资深电商视觉分析师。请严格基于用户提供的图片观察回答，"
+            "图片里没有的信息不要凭空想象或编造；看不清或无法判断时明确说"
+            "「无法判断」，不要猜测。输出必须是严格 JSON，不要附加 Markdown 或解释文字。"
+        )
     messages = [
         {"role": "system", "content": system_prompt},
         {"role": "user", "content": prompt},
     ]
 
+    used_model = model or getattr(client, "vision_model", "?")
     logger.info(
-        "[call_vision] 调用豆包视觉模型 vision_model=%s image_url=%s prompt_len=%d",
-        getattr(client, "vision_model", "?"),
+        "[call_vision] 调用豆包视觉模型 model=%s image_url=%s prompt_len=%d max_tokens=%d timeout=%d",
+        used_model,
         image_url[:120],
         len(prompt),
+        max_tokens,
+        timeout,
     )
     raw = client.vision_completion(
         messages=messages,
         images=[image_url],
-        temperature=0.2,
-        max_tokens=2048,
-        timeout=60,
+        temperature=temperature,
+        max_tokens=max_tokens,
+        timeout=timeout,
+        model=model,
     )
     if raw is None:
         logger.warning("[call_vision] 视觉模型返回 None (image_url=%s)", image_url[:80])
@@ -584,21 +617,23 @@ def call_video_generation(
     reference_images: list[str] | None = None,
     reference_audios: list[str] | None = None,
     reference_videos: list[str] | None = None,
-) -> str | None:
-    """调用 Seedance 2.5 生成视频（v1.6 单次出片版），返回本地 MP4 路径；失败返回 None。
+) -> dict | None:
+    """调用 Seedance 2.5 生成视频（v1.6.1 单次出片版）。
 
-    v1.6:
-    - 默认 generate_audio=True，模型原生合成环境音效/BGM；
-    - reference_audios 传 TTS 音频 URL 数组做口型驱动；
-    - reference_images 传产品素材 URL 数组做视觉参考；
-    - 单次最长 30 秒，不分段不拼接；
-    - image_url 存在时为「首帧图生视频」模式，自动不传 ratio（Bug #2110）。
+    成功返回 {"video_path": str, "usage": dict | None}（usage 含 completion_tokens），失败返回 None。
+
+    v1.6.1 关键约束（避免 20min 卡死）：
+    - 参考音频/视频/多图全部放进 content 数组并带 role=reference_audio/reference_video/reference_image；
+    - 纯首帧无参考时（first_frame 模式），Seedance 2.5 强制 ratio=adaptive；
+      传了参考音/视/多图时走 omni_reference 模式，ratio 可指定为 9:16（客户端内部自动判断）。
+    - ratio 默认 9:16（竖屏），客户端会根据是否有参考自动在 first_frame/adaptive 与 omni/9:16 间切换；
+      若创建任务因 ratio 报错（HTTP 400），客户端会自动回退到 adaptive 再试一次。
     """
     client = get_doubao_client()
     if not client.is_available:
         logger.warning("[ai_service] 豆包客户端未配置，跳过视频生成")
         return None
-    effective_ratio = None if image_url else ratio
+    effective_ratio = ratio or "9:16"
     try:
         kwargs: dict = dict(
             prompt=prompt,

scripts/render_env.sh

@@ -57,7 +57,7 @@ if [ "$TARGET_ENV" = "staging" ]; then
 fi
 
 # 共用 secrets 直接导出（如果存在）
-SHARED_SECRETS="OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET COSYVOICE_API_KEY DASHSCOPE_API_KEY MEDIAKIT_API_KEY DOUBAO_API_KEY DOUBAO_MODEL DOUBAO_BASE_URL DOUBAO_VISION_MODEL WECHAT_APP_ID WECHAT_APP_SECRET TIKHUB_API_KEY APIZERO_API_KEY GPU_WORKER_TOKEN"
+SHARED_SECRETS="OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET COSYVOICE_API_KEY DASHSCOPE_API_KEY MEDIAKIT_API_KEY DOUBAO_API_KEY DOUBAO_MODEL DOUBAO_FAST_MODEL DOUBAO_BASE_URL DOUBAO_VISION_MODEL DOUBAO_VISION_LITE_MODEL DOUBAO_VISION_USE_LITE WECHAT_APP_ID WECHAT_APP_SECRET TIKHUB_API_KEY APIZERO_API_KEY GPU_WORKER_TOKEN"
 for var in $SHARED_SECRETS; do
   value="${!var:-}"
   # 已经在环境中了，无需额外操作

tests/unit/test_1970_narrative_service.py

@@ -377,32 +377,12 @@ class TestPrepareNarrativeVoice:
         assert ei.value.status_code == 502
         assert "配音合成失败" in ei.value.message
 
-    def test_points_insufficient_402(self, monkeypatch):
-        class FakePoints:
-            def deduct_points(self, *a, **k):
-                return {"success": False, "balance": 0}
+    def test_no_points_service_invoked(self, monkeypatch):
+        """v1.6.2: 叙事配音已免费，不再实例化 PointsService / 扣点/退费。"""
+        # 确认 narrative_service 已不再暴露 PointsService
+        assert not hasattr(ns, "PointsService"), "narrative_service 不应再导入 PointsService"
 
-        monkeypatch.setattr(ns, "PointsService", lambda: FakePoints())
-        deps = self._deps(points_enabled=True)
-        with pytest.raises(NarrativeError) as ei:
-            prepare_narrative_voice(**deps)
-        assert ei.value.status_code == 402
-
-    def test_points_refund_on_failure(self, monkeypatch):
-        class FakePoints:
-            def __init__(self):
-                self.refunded = 0
-
-            def deduct_points(self, *a, **k):
-                return {"success": True, "balance": 100}
-
-            def refund_points(self, user_id, amount, source, db, ref_id="", **k):
-                self.refunded += amount
-
-        points = FakePoints()
-        monkeypatch.setattr(ns, "PointsService", lambda: points)
-
-        class FailingWorkflow:
+        class FakeWorkflow:
             def __init__(self, *, repository, cosyvoice_service):
                 pass
 
@@ -412,11 +392,22 @@ class TestPrepareNarrativeVoice:
             def process_synthesis_failure(self, job_id, error):
                 return None
 
-        monkeypatch.setattr(ns, "TTSWorkflowService", FailingWorkflow)
+        monkeypatch.setattr(ns, "TTSWorkflowService", FakeWorkflow)
         deps = self._deps(points_enabled=True)
-        with pytest.raises(NarrativeError):
+        with pytest.raises(NarrativeError) as ei:
             prepare_narrative_voice(**deps)
-        assert points.refunded > 0
+        # 走 502 业务错误路径，不再退费
+        assert ei.value.status_code == 502
+
+    def test_module_has_no_points_imports(self):
+        """模块源码不再包含扣点相关符号。"""
+        import inspect
+
+        src = inspect.getsource(ns)
+        assert "PointsService" not in src
+        assert "calculate_points_cost" not in src
+        assert "_points_scene" not in src
+        assert "_POINTS_SCENE" not in src
 
     def test_clone_source_resolves_profile(self, monkeypatch):
         captured = {}

tests/unit/test_2035_coverage.py

@@ -14,10 +14,13 @@ from packages.shared.ai_client import DoubaoClient
 class _FakeSettings:
     doubao_api_key = "test-key"
     doubao_model = "test-model"
+    doubao_fast_model = "test-fast-model"
     doubao_base_url = "https://ark.cn-beijing.volces.com/api/v3"
     doubao_timeout = 10
     doubao_max_retries = 0
     doubao_vision_model = "test-vision"
+    doubao_vision_lite_model = "test-vision-lite"
+    doubao_vision_use_lite = False
     doubao_embedding_model = "test-embedding"
 
 
@@ -84,7 +87,9 @@ _GEN_TASKS_PATH = Path(__file__).resolve().parents[2] / "apps/api/app/api/routes
 def _load_infer_func():
     src = _GEN_TASKS_PATH.read_text()
     start = src.index("# #2035：文案关键词")
-    end = src.index("from packages.middleware")
+    # 用紧跟 _infer_expected_categories 后的 logger 行作为结束锚点
+    end_marker = "\nlogger = logging.getLogger"
+    end = src.index(end_marker, start)
     code = src[start:end]
     ns: dict = {}
     exec(code, ns)

tests/unit/test_ai_avatar_render_points.py

@@ -1,48 +1,25 @@
-"""AI数字人渲染 积分扣点单元测试 (#1895 P2 step 2.6)"""
+"""AI 数字人渲染 — v1.6.2 起免费，不扣积分"""
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
 
-import pytest
-
-import packages.middleware.points_gate as _pg_module
-
-
-@pytest.fixture(autouse=True)
-def _enable(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-    yield
-
-
-class TestAiAvatarRenderPoints:
-    def test_ai_digital_human_per_unit(self):
+class TestAiAvatarRenderFree:
+    def test_ai_digital_human_returns_zero_cost(self):
         from packages.domain.points_rules import calculate_points_cost
 
-        cost = calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=1)
-        assert cost >= 15
+        assert calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=1) == 0
+        assert calculate_points_cost("ai_digital_human", is_member=True, duration_minutes=5) == 0
 
-    def test_decorator_attached(self):
+    def test_no_points_gate_decorator(self):
         from app.api.routes.ai_avatar_render import create_render_job
 
-        assert hasattr(create_render_job, "__wrapped__"), "missing @points_gate"
+        assert not hasattr(create_render_job, "__wrapped__")
 
-    def test_insufficient_raises_402(self):
-        from app.api.routes.ai_avatar_render import create_render_job
-        from app.schemas.ai_avatar_render import CreateAiAvatarRenderRequest
-        from fastapi import HTTPException
+    def test_module_has_no_points_imports(self):
+        import inspect
 
-        db = MagicMock()
-        cu = MagicMock()
-        cu.user.id = "u1"
-        cu.user.is_member = False
-        cu.user.member_type = None
-        svc = MagicMock()
-        body = CreateAiAvatarRenderRequest(lipsync_job_id="lip1")
-        with patch("packages.domain.points_service.PointsService") as MS:
-            msvc = MagicMock()
-            msvc.deduct_points.return_value = {"success": False, "balance": 0}
-            MS.return_value = msvc
-            with pytest.raises(HTTPException) as ei:
-                create_render_job(body=body, current_user=cu, svc=svc, db=db)
-            assert ei.value.status_code == 402
+        from app.api.routes import ai_avatar_render as mod
+
+        src = inspect.getsource(mod)
+        assert "PointsService" not in src
+        assert "points_gate" not in src

tests/unit/test_ai_client_video.py

@@ -112,10 +112,10 @@ class TestVideoGenerationHappyPath:
                 resolution="720p",
                 output_dir=str(tmp_path),
             )
-        assert out is not None
-        assert Path(out).exists()
-        assert Path(out).name == "seedance_task-001_abcd1234.mp4"
-        assert Path(out).read_bytes() == b"FAKEMP4DATA"
+        assert out is not None and isinstance(out, dict)
+        assert Path(out["video_path"]).exists()
+        assert Path(out["video_path"]).name == "seedance_task-001_abcd1234.mp4"
+        assert Path(out["video_path"]).read_bytes() == b"FAKEMP4DATA"
         assert calls["post"] == 1
         assert calls["get"] == 1
 
@@ -325,8 +325,8 @@ class TestVideoGenerationPollLoop:
                 doubao_video_poll_interval=0, doubao_video_timeout=200, doubao_video_model="seedance"
             )
             out = client.video_generation("p", output_dir=str(tmp_path))
-        assert out is not None
-        assert Path(out).read_bytes() == b"DATA"
+        assert out is not None and isinstance(out, dict)
+        assert Path(out["video_path"]).read_bytes() == b"DATA"
         # queued 和 running 各 sleep 一次
         assert len(sleeps) >= 2
 
@@ -373,8 +373,8 @@ class TestVideoGenerationPollLoop:
                 doubao_video_poll_interval=0, doubao_video_timeout=100, doubao_video_model="seedance"
             )
             out = client.video_generation("p", output_dir=str(tmp_path))
-        assert out is not None
-        assert Path(out).exists()
+        assert out is not None and isinstance(out, dict)
+        assert Path(out["video_path"]).exists()
         assert poll_calls["n"] == 2
 
     def test_default_output_dir_and_audio_watermark(self, tmp_path, monkeypatch):
@@ -442,8 +442,8 @@ class TestVideoGenerationPollLoop:
             out = client.video_generation(
                 "p", duration=3, ratio="1:1", resolution="480p", generate_audio=True, watermark=True
             )
-        assert out is not None
-        assert "/tmp/seedance_t-default_00000001.mp4" in out
+        assert out is not None and isinstance(out, dict)
+        assert out["video_path"] == "/tmp/seedance_t-default_00000001.mp4"
         assert captured["json"]["generate_audio"] is True
         assert captured["json"]["watermark"] is True
         assert captured["json"]["ratio"] == "1:1"

tests/unit/test_credit_system_switch.py

@@ -95,25 +95,29 @@ class TestCheckEndpointWhenDisabled:
         # 不再走免费额度判定
         svc.check_daily_free_clip.assert_not_called()
 
-    def test_unknown_scene_still_400_when_disabled(self):
-        """未知 scene 即使系统关闭也返回 400（参数校验先于开关）。"""
+    def test_unknown_scene_allowed_when_disabled(self):
+        """任意 scene_key（含未知/已下线）系统关闭时都返回 allowed=True, cost=0。"""
         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
+        svc = MagicMock()
+        svc.get_or_create_account.return_value = {"balance": 0}
+        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=PointsCheckRequest(scene_key="nope"), current_user=_make_cu(), db=MagicMock())
+        assert resp.allowed is True
+        assert resp.required_points == 0
 
     def test_check_enabled_calculates_cost(self):
-        """开关开启时保持原有计费校验。"""
+        """开关开启时保持原有计费校验（voice_clone_synth 正常计费）。"""
         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)
+        body = PointsCheckRequest(scene_key="voice_clone_synth", quantity=1, duration_minutes=1)
 
         with (
             patch("app.api.routes.points._credits_enabled", return_value=True),
@@ -123,6 +127,23 @@ class TestCheckEndpointWhenDisabled:
 
         assert resp.required_points == 2  # 免费用户 ceil(1*1.15)=2
 
+    def test_retired_scene_free_when_enabled(self):
+        """开关开启时，已下线场景返回 cost=0，直接放行。"""
+        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}
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=True),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            for scene in ["ai_voice", "ai_title", "ai_video", "ai_digital_human", "nope"]:
+                body = PointsCheckRequest(scene_key=scene, quantity=1)
+                resp = check_points(body=body, current_user=_make_cu(), db=MagicMock())
+                assert resp.required_points == 0, f"{scene} should be free"
+                assert resp.allowed is True
+
 
 # ── /points/deduct：关闭时 no-op，余额不变 ────────────────────────────────
 
@@ -217,13 +238,24 @@ class TestQueryEndpointsRemainAvailable:
 
 
 class TestBusinessRoutesBypassWhenDisabled:
-    def test_lipsync_route_skips_points(self):
-        """lipsync 创建任务路由：settings.points_enabled=False 时不构造 PointsService。"""
+    def test_lipsync_route_has_no_points_logic(self):
+        """lipsync 路由：已移除手动扣点代码（不导入 PointsService/calculate_points_cost）。"""
+        import inspect
+
         from app.api.routes import lipsync as lipsync_mod
 
-        assert bool(getattr(lipsync_mod.settings, "points_enabled", False)) is False
+        src = inspect.getsource(lipsync_mod)
+        assert "PointsService" not in src
+        assert "calculate_points_cost" not in src
+        assert "_points_deducted" not in src
+
+    def test_tts_route_has_no_points_logic(self):
+        """tts 路由：已移除手动扣点代码（不导入 PointsService/calculate_points_cost）。"""
+        import inspect
 
-    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
+        src = inspect.getsource(tts_mod)
+        assert "PointsService" not in src
+        assert "calculate_points_cost" not in src
+        assert "_points_deducted" not in src

tests/unit/test_generation_cover_points.py

@@ -1,28 +1,25 @@
-"""AI封面生成 积分扣点单元测试 (#1895 P2 step 2.7)"""
+"""封面生成 — v1.6.2 起免费，不扣积分"""
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
 
-import pytest
-
-import packages.middleware.points_gate as _pg_module
-
-
-@pytest.fixture(autouse=True)
-def _enable(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-    yield
-
-
-class TestGenerationCoverPoints:
-    def test_ai_cover_cost(self):
+class TestGenerationCoverFree:
+    def test_ai_cover_returns_zero_cost(self):
         from packages.domain.points_rules import calculate_points_cost
 
-        assert calculate_points_cost("ai_cover", is_member=False) == 2
-        assert calculate_points_cost("ai_cover", is_member=True, member_type="yearly") >= 0
+        assert calculate_points_cost("ai_cover", is_member=False, quantity=1) == 0
+        assert calculate_points_cost("ai_cover", is_member=True, quantity=10) == 0
 
-    def test_decorator_attached(self):
+    def test_no_points_gate_decorator(self):
         from app.api.routes.generation_cover import generate_cover
 
-        assert hasattr(generate_cover, "__wrapped__"), "missing @points_gate"
+        assert not hasattr(generate_cover, "__wrapped__")
+
+    def test_endpoint_has_no_points_logic(self):
+        import inspect
+
+        from app.api.routes import generation_cover as mod
+
+        src = inspect.getsource(mod)
+        assert "PointsService" not in src
+        assert "deduct_points" not in src

tests/unit/test_generation_preview_points.py

@@ -1,54 +1,31 @@
-"""视频预览生成 积分扣点单元测试 (#1895 P2 step 2.5)"""
+"""视频预览生成 — v1.6.2 起免费，不扣积分"""
 
 from __future__ import annotations
 
 from unittest.mock import MagicMock, patch
 
-import pytest
 
-import packages.middleware.points_gate as _pg_module
-
-
-@pytest.fixture(autouse=True)
-def _enable(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-    yield
-
-
-class TestGenerationPreviewPoints:
-    def test_ai_video_cost(self):
+class TestGenerationPreviewFree:
+    def test_ai_video_returns_zero_cost(self):
         from packages.domain.points_rules import calculate_points_cost
 
-        assert calculate_points_cost("ai_video", is_member=False) == 4
-        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
+        assert calculate_points_cost("ai_video", is_member=False) == 0
+        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly", duration_minutes=10) == 0
 
-    def test_insufficient_raises_402(self):
-        from app.api.routes.generation_preview import create_preview_generation_task
-        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
-        from fastapi import HTTPException
-
-        db = MagicMock()
-        cu = MagicMock()
-        cu.user.id = "u1"
-        cu.user.is_member = False
-        cu.user.member_type = None
-        req = CreatePreviewGenerationTaskRequest(template_id="t1", asset_ids=["a1"], preview_count=1)
-        with patch("packages.domain.points_service.PointsService") as MS:
-            svc = MagicMock()
-            svc.check_daily_free_clip.return_value = False
-            svc.deduct_points.return_value = {"success": False, "balance": 0}
-            MS.return_value = svc
-            with pytest.raises(HTTPException) as ei:
-                create_preview_generation_task(
-                    request=req,
-                    authenticated_user=cu,
-                    db=db,
-                    generation_task_repository=MagicMock(),
-                    asset_repo=MagicMock(),
-                )
-            assert ei.value.status_code == 402
-
-    def test_decorator_attached(self):
+    def test_no_points_gate_decorator(self):
+        """预览生成路由已移除 @points_gate。"""
         from app.api.routes.generation_preview import create_preview_generation_task
 
-        assert hasattr(create_preview_generation_task, "__wrapped__"), "missing @points_gate"
+        # 移除装饰器后 __wrapped__ 不再存在
+        assert not hasattr(create_preview_generation_task, "__wrapped__")
+
+    def test_endpoint_does_not_deduct_points(self):
+        """端点不再实例化 PointsService / 调用 deduct_points（直接走业务逻辑）。"""
+        import inspect
+
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        src = inspect.getsource(create_preview_generation_task)
+        assert "PointsService" not in src
+        assert "deduct_points" not in src
+        assert "calculate_points_cost" not in src

tests/unit/test_generation_tasks_points.py

@@ -1,63 +1,28 @@
-"""视频生成 积分扣点单元测试 (#1895 P2 step 2.4)"""
+"""智能混剪任务 — v1.6.2 起免费，不扣积分"""
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-import packages.middleware.points_gate as _pg_module
+from unittest.mock import MagicMock
 
 
-@pytest.fixture(autouse=True)
-def _enable(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-    yield
-
-
-class TestGenerationTasksPoints:
-    def test_ai_video_base_cost(self):
+class TestGenerationTasksFree:
+    def test_ai_video_returns_zero_cost(self):
         from packages.domain.points_rules import calculate_points_cost
 
-        assert calculate_points_cost("ai_video", is_member=False) == 4
-        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
+        assert calculate_points_cost("ai_video", is_member=False, duration_minutes=5) == 0
+        assert calculate_points_cost("ai_video", is_member=True, duration_minutes=10) == 0
 
-    def test_ai_video_quantity_scales(self):
-        from packages.domain.points_rules import calculate_points_cost
-
-        c1 = calculate_points_cost("ai_video", is_member=False, quantity=1)
-        c3 = calculate_points_cost("ai_video", is_member=False, quantity=3)
-        assert c3 > c1
-
-    def test_insufficient_raises_402(self):
-        from app.api.routes.generation_tasks import create_generation_task
-        from app.schemas.generation_task import CreateGenerationTaskRequest
-        from fastapi import HTTPException
-
-        db = MagicMock()
-        cu = MagicMock()
-        cu.user.id = "u1"
-        cu.user.is_member = False
-        cu.user.member_type = None
-        req = CreateGenerationTaskRequest(template_id="t1", asset_ids=["a1"], count=1)
-        with patch("packages.domain.points_service.PointsService") as MS:
-            svc = MagicMock()
-            svc.check_daily_free_clip.return_value = False
-            svc.deduct_points.return_value = {"success": False, "balance": 0}
-            MS.return_value = svc
-            with pytest.raises(HTTPException) as ei:
-                create_generation_task(
-                    request=req,
-                    authenticated_user=cu,
-                    db=db,
-                    generation_task_repository=MagicMock(),
-                    project_repository=MagicMock(),
-                    asset_library_repository=MagicMock(),
-                    asset_repository=MagicMock(),
-                )
-            assert ei.value.status_code == 402
-
-    def test_decorator_attached(self):
+    def test_no_points_gate_decorator(self):
         from app.api.routes.generation_tasks import create_generation_task
 
-        assert hasattr(create_generation_task, "__wrapped__"), "missing @points_gate"
+        assert not hasattr(create_generation_task, "__wrapped__")
+
+    def test_create_task_accepts_request_without_points_block(self):
+        """路由函数签名不再做扣点，但参数 points_enabled/is_member/member_type 仍保留以兼容调用方。"""
+        import inspect
+
+        from app.api.routes.generation_tasks import create_generation_task
+
+        sig = inspect.signature(create_generation_task)
+        # 函数存在
+        assert callable(create_generation_task)

tests/unit/test_health_routes.py

@@ -44,7 +44,7 @@ class TestCheckDatabase:
         assert result["type"] == "postgresql"
         assert result["message"] == "Database connection successful"
         mock_psycopg.connect.assert_called_once_with(
-            "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
+            "postgresql://test:test@localhost/test", connect_timeout=3
         )
         mock_cur.execute.assert_called_once_with("SELECT 1")
         mock_conn.close.assert_called_once()
@@ -96,7 +96,7 @@ class TestCheckMigrations:
         assert result["status"] == "healthy"
         assert result["message"] == "Database migrations applied"
         mock_psycopg.connect.assert_called_once_with(
-            "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
+            "postgresql://test:test@localhost/test", connect_timeout=3
         )
         mock_conn.close.assert_called_once()
 

tests/unit/test_lipsync_points.py

@@ -1,15 +1,16 @@
-"""lipsync 积分扣点单元测试 (#1895 P2 step 2.2)"""
+"""lipsync 口型同步 — v1.6.2 起免费，不扣积分"""
 
 from __future__ import annotations
 
 import math
-from unittest.mock import MagicMock
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
 
 
-def _make_cu(user_id="user-1", is_member=False, member_type=None):
+def _cu(user_id="u1", is_member=False, member_type=None):
     cu = MagicMock()
     cu.user.id = user_id
     cu.user.is_member = is_member
@@ -17,99 +18,6 @@ def _make_cu(user_id="user-1", is_member=False, member_type=None):
     return cu
 
 
-class TestLipsyncDurationEstimate:
-    @pytest.mark.parametrize(
-        "text,expected",
-        [
-            ("你好", 1.0),
-            ("你" * 240, 1.0),
-            ("你" * 241, 2.0),
-            ("你" * 1000, 5.0),
-        ],
-    )
-    def test_text_estimate(self, text, expected):
-        est = max(1.0, math.ceil(len(text) / 240))
-        assert est == expected
-
-    @pytest.mark.parametrize(
-        "seconds,expected",
-        [
-            (30, 1.0),
-            (60, 1.0),
-            (61, 2.0),
-            (120, 2.0),
-            (180, 3.0),
-        ],
-    )
-    def test_audio_duration_estimate(self, seconds, expected):
-        est = max(1.0, math.ceil(seconds / 60.0))
-        assert est == expected
-
-
-class TestLipsyncPointsDeduction:
-    def _deduct(self, text="你好", audio_duration=None, enabled=True, success=True, balance=100, **cu_kw):
-        from packages.domain.points_rules import calculate_points_cost
-
-        svc = MagicMock() if enabled else None
-        cu = _make_cu(**cu_kw)
-        if svc is None:
-            return 0, cu
-        if audio_duration and audio_duration > 0:
-            est = max(1.0, math.ceil(audio_duration / 60.0))
-        elif text:
-            est = max(1.0, math.ceil(len(text) / 240))
-        else:
-            est = 1.0
-        cost = calculate_points_cost(
-            "ai_digital_human",
-            is_member=getattr(cu.user, "is_member", False),
-            duration_minutes=est,
-            member_type=getattr(cu.user, "member_type", None),
-        )
-        svc.deduct_points.return_value = {"success": success, "balance": balance}
-        res = svc.deduct_points(cu.user.id, cost, "ai_digital_human", MagicMock())
-        if not res["success"]:
-            raise HTTPException(status_code=402, detail={"code": "INSUFFICIENT_POINTS"})
-        return cost, cu
-
-    def test_disabled(self):
-        cost, _ = self._deduct(enabled=False)
-        assert cost == 0
-
-    def test_short_text_min_1min(self):
-        cost, _ = self._deduct(text="你好")
-        assert cost >= 15  # 15 base/min for free user × 1.15
-
-    def test_audio_duration_used(self):
-        cost_long, _ = self._deduct(audio_duration=180)  # 3min
-        cost_short, _ = self._deduct(audio_duration=30)  # 1min
-        assert cost_long > cost_short
-
-    def test_insufficient_402(self):
-        with pytest.raises(HTTPException) as ei:
-            self._deduct(text="你" * 500, success=False, balance=0)
-        assert ei.value.status_code == 402
-
-    def test_member_cheaper(self):
-        cm, _ = self._deduct(text="你" * 500, is_member=True, member_type="yearly")
-        cf, _ = self._deduct(text="你" * 500, is_member=False)
-        assert cm < cf
-
-
-# ── 直接调用 create_lipsync_job 覆盖扣点/402/退费分支 ──
-import importlib
-from types import SimpleNamespace
-from unittest.mock import patch
-
-import packages.middleware.points_gate as _pg_module
-
-
-# Ensure the enable-gate fixture for lipsync also covers @points_gate (if any)
-# (the existing autouse _enable is below; importlib to avoid duplicate)
-def _do_enable(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-
-
 def _body(**kw):
     b = MagicMock()
     defaults = dict(
@@ -130,113 +38,76 @@ def _body(**kw):
     return b
 
 
-def _cu(user_id="u1", 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
-    return cu
+class TestLipsyncFree:
+    """lipsync 已移除手动扣点，业务异常仍按原状态码抛出。"""
 
+    def test_ai_digital_human_returns_zero_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        assert calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=1) == 0
+        assert calculate_points_cost("ai_digital_human", is_member=True, duration_minutes=10) == 0
+
+    def test_module_has_no_points_imports(self):
+        import inspect
+
+        from app.api.routes import lipsync as mod
+
+        src = inspect.getsource(mod)
+        assert "PointsService" not in src
+        assert "calculate_points_cost" not in src
+        assert "_points_deducted" not in src
+        assert "settings" not in src  # settings was only used for points_enabled
+
+    def test_docstring_at_top_of_create_lipsync_job(self):
+        """扣点块删除后，docstring 必须在函数体第一行（防止函数体中段 docstring 丢失）。"""
+        import ast
+        import inspect
 
-class TestLipsyncEndpointPoints:
-    def test_insufficient_raises_402(self, monkeypatch):
-        _do_enable(monkeypatch)
         from app.api.routes.lipsync import create_lipsync_job
 
-        db = MagicMock()
-        svc = MagicMock()
-        ps = MagicMock()
-        ps.deduct_points.return_value = {"success": False, "balance": 0}
-        fs = MagicMock(points_enabled=True)
-        with (
-            patch("app.api.routes.lipsync.PointsService", return_value=ps),
-            patch("app.api.routes.lipsync.settings", fs),
-        ):
-            with pytest.raises(HTTPException) as ei:
-                create_lipsync_job(body=_body(script_text="你" * 500), current_user=_cu(), db=db, svc=svc)
-            assert ei.value.status_code == 402
+        src = inspect.getsource(create_lipsync_job)
+        tree = ast.parse(src)
+        fn = tree.body[0]
+        # docstring 应为函数体第一条语句
+        assert (
+            isinstance(fn.body[0], ast.Expr)
+            and isinstance(fn.body[0].value, ast.Constant)
+            and isinstance(fn.body[0].value.value, str)
+        ), "create_lipsync_job docstring 不在函数体开头"
 
-    def test_value_error_refunds(self, monkeypatch):
-        _do_enable(monkeypatch)
+    def test_docstring_at_top_of_preview_tts(self):
+        import ast
+        import inspect
+
+        from app.api.routes.lipsync import preview_tts
+
+        src = inspect.getsource(preview_tts)
+        tree = ast.parse(src)
+        fn = tree.body[0]
+        assert (
+            isinstance(fn.body[0], ast.Expr)
+            and isinstance(fn.body[0].value, ast.Constant)
+            and isinstance(fn.body[0].value.value, str)
+        ), "preview_tts docstring 不在函数体开头"
+
+    def test_value_error_still_raises_400(self):
+        """业务异常仍抛 400（不再退费）。"""
         from app.api.routes.lipsync import create_lipsync_job
 
         db = MagicMock()
         svc = MagicMock()
         svc.create_job.side_effect = ValueError("bad input")
-        ps = MagicMock()
-        ps.deduct_points.return_value = {"success": True, "balance": 99}
-        fs = MagicMock(points_enabled=True)
-        with (
-            patch("app.api.routes.lipsync.PointsService", return_value=ps),
-            patch("app.api.routes.lipsync.settings", fs),
-        ):
-            with pytest.raises(HTTPException) as ei:
-                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
-            assert ei.value.status_code == 400
-            assert ps.refund_points.called
+        with pytest.raises(HTTPException) as ei:
+            create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
+        assert ei.value.status_code == 400
 
-    def test_mediakit_error_refunds(self, monkeypatch):
-        _do_enable(monkeypatch)
+    def test_success_returns_job(self):
         from app.api.routes.lipsync import create_lipsync_job
-        from app.services.mediakit_client import MediaKitError
-
-        db = MagicMock()
-        svc = MagicMock()
-        svc.create_job.side_effect = MediaKitError("fail", code="InvalidInput")
-        ps = MagicMock()
-        ps.deduct_points.return_value = {"success": True, "balance": 99}
-        fs = MagicMock(points_enabled=True)
-        with (
-            patch("app.api.routes.lipsync.PointsService", return_value=ps),
-            patch("app.api.routes.lipsync.settings", fs),
-        ):
-            with pytest.raises(HTTPException) as ei:
-                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
-            assert ei.value.status_code == 400
-            assert ps.refund_points.called
-
-    def test_generic_exception_refunds(self, monkeypatch):
-        _do_enable(monkeypatch)
-        from app.api.routes.lipsync import create_lipsync_job
-
-        db = MagicMock()
-        svc = MagicMock()
-        svc.create_job.side_effect = RuntimeError("boom")
-        ps = MagicMock()
-        ps.deduct_points.return_value = {"success": True, "balance": 99}
-        fs = MagicMock(points_enabled=True)
-        with (
-            patch("app.api.routes.lipsync.PointsService", return_value=ps),
-            patch("app.api.routes.lipsync.settings", fs),
-        ):
-            with pytest.raises(HTTPException) as ei:
-                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
-            assert ei.value.status_code == 400
-            assert ps.refund_points.called
-
-    def test_audio_duration_estimation(self, monkeypatch):
-        _do_enable(monkeypatch)
-        from app.api.routes.lipsync import create_lipsync_job
-
-        from packages.domain.points_rules import calculate_points_cost
 
         db = MagicMock()
         svc = MagicMock()
         job = SimpleNamespace(id="job-1", status="queued")
         svc.create_job.return_value = job
-        ps = MagicMock()
-        ps.deduct_points.return_value = {"success": True, "balance": 99}
-        fs = MagicMock(points_enabled=True)
-        with (
-            patch("app.api.routes.lipsync.PointsService", return_value=ps),
-            patch("app.api.routes.lipsync.settings", fs),
-        ):
-            create_lipsync_job(
-                body=_body(audio_url="http://x/a.mp3", audio_duration=180, script_text=None),
-                current_user=_cu(),
-                db=db,
-                svc=svc,
-            )
-            # 180 seconds -> 3 minutes; assert deduct called with cost >= 15*3
-            args = ps.deduct_points.call_args[0]
-            assert args[1] >= calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=3)
+        # 不再依赖 settings/PointsService patch
+        result = create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
+        assert result is job

tests/unit/test_points_gate.py

@@ -44,7 +44,7 @@ class TestExtractKwargs:
 
 class TestPointsGateSync:
     def test_no_user_raises_401(self):
-        @points_gate("ai_rewrite")
+        @points_gate("voice_clone_synth")
         def my_func(db=None):
             return "ok"
 
@@ -53,7 +53,7 @@ class TestPointsGateSync:
         assert exc_info.value.status_code == 401
 
     def test_no_db_raises_500(self):
-        @points_gate("ai_rewrite")
+        @points_gate("voice_clone_synth")
         def my_func(current_user=None, db=None):
             return "ok"
 
@@ -85,7 +85,14 @@ class TestPointsGateExecuteLogic:
         with patch("packages.domain.points_service.PointsService", return_value=mock_svc):
             with pytest.raises(HTTPException) as exc_info:
                 _execute_with_gate(
-                    my_func, (), {"current_user": cu, "db": db}, "ai_rewrite", None, None, None, is_async=False
+                    my_func,
+                    (),
+                    {"current_user": cu, "db": db},
+                    "voice_clone_synth",
+                    per_unit=10,
+                    unit_field=None,
+                    quantity_field=None,
+                    is_async=False,
                 )
             assert exc_info.value.status_code == 402
 
@@ -115,7 +122,7 @@ class TestPointsGateExecuteLogic:
                 my_func,
                 (),
                 {"current_user": cu, "db": db},
-                "ai_rewrite",
+                "voice_clone_synth",
                 per_unit=10,
                 unit_field=None,
                 quantity_field=None,
@@ -139,7 +146,7 @@ class TestPointsGateExecuteLogic:
                     failing_func,
                     (),
                     {"current_user": cu, "db": db},
-                    "ai_rewrite",
+                    "voice_clone_synth",
                     per_unit=10,
                     unit_field=None,
                     quantity_field=None,
@@ -147,21 +154,21 @@ class TestPointsGateExecuteLogic:
                 )
             mock_svc.refund_points.assert_called_once()
 
-    def test_ai_video_free_quota_for_free_user(self):
+    def test_retired_scene_passes_through_with_zero_deduction(self):
+        """已下线场景（如 ai_video/ai_rewrite/ai_voice 等）直接放行，不扣积分。"""
         cu = _make_current_user(is_member=False)
         db = MagicMock()
-        mock_svc = MagicMock()
-        mock_svc.check_daily_free_clip.return_value = True
-        mock_svc.record_daily_free_clip.return_value = True
 
         def my_func(current_user=cu, db=db, **kwargs):
-            return kwargs.get("_is_free_quota", False)
+            return kwargs.get("_points_deducted", -1)
 
-        with patch("packages.domain.points_service.PointsService", return_value=mock_svc):
+        # 不应调用 PointsService
+        with patch("packages.domain.points_service.PointsService") as mock_svc_cls:
             result = _execute_with_gate(
                 my_func, (), {"current_user": cu, "db": db}, "ai_video", None, None, None, is_async=False
             )
-            assert result is True
+            assert result == 0
+            mock_svc_cls.assert_not_called()
 
 
 class TestPointsGateAsync:
@@ -172,7 +179,7 @@ class TestPointsGateAsync:
         mock_svc = MagicMock()
         mock_svc.deduct_points.return_value = {"success": True, "balance": 90, "transaction_id": "t1"}
 
-        @points_gate("ai_rewrite", per_unit=5)
+        @points_gate("voice_clone_synth", per_unit=5)
         async def my_async_func(current_user=None, db=None, **kwargs):
             return kwargs.get("_points_deducted", 0)
 

tests/unit/test_points_routes.py

@@ -2,7 +2,7 @@
 
 覆盖:
 - P0-1: POST /points/recharge 返回 pay_params / points_amount / expire_at
-- P0-2: POST /points/check 未知 scene_key 返回 400（非 500）
+- P0-2: POST /points/check 任意 scene_key 均可查询（已下线场景返回 cost=0，不报错）
 - P1-3: GET /points/rules 返回 description 字段
 - P1-6: GET /subscription/plans 返回档位列表
 - P1-7: multiplier 实际扣费一致（calculate_points_cost 统一应用）
@@ -76,39 +76,40 @@ class TestRechargeOrderResponse:
         assert exc.value.status_code == 400
 
 
-# ── P0-2: check unknown scene → 400 ───────────────────────────────────
+# ── P0-2: check 任意 scene_key（已下线场景返回 cost=0） ──────────────────
 
 
 class TestCheckPointsUnknownScene:
-    def test_unknown_scene_returns_400_not_500(self):
-        """未知 scene_key（如 ai_script）应返回 400 UNKNOWN_SCENE，而不是 500。"""
+    def test_unknown_scene_returns_zero_cost_not_error(self):
+        """任意 scene_key 均可查询，已下线/未知场景返回 cost=0（免费放行）。"""
         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_script", quantity=1)
 
-        with pytest.raises(HTTPException) as exc:
-            check_points(body=body, current_user=cu, db=db)
-        assert exc.value.status_code == 400
-        detail = exc.value.detail
-        assert detail["code"] == "UNKNOWN_SCENE"
-        assert "ai_script" in detail["message"]
-        assert "ai_voice" in detail["valid_scenes"]
-        assert "ai_title" in detail["valid_scenes"]
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=True),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            for scene in ["ai_script", "ai_voice", "ai_video", "ai_title", "ai_cover", "nonexistent"]:
+                body = PointsCheckRequest(scene_key=scene, quantity=1)
+                resp = check_points(body=body, current_user=cu, db=db)
+                assert resp.required_points == 0, f"{scene} should be free"
+                assert resp.allowed is True
 
-    def test_known_scene_still_works(self):
-        """合法 scene_key 正常返回，免费用户 ai_voice 1 分钟 = 2 积分。"""
+    def test_voice_clone_synth_still_charges(self):
+        """合法付费场景 voice_clone_synth 正常计费：免费用户 1 分钟 = ceil(1*1.15)=2 积分。"""
         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": 50}
         db = MagicMock()
         cu = _make_cu()
-        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=1)
+        body = PointsCheckRequest(scene_key="voice_clone_synth", quantity=1, duration_minutes=1)
 
         with (
             patch("app.api.routes.points._credits_enabled", return_value=True),
@@ -128,7 +129,9 @@ class TestPointsRulesDescription:
         from app.api.routes.points import get_rules
 
         resp = get_rules(_current_user=_make_cu())
-        assert len(resp.rules) >= 9
+        # 场景列表包含 voice_clone_train / voice_clone_synth / viral_video（爆款视频为动态定价）
+        keys = {r.scene_key for r in resp.rules}
+        assert {"voice_clone_train", "voice_clone_synth", "viral_video"}.issubset(keys)
         for rule in resp.rules:
             assert rule.description, f"{rule.scene_key} missing description"
             assert isinstance(rule.description, str)
@@ -202,11 +205,18 @@ class TestSubscriptionPlans:
 
 
 class TestMultiplierConsistency:
-    def test_free_user_ai_title_costs_2(self):
-        """ai_title base=1，免费用户 ceil(1*1.15)=2。"""
+    def test_free_user_voice_clone_synth_1min_costs_2(self):
+        """voice_clone_synth base=1，免费用户 ceil(1*1.15)=2。"""
         from packages.domain.points_rules import calculate_points_cost
 
-        assert calculate_points_cost("ai_title", is_member=False, quantity=1) == 2
+        assert calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=1) == 2
+
+    def test_retired_scenes_return_zero(self):
+        """已下线场景（ai_voice/ai_title/ai_cover/ai_rewrite 等）calculate_points_cost 统一返回 0。"""
+        from packages.domain.points_rules import calculate_points_cost
+
+        for scene in ["ai_voice", "ai_title", "ai_cover", "ai_rewrite", "ai_video", "ai_digital_human"]:
+            assert calculate_points_cost(scene, is_member=False, quantity=1) == 0
 
     def test_check_matches_direct_calculation(self):
         """check 端点 required_points 与 calculate_points_cost 结果一致。"""
@@ -216,15 +226,14 @@ class TestMultiplierConsistency:
         from packages.domain.points_rules import calculate_points_cost
 
         svc = MagicMock()
-        svc.check_daily_free_clip.return_value = False
         svc.get_or_create_account.return_value = {"balance": 999}
         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"]:
-                body = PointsCheckRequest(scene_key=scene, quantity=1)
+            for scene in ["voice_clone_synth", "voice_clone_train", "ai_voice", "ai_video", "ai_title"]:
+                body = PointsCheckRequest(scene_key=scene, quantity=1, duration_minutes=1)
                 with patch("app.api.routes.points._get_service", return_value=svc):
                     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, duration_minutes=1)
                 assert resp.required_points == expected, f"{scene}: got {resp.required_points}, expected {expected}"

tests/unit/test_points_rules.py

@@ -1,4 +1,4 @@
-"""积分消耗规则单元测试 (#1895)"""
+"""积分消耗规则单元测试 (#1895) — v1.6.2: 仅保留 voice_clone 相关"""
 
 from __future__ import annotations
 
@@ -7,7 +7,6 @@ import math
 import pytest
 
 from packages.domain.points_rules import (
-    DAILY_FREE_CLIP_LIMIT,
     FREE_USER_MULTIPLIER,
     MEMBER_DISCOUNT,
     MEMBERSHIP_PRICES,
@@ -20,8 +19,22 @@ from packages.domain.points_rules import (
 class TestPointsScenesConfig:
     """场景配置完整性"""
 
-    def test_all_nine_scenes_defined(self):
-        assert len(POINTS_SCENES) == 9
+    def test_registered_scenes_include_voice_clone_and_viral_video(self):
+        """场景配置：包含声音克隆（训练/合成）+ 爆款视频（动态定价）。"""
+        assert {"voice_clone_train", "voice_clone_synth", "viral_video"}.issubset(set(POINTS_SCENES.keys()))
+
+    def test_viral_video_scene_is_dynamic_with_zero_base(self):
+        """viral_video 必须注册但 base_points=0 且 dynamic=True，不使用 @points_gate。"""
+        vv = POINTS_SCENES["viral_video"]
+        assert vv["base_points"] == 0
+        assert vv["dynamic"] is True
+        assert vv["unit"] == "次"
+        assert vv["name"] == "爆款视频"
+
+    def test_voice_clone_scenes_defined(self):
+        # 保留声音克隆两个场景
+        assert "voice_clone_train" in POINTS_SCENES
+        assert "voice_clone_synth" in POINTS_SCENES
 
     def test_required_keys_present(self):
         for key, scene in POINTS_SCENES.items():
@@ -32,8 +45,14 @@ class TestPointsScenesConfig:
     def test_voice_clone_train_is_free(self):
         assert POINTS_SCENES["voice_clone_train"]["base_points"] == 0
 
-    def test_ai_video_has_extra_per_30s(self):
-        assert POINTS_SCENES["ai_video"]["extra_per_30s"] == 1
+    def test_voice_clone_synth_is_per_minute(self):
+        assert POINTS_SCENES["voice_clone_synth"]["base_points"] == 1
+        assert POINTS_SCENES["voice_clone_synth"]["unit"] == "分钟"
+
+    def test_calculate_points_cost_returns_zero_for_dynamic_viral_video(self):
+        """calculate_points_cost 对动态场景 viral_video 必须返回 0（由业务侧手动计算）。"""
+        assert calculate_points_cost("viral_video", is_member=False) == 0.0
+        assert calculate_points_cost("viral_video", is_member=True, member_type="monthly") == 0.0
 
 
 class TestPointsPackages:
@@ -52,43 +71,23 @@ class TestMembershipPrices:
         assert MEMBERSHIP_PRICES["yearly"]["price_cents"] == 15900
 
 
-class TestDailyFreeLimit:
-    def test_limit_is_2(self):
-        assert DAILY_FREE_CLIP_LIMIT == 2
-
-
 class TestCalculatePointsCost:
     """核心计费逻辑"""
 
-    # ── 按次计费 ──
+    # ── 声音克隆合成（按时长计费） ──
 
-    def test_per_time_base_cost(self):
-        # ai_rewrite: 1积分/次，免费用户 ceil(1 * 1.15) = 2
-        cost = calculate_points_cost("ai_rewrite", is_member=False, quantity=1)
+    def test_voice_clone_synth_base(self):
+        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=3)
+        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
+
+    def test_voice_clone_synth_rounds_up(self):
+        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=2.3)
+        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
+
+    def test_voice_clone_synth_minimum_1_minute(self):
+        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=0.1)
         assert cost == math.ceil(1 * FREE_USER_MULTIPLIER)
 
-    def test_per_time_multiple(self):
-        # ai_cover: 1积分/张，3张 → base=3, free: ceil(3*1.15)=4
-        cost = calculate_points_cost("ai_cover", is_member=False, quantity=3)
-        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
-
-    # ── 按时长计费 ──
-
-    def test_per_minute_base(self):
-        # ai_voice: 1积分/分钟，3分钟 → base=3, free: ceil(3*1.15)=4
-        cost = calculate_points_cost("ai_voice", is_member=False, duration_minutes=3)
-        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
-
-    def test_per_minute_rounds_up(self):
-        # 2.3分钟 → ceil(2.3)=3分钟 → base=3
-        cost = calculate_points_cost("ai_voice", is_member=False, duration_minutes=2.3)
-        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
-
-    def test_digital_human_expensive(self):
-        # ai_digital_human: 15积分/分钟，1分钟 → base=15, free: ceil(15*1.15)=18
-        cost = calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=1)
-        assert cost == 18
-
     # ── 免费场景 ──
 
     def test_voice_clone_train_free(self):
@@ -99,42 +98,436 @@ class TestCalculatePointsCost:
         cost = calculate_points_cost("voice_clone_train", is_member=True)
         assert cost == 0
 
-    # ── 混剪额外逻辑 ──
-
-    def test_ai_video_short_no_extra(self):
-        # 20s (0.33min) ≤ 30s，不额外加积分，base=3, free: ceil(3*1.15)=4
-        cost = calculate_points_cost("ai_video", is_member=False, quantity=1, duration_minutes=0.33)
-        assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
-
-    def test_ai_video_long_extra_charge(self):
-        # 80s → base=3 + extra ceil((80-30)/30)=2 → total_base=5, free: ceil(5*1.15)=6
-        cost = calculate_points_cost("ai_video", is_member=False, quantity=1, duration_minutes=80 / 60)
-        assert cost == math.ceil(5 * FREE_USER_MULTIPLIER)
-
     # ── 会员折扣 ──
 
     def test_monthly_member_discount(self):
-        # ai_voice 1分钟 base=1, 月卡0.9 → floor(1*0.9)=1 → max(1,1)=1
-        cost = calculate_points_cost("ai_voice", is_member=True, duration_minutes=1, member_type="monthly")
-        assert cost == max(1, math.floor(1 * 0.9))
+        cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1, member_type="monthly")
+        assert cost == max(1, math.floor(1 * MEMBER_DISCOUNT["monthly"]))
 
     def test_yearly_member_deep_discount(self):
-        # ai_digital_human 2分钟 base=30, 年卡0.8 → floor(30*0.8)=24
         cost = calculate_points_cost(
-            "ai_digital_human",
+            "voice_clone_synth",
             is_member=True,
             duration_minutes=2,
             member_type="yearly",
         )
-        assert cost == max(1, math.floor(30 * 0.8))
+        assert cost == max(1, math.floor(2 * MEMBER_DISCOUNT["yearly"]))
 
     def test_member_without_type_no_discount(self):
-        # is_member=True 但没传 member_type → 不按会员折扣
-        cost = calculate_points_cost("ai_voice", is_member=True, duration_minutes=1)
-        assert cost == 1  # base=1, no discount applied
+        cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1)
+        assert cost == 1
 
-    # ── 异常 ──
+    # ── 已下线/未知场景（向后兼容：返回 0） ──
 
-    def test_unknown_scene_raises(self):
-        with pytest.raises(ValueError, match="Unknown points scene"):
-            calculate_points_cost("nonexistent_scene", is_member=False)
+    @pytest.mark.parametrize(
+        "scene",
+        [
+            "ai_voice",
+            "ai_video",
+            "ai_digital_human",
+            "ai_rewrite",
+            "ai_cover",
+            "ai_title",
+            "douyin_extract",
+            "nonexistent",
+        ],
+    )
+    def test_retired_scenes_return_zero(self, scene):
+        assert calculate_points_cost(scene, is_member=False) == 0
+        assert calculate_points_cost(scene, is_member=True, duration_minutes=10) == 0
+
+
+# ============ 爆款视频动态定价 (#2151) ============
+
+
+class TestResolveVideoDimensions:
+    """resolve_video_dimensions(): 分辨率别名、比例、默认兜底。"""
+
+    def test_1080p_16_9(self):
+        """1080p + 16:9 → w=1920, h=1080。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("1080p", "16:9")
+        assert (w, h) == (1920, 1080)
+
+    def test_480p_16_9(self):
+        """480p + 16:9 → 854×480（ceil(480*16/9)=854，偶对齐）。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("480p", "16:9")
+        assert (w, h) == (854, 480)
+
+    def test_720p_1_1(self):
+        """1:1 正方形 → w == h。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("720p", "1:1")
+        assert (w, h) == (720, 720)
+
+    def test_1080p_1_1(self):
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("1080p", "1:1")
+        assert (w, h) == (1080, 1080)
+
+    def test_resolution_aliases(self):
+        """中文/英文别名应正确映射到对应高度。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        cases = [
+            ("普清", 480),
+            ("sd", 480),
+            ("low", 480),
+            ("default", 480),
+            ("高清", 720),
+            ("medium", 720),
+            ("hd", 720),
+            ("超清", 1080),
+            ("fhd", 1080),
+            ("ultra", 1080),
+            ("全能", 1080),
+            ("high", 1080),
+        ]
+        for alias, expected_h in cases:
+            _, h = resolve_video_dimensions(alias, "1:1")
+            assert h == expected_h, f"{alias} -> h={h}, expected {expected_h}"
+
+    def test_unknown_resolution_falls_back_to_720p(self):
+        """未知分辨率字符串兜底到 720p。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("2160p", "1:1")
+        assert h == 720
+        assert w == 720
+
+    def test_empty_resolution_defaults_to_720p_9_16(self):
+        """空 resolution + 空 ratio → 默认 720p + 9:16 竖屏 (720×1280)。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("", "")
+        assert (w, h) == (720, 1280)
+
+    def test_none_resolution_default_ratio(self):
+        """None resolution + None ratio → 720p + 9:16 竖屏默认。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions(None, None)
+        assert (w, h) == (720, 1280)
+
+    def test_720p_9_16_portrait(self):
+        """720p + 9:16 竖屏 → 短边是 width=720，height=1280（v10实测）。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("720p", "9:16")
+        assert (w, h) == (720, 1280)
+
+    def test_1080p_9_16_portrait(self):
+        """1080p + 9:16 竖屏 → 1080×1920。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("1080p", "9:16")
+        assert (w, h) == (1080, 1920)
+
+    def test_480p_9_16_portrait(self):
+        """480p + 9:16 竖屏 → 480×854。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("480p", "9:16")
+        assert (w, h) == (480, 854)
+
+    def test_all_dimensions_even(self):
+        """所有返回尺寸都应是偶数（视频编码要求）。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        for res in ("480p", "720p", "1080p", "普清", "高清", "超清"):
+            for ratio in ("16:9", "9:16", "1:1"):
+                w, h = resolve_video_dimensions(res, ratio)
+                assert w % 2 == 0 and h % 2 == 0, f"{res}/{ratio} -> ({w},{h}) not even"
+
+    def test_whitespace_resolution_case_insensitive(self):
+        """前后空格 + 大写应被规范化处理。"""
+        from packages.domain.points_rules import resolve_video_dimensions
+
+        w, h = resolve_video_dimensions("  1080P  ", " 16:9 ")
+        assert (w, h) == (1920, 1080)
+
+
+class TestMatchModelPrefix:
+    """_match_model_prefix() 前缀匹配 + 兜底。"""
+
+    def test_seedance_2_5_exact(self):
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("seedance-2.5") == "seedance-2.5"
+
+    def test_seedance_2_5_with_variant(self):
+        """带后缀版本号（如 seedance-2.5-pro）仍匹配 seedance-2.5。"""
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("seedance-2.5-pro") == "seedance-2.5"
+
+    def test_seedance_2_0_exact(self):
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("seedance-2.0") == "seedance-2.0"
+
+    def test_seedance_2_0_with_variant(self):
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("seedance-2.0-lite") == "seedance-2.0"
+
+    def test_unknown_model_falls_back_to_2_5(self):
+        """未知模型前缀兜底 seedance-2.5。"""
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("kling-v1") == "seedance-2.5"
+        assert _match_model_prefix("") == "seedance-2.5"
+        assert _match_model_prefix(None) == "seedance-2.5"
+
+    def test_case_insensitive(self):
+        from packages.domain.points_rules import _match_model_prefix
+
+        assert _match_model_prefix("SEEDANCE-2.0") == "seedance-2.0"
+
+
+class TestInferResolutionKey:
+    """_infer_resolution_key(w, h): 按短边 1000+/650-999/<650 三档。"""
+
+    def test_short_side_ge_1000_is_1080p(self):
+        from packages.domain.points_rules import _infer_resolution_key
+
+        assert _infer_resolution_key(1920, 1080) == "1080p"  # 横屏
+        assert _infer_resolution_key(1080, 1920) == "1080p"  # 竖屏
+        assert _infer_resolution_key(1080, 1080) == "1080p"  # 方屏
+
+    def test_short_side_650_to_999_is_720p(self):
+        from packages.domain.points_rules import _infer_resolution_key
+
+        assert _infer_resolution_key(1280, 720) == "720p"
+        assert _infer_resolution_key(720, 1280) == "720p"
+        assert _infer_resolution_key(720, 720) == "720p"
+
+    def test_short_side_lt_650_is_480p(self):
+        from packages.domain.points_rules import _infer_resolution_key
+
+        assert _infer_resolution_key(854, 480) == "480p"
+        assert _infer_resolution_key(480, 854) == "480p"
+        assert _infer_resolution_key(480, 480) == "480p"
+        # 极小值兜底
+        assert _infer_resolution_key(1, 1) == "480p"
+
+    def test_portrait_1280_height_is_720p_short_side(self):
+        """竖屏 720×1280 短边=720，应识别为 720p 而非 1080p（老bug回归）。"""
+        from packages.domain.points_rules import _infer_resolution_key
+
+        assert _infer_resolution_key(720, 1280) == "720p"
+
+
+class TestCalculateViralVideoCredits:
+    """calculate_viral_video_credits()：爆款视频动态定价核心函数。"""
+
+    def test_default_args_returns_float(self):
+        """默认参数返回 float。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        credits = calculate_viral_video_credits(15, 1280, 720)
+        assert isinstance(credits, float)
+
+    def test_return_is_rounded_to_two_decimals(self):
+        """round(..., 2) 后值本身就是两位小数（再 round 不变化）。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        for dur, w, h in [(15, 1280, 720), (5, 854, 480), (30, 1920, 1080), (10, 720, 720)]:
+            credits = calculate_viral_video_credits(dur, w, h)
+            assert round(credits, 2) == credits
+
+    def test_has_video_input_uses_lower_price(self):
+        """has_video_input=True 时使用参考视频价格（有视频输入便宜）。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        no_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=False)
+        with_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=True)
+        assert with_input < no_input
+
+    def test_unknown_model_falls_back_to_seedance_2_5(self):
+        """未知 model 前缀兜底到 seedance-2.5 价格，与默认等价。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        unknown = calculate_viral_video_credits(15, 1280, 720, model="unknown-model")
+        default = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5")
+        assert unknown == default
+
+    def test_actual_tokens_overrides_calculation(self):
+        """传入 actual_tokens>0 时用它替代公式计算的 tokens。"""
+        from packages.domain.points_rules import (
+            VIRAL_VIDEO_FIXED_COST,
+            VIRAL_VIDEO_MODEL_PRICES,
+            VIRAL_VIDEO_PROFIT_MULTIPLIER,
+            calculate_viral_video_credits,
+        )
+
+        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
+        actual_tokens = 2_000_000
+        expected = round(
+            (actual_tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2
+        )
+        credits = calculate_viral_video_credits(15, 1280, 720, actual_tokens=actual_tokens)
+        assert credits == expected
+
+    def test_zero_duration_width_height_defensive_max1(self):
+        """duration/width/height 为 0/None 时 max(1,...) 防御，结果>0。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        c_zero = calculate_viral_video_credits(0, 0, 0)
+        assert c_zero > 0
+        c_none = calculate_viral_video_credits(None, None, None)
+        assert c_none > 0
+        c_one = calculate_viral_video_credits(1, 1, 1)
+        assert c_none == c_one
+
+    def test_non_default_fps_affects_tokens(self):
+        """fps 非默认值(30) 应比默认(24) 积分高。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        c24 = calculate_viral_video_credits(15, 1280, 720, fps=24)
+        c30 = calculate_viral_video_credits(15, 1280, 720, fps=30)
+        assert c30 > c24
+
+    def test_seedance_2_0_priced_lower_than_2_5_at_1080p(self):
+        """seedance-2.0 在 1080p 无视频输入时定价低于 seedance-2.5。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        c20 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.0", has_video_input=False)
+        c25 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.5", has_video_input=False)
+        assert c20 < c25
+
+    def test_formula_includes_fixed_cost_and_multiplier(self):
+        """手算公式结果应与函数返回一致（固定成本 + 利润系数）。"""
+        from packages.domain.points_rules import (
+            VIRAL_VIDEO_FIXED_COST,
+            VIRAL_VIDEO_FPS,
+            VIRAL_VIDEO_MODEL_PRICES,
+            VIRAL_VIDEO_PROFIT_MULTIPLIER,
+            calculate_viral_video_credits,
+        )
+
+        dur, w, h = 10, 1280, 720
+        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
+        tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
+        expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
+        assert calculate_viral_video_credits(dur, w, h) == expected
+
+    def test_seedance_2_0_with_video_input_falls_back_to_seedance_2_5_price(self):
+        """seedance-2.0 + has_video_input=True 组合不在价格表，走 line 111 fallback 到 seedance-2.5 的 720p False 价格。"""
+        from packages.domain.points_rules import (
+            VIRAL_VIDEO_FIXED_COST,
+            VIRAL_VIDEO_FPS,
+            VIRAL_VIDEO_MODEL_PRICES,
+            VIRAL_VIDEO_PROFIT_MULTIPLIER,
+            calculate_viral_video_credits,
+        )
+
+        dur, w, h = 10, 1280, 720
+        # 兜底价格 = seedance-2.5/720p/False = 70.0
+        price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
+        assert price == 70.0
+        tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
+        expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
+        credits = calculate_viral_video_credits(dur, w, h, model="seedance-2.0", has_video_input=True)
+        assert credits == expected
+
+    def test_fps_zero_or_none_falls_back_to_default(self):
+        """fps=0/None 时 int(fps or 24) 兜底到默认 24，结果与 fps=24 一致。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        c_default = calculate_viral_video_credits(10, 1280, 720, fps=24)
+        c_zero = calculate_viral_video_credits(10, 1280, 720, fps=0)
+        c_none = calculate_viral_video_credits(10, 1280, 720, fps=None)
+        assert c_zero == c_default
+        assert c_none == c_default
+
+    # ──────── P0 计费回归：短边规则 + 价格精确断言 ────────
+
+    def test_15s_720p_portrait_is_29_68(self):
+        """P0 回归：15s/720p/9:16 竖屏 (720×1280) 必须 =29.68 积分。"""
+        from packages.domain.points_rules import (
+            calculate_viral_video_credits,
+            resolve_video_dimensions,
+        )
+
+        w, h = resolve_video_dimensions("720p", "9:16")
+        assert (w, h) == (720, 1280)
+        assert calculate_viral_video_credits(15, w, h) == 29.68
+
+    def test_30s_1080p_portrait_is_146_14(self):
+        """P0 回归：30s/1080p/9:16 竖屏 (1080×1920) =146.14 积分。"""
+        from packages.domain.points_rules import (
+            calculate_viral_video_credits,
+            resolve_video_dimensions,
+        )
+
+        w, h = resolve_video_dimensions("1080p", "9:16")
+        assert (w, h) == (1080, 1920)
+        assert calculate_viral_video_credits(30, w, h) == 146.14
+
+    def test_portrait_landscape_same_pixels_same_price(self):
+        """相同像素数（横竖屏旋转）积分一致。"""
+        from packages.domain.points_rules import calculate_viral_video_credits
+
+        assert calculate_viral_video_credits(15, 1280, 720) == calculate_viral_video_credits(15, 720, 1280)
+
+
+class TestViralVideoCreditsWithBreakdown:
+    """calculate_viral_video_credits_with_breakdown：返回 (credits, breakdown_dict)。"""
+
+    def test_returns_credits_matching_plain_version(self):
+        """新函数返回的 credits 必须与 calculate_viral_video_credits 完全一致，且 breakdown 字段齐全。"""
+        from packages.domain.points_rules import (
+            calculate_viral_video_credits,
+            calculate_viral_video_credits_with_breakdown,
+        )
+
+        for dur, w, h, model, hvi in [
+            (15, 1280, 720, "seedance-2.5", False),
+            (10, 720, 1280, "seedance-2.0", False),
+            (30, 1920, 1080, "seedance-2.5", False),
+            (5, 480, 480, "", False),
+        ]:
+            c1 = calculate_viral_video_credits(dur, w, h, model=model, has_video_input=hvi)
+            c2, bd = calculate_viral_video_credits_with_breakdown(dur, w, h, model=model, has_video_input=hvi)
+            assert c1 == c2
+            assert isinstance(bd, dict)
+            for key in (
+                "tokens",
+                "video_cost",
+                "fixed_cost",
+                "profit_multiplier",
+                "model_price",
+                "width",
+                "height",
+                "fps",
+            ):
+                assert key in bd, f"breakdown missing key: {key}"
+            assert bd["fixed_cost"] == 0.15
+            assert bd["profit_multiplier"] == 1.3
+            assert bd["width"] == w
+            assert bd["height"] == h
+            assert bd["fps"] == 24
+            assert bd["tokens"] > 0
+            assert bd["model_price"] > 0
+            expected = round((bd["video_cost"] + bd["fixed_cost"]) * bd["profit_multiplier"], 2)
+            assert expected == c2
+
+    def test_actual_tokens_overrides_computed(self):
+        """actual_tokens 传入时应覆盖按公式计算的 tokens。"""
+        from packages.domain.points_rules import calculate_viral_video_credits_with_breakdown
+
+        c, bd = calculate_viral_video_credits_with_breakdown(
+            15,
+            1280,
+            720,
+            actual_tokens=1_000_000,
+        )
+        assert bd["tokens"] == 1_000_000.0
+        # video_cost = 1M/1M * 70 = 70; total = (70+0.15)*1.3 = 91.195 → 91.20
+        assert c == 91.20

tests/unit/test_points_service.py

@@ -72,7 +72,7 @@ class TestCheckBalance:
 
 class TestDeductPoints:
     def test_deduct_fails_insufficient_balance(self, service, db_session, user_id):
-        result = service.deduct_points(user_id, 100, "ai_voice", db_session)
+        result = service.deduct_points(user_id, 100, "voice_clone_synth", db_session)
         assert result["success"] is False
         assert result["transaction_id"] is None
 
@@ -80,13 +80,13 @@ class TestDeductPoints:
         # 先充值
         service.add_points(user_id, 50, "recharge", db_session)
         # 再扣减
-        result = service.deduct_points(user_id, 20, "ai_voice", db_session)
+        result = service.deduct_points(user_id, 20, "voice_clone_synth", db_session)
         assert result["success"] is True
         assert result["balance"] == 30
 
     def test_deduct_creates_transaction(self, service, db_session, user_id):
         service.add_points(user_id, 100, "recharge", db_session)
-        result = service.deduct_points(user_id, 30, "ai_voice", db_session)
+        result = service.deduct_points(user_id, 30, "voice_clone_synth", db_session)
         assert result["success"] is True
 
         txns = service.get_transactions(user_id, db_session)
@@ -111,14 +111,14 @@ class TestAddPoints:
 class TestRefundPoints:
     def test_refund_adds_back(self, service, db_session, user_id):
         service.add_points(user_id, 100, "recharge", db_session)
-        service.deduct_points(user_id, 20, "ai_voice", db_session)
-        result = service.refund_points(user_id, 20, "ai_voice", db_session)
+        service.deduct_points(user_id, 20, "voice_clone_synth", db_session)
+        result = service.refund_points(user_id, 20, "voice_clone_synth", db_session)
         assert result["success"] is True
         assert result["balance"] == 100
 
     def test_refund_creates_refund_transaction(self, service, db_session, user_id):
         service.add_points(user_id, 100, "recharge", db_session)
-        service.refund_points(user_id, 10, "ai_rewrite", db_session)
+        service.refund_points(user_id, 10, "voice_clone_synth", db_session)
 
         txns = service.get_transactions(user_id, db_session)
         refund_txns = [t for t in txns["items"] if t["type"] == "add" and "refund" in t["source"]]
@@ -145,20 +145,14 @@ class TestGetTransactions:
 
 
 class TestGetDailyUsage:
-    def test_zero_usage(self, service, db_session, user_id):
-        with patch("packages.domain.points_service._get_redis_client", return_value=None):
-            result = service.get_daily_usage(user_id, db_session)
-        assert result["free_clips_used"] == 0
-        assert result["free_clips_limit"] == 2
-        assert result["free_clips_remaining"] == 2
-        assert "reset_at" in result
+    """智能混剪已免费，get_daily_usage 返回 unlimited（-1）占位。"""
 
-    def test_after_recording(self, service, db_session, user_id):
-        with patch("packages.domain.points_service._get_redis_client", return_value=None):
-            service.record_daily_free_clip(user_id, db_session)
-            result = service.get_daily_usage(user_id, db_session)
-        assert result["free_clips_used"] == 1
-        assert result["free_clips_remaining"] == 1
+    def test_returns_unlimited(self, service, db_session, user_id):
+        result = service.get_daily_usage(user_id, db_session)
+        assert result["free_clips_used"] == 0
+        assert result["free_clips_limit"] == -1  # -1 表示 unlimited
+        assert result["free_clips_remaining"] == -1
+        assert "reset_at" in result
 
 
 class TestCreateOrder:
@@ -185,3 +179,172 @@ class TestCreateOrder:
     def test_unknown_order_type_raises(self, service, db_session, user_id):
         with pytest.raises(ValueError, match="Unknown order type"):
             service.create_order(user_id, "insurance", "basic", db_session)
+
+
+# ============ 爆款视频（viral_video）动态定价方法 ============
+
+
+class TestDeductViralVideo:
+    """deduct_viral_video(): 预扣积分，委托给 deduct_points。"""
+
+    def test_delegates_to_deduct_points_with_correct_args(self, service, db_session, user_id):
+        """deduct_viral_video 应以 source='viral_video', ref_id=job_id 调用 deduct_points。"""
+        from unittest.mock import MagicMock
+
+        expected = {"success": True, "balance": 50.0, "transaction_id": "t1"}
+        with patch.object(service, "deduct_points", return_value=expected) as mock_dp:
+            result = service.deduct_viral_video(user_id, 10.5, "job-abc", db_session)
+
+        assert result == expected
+        mock_dp.assert_called_once()
+        kwargs = mock_dp.call_args.kwargs
+        assert kwargs["user_id"] == user_id
+        assert kwargs["amount"] == 10.5
+        assert kwargs["source"] == "viral_video"
+        assert kwargs["db"] is db_session
+        assert kwargs["description"] == "爆款视频生成"
+        assert kwargs["ref_id"] == "job-abc"
+
+    def test_none_credits_coerced_to_zero(self, service, db_session, user_id):
+        """credits=None 时应被 float(credits or 0) 转为 0，不抛异常。"""
+
+        with patch.object(
+            service, "deduct_points", return_value={"success": True, "balance": 0, "transaction_id": "t"}
+        ) as mock_dp:
+            service.deduct_viral_video(user_id, None, "job-nil", db_session)
+        assert mock_dp.call_args.kwargs["amount"] == 0.0
+
+
+class TestSettleViralVideo:
+    """settle_viral_video(): 多退少补结算。"""
+
+    def test_no_action_when_diff_below_epsilon(self, service, db_session, user_id):
+        """|diff|<0.01 时返回 action=none，不调 refund/deduct。"""
+
+        with (
+            patch.object(service, "refund_points") as mock_refund,
+            patch.object(service, "deduct_points") as mock_deduct,
+        ):
+            result = service.settle_viral_video(user_id, estimated=10.00, actual=10.001, txn_id="t1", db=db_session)
+
+        assert result["success"] is True
+        assert result["action"] == "none"
+        assert result["diff"] == 0.0
+        mock_refund.assert_not_called()
+        mock_deduct.assert_not_called()
+
+    def test_refund_when_actual_less_than_estimated(self, service, db_session, user_id):
+        """actual<estimated 时走 refund_points，返回 action=refund。"""
+
+        refund_res = {"success": True, "balance": 60.0, "transaction_id": "tr-1"}
+        with patch.object(service, "refund_points", return_value=refund_res) as mock_refund:
+            result = service.settle_viral_video(user_id, estimated=20.0, actual=15.0, txn_id="t2", db=db_session)
+
+        assert result["success"] is True
+        assert result["action"] == "refund"
+        assert result["amount"] == 5.0
+        assert result["diff"] == -5.0
+        mock_refund.assert_called_once()
+        rk = mock_refund.call_args.kwargs
+        assert rk["user_id"] == user_id
+        assert rk["amount"] == 5.0
+        assert rk["source"] == "viral_video"
+        assert rk["ref_id"] == "t2"
+        assert rk["description"] == "爆款视频结算退费"
+
+    def test_refund_exception_returns_failure(self, service, db_session, user_id):
+        """refund_points 抛异常时，应捕获并返回 success=False。"""
+
+        with patch.object(service, "refund_points", side_effect=RuntimeError("db down")):
+            result = service.settle_viral_video(user_id, estimated=20.0, actual=10.0, txn_id="t3", db=db_session)
+
+        assert result["success"] is False
+        assert result["action"] == "refund"
+
+    def test_deduct_when_actual_greater_than_estimated_success(self, service, db_session, user_id):
+        """actual>estimated 且补扣成功 → action=deduct, success=True。"""
+
+        deduct_res = {"success": True, "balance": 40.0, "transaction_id": "td-1"}
+        with patch.object(service, "deduct_points", return_value=deduct_res) as mock_deduct:
+            result = service.settle_viral_video(user_id, estimated=15.0, actual=20.0, txn_id="t4", db=db_session)
+
+        assert result["success"] is True
+        assert result["action"] == "deduct"
+        assert result["amount"] == 5.0
+        assert result["diff"] == 5.0
+        mock_deduct.assert_called_once()
+        dk = mock_deduct.call_args.kwargs
+        assert dk["amount"] == 5.0
+        assert dk["source"] == "viral_video"
+        assert dk["ref_id"] == "t4"
+
+    def test_deduct_insufficient_balance_returns_success_false_not_raise(self, service, db_session, user_id):
+        """actual>estimated 补扣时余额不足（success=False）应记录 warning 但不抛异常。"""
+        import logging
+
+        deduct_res = {"success": False, "balance": 2.0, "transaction_id": None}
+        with (
+            patch.object(service, "deduct_points", return_value=deduct_res),
+            patch("packages.domain.points_service.logger") as mock_logger,
+        ):
+            result = service.settle_viral_video(user_id, estimated=15.0, actual=20.0, txn_id="t5", db=db_session)
+
+        # 即使补扣失败，函数也返回 action=deduct 但 success=False（不阻塞任务完成）
+        assert result["success"] is False
+        assert result["action"] == "deduct"
+        assert result["amount"] == 5.0
+        # 应打印 warning
+        mock_logger.warning.assert_called_once()
+
+    def test_deduct_exception_returns_failure(self, service, db_session, user_id):
+        """deduct_points 抛异常时应捕获并返回 success=False。"""
+
+        with patch.object(service, "deduct_points", side_effect=RuntimeError("db boom")):
+            result = service.settle_viral_video(user_id, estimated=10.0, actual=20.0, txn_id="t6", db=db_session)
+
+        assert result["success"] is False
+        assert result["action"] == "deduct"
+        assert result["diff"] == 10.0
+
+
+class TestRefundViralVideo:
+    """refund_viral_video(): 爆款视频失败全额退款。"""
+
+    def test_zero_amount_returns_none_action(self, service, db_session, user_id):
+        """amount<=0 直接返回 none action，不调 refund_points。"""
+
+        with patch.object(service, "refund_points") as mock_refund:
+            r1 = service.refund_viral_video(user_id, 0, "t0", db_session)
+            r2 = service.refund_viral_video(user_id, None, "t0", db_session)
+            r3 = service.refund_viral_video(user_id, -1.5, "t0", db_session)
+
+        assert r1 == {"success": True, "action": "none", "amount": 0.0}
+        assert r2 == {"success": True, "action": "none", "amount": 0.0}
+        assert r3["action"] == "none"
+        mock_refund.assert_not_called()
+
+    def test_success_path_delegates_to_refund_points(self, service, db_session, user_id):
+        """成功路径：透传 user_id/amount/ref_id=txn_id/source=viral_video。"""
+
+        expected = {"success": True, "balance": 80.0, "transaction_id": "rf-1"}
+        with patch.object(service, "refund_points", return_value=expected) as mock_refund:
+            result = service.refund_viral_video(user_id, 30.0, "txn-xyz", db_session)
+
+        assert result == expected
+        mock_refund.assert_called_once()
+        rk = mock_refund.call_args.kwargs
+        assert rk["user_id"] == user_id
+        assert rk["amount"] == 30.0
+        assert rk["source"] == "viral_video"
+        assert rk["ref_id"] == "txn-xyz"
+        assert rk["description"] == "爆款视频失败退款"
+
+    def test_exception_returns_failure(self, service, db_session, user_id):
+        """refund_points 抛异常时返回 success=False/action=refund。"""
+
+        with patch.object(service, "refund_points", side_effect=RuntimeError("conn lost")):
+            result = service.refund_viral_video(user_id, 25.0, "txn-err", db_session)
+
+        assert result["success"] is False
+        assert result["action"] == "refund"
+        assert result["amount"] == 25.0

tests/unit/test_scripts_ai_points.py

@@ -1,76 +1,31 @@
-"""scripts_ai 积分扣点单元测试 (#1895 P2 step 2.3)"""
+"""scripts_ai (抖音解析/改写/标题) — v1.6.2 起全部免费，不扣积分"""
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
 
-import pytest
-from fastapi import HTTPException
+class TestScriptsAiFree:
+    """三个端点都已移除 @points_gate，不再扣点。"""
 
-import packages.middleware.points_gate as _pg_module
+    def test_all_scenes_return_zero_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
 
+        for scene in ("douyin_extract", "ai_rewrite", "ai_title"):
+            assert calculate_points_cost(scene, is_member=False) == 0
+            assert calculate_points_cost(scene, is_member=True) == 0
 
-def _make_cu(user_id="u1", 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
-    return cu
-
-
-@pytest.fixture(autouse=True)
-def _enable_gate(monkeypatch):
-    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
-    yield
-
-
-class TestScriptsAiPointsGate:
-    """测试 scripts_ai 三个端点都挂了 @points_gate 并正确扣费。"""
-
-    @pytest.mark.parametrize(
-        "scene,endpoint_fn_name",
-        [
-            ("douyin_extract", "extract_from_douyin"),
-            ("ai_rewrite", "ai_rewrite"),
-            ("ai_title", "ai_generate_titles"),
-        ],
-    )
-    def test_insufficient_points_raises_402(self, scene, endpoint_fn_name):
-        """积分不足时抛 402。"""
+    def test_no_points_gate_decorators(self):
         from app.api.routes import scripts_ai
-        from app.schemas.scripts_ai import (
-            AiGenerateTitlesRequest,
-            AiRewriteRequest,
-            ExtractFromDouyinRequest,
-        )
 
-        fn = getattr(scripts_ai, endpoint_fn_name)
-        db = MagicMock()
-        cu = _make_cu()
-        if scene == "douyin_extract":
-            req = ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
-        elif scene == "ai_rewrite":
-            req = AiRewriteRequest(content="测试文案")
-        else:
-            req = AiGenerateTitlesRequest(content="测试文案", count=3)
+        for fn_name in ("extract_from_douyin", "ai_rewrite", "ai_generate_titles"):
+            fn = getattr(scripts_ai, fn_name)
+            assert not hasattr(fn, "__wrapped__"), f"{fn_name} still has @points_gate"
 
-        with patch("packages.domain.points_service.PointsService") as MockSvc:
-            svc = MagicMock()
-            svc.deduct_points.return_value = {"success": False, "balance": 0}
-            MockSvc.return_value = svc
-            with pytest.raises(HTTPException) as ei:
-                fn(request=req, current_user=cu, db=db)
-            assert ei.value.status_code == 402
+    def test_module_no_points_imports(self):
+        import inspect
 
-    def test_disabled_passthrough_no_user_error(self, monkeypatch):
-        """关闭时不需要 user/db 也能被装饰器透传（验证 gate 关闭零副作用）。"""
         from app.api.routes import scripts_ai
-        from app.schemas.scripts_ai import AiRewriteRequest
 
-        monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
-        fn = scripts_ai.ai_rewrite
-        # 不带 db/current_user 也应透传（后续业务逻辑可能报错但不是 401/500 gate 错误）
-        with pytest.raises(Exception) as ei:
-            fn(request=AiRewriteRequest(content="x"), current_user=None, db=None)
-        # 不应是 gate 抛的 401/500
-        assert isinstance(ei.value, AttributeError) or ei.value.status_code not in (401, 500)
+        src = inspect.getsource(scripts_ai)
+        assert "PointsService" not in src
+        assert "points_gate" not in src
+        assert "calculate_points_cost" not in src

tests/unit/test_tts_voice_clone_points.py

@@ -1,6 +1,6 @@
-"""TTS + voice_clone 积分扣点单元测试 (#1895 P2 step 2.1)
+"""TTS (免费) + voice_clone 预览 (扣点) 单测 (#1895 P2 step 2.1)
 
-覆盖 synthesize / voice_clone preview 在积分开关下的扣点、余额不足、失败退费、会员折扣等分支。
+v1.6.2: TTS 合成/预览（ai_voice）已免费，不再扣点；voice_clone 预览（voice_clone_synth）仍保持 1积分/分钟扣点。
 """
 
 from __future__ import annotations
@@ -44,21 +44,10 @@ def _make_request(text="你好世界", voice_id="v1", **kw):
     return r
 
 
-def _est_minutes(chars: int) -> float:
-    return max(1.0, math.ceil(chars / 240))
+class TestTtsSynthesizeFree:
+    """TTS synthesize/preview 已移除手动扣点，不再实例化 PointsService。"""
 
-
-class TestEstimateMinutes:
-    @pytest.mark.parametrize(
-        "chars,expected",
-        [(1, 1.0), (240, 1.0), (241, 2.0), (480, 2.0), (481, 3.0), (1000, 5.0)],
-    )
-    def test_estimate(self, chars, expected):
-        assert _est_minutes(chars) == expected
-
-
-class TestTtsSynthesizePointsDeduction:
-    def _setup(self, text="你好", deduct_success=True, balance=0, start_synth_raises=None, send_task_raises=None):
+    def _setup(self, start_synth_raises=None):
         db = MagicMock()
         cu = _make_cu()
         repo = MagicMock()
@@ -77,43 +66,27 @@ class TestTtsSynthesizePointsDeduction:
             wf.start_synthesis.side_effect = start_synth_raises
         vc_repo = MagicMock()
         vc_repo.get.return_value = None
-        svc = MagicMock()
-        svc.deduct_points.return_value = {"success": deduct_success, "balance": balance}
-        fake_settings = MagicMock(points_enabled=True)
-        return db, cu, repo, uc, wf, vc_repo, svc, fake_settings, job
+        return db, cu, repo, uc, wf, vc_repo, job
 
-    def test_insufficient_raises_402(self):
-        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(text="你好" * 200, deduct_success=False, balance=0)
+    def test_module_has_no_points_imports(self):
+        import inspect
+
+        from app.api.routes import tts as mod
+
+        src = inspect.getsource(mod)
+        assert "PointsService" not in src
+        assert "calculate_points_cost" not in src
+        assert "_points_deducted" not in src
+        assert "import math" not in src
+
+    def test_success_returns_job_without_points(self):
+        db, cu, repo, uc, wf, vc_repo, job = self._setup()
         from app.api.routes.tts import synthesize
 
         with (
             patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
             patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
-            patch("app.api.routes.tts.PointsService", return_value=svc),
-            patch("app.api.routes.tts.settings", fs),
-        ):
-            with pytest.raises(HTTPException) as ei:
-                synthesize(
-                    request=_make_request(text="你好" * 200),
-                    authenticated_user=cu,
-                    db=db,
-                    repository=repo,
-                    cosyvoice_service=MagicMock(),
-                    voice_clone_repo=vc_repo,
-                )
-            assert ei.value.status_code == 402
-            assert ei.value.detail["code"] == "INSUFFICIENT_POINTS"
-
-    def test_success_deducts_points(self):
-        db, cu, repo, uc, wf, vc_repo, svc, fs, job = self._setup()
-        from app.api.routes.tts import synthesize
-
-        with (
-            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
-            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
-            patch("app.api.routes.tts.PointsService", return_value=svc),
-            patch("app.api.routes.tts.celery_app.send_task") as _st,
-            patch("app.api.routes.tts.settings", fs),
+            patch("app.api.routes.tts.celery_app.send_task"),
         ):
             resp = synthesize(
                 request=_make_request(text="测试"),
@@ -123,60 +96,17 @@ class TestTtsSynthesizePointsDeduction:
                 cosyvoice_service=MagicMock(),
                 voice_clone_repo=vc_repo,
             )
-            svc.deduct_points.assert_called_once()
             assert resp.job_id == job.id
 
-    def test_synthesis_failure_refunds(self):
-        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(start_synth_raises=RuntimeError("boom"))
-        from app.api.routes.tts import synthesize
-
-        with (
-            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
-            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
-            patch("app.api.routes.tts.PointsService", return_value=svc),
-            patch("app.api.routes.tts.celery_app.send_task"),
-            patch("app.api.routes.tts.settings", fs),
-        ):
-            synthesize(
-                request=_make_request(text="测试"),
-                authenticated_user=cu,
-                db=db,
-                repository=repo,
-                cosyvoice_service=MagicMock(),
-                voice_clone_repo=vc_repo,
-            )
-            assert svc.refund_points.called
-
-    def test_celery_send_failure_refunds(self):
-        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(send_task_raises=RuntimeError("celery down"))
-        from app.api.routes.tts import synthesize
-
-        with (
-            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
-            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
-            patch("app.api.routes.tts.PointsService", return_value=svc),
-            patch("app.api.routes.tts.celery_app.send_task", side_effect=RuntimeError("celery down")),
-            patch("app.api.routes.tts.settings", fs),
-        ):
-            synthesize(
-                request=_make_request(text="测试"),
-                authenticated_user=cu,
-                db=db,
-                repository=repo,
-                cosyvoice_service=MagicMock(),
-                voice_clone_repo=vc_repo,
-            )
-            assert svc.refund_points.called
-
-    def test_member_cheaper(self):
+    def test_ai_voice_cost_zero(self):
         from packages.domain.points_rules import calculate_points_cost
 
-        cf = calculate_points_cost("ai_voice", is_member=False, duration_minutes=2)
-        cm = calculate_points_cost("ai_voice", is_member=True, member_type="monthly", duration_minutes=2)
-        assert cm < cf
+        assert calculate_points_cost("ai_voice", is_member=False, duration_minutes=10) == 0
 
 
 class TestVoiceClonePreviewPoints:
+    """voice_clone 预览（voice_clone_synth）保持 1 积分/分钟扣点。"""
+
     def _setup(self, text="你好", deduct_success=True, balance=0, synth_raises=None):
         db = MagicMock()
         cu = _make_cu()
@@ -265,3 +195,10 @@ class TestVoiceClonePreviewPoints:
             )
             svc.deduct_points.assert_called_once()
             assert resp.audio_url.startswith("http")
+
+    def test_member_cheaper(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        cf = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=2)
+        cm = calculate_points_cost("voice_clone_synth", is_member=True, member_type="monthly", duration_minutes=2)
+        assert cm < cf

tests/unit/test_viral_video.py

@@ -17,7 +17,6 @@ import pytest
 from pydantic import ValidationError
 
 from packages.domain.viral_video import (
-    CREDITS_VIRAL_VIDEO_COST,
     STAGE_LABELS,
     FusionLevel,
     StyleStrength,
@@ -129,9 +128,6 @@ class TestViralVideoJobDefaults:
         assert job.result_video_url == ""
         assert job.error_msg == ""
 
-    def test_credits_cost_constant(self):
-        assert CREDITS_VIRAL_VIDEO_COST == 50
-
 
 class TestViralVideoStage:
     """阶段枚举测试。"""
@@ -559,7 +555,7 @@ class TestPipelineIntegration:
         mock_review.return_value = {"passed": True, "score": 90}
         mock_tts.return_value = None  # TTS 失败也能走下去（Seedance generate_audio=True 会自己合成音效）
         mock_tts_upload.return_value = None
-        mock_render.return_value = "/tmp/video.mp4"
+        mock_render.return_value = ("/tmp/video.mp4", {"completion_tokens": 1000000})
         mock_upload.return_value = "https://oss.example.com/final.mp4"
 
         result = resume_viral_video_pipeline.run("job-001")
@@ -567,4 +563,4 @@ class TestPipelineIntegration:
         assert result["ok"] is True
         assert result["video_url"] == "https://oss.example.com/final.mp4"
         assert job.status == ViralVideoStatus.COMPLETED
-        assert job.credits_cost == CREDITS_VIRAL_VIDEO_COST
+        assert isinstance(job.credits_cost, float)

tests/unit/test_viral_video_p0.py

@@ -212,10 +212,13 @@ class TestCallVideoGeneration:
         with patch("packages.shared.ai_service.get_doubao_client") as mock_get:
             mock_client = MagicMock()
             mock_client.is_available = True
-            mock_client.video_generation.return_value = str(out)
+            mock_client.video_generation.return_value = {
+                "video_path": str(out),
+                "usage": {"completion_tokens": 1000000},
+            }
             mock_get.return_value = mock_client
             result = call_video_generation(prompt="测试", image_url="https://img/x.jpg", duration=5, ratio="9:16")
-            assert result == str(out)
+            assert result is not None and result["video_path"] == str(out)
             mock_client.video_generation.assert_called_once()
             kwargs = mock_client.video_generation.call_args.kwargs
             assert kwargs["prompt"] == "测试"
@@ -238,7 +241,10 @@ class TestCallVideoGenerationV16:
         with patch("packages.shared.ai_service.get_doubao_client") as mock_get:
             mock_client = MagicMock()
             mock_client.is_available = True
-            mock_client.video_generation.return_value = str(out)
+            mock_client.video_generation.return_value = {
+                "video_path": str(out),
+                "usage": {"completion_tokens": 1500000},
+            }
             mock_get.return_value = mock_client
             result = call_video_generation(
                 prompt="测试",
@@ -251,10 +257,10 @@ class TestCallVideoGenerationV16:
                 generate_audio=True,
                 model="doubao-seedance-2-5-260628",
             )
-            assert result == str(out)
+            assert result is not None and result["video_path"] == str(out)
             kwargs = mock_client.video_generation.call_args.kwargs
-            # 首帧模式不传 ratio（Bug #2110）
-            assert "ratio" not in kwargs
+            # 图生视频也必须传 ratio（避免首帧方图导致默认输出 1:1）
+            assert kwargs.get("ratio") == "9:16", f"ratio 应透传，got {kwargs.get('ratio')!r}"
             assert kwargs["image_url"] == "https://img/x.jpg"
             assert kwargs["reference_audios"] == ["https://oss/tts.mp3"]
             assert kwargs["reference_images"] == ["https://img/r1.jpg"]
@@ -270,7 +276,7 @@ class TestCallVideoGenerationV16:
         with patch("packages.shared.ai_service.get_doubao_client") as mock_get:
             mock_client = MagicMock()
             mock_client.is_available = True
-            mock_client.video_generation.return_value = str(out)
+            mock_client.video_generation.return_value = {"video_path": str(out), "usage": None}
             mock_get.return_value = mock_client
             call_video_generation(prompt="测试", duration=10, ratio="16:9")
             kwargs = mock_client.video_generation.call_args.kwargs

tests/unit/test_viral_video_routes.py

@@ -14,6 +14,8 @@ from __future__ import annotations
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 
 def _auth_user(uid: str = "u1"):
     return SimpleNamespace(user=SimpleNamespace(id=uid))
@@ -60,7 +62,12 @@ def _make_job(job_id: str = "job-1", user_id: str = "u1", status: str = "pending
         "voice_mode": "global",
         "video_ratio": "9:16",
         "video_model": "",
-        "credits_cost": 0,
+        "video_resolution": "720p",
+        "credits_prepaid": 0.0,
+        "credits_transaction_id": "",
+        "credits_cost": 0.0,
+        "current_stage": "",
+        "phase_message": "",
         "updated_at": None,
         "is_terminal": False,
         "effective_copy_text": "",
@@ -131,6 +138,163 @@ class TestRetryViralVideo:
         mock_send.assert_called_once_with("worker.run_viral_video_pipeline", args=["job-retry"])
         assert resp.id == "job-retry"
 
+    def test_retry_without_body_keeps_original_params(self):
+        """不传 body 时，保持原参数且不调用积分服务。"""
+        from app.api.routes import viral_video as vv_mod
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-retry2", user_id="u1", status=ViralVideoStatus.FAILED,
+            duration=15, video_ratio="9:16", video_resolution="720p", video_model="seedance-2.5",
+            credits_prepaid=5.0, credits_transaction_id="txn1", retry_count=0,
+        )
+        repo = MagicMock()
+        repo.get.return_value = job
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService") as MockSvc,
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            # request=None (未传 body)
+            resp = vv_mod.retry_viral_video_job("job-retry2", None, authenticated_user=user, session=session)
+
+        MockSvc.assert_not_called()
+        assert resp.id == "job-retry2"
+        assert job.status == ViralVideoStatus.PENDING
+        assert job.duration == 15  # 参数不变
+
+    def test_retry_insufficient_points_raises_402(self):
+        """参数变更导致新预估更高且余额不足时，抛 402 阻止重试。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import RetryViralVideoRequest
+        from fastapi import HTTPException
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-retry3a", user_id="u1", status=ViralVideoStatus.FAILED,
+            duration=15, video_ratio="9:16", video_resolution="720p", video_model="seedance-2.5",
+            credits_prepaid=5.0, credits_transaction_id="txn-old", retry_count=0,
+        )
+        repo = MagicMock()
+        repo.get.return_value = job
+
+        fake_svc = MagicMock()
+        fake_svc.deduct_viral_video.return_value = {"success": False, "balance": 1.0}
+        req = RetryViralVideoRequest(duration=30, video_resolution="1080p", video_ratio="16:9", video_model="seedance-2.5")
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService", return_value=fake_svc),
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(1920, 1080)),
+            patch("packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                  return_value=(15.0, {})),
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            with pytest.raises(HTTPException) as exc:
+                vv_mod.retry_viral_video_job("job-retry3a", req, authenticated_user=user, session=session)
+            assert exc.value.status_code == 402
+        fake_svc.deduct_viral_video.assert_called_once()
+
+    def test_retry_higher_estimation_calls_deduct_delta(self):
+        """参数变更新预估更高时调用 deduct_viral_video 补扣差额，并更新 job 参数。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import RetryViralVideoRequest
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-retry3b", user_id="u1", status=ViralVideoStatus.FAILED,
+            duration=15, video_ratio="9:16", video_resolution="720p", video_model="seedance-2.5",
+            credits_prepaid=5.0, credits_transaction_id="txn-old", retry_count=0,
+        )
+        # 用 SimpleNamespace 让属性真正可写
+        from types import SimpleNamespace
+        job.credits_prepaid = 5.0
+        repo = MagicMock()
+        repo.get.return_value = job
+
+        fake_svc = MagicMock()
+        fake_svc.deduct_viral_video.return_value = {"success": True, "balance": 50.0, "transaction_id": "txn-new"}
+        req = RetryViralVideoRequest(duration=30, video_resolution="1080p", video_ratio="16:9", video_model="seedance-2.5")
+
+        new_est = 15.0
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService", return_value=fake_svc),
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(1920, 1080)),
+            patch("packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                  return_value=(new_est, {})),
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            resp = vv_mod.retry_viral_video_job("job-retry3b", req, authenticated_user=user, session=session)
+
+        assert job.duration == 30
+        assert job.video_resolution == "1080p"
+        assert job.video_ratio == "16:9"
+        # 补扣差额 = 15-5 = 10
+        fake_svc.deduct_viral_video.assert_called_once()
+        call_args = fake_svc.deduct_viral_video.call_args
+        assert call_args.args[1] == 10.0  # credits 是位置参数
+        assert resp.id == "job-retry3b"
+
+    def test_retry_lower_estimation_calls_refund_delta(self):
+        """参数变更新预估更低时，调用 refund_points 退还差额，并更新 job 参数。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import RetryViralVideoRequest
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-retry4", user_id="u1", status=ViralVideoStatus.FAILED,
+            duration=20, video_ratio="16:9", video_resolution="1080p", video_model="seedance-2.5",
+            credits_prepaid=10.0, credits_transaction_id="txn-old", retry_count=0,
+        )
+        job.credits_prepaid = 10.0
+        repo = MagicMock()
+        repo.get.return_value = job
+
+        fake_svc = MagicMock()
+        fake_svc.refund_points.return_value = {"success": True}
+        req = RetryViralVideoRequest(duration=5, video_resolution="480p", video_ratio="9:16")
+
+        new_est = 3.0
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService", return_value=fake_svc),
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(270, 480)),
+            patch("packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                  return_value=(new_est, {})),
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            vv_mod.retry_viral_video_job("job-retry4", req, authenticated_user=user, session=session)
+
+        assert job.duration == 5
+        assert job.video_resolution == "480p"
+        assert job.video_ratio == "9:16"
+        fake_svc.refund_points.assert_called_once()
+        call_args = fake_svc.refund_points.call_args
+        # 退差额 = 10-3 = 7
+        assert call_args.kwargs["amount"] == 7.0
+
 
 # ── confirm-intent ──────────────────────────────────────────────────────
 
@@ -397,3 +561,306 @@ class TestConfirmCopy:
             with pytest.raises(HTTPException) as exc:
                 vv_mod.confirm_copy("job-cc2", ConfirmCopyRequest(), authenticated_user=user, session=session)
             assert exc.value.status_code == 409
+
+
+# ── confirm-copy 积分预扣 + estimate-credits 端点 (#2151) ──────────────
+
+
+class TestConfirmCopyPointsDeduction:
+    """confirm_copy 中积分预扣分支（points_enabled=True）。"""
+
+    def test_points_enabled_deducts_successfully(self):
+        """points_enabled=True + 未预付 → 计算预估积分 → deduct_viral_video → 写入 credits_prepaid。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmCopyRequest
+        from fastapi import HTTPException
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-pay", user_id="u1", status=ViralVideoStatus.COPY_GENERATED)
+        # 默认 credits_prepaid=0, credits_cost=0 → 触发预扣
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmCopyRequest(edited_copy="改好的文案")
+
+        mock_svc = MagicMock()
+        mock_svc.deduct_viral_video.return_value = {"success": True, "balance": 100.0, "transaction_id": "txn-1"}
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch.object(vv_mod, "_settings", None, create=True),  # ensure not cached
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_rules.calculate_viral_video_credits", return_value=5.2),
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(720, 1280)),
+            patch("packages.domain.points_service.PointsService", return_value=mock_svc),
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            resp = vv_mod.confirm_copy("job-pay", req, authenticated_user=user, session=session)
+
+        # deduct_viral_video 被调用
+        mock_svc.deduct_viral_video.assert_called_once()
+        call_args = mock_svc.deduct_viral_video.call_args
+        assert call_args.args[0] == "u1"  # user_id
+        assert call_args.args[1] == 5.2   # credits
+        assert call_args.args[2] == "job-pay"  # job_id
+        # credits_prepaid / credits_transaction_id 被写入
+        assert job.credits_prepaid == 5.2
+        assert job.credits_transaction_id == "txn-1"
+        assert resp.id == "job-pay"
+        # resume + repo.update 至少调用过（其中一次是 credits 字段更新，一次是 resume 后）
+        job.resume_from_copy_generated.assert_called_once_with(edited_copy="改好的文案")
+
+    def test_points_enabled_insufficient_balance_raises_402(self):
+        """余额不足（deduct_viral_video 返回 success=False）→ HTTP 402。"""
+        import pytest
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmCopyRequest
+        from fastapi import HTTPException
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-402", user_id="u1", status=ViralVideoStatus.COPY_GENERATED)
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmCopyRequest()
+
+        mock_svc = MagicMock()
+        mock_svc.deduct_viral_video.return_value = {"success": False, "balance": 1.5, "transaction_id": None}
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_rules.calculate_viral_video_credits", return_value=10.0),
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(720, 1280)),
+            patch("packages.domain.points_service.PointsService", return_value=mock_svc),
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            with pytest.raises(HTTPException) as exc:
+                vv_mod.confirm_copy("job-402", req, authenticated_user=user, session=session)
+
+        assert exc.value.status_code == 402
+        detail = exc.value.detail
+        assert detail["code"] == "INSUFFICIENT_POINTS"
+        assert detail["required"] == 10.0
+        assert detail["balance"] == 1.5
+        # 预扣失败不应调用 resume 或 send_task
+        job.resume_from_copy_generated.assert_not_called()
+
+    def test_already_paid_skips_deduction(self):
+        """credits_prepaid>0（已经扣过费/重试场景） → 跳过预扣，不调用 PointsService。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmCopyRequest
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-paid",
+            user_id="u1",
+            status=ViralVideoStatus.COPY_GENERATED,
+            credits_prepaid=8.5,
+            credits_transaction_id="txn-old",
+        )
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmCopyRequest(edited_copy="继续")
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService") as MockSvc,
+            patch.object(vv_mod.celery_app, "send_task") as mock_send,
+        ):
+            mock_settings.points_enabled = True
+            resp = vv_mod.confirm_copy("job-paid", req, authenticated_user=user, session=session)
+
+        # PointsService 不应被实例化（没预扣）
+        MockSvc.assert_not_called()
+        job.resume_from_copy_generated.assert_called_once_with(edited_copy="继续")
+        mock_send.assert_called_once_with("worker.run_viral_video_render", args=["job-paid"])
+        assert resp.id == "job-paid"
+        # credits_prepaid 保持不变
+        assert job.credits_prepaid == 8.5
+
+    def test_already_paid_via_credits_cost_skips_deduction(self):
+        """credits_cost>0 也算已付费（兼容旧字段），跳过预扣。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmCopyRequest
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(
+            job_id="job-paid2",
+            user_id="u1",
+            status=ViralVideoStatus.COPY_GENERATED,
+            credits_prepaid=0,
+            credits_cost=7.0,
+        )
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmCopyRequest()
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService") as MockSvc,
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = True
+            vv_mod.confirm_copy("job-paid2", req, authenticated_user=user, session=session)
+
+        MockSvc.assert_not_called()
+
+    def test_points_disabled_skips_deduction(self):
+        """points_enabled=False 时不进入预扣逻辑，保持原流程。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmCopyRequest
+
+        from packages.domain.viral_video import ViralVideoStatus
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-free", user_id="u1", status=ViralVideoStatus.COPY_GENERATED)
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmCopyRequest()
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch("app.config.settings") as mock_settings,
+            patch("packages.domain.points_service.PointsService") as MockSvc,
+            patch.object(vv_mod.celery_app, "send_task"),
+        ):
+            mock_settings.points_enabled = False
+            vv_mod.confirm_copy("job-free", req, authenticated_user=user, session=session)
+
+        MockSvc.assert_not_called()
+        job.resume_from_copy_generated.assert_called_once()
+
+
+class TestEstimateCredits:
+    """POST /estimate-credits: 纯计算预估积分。"""
+
+    def test_estimate_returns_float_with_breakdown(self):
+        """正常参数应返回 estimated_credits(float, >0, 两位小数) + formula_breakdown。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import EstimateCreditsRequest
+
+        req = EstimateCreditsRequest(model="seedance-2.5", resolution="720p", ratio="9:16", duration=15)
+        user = _auth_user("u1")
+
+        resp = vv_mod.estimate_credits(req, authenticated_user=user)
+        assert isinstance(resp.estimated_credits, float)
+        assert resp.estimated_credits > 0
+        assert round(resp.estimated_credits, 2) == resp.estimated_credits
+        # formula_breakdown 必须返回并包含全部字段
+        bd = resp.formula_breakdown
+        assert bd.tokens > 0
+        assert bd.video_cost >= 0
+        assert bd.fixed_cost > 0
+        assert bd.profit_multiplier == 1.3
+        assert bd.model_price > 0
+        assert bd.width > 0
+        assert bd.height > 0
+        assert bd.fps > 0
+
+    def test_estimate_uses_dimensions_resolver_and_with_breakdown(self):
+        """estimate_credits 调用 resolve_video_dimensions 与 calculate_viral_video_credits_with_breakdown。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import EstimateCreditsRequest
+
+        req = EstimateCreditsRequest(model="seedance-2.5", resolution="1080p", ratio="16:9", duration=20)
+        user = _auth_user("u1")
+        fake_bd = {
+            "tokens": 1000.0, "video_cost": 1.0, "fixed_cost": 0.15,
+            "profit_multiplier": 1.3, "model_price": 70.0,
+            "width": 1920, "height": 1080, "fps": 24,
+        }
+
+        with (
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(1920, 1080)) as mock_res,
+            patch(
+                "packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                return_value=(8.88, fake_bd),
+            ) as mock_calc,
+        ):
+            resp = vv_mod.estimate_credits(req, authenticated_user=user)
+
+        mock_res.assert_called_once_with("1080p", "16:9")
+        mock_calc.assert_called_once()
+        args, kwargs = mock_calc.call_args
+        assert args[0] == 20
+        assert args[1] == 1920
+        assert args[2] == 1080
+        assert args[3] == "seedance-2.5"
+        assert resp.estimated_credits == 8.88
+        assert resp.formula_breakdown.width == 1920
+        assert resp.formula_breakdown.height == 1080
+        assert resp.formula_breakdown.model_price == 70.0
+
+    def test_estimate_empty_model_defaults_to_seedance_2_5(self):
+        """model 为空字符串时，传入 calculate 的 model 参数应为 'seedance-2.5'。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import EstimateCreditsRequest
+
+        req = EstimateCreditsRequest(model="", resolution="720p", ratio="9:16", duration=10)
+        user = _auth_user("u1")
+        fake_bd = {
+            "tokens": 500.0, "video_cost": 0.5, "fixed_cost": 0.15,
+            "profit_multiplier": 1.3, "model_price": 70.0,
+            "width": 720, "height": 1280, "fps": 24,
+        }
+
+        with (
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(720, 1280)),
+            patch(
+                "packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                return_value=(3.5, fake_bd),
+            ) as mock_calc,
+        ):
+            resp = vv_mod.estimate_credits(req, authenticated_user=user)
+
+        args, kwargs = mock_calc.call_args
+        assert args[3] == "seedance-2.5"
+        assert resp.estimated_credits == 3.5
+        assert resp.formula_breakdown.height == 1280
+
+    def test_estimate_accepts_video_model_alias(self):
+        """前端传 video_model/video_resolution/video_ratio（别名）也应被正确解析。"""
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import EstimateCreditsRequest
+
+        req = EstimateCreditsRequest.model_validate(
+            {"video_model": "seedance-2.0", "video_resolution": "480p", "video_ratio": "1:1", "duration": 5}
+        )
+        user = _auth_user("u1")
+        fake_bd = {
+            "tokens": 100.0, "video_cost": 0.1, "fixed_cost": 0.15,
+            "profit_multiplier": 1.3, "model_price": 46.0,
+            "width": 480, "height": 480, "fps": 24,
+        }
+
+        with (
+            patch("packages.domain.points_rules.resolve_video_dimensions", return_value=(480, 480)) as mock_res,
+            patch(
+                "packages.domain.points_rules.calculate_viral_video_credits_with_breakdown",
+                return_value=(1.0, fake_bd),
+            ) as mock_calc,
+        ):
+            vv_mod.estimate_credits(req, authenticated_user=user)
+
+        mock_res.assert_called_once_with("480p", "1:1")
+        args, kwargs = mock_calc.call_args
+        assert args[0] == 5
+        assert args[1] == 480
+        assert args[2] == 480
+        assert args[3] == "seedance-2.0"