mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

爆款开头API开发完成文件追加

Browse Source
This commit is contained in:
GinHa
2026-06-10 15:23:03 +08:00
parent b4a9ca9e9b
commit fa2e298efa
12 changed files with 115 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/tasks/__init__.py
+1
View File
@@ -10,6 +10,7 @@ try:
generation_poll_tasks,
generation_download_tasks,
generation_recovery_tasks,
hot_opening_replicate_tasks
)
except Exception:
pass
video-gen-api/app/tasks/celery_app.py
+2
View File
@@ -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"},
video-gen-api/app/tasks/generation_create_tasks.py
+17 -1
View File
@@ -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:
video-gen-api/app/tasks/generation_download_tasks.py
+12 -2
View File
@@ -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",
video-gen-api/app/tasks/generation_poll_tasks.py
+14 -1
View File
@@ -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: