diff --git a/video-gen-api/app/enums/celery_queue.py b/video-gen-api/app/enums/celery_queue.py index 04b7493d..f3c78967 100644 --- a/video-gen-api/app/enums/celery_queue.py +++ b/video-gen-api/app/enums/celery_queue.py @@ -47,6 +47,7 @@ class CeleryTaskName(str, Enum): PRIVATE_PORTRAIT_DELETE_PROJECT = "private_portrait.delete_project_remote" PRIVATE_PORTRAIT_RECOVER_REMOTE_DELETES = "private_portrait.recover_remote_deletes" CREDIT_MAINTENANCE = "credit.maintenance_once" + PAYMENT_EXPIRE_PENDING = "payment.expire_pending_orders" VP_V3_POLL_ASSET = "vp_v3.asset.poll_status" VP_V3_SYNC_DUE_ASSETS = "vp_v3.sync_due_assets" VP_V3_DELETE_ASSET = "vp_v3.asset.delete_remote" diff --git a/video-gen-api/app/main.py b/video-gen-api/app/main.py index 06a8f2ee..02e701ba 100644 --- a/video-gen-api/app/main.py +++ b/video-gen-api/app/main.py @@ -32,29 +32,6 @@ async def lifespan(app: FastAPI): await init_redis() # await _seed_data() - # Background task: auto-expire pending payment orders and sync status - async def _order_expiry_loop(): - from app.services.payment import expire_all_pending_orders, sync_pending_orders - from logging import getLogger - bg_logger = getLogger("payment") - while True: - try: - async with async_session() as db: - # 同步待支付订单状态（检查支付宝实际支付状态 - sync_count = await sync_pending_orders(db) - if sync_count > 0: - bg_logger.info(f"Synced {sync_count} pending payment order(s)") - - # 自动过期订单 - n = await expire_all_pending_orders(db) - if n > 0: - bg_logger.info(f"Auto-expired {n} pending payment order(s)") - except Exception as e: - bg_logger.error(f"Order expiry loop error: {e}") - await asyncio.sleep(60) # check every minute - - expiry_task = asyncio.create_task(_order_expiry_loop()) - # 启动token刷新定时任务（每5分钟检查一次，小于800秒有效期的token进行刷新） from app.tasks.token_refresh_task import token_refresh_scheduler token_refresh_task = asyncio.create_task(token_refresh_scheduler()) @@ -108,7 +85,6 @@ async def lifespan(app: FastAPI): await consumption_queue_task consumption_schedule_task.cancel() pre_test_poll_task.cancel() - expiry_task.cancel() token_refresh_task.cancel() await close_database() await close_redis() diff --git a/video-gen-api/app/tasks/celery_app.py b/video-gen-api/app/tasks/celery_app.py index cfc7e51c..f1d8f246 100644 --- a/video-gen-api/app/tasks/celery_app.py +++ b/video-gen-api/app/tasks/celery_app.py @@ -36,6 +36,7 @@ CELERY_TASK_IMPORTS = ( "app.tasks.vp_v3_asset_tasks", "app.tasks.celery_runtime_tasks", "app.tasks.credit_tasks", + "app.tasks.payment_tasks", "app.tasks.api_generation_tasks", "app.tasks.api_recovery_tasks", "app.tasks.api_upscale_tasks", @@ -138,6 +139,14 @@ def _beat_schedule() -> dict: "schedule": 60, "options": {"queue": CeleryQueue.GEN_CREDIT_MAINTENANCE.value}, } + schedule["payment-expire-pending-orders-every-minute"] = { + "task": CeleryTaskName.PAYMENT_EXPIRE_PENDING.value, + "schedule": 60, + "options": { + "queue": RECOVERY_QUEUE, + "priority": settings.DOWNLOAD_TASK_PRIORITY_RECOVER, + }, + } schedule["vp-v3-sync-due-assets-every-minute"] = { "task": CeleryTaskName.VP_V3_SYNC_DUE_ASSETS.value, "schedule": 60, @@ -210,6 +219,7 @@ if broker_url: CeleryTaskName.SHOT_ANALYSIS_RECOVERY.value: {"ignore_result": True}, CeleryTaskName.SHOT_SPLIT_RECOVERY.value: {"ignore_result": True}, CeleryTaskName.CREDIT_MAINTENANCE.value: {"ignore_result": True}, + CeleryTaskName.PAYMENT_EXPIRE_PENDING.value: {"ignore_result": True}, }, worker_prefetch_multiplier=1, worker_cancel_long_running_tasks_on_connection_loss=True, @@ -270,6 +280,7 @@ if broker_url: CeleryTaskName.PRIVATE_PORTRAIT_DELETE_PROJECT.value: {"queue": CeleryQueue.GEN_PRIVATE_PORTRAIT.value}, CeleryTaskName.PRIVATE_PORTRAIT_RECOVER_REMOTE_DELETES.value: {"queue": CeleryQueue.GEN_PRIVATE_PORTRAIT.value}, CeleryTaskName.CREDIT_MAINTENANCE.value: {"queue": CeleryQueue.GEN_CREDIT_MAINTENANCE.value}, + CeleryTaskName.PAYMENT_EXPIRE_PENDING.value: {"queue": RECOVERY_QUEUE}, CeleryTaskName.VP_V3_POLL_ASSET.value: {"queue": CeleryQueue.GEN_PRIVATE_PORTRAIT.value}, CeleryTaskName.VP_V3_SYNC_DUE_ASSETS.value: {"queue": CeleryQueue.GEN_PRIVATE_PORTRAIT.value}, CeleryTaskName.VP_V3_DELETE_ASSET.value: {"queue": CeleryQueue.GEN_PRIVATE_PORTRAIT.value}, diff --git a/video-gen-api/app/tasks/payment_tasks.py b/video-gen-api/app/tasks/payment_tasks.py new file mode 100644 index 00000000..44f9c854 --- /dev/null +++ b/video-gen-api/app/tasks/payment_tasks.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import logging +from typing import Any + +from app.models.base import async_session +from app.services.payment import expire_all_pending_orders +from app.tasks.async_runner import run_async +from app.tasks.celery_app import celery_app + +logger = logging.getLogger("payment") + + +async def _expire_pending_payment_orders_once() -> dict[str, Any]: + """服务端定时清理已超过支付时限的 pending 订单。 + + 具体的支付渠道关闭确认、订单取消以及升级预留释放仍完全复用 + ``expire_all_pending_orders``，这里不复制支付状态机，避免两套逻辑分叉。 + """ + async with async_session() as db: + try: + expired_count = await expire_all_pending_orders(db) + if expired_count > 0: + logger.info( + "PAYMENT_EXPIRE_TASK_COMPLETED expired_count=%s", + expired_count, + ) + return {"expired_count": int(expired_count)} + except Exception: + await db.rollback() + logger.exception("PAYMENT_EXPIRE_TASK_FAILED") + raise + + +if celery_app: + + @celery_app.task( + name="payment.expire_pending_orders", + bind=True, + soft_time_limit=540, + time_limit=600, + ignore_result=True, + ) + def expire_pending_payment_orders(self) -> dict[str, Any]: + return run_async(_expire_pending_payment_orders_once()) + +else: + + class _DisabledTask: + def delay(self, *args: Any, **kwargs: Any) -> None: + raise RuntimeError("Celery is disabled") + + def apply_async(self, *args: Any, **kwargs: Any) -> None: + raise RuntimeError("Celery is disabled") + + expire_pending_payment_orders = _DisabledTask()