feat(points): P2 step 2.6 - ai_avatar_render 数字人渲染 points gate (#1943)

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

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

@@ -29,6 +29,8 @@ 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()
@@ -42,10 +44,12 @@ 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),
     svc: AiAvatarRenderService = Depends(_get_service),
+    db: Session = Depends(get_db_session),
 ):
     """提交 AI 数字人渲染任务.
 

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

@@ -43,6 +43,7 @@ from packages.application import (
     GetGenerationTaskUseCase,
     ListGeneratedVideosByTaskUseCase,
 )
+from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -271,6 +272,7 @@ 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

@@ -42,6 +42,7 @@ from packages.application import (
     ListGeneratedVideosByTaskUseCase,
 )
 from packages.domain.smart_match import smart_select_assets
+from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -211,6 +212,7 @@ 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,9 +12,11 @@
 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,
@@ -30,6 +32,9 @@ 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()
@@ -53,8 +58,40 @@ def _get_service(
 def create_lipsync_job(
     body: CreateLipsyncJobRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
+    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"],
+                },
+            )
     """提交对口型任务.
 
     三种模式：
@@ -66,7 +103,7 @@ def create_lipsync_job(
     """
     try:
         job = svc.create_job(
-            user_id=current_user.user.id,
+            user_id=user_id,
             video_url=body.video_url,
             audio_url=body.audio_url,
             audio_duration=body.audio_duration,
@@ -79,8 +116,18 @@ 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
@@ -96,11 +143,24 @@ 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
 
 
@@ -111,8 +171,34 @@ def create_lipsync_job(
 def preview_tts(
     body: AiAvatarTtsPreviewRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
+    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 时长 → 句子时间戳计算，
@@ -121,13 +207,18 @@ def preview_tts(
     """
     try:
         result = svc.preview_tts(
-            user_id=current_user.user.id,
+            user_id=user_id,
             voice_id=body.voice_id,
             script_text=body.script_text,
             speed=body.speed,
             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
@@ -142,6 +233,11 @@ 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/scripts_ai.py

@@ -13,6 +13,7 @@ import re
 import tempfile
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session
 from app.schemas.scripts_ai import (
     AiGenerateTitlesRequest,
     AiGenerateTitlesResponse,
@@ -27,7 +28,9 @@ from app.services.script_asr_service import (
     transcribe_to_text,
 )
 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__)
@@ -62,9 +65,11 @@ def _validate_douyin_url(url: str) -> None:
     "/extract-from-douyin",
     response_model=ExtractFromDouyinResponse,
 )
+@points_gate("douyin_extract")
 def extract_from_douyin(
     request: ExtractFromDouyinRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> ExtractFromDouyinResponse:
     """从抖音视频下载无水印视频并通过 ASR 提取文案."""
     source_url = request.url.strip()
@@ -138,9 +143,11 @@ def extract_from_douyin(
     "/ai-rewrite",
     response_model=AiRewriteResponse,
 )
+@points_gate("ai_rewrite")
 def ai_rewrite(
     request: AiRewriteRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> AiRewriteResponse:
     """使用豆包大模型改写文案."""
     content = (request.content or "").strip()
@@ -206,9 +213,11 @@ def ai_rewrite(
     "/ai-generate-titles",
     response_model=AiGenerateTitlesResponse,
 )
+@points_gate("ai_title")
 def ai_generate_titles(
     request: AiGenerateTitlesRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> AiGenerateTitlesResponse:
     """使用现有 generate_smart_titles 生成标题."""
     content = (request.content or "").strip()

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

@@ -4,12 +4,14 @@ 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 (
@@ -51,6 +53,8 @@ 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
@@ -128,6 +132,7 @@ def _to_response(job, sign_url=None) -> TTSJobResponse:
 def synthesize(
     request: TTSSynthesizeRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
     cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
     voice_clone_repo=Depends(get_voice_clone_profile_repository),
@@ -139,6 +144,31 @@ 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
@@ -198,6 +228,7 @@ def synthesize(
         cosyvoice_service=cosyvoice_service,
     )
 
+    synthesis_error: Exception | None = None
     try:
         job = workflow.start_synthesis(job.id)
     except Exception as e:
@@ -205,10 +236,17 @@ 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":
@@ -223,10 +261,17 @@ def synthesize(
                 celery_app.send_task("worker.process_tts_synthesis", args=[job.id])
         except Exception as e:
             # Celery 调度失败，标记 job 为 failed
+            # e used below for refund context
             try:
                 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,
@@ -553,6 +598,7 @@ def save_tts_job_to_library(
 def preview_tts(
     request: TTSPreviewRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
     voice_clone_repo=Depends(get_voice_clone_profile_repository),
 ) -> TTSPreviewResponse:
@@ -561,6 +607,31 @@ 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)
@@ -586,16 +657,16 @@ def preview_tts(
             emotion=request.emotion,
             language=getattr(request, "language", "zh-CN"),
         )
-    except CosyVoiceError as e:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"TTS 合成失败: {e}",
-        ) from e
-    except ValueError as e:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=str(e),
-        ) from e
+    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
 
     return TTSPreviewResponse(
         audio_url=result.audio_url,

apps/api/app/api/routes/voice_clones.py

@@ -3,14 +3,17 @@
 from __future__ import annotations
 
 import logging
+import math
 from typing import 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 (
     get_asset_repository,
     get_cosyvoice_service,
+    get_db_session,
     get_project_repository,
     get_voice_clone_profile_repository,
 )
@@ -22,6 +25,7 @@ from app.schemas.voice_clone import (
     VoiceCloneStatusResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
     SQLAlchemyVoiceCloneProfileRepository,
@@ -38,6 +42,11 @@ from packages.application.voice_clone.use_cases import (
 from packages.application.voice_clone.workflow import (
     VoiceCloneWorkflowService,
 )
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
+
+# remove duplicate
+_DUMMY_DELETED = ()
 from packages.ports.asset_repository import AssetRepository
 from packages.ports.project_repository import ProjectRepository
 from packages.shared.storage import SharedStorageService
@@ -339,6 +348,7 @@ def get_voice_clone_preview(
         description="情绪：neutral/happy/sad/angry/surprised/fearful/disgusted，兼容旧值 natural/excited/calm/friendly，空为默认自然",
     ),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
     cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service),
 ) -> VoiceClonePreviewResponse:
@@ -350,6 +360,31 @@ def get_voice_clone_preview(
     """
     import time
 
+    user_id = authenticated_user.user.id
+    _points_deducted = 0
+    _points_scene = "voice_clone_synth"
+    _points_svc = PointsService() if settings.points_enabled else None
+    _preview_text_for_points = text.strip() or CLONE_PREVIEW_TEMPLATE
+    if _points_svc is not None:
+        est_minutes = max(1.0, math.ceil(len(_preview_text_for_points) / 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"],
+                },
+            )
+
     if emotion not in _ALLOWED_PREVIEW_EMOTIONS:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -393,9 +428,14 @@ def get_voice_clone_preview(
             speed=speed,
             emotion=emotion,
         )
-    except CosyVoiceError as e:
-        raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e
-    except ValueError as e:
+    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"克隆音色试听失败退积分异常: clone_id={clone_id}, err={refund_err}")
+        if isinstance(e, CosyVoiceError):
+            raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
 
     # 缓存（仅默认参数组合）