fix(points) P0: 分辨率映射改为短边规则，修复竖屏积分少算68%的计费bug (#2154)

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

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,
@@ -140,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,
@@ -193,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,
     )
 
@@ -235,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)
@@ -298,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)
@@ -330,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)
 
@@ -344,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,
@@ -397,10 +470,17 @@ 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 任务强制重置后重试）。"""
+    """重试失败的爆款视频任务（也支持对僵尸/超时 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)
@@ -421,6 +501,100 @@ def retry_viral_video_job(
     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
@@ -435,7 +609,10 @@ def retry_viral_video_job(
     # 重新入队
     try:
         celery_app.send_task("worker.run_viral_video_pipeline", args=[job.id])
-        logger.info("[爆款视频] 重试入队: job_id=%s retry_count=%d stale=%s", job.id, job.retry_count, is_stale_running)
+        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
@@ -218,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
@@ -250,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/ViralVideoPage.tsx

@@ -1,4 +1,5 @@
 import React, { useCallback, useEffect, useRef, useState } from "react"
+import axios from "axios"
 import {
   PlusOutlined,
   CloseOutlined,
@@ -46,6 +47,7 @@ import {
   analyzeViralImages,
   generateViralCopy,
   confirmViralCopy,
+  estimateViralVideoCredits,
 } from "@/api/viral-video"
 import { useViralVideoPolling } from "./hooks/useViralVideoPolling"
 import CloneModal from "@/components/voice/CloneModal"
@@ -123,6 +125,9 @@ type TabTask = {
   videoModel: string
   quality: string
   upscale: string
+  // 积分预估
+  estimatedCredits: number | null // null=加载中/未发起；浮点数，展示时 toFixed(2)
+  estimatedCreditsError: boolean // true=接口失败显示 --
   // UI 状态
   uiStep: UIStep
   imageAnalysis: ImageAnalysisResult | null
@@ -395,6 +400,8 @@ const emptyTask = (id: string, title: string): TabTask => ({
   videoModel: "seedance-2.5",
   quality: "480p",
   upscale: "default",
+  estimatedCredits: null,
+  estimatedCreditsError: false,
   uiStep: "step1_upload",
   imageAnalysis: null,
   storyboard: null,
@@ -565,6 +572,50 @@ const ViralVideoPage: React.FC = () => {
   )
   useViralVideoPolling(task.jobId, onPollUpdate)
 
+  /* ── 积分动态预估：STEP2文案生成完成后首次调用；STEP3参数变化时防抖300ms刷新 ── */
+  const creditsTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null)
+  const creditsAbortRef = useRef<number>(0)
+  useEffect(() => {
+    const creditsReady =
+      task.uiStep === "step2_copy_ready" ||
+      task.uiStep === "step3_ready" ||
+      task.uiStep === "step3_generating" ||
+      task.uiStep === "step3_done" ||
+      task.uiStep === "failed"
+    if (!creditsReady) {
+      // 未到可预估阶段：清空状态
+      if (task.estimatedCredits !== null || task.estimatedCreditsError) {
+        setTask({ estimatedCredits: null, estimatedCreditsError: false })
+      }
+      return
+    }
+    // 防抖 300ms
+    if (creditsTimerRef.current) clearTimeout(creditsTimerRef.current)
+    const seq = ++creditsAbortRef.current
+    creditsTimerRef.current = setTimeout(async () => {
+      try {
+        const res = await estimateViralVideoCredits({
+          video_model: task.videoModel,
+          resolution: task.quality,
+          video_ratio: task.videoRatio,
+          duration: task.duration,
+        })
+        if (seq !== creditsAbortRef.current) return
+        setTask({
+          estimatedCredits: res.estimated_credits,
+          estimatedCreditsError: false,
+        })
+      } catch {
+        if (seq !== creditsAbortRef.current) return
+        setTask({ estimatedCredits: null, estimatedCreditsError: true })
+      }
+    }, 300)
+    return () => {
+      if (creditsTimerRef.current) clearTimeout(creditsTimerRef.current)
+    }
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [task.videoModel, task.quality, task.videoRatio, task.duration, task.uiStep])
+
   /* ── OSS 上传 ── */
   const uploadFile = useCallback(
     async (file: File, kind: "image" | "video" | "voice", onProgress?: (pct: number) => void) => {
@@ -947,8 +998,21 @@ const ViralVideoPage: React.FC = () => {
       setTask({ job, jobId: job.id })
       message.success("视频已提交生成，预计 1-3 分钟…")
     } catch (err: unknown) {
-      message.error(err instanceof Error ? err.message : "提交失败")
-      setTask({ uiStep: "failed", videoError: err instanceof Error ? err.message : "提交失败" })
+      // 402 Payment Required → 积分不足
+      if (axios.isAxiosError(err) && err.response?.status === 402) {
+        const detail =
+          (err.response.data as { detail?: string; message?: string; msg?: string } | undefined)
+            ?.detail ||
+          (err.response.data as { detail?: string; message?: string; msg?: string } | undefined)
+            ?.message ||
+          "积分不足，请充值"
+        message.error(detail)
+        setTask({ uiStep: "step3_ready", videoError: detail })
+        return
+      }
+      const msg = err instanceof Error ? err.message : "提交失败"
+      message.error(msg)
+      setTask({ uiStep: "failed", videoError: msg })
     }
   }
 
@@ -2031,13 +2095,21 @@ const ViralVideoPage: React.FC = () => {
             </div>
           </div>
 
-          <div className="vv-credits">
-            <span>
-              <SoundOutlined style={{ marginRight: 4 }} />
-              预计消耗
-            </span>
-            <strong>50 积分</strong>
-          </div>
+          {step3Enabled && (
+            <div className="vv-credits">
+              <span>
+                <SoundOutlined style={{ marginRight: 4 }} />
+                预计消耗
+              </span>
+              <strong>
+                {task.estimatedCreditsError
+                  ? "-- 积分"
+                  : task.estimatedCredits === null
+                    ? "… 积分"
+                    : `${task.estimatedCredits.toFixed(2)} 积分`}
+              </strong>
+            </div>
+          )}
 
           {/* 视频预览区 */}
           <div className="vv-section">

apps/worker/worker_app/tasks/viral_video.py

@@ -37,7 +37,6 @@ from packages.adapters.sqlalchemy_impl.viral_video_repository import (
     SQLAlchemyViralVideoJobRepository,
 )
 from packages.domain.viral_video import (
-    CREDITS_VIRAL_VIDEO_COST,
     STAGE_LABELS,
     ViralVideoJob,
     ViralVideoStage,
@@ -629,7 +628,7 @@ _SCRIPT_GENERATION_PROMPT = """你是资深短视频导演，为 Seedance 2.5（
 6. hard_constraints/negative_prompts 保留默认项可追加，不要删减。
 7. voiceover_script 为纯口播文本（无标记/括号/前缀），{duration}秒约{approx_chars}字。
 8. 严格按上方「爆款结构」的节奏/段落顺序编排（钩子/痛点/反转/案例/行动号召与结构对齐）。
-9. 输出前自检：口播对白禁止错别字和语病（特别注意"很/最"等常见误用），同音字错误一律修正。
+9. 输出前自检：口播对白禁止错别字和语病，**严禁使用"很近"，正确用词是"最近"**（指"最近一段时间/最近在用"，绝不能写成"很近"）；其他同音字、形近字错误一律修正。
 10. 必须使用产品信息中真实的品牌、品名和外观特征，不要编造与产品无关的内容。"""
 
 
@@ -921,8 +920,20 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
         voiceover_len = len(voiceover)
         shots_cnt = len((normalized or {}).get("shots") or [])
         # 判定是否"退化到兜底质量"：口播过短（<20字）或镜头数<1；正常的短口播（如15s视频~40字）不视为兜底
+        # v1.6.1 双保险：先做 hard fix 字符串替换（"很近" → "最近"），再做不合格判定
+        if "很近" in voiceover:
+            logger.warning("[爆款视频] 编导脚本含错别字'很近'，hard fix 替换为'最近' label=%s", label)
+            voiceover = voiceover.replace("很近", "最近")
+            normalized["voiceover_script"] = voiceover
+            # 同时在 shots 对白里替换
+            for sh in normalized.get("shots") or []:
+                if isinstance(sh, dict):
+                    sd = sh.get("scene_and_dialogue") or ""
+                    if "很近" in sd:
+                        sh["scene_and_dialogue"] = sd.replace("很近", "最近")
         fallback_marker = "我最近在用的好物" in voiceover  # _fallback_script 的特征串
-        is_fallback = fallback_marker or shots_cnt < 1 or voiceover_len < 20
+        has_typo_henjin = "很近" in voiceover  # v1.6.1: 错别字"很近"视为不合格，触发重试
+        is_fallback = fallback_marker or shots_cnt < 1 or voiceover_len < 20 or has_typo_henjin
         logger.info(
             "[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s raw_type=%s",
             label,
@@ -1098,14 +1109,18 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str:
     return "\n".join(lines)
 
 
-def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | None) -> str:
-    """步骤 6: v1.6 单次 Seedance 生成（不再分段/拼接）。"""
+def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | None) -> tuple[str, dict | None]:
+    """步骤 6: v1.6 单次 Seedance 生成（不再分段/拼接）。
+
+    返回 (本地视频路径, usage dict|None)。失败抛异常。
+    """
     from packages.shared.ai_service import call_video_generation
 
     prompt = _assemble_seedance_prompt(copy_result, job)
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
     ratio = getattr(job, "video_ratio", None) or "9:16"
     model = getattr(job, "video_model", "") or None
+    resolution = getattr(job, "video_resolution", "720p") or "720p"
 
     # reference_audios: TTS 音频驱动口型
     ref_audios = [tts_audio_url] if tts_audio_url else []
@@ -1129,12 +1144,12 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
     )
     logger.info("[爆款视频] Seedance prompt (前300字): %s", prompt[:300])
 
-    video_path = call_video_generation(
+    result = call_video_generation(
         prompt=prompt,
         image_url=first_image,
         duration=dur,
         ratio=ratio,
-        resolution="720p",
+        resolution=resolution,
         output_dir=str(tmpdir),
         model=model,
         generate_audio=True,  # Seedance 原生生成环境音效/BGM；口型由 reference_audios 的 TTS 驱动
@@ -1142,10 +1157,16 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non
         reference_audios=ref_audios,
         reference_videos=ref_videos,
     )
+    if not result or not isinstance(result, dict):
+        raise RuntimeError("Seedance 视频生成失败：返回为空")
+    video_path = result.get("video_path") or ""
+    usage = result.get("usage")
     if not video_path or not Path(video_path).exists() or Path(video_path).stat().st_size == 0:
         raise RuntimeError("Seedance 视频生成失败：返回空文件或路径不存在")
-    logger.info("[爆款视频] Seedance 单次生成完成: %s size=%d", video_path, Path(video_path).stat().st_size)
-    return str(video_path)
+    logger.info(
+        "[爆款视频] Seedance 单次生成完成: %s size=%d usage=%s", video_path, Path(video_path).stat().st_size, usage
+    )
+    return str(video_path), (usage if isinstance(usage, dict) else None)
 
 
 def _step_upload(job: ViralVideoJob, video_path: str) -> str:
@@ -1543,6 +1564,90 @@ def _quick_compliance_blacklist_check(copy_result: dict) -> None:
                     copy_result[k] = copy_result[k].replace(bk, bv)
 
 
+def _try_refund_viral_video(job: ViralVideoJob) -> None:
+    """爆款视频生成失败：若已预扣积分则全额退款。"""
+    try:
+        from packages.shared import get_shared_settings
+
+        _s = get_shared_settings()
+        if not _s.points_enabled:
+            return
+        prepaid = float(getattr(job, "credits_prepaid", 0) or 0)
+        if prepaid <= 0:
+            return
+        from packages.domain.points_service import PointsService
+
+        svc = PointsService()
+        # 使用独立 session（避免污染外层事务）
+        ssn = SessionLocal()
+        try:
+            svc.refund_viral_video(
+                job.user_id,
+                prepaid,
+                getattr(job, "credits_transaction_id", "") or "",
+                ssn,
+            )
+            job.credits_prepaid = 0.0
+        finally:
+            ssn.close()
+    except Exception:
+        logger.exception("[爆款视频] 失败退款异常 job_id=%s", job.id)
+
+
+def _settle_viral_video(job: ViralVideoJob, usage: dict | None) -> None:
+    """爆款视频生成成功：按实际 usage 结算，多退少补，写 credits_cost。"""
+    try:
+        from packages.shared import get_shared_settings
+
+        _s = get_shared_settings()
+        if not _s.points_enabled:
+            job.credits_cost = 0.0
+            job.credits_prepaid = 0.0
+            return
+        prepaid = float(getattr(job, "credits_prepaid", 0) or 0)
+        if prepaid <= 0:
+            job.credits_cost = 0.0
+            return
+        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",
+            getattr(job, "video_ratio", "9:16") or "9:16",
+        )
+        actual_tokens = None
+        if isinstance(usage, dict):
+            at = usage.get("completion_tokens")
+            if isinstance(at, (int, float)) and at > 0:
+                actual_tokens = int(at)
+        actual_credits = calculate_viral_video_credits(
+            int(getattr(job, "duration", 15) or 15),
+            w,
+            h,
+            getattr(job, "video_model", "") or "seedance-2.5",
+            actual_tokens=actual_tokens,
+        )
+        svc = PointsService()
+        ssn = SessionLocal()
+        try:
+            svc.settle_viral_video(
+                job.user_id,
+                prepaid,
+                actual_credits,
+                getattr(job, "credits_transaction_id", "") or "",
+                ssn,
+            )
+            job.credits_cost = actual_credits
+            job.credits_prepaid = 0.0
+        finally:
+            ssn.close()
+    except Exception:
+        logger.exception("[爆款视频] 积分结算异常 job_id=%s", job.id)
+        # 结算异常不阻塞任务完成：保守按预扣值记 credits_cost
+        job.credits_cost = float(getattr(job, "credits_prepaid", 0) or 0)
+        job.credits_prepaid = 0.0
+
+
 def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
     """v1.6.1 阶段3：出片前合规审核（LLM 深度）→ TTS → Seedance → Upload → Completed。
 
@@ -1584,9 +1689,17 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
     tts_url = _upload_tts_to_oss(job, tts_path)
     _emit_progress(job_id, ViralVideoStage.TTS, 78.0, "配音完成", {"has_tts": tts_url is not None})
 
-    # Step 6: 单次 Seedance
+    # Step 6: 单次 Seedance（失败自动退款）
     _set_stage(job, repo, session, ViralVideoStage.RENDERING, "正在生成视频（约1-3分钟）...")
-    video_path = _step_render(job, copy_result, tts_url)
+    video_path = None
+    usage = None
+    try:
+        video_path, usage = _step_render(job, copy_result, tts_url)
+    except Exception as e:
+        logger.error("[爆款视频][阶段3] Seedance 生成失败，触发退款: %s", e, exc_info=True)
+        # 退款
+        _try_refund_viral_video(job)
+        raise
     _emit_progress(job_id, ViralVideoStage.RENDERING, 92.0, "视频生成完成")
 
     # Step 7: Upload
@@ -1598,7 +1711,8 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict:
     if video_url:
         _wait_oss_ready(video_url, timeout_sec=10)
 
-    job.credits_cost = CREDITS_VIRAL_VIDEO_COST
+    # 积分结算：按实际 tokens 多退少补
+    _settle_viral_video(job, usage)
     job.mark_completed(video_url)
     job.current_stage = ViralVideoStage.UPLOADING
     job.phase_message = "视频生成完成"
@@ -1636,6 +1750,23 @@ def run_viral_video_render(self: Task, job_id: str) -> dict:
         raise
     except Exception as e:
         logger.error("[爆款视频][阶段3] 异常: %s", e, exc_info=True)
+        # 兜底：任何阶段3异常都尝试退款（_step_render 内部异常已经退过，但 upload 等后续失败也需退）
+        try:
+            if session is not None:
+                job_safe = None
+                try:
+                    repo_safe = SQLAlchemyViralVideoJobRepository(session)
+                    job_safe = repo_safe.get(job_id)
+                except Exception:
+                    pass
+                if job_safe is not None and float(getattr(job_safe, "credits_prepaid", 0) or 0) > 0:
+                    _try_refund_viral_video(job_safe)
+                    try:
+                        repo_safe.update(job_safe)
+                    except Exception:
+                        pass
+        except Exception:
+            logger.exception("[爆款视频][阶段3] 兜底退款异常")
         _mark_failed_and_notify(job_id, session, None, None, str(e), ViralVideoStage.RENDERING)
         return {"ok": False, "job_id": job_id, "error": str(e)}
     finally:

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))
@@ -961,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

@@ -46,7 +46,10 @@ def _to_domain(model: ViralVideoJobModel) -> ViralVideoJob:
         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,
@@ -95,7 +98,10 @@ class SQLAlchemyViralVideoJobRepository:
             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,
@@ -121,7 +127,10 @@ class SQLAlchemyViralVideoJobRepository:
         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/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: "视频风格分析",
@@ -121,7 +119,10 @@ class ViralVideoJob:
     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

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

@@ -261,8 +261,11 @@ 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 生视频（异步任务→轮询→下载）。
+
+        成功返回 {"video_path": str, "usage": dict | None}，失败返回 None。
+        usage 是 Seedance 返回的计费信息（含 completion_tokens）。
 
         【v1.6.1 修复】严格按官方 content 数组协议构造请求：
         - 所有参考（图/音/视）必须放进 content 数组并带 role 字段，不能放顶层 reference_audios/reference_videos（非官方字段，会被忽略或导致异常）。
@@ -420,6 +423,7 @@ class DoubaoClient:
         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:
@@ -437,8 +441,9 @@ 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", task_id, poll_count)
+                        logger.info("Seedance 任务成功: task_id=%s polls=%d usage=%s", task_id, poll_count, usage)
                         break
                     last_err = RuntimeError(f"task succeeded but no video_url: {str(data)[:300]}")
                     logger.error("Seedance succeeded 但无 video_url: %s", last_err)
@@ -500,7 +505,7 @@ class DoubaoClient:
                 except Exception:
                     pass
                 return None
-            return local_path
+            return {"video_path": local_path, "usage": usage}
         except Exception as e:
             logger.error("Seedance 视频下载失败: %s", e, exc_info=True)
             return None

packages/shared/ai_service.py

@@ -617,8 +617,10 @@ 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.1 单次出片版），返回本地 MP4 路径；失败返回 None。
+) -> dict | None:
+    """调用 Seedance 2.5 生成视频（v1.6.1 单次出片版）。
+
+    成功返回 {"video_path": str, "usage": dict | None}（usage 含 completion_tokens），失败返回 None。
 
     v1.6.1 关键约束（避免 20min 卡死）：
     - 参考音频/视频/多图全部放进 content 数组并带 role=reference_audio/reference_video/reference_image；

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

@@ -87,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_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,7 +257,7 @@ 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（避免首帧方图导致默认输出 1:1）
             assert kwargs.get("ratio") == "9:16", f"ratio 应透传，got {kwargs.get('ratio')!r}"
@@ -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,10 @@ 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,
@@ -133,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 ──────────────────────────────────────────────────────
 
@@ -399,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"