diff --git a/video-gen-api/app/api/v1/__init__.py b/video-gen-api/app/api/v1/__init__.py index cdbe67ce..e80ee5b1 100644 --- a/video-gen-api/app/api/v1/__init__.py +++ b/video-gen-api/app/api/v1/__init__.py @@ -15,6 +15,7 @@ from app.api.v1.recharge_packages import router as recharge_packages_router from app.api.v1.video_engines import router as video_engines_router from app.api.v1.image_engines import router as image_engines_router from app.api.v1.generation_ai import router as generation_ai_router +from app.api.v1.hot_opening_replicate import router as hot_opening_replicate_router from app.api.v1.test import router as test_router from app.api.v1.user_oauth import router as user_oauth_router from app.api.v1.user_oauth_app import router as user_oauth_app_router @@ -35,6 +36,7 @@ api_router.include_router(recharge_packages_router) api_router.include_router(video_engines_router) api_router.include_router(image_engines_router) api_router.include_router(generation_ai_router) +api_router.include_router(hot_opening_replicate_router) api_router.include_router(test_router) api_router.include_router(user_oauth_router) api_router.include_router(user_oauth_app_router) diff --git a/video-gen-api/app/config.py b/video-gen-api/app/config.py index 6ba4706b..334175a3 100644 --- a/video-gen-api/app/config.py +++ b/video-gen-api/app/config.py @@ -87,7 +87,7 @@ class Settings(BaseSettings): # ChatAPI async generation pipeline settings CELERY_BROKER_URL: str = "" CELERY_RESULT_BACKEND: str = "" - CHATAPI_REQUEST_TIMEOUT_SECONDS: int = 120 + CHATAPI_REQUEST_TIMEOUT_SECONDS: int = 180 CHATAPI_VIDEO_FPS: float = 0.5 CHATAPI_ASYNC_MAX_RETRIES: int = 3 CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS: int = 30 @@ -128,5 +128,11 @@ class Settings(BaseSettings): RESOURCE_SIGN_ARG_EXPIRE: str = "exp" RESOURCE_SIGN_ARG_SIGNATURE: str = "sign" + # 爆款开头复刻默认配置。素材校验由前端完成，后端只接收已有上传接口返回的链接。 + HOT_OPENING_DEFAULT_VIDEO_DURATION: int = 4 + HOT_OPENING_DEFAULT_VIDEO_RATIO: str = "9:16" + HOT_OPENING_DEFAULT_VIDEO_RESOLUTION: str = "480p" + HOT_OPENING_DEFAULT_TARGET_PLATFORM: str = "抖音" + settings = Settings() diff --git a/video-gen-api/app/models/__init__.py b/video-gen-api/app/models/__init__.py index 0e056fc3..82311bb8 100644 --- a/video-gen-api/app/models/__init__.py +++ b/video-gen-api/app/models/__init__.py @@ -21,6 +21,8 @@ from app.models.chat_provider_call_log import ChatProviderCallLog from app.models.generated_resource import GeneratedResource from app.models.user_resource_month_stat import UserResourceMonthStat from app.models.user_resource_total_stat import UserResourceTotalStat +from app.models.module_generation_project import ModuleGenerationProject +from app.models.module_generation_step import ModuleGenerationStep from app.models.user_oauth import UserOAuth from app.models.user_oauth_account import UserOAuthAccount from app.models.user_oauth_app import UserOAuthApp @@ -34,5 +36,6 @@ __all__ = [ "MenuConfig", "RechargePackage", "OperationLog", "ChatGenerationTask", "ChatGenerationTaskEvent", "ChatProviderCallLog", "GeneratedResource", "UserResourceMonthStat", "UserResourceTotalStat", + "ModuleGenerationProject", "ModuleGenerationStep", "UserOAuth", "UserOAuthAccount", "UserOAuthApp", ] diff --git a/video-gen-api/app/models/chat_generation_task.py b/video-gen-api/app/models/chat_generation_task.py index 2bd74af6..7307e1f8 100644 --- a/video-gen-api/app/models/chat_generation_task.py +++ b/video-gen-api/app/models/chat_generation_task.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text +from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text, text from sqlalchemy.orm import Mapped, mapped_column from app.models.base import Base, TimestampMixin, SoftDeleteMixin @@ -18,7 +18,14 @@ class ChatGenerationTask(Base, TimestampMixin, SoftDeleteMixin): __table_args__ = ( # 防止前端按钮连点/网络重试时同一个 idempotency_key 并发创建多条任务。 # nullable unique 兼容不传 idempotency_key 的普通请求。 - Index("uq_chat_generation_tasks_user_mode_idempotency", "user_id", "generation_mode", "idempotency_key", unique=True), + Index( + "uq_chat_generation_tasks_user_mode_idempotency", + "user_id", + "generation_mode", + "idempotency_key", + unique=True, + postgresql_where=text("deleted_at IS NULL AND idempotency_key IS NOT NULL"), + ), ) diff --git a/video-gen-api/app/services/generation_billing_service.py b/video-gen-api/app/services/generation_billing_service.py index 74d1087c..51604d16 100644 --- a/video-gen-api/app/services/generation_billing_service.py +++ b/video-gen-api/app/services/generation_billing_service.py @@ -20,6 +20,7 @@ CHARGE_MEDIA = "media" OWNER_GENERATION_RECORD = "generation_record" OWNER_CHAT_GENERATION_TASK = "chat_generation_task" +OWNER_MODULE_GENERATION_STEP = "module_generation_step" _BIZ_KEY_PATTERN = re.compile( r"^(?P[^:]+):(?P[^:]+):attempt:(?P\d+):(?P[^:]+):(?Pcharge|refund)$" @@ -287,6 +288,45 @@ async def charge_chatapi_prompt_usage( return BillingSummary(record_id=record.id, user_id=record.user_id, items=items) +async def charge_module_prompt_usage( + db: AsyncSession, + *, + user_id: str, + step_id: str, + usage: Mapping[str, Any], + description: str, + attempt_no: int = 1, +) -> BillingSummary: + """爆款开头复刻模块图片/视频 AI 提词扣文本积分。 + + 文本提词属于已经发生的 LLM 消费： + - 调用成功后按 input_tokens + output_tokens 扣费。 + - 不参与后续图片/视频媒体生成失败退款。 + - 通过 module_generation_step:{step_id}:attempt:1:text_prompt:charge 幂等。 + """ + input_tokens = _safe_int(usage.get("input_tokens")) + output_tokens = _safe_int(usage.get("output_tokens")) + text_credits = await calc_text_credits(db, input_tokens, output_tokens) + biz_key = build_credit_biz_key( + owner_type=OWNER_MODULE_GENERATION_STEP, + owner_id=step_id, + attempt_no=attempt_no, + charge_kind=CHARGE_TEXT_PROMPT, + action="charge", + ) + item = await deduct_credits_locked_once( + db, + user_id=user_id, + amount=text_credits, + description=description, + related_id=step_id, + charge_key=CHARGE_TEXT_PROMPT, + biz_key=biz_key, + attempt_no=attempt_no, + ) + return BillingSummary(record_id=step_id, user_id=user_id, items=[item]) + + async def charge_generation_media_by_params( db: AsyncSession, *, diff --git a/video-gen-api/app/services/generation_recovery_service.py b/video-gen-api/app/services/generation_recovery_service.py index cb54a932..33ffd7a3 100644 --- a/video-gen-api/app/services/generation_recovery_service.py +++ b/video-gen-api/app/services/generation_recovery_service.py @@ -17,10 +17,13 @@ from app.services.celery_download_recovery_service import ( remove_download_active, ) from app.services.generation_log_service import log_task_event +from app.services.generation_module_hook_service import notify_chat_generation_task_finished from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once logger = logging.getLogger("video_gen") +ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate"} + def _now() -> datetime: return datetime.now(timezone.utc) @@ -73,7 +76,7 @@ async def recover_one_download_task( if not task: return "skip_missing_task" - if task.generation_mode != "chatapi_async": + if task.generation_mode not in ALLOWED_GENERATION_MODES: await remove_download_active(task.id) return "clean_invalid_mode" if _is_final_task_state(task): @@ -218,7 +221,7 @@ async def recover_download_tasks_once(db: AsyncSession) -> dict[str, Any]: select(ChatGenerationTask) .where( ChatGenerationTask.deleted_at.is_(None), - ChatGenerationTask.generation_mode == "chatapi_async", + ChatGenerationTask.generation_mode.in_(["chatapi_async", "hot_opening_replicate"]), ChatGenerationTask.status == "generating", ChatGenerationTask.remote_result_url.is_not(None), ChatGenerationTask.pipeline_stage.in_( @@ -263,7 +266,7 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]: select(ChatGenerationTask) .where( ChatGenerationTask.deleted_at.is_(None), - ChatGenerationTask.generation_mode == "chatapi_async", + ChatGenerationTask.generation_mode.in_(["chatapi_async", "hot_opening_replicate"]), ChatGenerationTask.status == "generating", ChatGenerationTask.pipeline_stage.in_( [ @@ -290,6 +293,7 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]: error_message="任务超时", pipeline_stage="timeout", ) + await notify_chat_generation_task_finished(db, task) await db.commit() await log_task_event( task, diff --git a/video-gen-api/app/services/generation_refund_service.py b/video-gen-api/app/services/generation_refund_service.py index 1ebb9526..0b13b644 100644 --- a/video-gen-api/app/services/generation_refund_service.py +++ b/video-gen-api/app/services/generation_refund_service.py @@ -180,7 +180,7 @@ async def mark_chat_generation_task_failed_and_refund_once( select(ChatGenerationTask) .where( ChatGenerationTask.id == task_id, - ChatGenerationTask.generation_mode == "chatapi_async", + ChatGenerationTask.generation_mode.in_(["chatapi_async", "hot_opening_replicate"]), ChatGenerationTask.deleted_at.is_(None), ) .with_for_update() diff --git a/video-gen-api/app/tasks/__init__.py b/video-gen-api/app/tasks/__init__.py index 8ff0cbff..1685b2f7 100644 --- a/video-gen-api/app/tasks/__init__.py +++ b/video-gen-api/app/tasks/__init__.py @@ -10,6 +10,7 @@ try: generation_poll_tasks, generation_download_tasks, generation_recovery_tasks, + hot_opening_replicate_tasks ) except Exception: pass diff --git a/video-gen-api/app/tasks/celery_app.py b/video-gen-api/app/tasks/celery_app.py index 4c6c59be..ed4b567d 100644 --- a/video-gen-api/app/tasks/celery_app.py +++ b/video-gen-api/app/tasks/celery_app.py @@ -49,6 +49,8 @@ if broker_url: "generation.chatapi_create_generation_task": {"queue": "gen_chatapi_create"}, "generation.poll_generation_task": {"queue": "gen_provider_poll"}, "generation.download_generation_result_task": {"queue": "gen_result_download"}, + "hot_opening.start_image_prompt_optimize": {"queue": "gen_chatapi_create"}, + "hot_opening.start_video_prompt_optimize": {"queue": "gen_chatapi_create"}, "generation.recover_download_tasks_once": {"queue": "gen_result_download"}, "generation.recover_generation_tasks_once": {"queue": "gen_result_download"}, "app.tasks.cleanup.*": {"queue": "default"}, diff --git a/video-gen-api/app/tasks/generation_create_tasks.py b/video-gen-api/app/tasks/generation_create_tasks.py index ea83e2f7..29a5d44e 100644 --- a/video-gen-api/app/tasks/generation_create_tasks.py +++ b/video-gen-api/app/tasks/generation_create_tasks.py @@ -13,6 +13,8 @@ from app.services.generation_refund_service import mark_chat_generation_task_fai from app.services.generation_provider_service import create_provider_task from app.tasks.celery_app import celery_app +ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate"} + def _get_first_value(obj: Any, *field_names: str) -> Optional[Any]: """ @@ -63,6 +65,14 @@ def _build_optimized_prompt_by_params(task: ChatGenerationTask) -> str: base_prompt = original_prompt.rstrip("，,。；; \n\t") gen_type = (_to_clean_str(getattr(task, "gen_type", None)) or "").lower() + generation_mode = _to_clean_str(getattr(task, "generation_mode", None)) or "" + + # 爆款开头复刻第5步的视频生成，original_prompt 已经是视频提词 JSON schema。 + # 不能再追加“时长/比例/分辨率”中文参数，否则会污染 schema。 + if generation_mode == "hot_opening_replicate" and gen_type == "video": + stripped = base_prompt.strip() + if stripped.startswith("{") or stripped.startswith("["): + return base_prompt duration = _get_first_value(task, "duration") aspect_ratio = _get_first_value(task, "aspect_ratio") @@ -113,7 +123,7 @@ async def _run(task_id: str): ).with_for_update().limit(1)) task = result.scalar_one_or_none() - if not task or task.generation_mode != "chatapi_async": + if not task or task.generation_mode not in ALLOWED_GENERATION_MODES: return if task.status != "generating": @@ -128,6 +138,9 @@ async def _run(task_id: str): ) await db.commit() await log_task_event(task, event_type="TASK_TIMEOUT", to_status="failed", to_stage="timeout") + from app.services.generation_module_hook_service import notify_chat_generation_task_finished + await notify_chat_generation_task_finished(db, task) + await db.commit() return if task.pipeline_stage not in ("queued", "preparing", "creating_provider_task"): @@ -253,6 +266,9 @@ async def _run(task_id: str): ) await db.commit() await log_task_event(task, event_type="TASK_FAILED", message=task.error_message) + from app.services.generation_module_hook_service import notify_chat_generation_task_finished + await notify_chat_generation_task_finished(db, task) + await db.commit() if celery_app: diff --git a/video-gen-api/app/tasks/generation_download_tasks.py b/video-gen-api/app/tasks/generation_download_tasks.py index 61543671..3a27f659 100644 --- a/video-gen-api/app/tasks/generation_download_tasks.py +++ b/video-gen-api/app/tasks/generation_download_tasks.py @@ -21,6 +21,8 @@ from app.services.generation_refund_service import mark_chat_generation_task_fai from app.services.resource_accounting_service import record_chat_task_generated_resource from app.tasks.celery_app import celery_app +ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate"} + DOWNLOAD_QUEUE = "gen_result_download" DOWNLOAD_STAGE_QUEUED = "download_queued" DOWNLOAD_STAGE_DOWNLOADING = "downloading" @@ -108,7 +110,7 @@ async def enqueue_download_task( countdown: int | None = None, ) -> str | None: """统一投递图片/视频下载任务，并同步 DB + Redis active 注册表。""" - if not task or task.generation_mode != "chatapi_async": + if not task or task.generation_mode not in ALLOWED_GENERATION_MODES: return None if task.status != "generating": return None @@ -159,7 +161,7 @@ async def _reload_task(db: AsyncSession, task_id: str) -> ChatGenerationTask | N async def _claim_download_lease(db: AsyncSession, task: ChatGenerationTask) -> bool: now = _now() - if not task or task.generation_mode != "chatapi_async": + if not task or task.generation_mode not in ALLOWED_GENERATION_MODES: return False if task.status != "generating": return False @@ -310,6 +312,10 @@ async def _run(task_id: str): await db.commit() await remove_download_active(task.id) + from app.services.generation_module_hook_service import notify_chat_generation_task_finished + await notify_chat_generation_task_finished(db, task) + await db.commit() + await log_task_event( task, event_type="DOWNLOAD_SUCCESS", @@ -348,6 +354,10 @@ async def _run(task_id: str): await remove_download_active(task.id) + from app.services.generation_module_hook_service import notify_chat_generation_task_finished + await notify_chat_generation_task_finished(db, task) + await db.commit() + await log_task_event( task, event_type="DOWNLOAD_FAILED", diff --git a/video-gen-api/app/tasks/generation_poll_tasks.py b/video-gen-api/app/tasks/generation_poll_tasks.py index a6797593..08835b8b 100644 --- a/video-gen-api/app/tasks/generation_poll_tasks.py +++ b/video-gen-api/app/tasks/generation_poll_tasks.py @@ -13,6 +13,8 @@ from app.services.generation_refund_service import mark_chat_generation_task_fai from app.services.generation_provider_service import poll_provider_task from app.tasks.celery_app import celery_app +ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate"} + def _is_success(status: str) -> bool: return status in ("succeeded", "success", "completed", "done") @@ -29,6 +31,12 @@ def _engine_snapshot(task: ChatGenerationTask) -> dict: return {} +async def _notify_finished(db, task: ChatGenerationTask) -> None: + from app.services.generation_module_hook_service import notify_chat_generation_task_finished + + await notify_chat_generation_task_finished(db, task) + + async def _reload_task(db, task_id: str) -> ChatGenerationTask | None: """ rollback 后重新查询任务对象。 @@ -54,7 +62,7 @@ async def _run(task_id: str): ChatGenerationTask.deleted_at.is_(None), ).with_for_update().limit(1)) task = result.scalar_one_or_none() - if not task or task.generation_mode != "chatapi_async": + if not task or task.generation_mode not in ALLOWED_GENERATION_MODES: return # 只处理正在生成，且处于远程等待/轮询中的任务。 @@ -68,6 +76,7 @@ async def _run(task_id: str): error_message="任务轮询超时", pipeline_stage="timeout", ) + await _notify_finished(db, task) await db.commit() await log_task_event(task, event_type="TASK_TIMEOUT", to_status="failed", to_stage="timeout") return @@ -79,6 +88,7 @@ async def _run(task_id: str): error_message="缺少外部任务ID", pipeline_stage="failed", ) + await _notify_finished(db, task) await db.commit() await log_task_event(task, event_type="POLL_FAILED", message=task.error_message) return @@ -130,6 +140,7 @@ async def _run(task_id: str): error_message="供应商任务成功但未返回结果URL", pipeline_stage="failed", ) + await _notify_finished(db, task) await db.commit() await log_task_event(task, event_type="POLL_FAILED", message=task.error_message) return @@ -153,6 +164,7 @@ async def _run(task_id: str): error_message=poll_result.get("error") or f"供应商任务失败: {status}", pipeline_stage="failed", ) + await _notify_finished(db, task) await db.commit() await log_task_event(task, event_type="POLL_FAILED", message=task.error_message, detail=poll_result) return @@ -194,6 +206,7 @@ async def _run(task_id: str): error_message=error_message, pipeline_stage="failed", ) + await _notify_finished(db, task) await db.commit() await log_task_event(task, event_type="POLL_FAILED", message=task.error_message) else: