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

会员积分改版V8 追加订单自动过期celery定时任务

Browse Source
This commit is contained in:
GinHa
2026-08-11 14:11:59 +08:00
parent a49ed802f4
commit 81ef619ad3
4 changed files with 68 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/enums/celery_queue.py
+1
View File
@@ -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"
video-gen-api/app/main.py
-24
View File
@@ -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()
video-gen-api/app/tasks/celery_app.py
+11
View File
@@ -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},
video-gen-api/app/tasks/payment_tasks.py
+56
View File
@@ -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()