fix(app): 修复定时任务同步待支付订单的上下文问题
- 将 sync_pending_orders 增加 app 参数以使用 Flask 应用上下文 - 在函数内部使用 app.app_context() 包裹数据库操作,避免上下文错误 - 保持原有逻辑查询并更新30分钟内的待支付订单状态 - 增加日志记录和异常捕获,确保任务稳定运行 - 调度器调用时传入 app 作为参数以支持上下文执行
This commit is contained in:
parent
1196809c6a
commit
93d5c503b2
96
app.py
96
app.py
@ -15,64 +15,65 @@ import logging
|
|||||||
import models
|
import models
|
||||||
|
|
||||||
# 定时任务函数
|
# 定时任务函数
|
||||||
def sync_pending_orders():
|
def sync_pending_orders(app):
|
||||||
"""定时任务: 检查并同步30分钟内的待支付订单"""
|
"""定时任务: 检查并同步30分钟内的待支付订单"""
|
||||||
from models import Order, User, get_bj_now
|
with app.app_context():
|
||||||
from services.alipay_service import AlipayService
|
from models import Order, User, get_bj_now
|
||||||
from services.logger import system_logger
|
from services.alipay_service import AlipayService
|
||||||
from datetime import timedelta
|
from services.logger import system_logger
|
||||||
|
from datetime import timedelta
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# 查询最还30分钟内的待支付订单
|
# 查询最还30分钟内的待支付订单
|
||||||
thirty_min_ago = get_bj_now() - timedelta(minutes=30)
|
thirty_min_ago = get_bj_now() - timedelta(minutes=30)
|
||||||
pending_orders = Order.query.filter(
|
pending_orders = Order.query.filter(
|
||||||
Order.status == 'PENDING',
|
Order.status == 'PENDING',
|
||||||
Order.created_at >= thirty_min_ago
|
Order.created_at >= thirty_min_ago
|
||||||
).all()
|
).all()
|
||||||
|
|
||||||
if not pending_orders:
|
if not pending_orders:
|
||||||
return
|
return
|
||||||
|
|
||||||
alipay_service = AlipayService()
|
alipay_service = AlipayService()
|
||||||
updated_count = 0
|
updated_count = 0
|
||||||
|
|
||||||
for order in pending_orders:
|
for order in pending_orders:
|
||||||
try:
|
try:
|
||||||
# 查询订单状态
|
# 查询订单状态
|
||||||
alipay_result = alipay_service.query_order_status(order.out_trade_no)
|
alipay_result = alipay_service.query_order_status(order.out_trade_no)
|
||||||
|
|
||||||
if alipay_result and alipay_result.get('trade_status') in ['TRADE_SUCCESS', 'TRADE_FINISHED']:
|
if alipay_result and alipay_result.get('trade_status') in ['TRADE_SUCCESS', 'TRADE_FINISHED']:
|
||||||
# 使用行级锁重新查询,防止并发问题
|
# 使用行级锁重新查询,防止并发问题
|
||||||
order_locked = Order.query.filter_by(out_trade_no=order.out_trade_no).with_for_update().first()
|
order_locked = Order.query.filter_by(out_trade_no=order.out_trade_no).with_for_update().first()
|
||||||
|
|
||||||
# 二次校验状态,防止异步回调已经处理
|
# 二次校验状态,防止异步回调已经处理
|
||||||
if order_locked and order_locked.status == 'PENDING':
|
if order_locked and order_locked.status == 'PENDING':
|
||||||
# 更新订单
|
# 更新订单
|
||||||
order_locked.status = 'PAID'
|
order_locked.status = 'PAID'
|
||||||
order_locked.trade_no = alipay_result.get('trade_no')
|
order_locked.trade_no = alipay_result.get('trade_no')
|
||||||
if not order_locked.paid_at:
|
if not order_locked.paid_at:
|
||||||
order_locked.paid_at = get_bj_now()
|
order_locked.paid_at = get_bj_now()
|
||||||
|
|
||||||
# 增加用户积分
|
# 增加用户积分
|
||||||
user = db.session.get(User, order_locked.user_id)
|
user = db.session.get(User, order_locked.user_id)
|
||||||
if user:
|
if user:
|
||||||
user.points += order_locked.points
|
user.points += order_locked.points
|
||||||
system_logger.info(f"定时任务-订单支付成功", order_id=order_locked.out_trade_no, points=order_locked.points, user_id=user.id)
|
system_logger.info(f"定时任务-订单支付成功", order_id=order_locked.out_trade_no, points=order_locked.points, user_id=user.id)
|
||||||
updated_count += 1
|
updated_count += 1
|
||||||
|
|
||||||
db.session.commit()
|
db.session.commit()
|
||||||
elif order_locked and order_locked.status == 'PAID':
|
elif order_locked and order_locked.status == 'PAID':
|
||||||
# 订单已经被处理,跳过
|
# 订单已经被处理,跳过
|
||||||
logging.info(f"定时任务-订单 {order.out_trade_no} 已被处理,跳过")
|
logging.info(f"定时任务-订单 {order.out_trade_no} 已被处理,跳过")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
db.session.rollback()
|
db.session.rollback()
|
||||||
logging.error(f"定时任务处理订单 {order.out_trade_no} 失败: {str(e)}")
|
logging.error(f"定时任务处理订单 {order.out_trade_no} 失败: {str(e)}")
|
||||||
|
|
||||||
if updated_count > 0:
|
if updated_count > 0:
|
||||||
logging.info(f"定时任务完成,帮助更新了{updated_count}个订单")
|
logging.info(f"定时任务完成,帮助更新了{updated_count}个订单")
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(f"定时任务异常: {str(e)}", exc_info=True)
|
logging.error(f"定时任务异常: {str(e)}", exc_info=True)
|
||||||
|
|
||||||
# 导入模律(必需在 db.create_all() 之前导入)
|
# 导入模律(必需在 db.create_all() 之前导入)
|
||||||
|
|
||||||
@ -181,6 +182,7 @@ def create_app():
|
|||||||
# 每分钟检查一次待支付订单
|
# 每分钟检查一次待支付订单
|
||||||
scheduler.add_job(
|
scheduler.add_job(
|
||||||
func=sync_pending_orders,
|
func=sync_pending_orders,
|
||||||
|
args=[app],
|
||||||
trigger=IntervalTrigger(minutes=1),
|
trigger=IntervalTrigger(minutes=1),
|
||||||
id='sync_pending_orders',
|
id='sync_pending_orders',
|
||||||
name='同步待支付订单',
|
name='同步待支付订单',
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user