```

feat(app): 添加定时任务检查待支付订单状态

- 集成 APScheduler 背景任务调度器，每分钟检查一次30分钟内
  的待支付订单
- 实现 sync_pending_orders 函数，主动查询支付宝订单状态，
  防止异步回调丢失导致订单状态不更新
- 添加 Redis 分布式锁机制，防止并发处理同一订单
- 在应用启动时自动启动定时任务调度器

fix(alipay): 修复支付宝签名验证和订单查询功能

- 修复签名验证时遗漏 sign_type 字段的问题
- 新增 query_order_status 方法，支持主动查询支付宝订单状态
- 更新支付宝回调和通知URL地址

feat(task): 优化异步任务状态跟踪和图片压缩功能

- 集成 Redis 存储任务状态，实时跟踪生图和视频任务进度
- 添加图片自动压缩功能，超过2K分辨率的图片自动压缩
- 优化任务状态更新，提供更准确的进度反馈

chore(config): 更新支付宝回调地址配置
```

app.py

@@ -5,12 +5,86 @@ from blueprints.auth import auth_bp
 from blueprints.api import api_bp
 from blueprints.admin import admin_bp
 from blueprints.payment import payment_bp
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.interval import IntervalTrigger
 import threading
 import time
+import logging
 
-# 导入模型（必须在 db.create_all() 之前导入）
+# 导入模律（必需在 db.create_all() 之前导入）
 import models
 
+# 定时任务函数
+def sync_pending_orders(app):
+    """定时任务: 检查并同步30分钟内的待支付订单"""
+    with app.app_context():
+        from models import Order, User, get_bj_now
+        from services.alipay_service import AlipayService
+        from services.logger import system_logger
+        from datetime import timedelta
+        
+        try:
+            # 查询最还30分钟内的待支付订单
+            thirty_min_ago = get_bj_now() - timedelta(minutes=30)
+            pending_orders = Order.query.filter(
+                Order.status == 'PENDING',
+                Order.created_at >= thirty_min_ago
+            ).all()
+            
+            if not pending_orders:
+                return
+            
+            alipay_service = AlipayService()
+            updated_count = 0
+            
+            for order in pending_orders:
+                try:
+                    # 使用防止并发的锁
+                    lock_key = f"lock:order:{order.out_trade_no}"
+                    # 尝试获取锁，等待3秒。如果获取不到，说明正在处理，本次跳过
+                    with redis_client.lock(lock_key, timeout=10, blocking_timeout=3):
+                        # 查询订单状态
+                        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']:
+                            # 使用行级锁重新查询，防止并发问题 (数据库层面的双重保障)
+                            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':
+                                # 更新订单
+                                order_locked.status = 'PAID'
+                                order_locked.trade_no = alipay_result.get('trade_no')
+                                if not order_locked.paid_at:
+                                    order_locked.paid_at = get_bj_now()
+                                
+                                # 增加用户积分
+                                user = db.session.get(User, order_locked.user_id)
+                                if user:
+                                    user.points += order_locked.points
+                                    system_logger.info(f"定时任务-订单支付成功", order_id=order_locked.out_trade_no, points=order_locked.points, user_id=user.id)
+                                    updated_count += 1
+                                
+                                db.session.commit()
+                            elif order_locked and order_locked.status == 'PAID':
+                                # 订单已经被处理，跳过
+                                logging.info(f"定时任务-订单 {order.out_trade_no} 已被处理，跳过")
+                except Exception as e:
+                    # Redis lock error or other errors
+                    if "LockError" in str(e) or "BlockingIOError" in str(e):
+                        logging.info(f"定时任务-订单 {order.out_trade_no} 锁定失败或正在处理中")
+                    else:
+                        db.session.rollback()
+                        logging.error(f"定时任务处理订单 {order.out_trade_no} 失败: {str(e)}")
+            
+            if updated_count > 0:
+                logging.info(f"定时任务完成，帮助更新了{updated_count}个订单")
+        
+        except Exception as e:
+            logging.error(f"定时任务异常: {str(e)}", exc_info=True)
+
+# 导入模律（必需在 db.create_all() 之前导入）
+
 def create_app():
     app = Flask(__name__)
     app.config.from_object(Config)
@@ -110,6 +184,23 @@ def create_app():
         db.create_all()
         print("✅ 数据库表已就绪")
         
+        # 创建并启动定时任务调度器
+        try:
+            scheduler = BackgroundScheduler(daemon=True)
+            # 每分钟检查一次待支付订单
+            scheduler.add_job(
+                func=sync_pending_orders,
+                args=[app],
+                trigger=IntervalTrigger(minutes=1),
+                id='sync_pending_orders',
+                name='同步待支付订单',
+                replace_existing=True
+            )
+            scheduler.start()
+            print("🚀 定时任务引擎已启动，将每分钟检查一次待支付订单")
+        except Exception as e:
+            print(f"⚠️  定时任务启动失败: {str(e)}")
+        
     return app
 
 app = create_app()

blueprints/payment.py

@@ -1,5 +1,5 @@
 from flask import Blueprint, request, redirect, url_for, session, jsonify, render_template
-from extensions import db
+from extensions import db, redis_client
 from models import Order, User, to_bj_time, get_bj_now
 from services.alipay_service import AlipayService
 from services.logger import system_logger
@@ -76,6 +76,42 @@ def payment_return():
         out_trade_no = data.get('out_trade_no')
         
         if success:
+            # 同步回调也进行订单处理，防止异步回调延迟或失败
+            out_trade_no = data.get('out_trade_no')
+            trade_no = data.get('trade_no')
+            
+            try:
+                # 使用分布式锁防止并发
+                lock_key = f"lock:order:{out_trade_no}"
+                with redis_client.lock(lock_key, timeout=10, blocking_timeout=3):
+                    # 查询订单 (加锁防止并发导致双重发放)
+                    order = Order.query.filter_by(out_trade_no=out_trade_no).with_for_update().first()
+                    
+                    # 如果订单存在且状态为PENDING，则更新为PAID
+                    if order and order.status == 'PENDING':
+                        order.status = 'PAID'
+                        order.trade_no = trade_no
+                        order.paid_at = get_bj_now()
+                        
+                        # 增加用户积分
+                        user = db.session.get(User, order.user_id)
+                        if user:
+                            user.points += order.points
+                            system_logger.info(f"同步回调-订单支付成功", order_id=out_trade_no, points=order.points, user_id=user.id)
+                        
+                        db.session.commit()
+                    elif order:
+                        # 订单已经是完成状态，不做处理
+                        pass
+                    else:
+                        system_logger.warning(f"同步回调-未找到订单", order_id=out_trade_no)
+                    
+            except Exception as e:
+                db.session.rollback()
+                # 忽略锁错误，说明可能已经在处理
+                if "LockError" not in str(e) and "BlockingIOError" not in str(e):
+                    system_logger.error(f"同步回调-订单状态更新失败: {str(e)}")
+            
             return redirect(url_for('auth.buy_page', success='true', out_trade_no=out_trade_no))
         else:
             system_logger.warning(f"支付同步回调验证失败", order_id=out_trade_no)
@@ -133,11 +169,121 @@ def api_payment_history():
             "paid_at": o.paid_at_bj.strftime('%Y-%m-%d %H:%M:%S') if o.paid_at else None
         } for o in orders]
     })
+@payment_bp.route('/api/sync_order', methods=['POST'])
+def api_sync_order():
+    """主动查询订单状态并同步 - 用于用户手动关闭支付页面后检查是否支付成功"""
+    if 'user_id' not in session:
+        return jsonify({'code': 401, 'msg': '请先登录'}), 401
+    
+    out_trade_no = request.form.get('out_trade_no')
+    if not out_trade_no:
+        return jsonify({'code': 400, 'msg': '订单号不能为空'}), 400
+    
+    try:
+        # 查询订单
+        order = Order.query.filter_by(out_trade_no=out_trade_no).first()
+        if not order:
+            return jsonify({'code': 404, 'msg': '订单不存在'}), 404
+        
+        # 只有当前用户才能查询自己的订单
+        if order.user_id != session['user_id']:
+            return jsonify({'code': 403, 'msg': '无权限访问此订单'}), 403
+        
+        # 如果订单已经是PAID或FAILED状态，直接返回
+        if order.status in ['PAID', 'FAILED']:
+            return jsonify({
+                'code': 200,
+                'msg': '订单状态已确定',
+                'status': order.status,
+                'paid_at': order.paid_at_bj.strftime('%Y-%m-%d %H:%M:%S') if order.paid_at else None
+            })
+        
+        # 向支付宝查询订单状态
+        alipay_service = AlipayService()
+        alipay_result = alipay_service.query_order_status(out_trade_no)
+        
+        if not alipay_result:
+            return jsonify({'code': 500, 'msg': '查询支付宝订单失败，请稍后重试'}), 500
+        
+        trade_status = alipay_result.get('trade_status')
+        
+        # 如果支付宝显示已支付，更新本地订单状态
+        if trade_status in ['TRADE_SUCCESS', 'TRADE_FINISHED']:
+            try:
+                lock_key = f"lock:order:{out_trade_no}"
+                with redis_client.lock(lock_key, timeout=10, blocking_timeout=3):
+                    # 使用行级锁重新查询，防止并发问题
+                    order_locked = Order.query.filter_by(out_trade_no=out_trade_no).with_for_update().first()
+                    
+                    # 二次校验状态，防止异步回调/定时任务已经处理
+                    if order_locked and order_locked.status == 'PENDING':
+                        order_locked.status = 'PAID'
+                        order_locked.trade_no = alipay_result.get('trade_no')
+                        if not order_locked.paid_at:
+                            order_locked.paid_at = get_bj_now()
+                        
+                        # 增加用户积分
+                        user = db.session.get(User, order_locked.user_id)
+                        if user:
+                            user.points += order_locked.points
+                            system_logger.info(f"主动查询-订单支付成功", order_id=out_trade_no, points=order_locked.points, user_id=user.id)
+                        
+                        db.session.commit()
+                        
+                        return jsonify({
+                            'code': 200,
+                            'msg': '订单已支付，积分已增加',
+                            'status': 'PAID',
+                            'points': order_locked.points,
+                            'paid_at': order_locked.paid_at_bj.strftime('%Y-%m-%d %H:%M:%S')
+                        })
+                    elif order_locked and order_locked.status == 'PAID':
+                        # 订单已经被处理，直接返回
+                        return jsonify({
+                            'code': 200,
+                            'msg': '订单已支付',
+                            'status': 'PAID',
+                            'points': order_locked.points,
+                            'paid_at': order_locked.paid_at_bj.strftime('%Y-%m-%d %H:%M:%S') if order_locked.paid_at else None
+                        })
+            except Exception as e:
+                if "LockError" in str(e) or "BlockingIOError" in str(e):
+                    # 如果获取锁失败，可能是因为正在处理中，返回成功状态（前端会重试或刷新）
+                    return jsonify({'code': 200, 'msg': '处理中', 'status': 'PAID'})
+                else:
+                    db.session.rollback()
+                    raise e
+        
+        # 支付宝显示未支付
+        elif trade_status in ['TRADE_CLOSED', 'WAIT_BUYER_PAY']:
+            return jsonify({
+                'code': 200,
+                'msg': '订单未支付',
+                'status': 'PENDING',
+                'trade_status': trade_status
+            })
+        
+        else:
+            return jsonify({
+                'code': 200,
+                'msg': f'订单状态: {trade_status}',
+                'status': order.status,
+                'trade_status': trade_status
+            })
+        
+    except Exception as e:
+        system_logger.error(f"主动查询订单异常: {str(e)}")
+        db.session.rollback()
+        return jsonify({'code': 500, 'msg': f'查询失败: {str(e)}'}), 500
+
 @payment_bp.route('/notify', methods=['POST'])
 def payment_notify():
     """支付宝异步通知"""
     try:
         data = request.form.to_dict()
+        # 记录异步通知到系统日志，而不是本地文件
+        system_logger.info(f"支付宝异步通知接收", extra_data=str(data))
+        
         signature = data.get("sign")
         
         if not signature:
@@ -150,6 +296,36 @@ def payment_notify():
             out_trade_no = data.get('out_trade_no')
             trade_no = data.get('trade_no')
             
+            # 加锁查询，确保并发安全
+            try:
+                lock_key = f"lock:order:{out_trade_no}"
+                with redis_client.lock(lock_key, timeout=10, blocking_timeout=3):
+                    order = Order.query.filter_by(out_trade_no=out_trade_no).with_for_update().first()
+                    if order and order.status == 'PENDING':
+                        order.status = 'PAID'
+                        order.trade_no = trade_no
+                        order.paid_at = get_bj_now()
+                        
+                        user = db.session.get(User, order.user_id)
+                        if user:
+                            user.points += order.points
+                            system_logger.info(f"订单支付成功", order_id=out_trade_no, points=order.points, user_id=user.id)
+                        
+                        db.session.commit()
+                        return "success"
+                    elif order:
+                        return "success"
+                    else:
+                        return "fail"
+            except Exception as e:
+                # 如果锁获取失败，暂时返回fail让支付宝重试，或者返回success如果确信正在处理？
+                # 返回fail让支付宝重试比较稳妥，因为可能正在处理但还没提交
+                system_logger.warning(f"处理异步通知锁定失败: {str(e)}")
+                if "LockError" in str(e) or "BlockingIOError" in str(e):
+                    # 正在处理中，告诉支付宝由于并发我们正在处理，稍后重试（或视为成功？）
+                    # 如果返回fail，支付宝会重试。
+                    return "fail"
+                raise e
             order = Order.query.filter_by(out_trade_no=out_trade_no).first()
             if order and order.status == 'PENDING':
                 order.status = 'PAID'
@@ -212,3 +388,31 @@ def payment_notify():
         db.session.rollback()
         return "fail"
 
+
+@payment_bp.route('/api/query/<out_trade_no>', methods=['GET'])
+def api_query_order(out_trade_no):
+    """简单查询接口 - 获取订单当前状态而不自动更新"""
+    if 'user_id' not in session:
+        return jsonify({'code': 401, 'msg': '请先登录'}), 401
+    
+    try:
+        order = Order.query.filter_by(out_trade_no=out_trade_no).first()
+        if not order:
+            return jsonify({'code': 404, 'msg': '订单不存在'}), 404
+        
+        if order.user_id != session['user_id']:
+            return jsonify({'code': 403, 'msg': '无权限访问'}), 403
+        
+        return jsonify({
+            'code': 200,
+            'out_trade_no': order.out_trade_no,
+            'status': order.status,
+            'amount': float(order.amount),
+            'points': order.points,
+            'trade_no': order.trade_no,
+            'created_at': order.created_at_bj.strftime('%Y-%m-%d %H:%M:%S'),
+            'paid_at': order.paid_at_bj.strftime('%Y-%m-%d %H:%M:%S') if order.paid_at else None
+        })
+        
+    except Exception as e:
+        return jsonify({'code': 500, 'msg': f'查询失败: {str(e)}'}), 500

config.py

@@ -59,8 +59,8 @@ MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC+BMpGTJMzDoOnjyGh69rDLbV/8rlB
     ALIPAY_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
 MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAlDx4KdOtOQE+tBq6jHKKFenRaRe2gbBnleBk++5gki9IQuxVyZUGTJixstf2gELFHWrGanpnwmGggXsqG+Rm5ZLJOlmFM1k0XeAIDvi6tP/rM+ZDFSu1bMBYtT5vzgVZC7mzIvOp9gsT/puqd3aNZmlviLD0R6OYN0zvFX+5qADZV7A9ziA+nXPFSHreBh7yY/q9ophVZNeHGPoYkDVI5++RrF1cALKOdit0giN5vxpe3ch9z3E6+FZg3LiP+1RW3tMiDQfp/SlVs6bNhLUtmlI5r7+mtFCKDUCEpnQ3S9e0II6rzyVXRyKCFs7qi5YzyhhmO3tJJoe9ilEFyNzfRQIDAQAB
 -----END PUBLIC KEY-----"""  # 支付宝公钥
-    ALIPAY_RETURN_URL = "http://331002.xyz:2010/payment/return"  # 支付成功跳转地址
-    ALIPAY_NOTIFY_URL = "http://331002.xyz:2010/payment/notify"  # 支付异步通知地址
+    ALIPAY_RETURN_URL = "https://860576.xyz/payment/return"  # 支付成功跳转地址
+    ALIPAY_NOTIFY_URL = "https://860576.xyz/payment/notify"  # 支付异步通知地址
     ALIPAY_DEBUG = False  # 是否使用沙箱环境
 
     # 开发模式配置

services/alipay_service.py

@@ -52,6 +52,7 @@ class AlipayService:
             # python-alipay-sdk 的 verify 方法会自动处理 sign 的移除
             # 但为了保险，我们手动移除它，保留其他所有字段
             verify_data.pop('sign', None)
+            verify_data.pop('sign_type', None)
             
             alipay = self.get_alipay_client()
             # 对于同步回调，sign_type 实际上是参与签名的（某些版本/接口）
@@ -76,3 +77,21 @@ class AlipayService:
         except Exception as e:
             logger.error(f"验证签名时发生异常: {str(e)}", exc_info=True)
             return False
+    
+    def query_order_status(self, out_trade_no):
+        """查询订单状态 - 主动查询支付宝获取真实支付状态"""
+        try:
+            alipay = self.get_alipay_client()
+            # 调用支付宝订单查询接口
+            result = alipay.api_alipay_trade_query(out_trade_no=out_trade_no)
+            
+            if result:
+                logger.info(f"订单查询成功: {out_trade_no}, 状态: {result.get('trade_status')}")
+                return result
+            else:
+                logger.warning(f"订单查询返回为空: {out_trade_no}")
+                return None
+                
+        except Exception as e:
+            logger.error(f"查询订单状态异常: {str(e)}", exc_info=True)
+            return None

services/generation_service.py

@@ -1,5 +1,6 @@
 from config import Config
 from models import SystemDict, GenerationRecord, User, db
+from extensions import redis_client
 from services.logger import system_logger
 from services.task_service import process_image_generation, process_video_generation
 import requests
@@ -124,6 +125,8 @@ def start_async_image_task(app, user_id, payload, api_key, target_api, cost, mod
     log_msg = "用户发起验光单解读" if payload.get('prompt') == "解读验光单" else "用户发起生图任务"
     system_logger.info(log_msg, model=model_value, mode=mode)
     
+    redis_client.setex(f"task:{task_id}", 3600, json.dumps({"status": "queued", "message": "任务已提交，等待处理..."}))
+    
     threading.Thread(
         target=process_image_generation,
         args=(app, user_id, task_id, payload, api_key, target_api, cost, use_trial)
@@ -151,6 +154,8 @@ def start_async_video_task(app, user_id, payload, cost, model_value):
     
     system_logger.info("用户发起视频生成任务 (积分模式)", model=model_value, cost=cost)
     
+    redis_client.setex(f"task:{task_id}", 3600, json.dumps({"status": "queued", "message": "视频任务已提交，准备开始导演..."}))
+
     threading.Thread(
         target=process_video_generation,
         args=(app, user_id, task_id, payload, api_key, cost, True) # 视频目前默认为积分模式

services/task_service.py

@@ -93,6 +93,8 @@ def sync_images_background(app, record_id, raw_urls):
 def process_image_generation(app, user_id, task_id, payload, api_key, target_api, cost, use_trial=False):
     """异步执行图片生成并存入 Redis"""
     with app.app_context():
+        # 更新状态为处理中
+        redis_client.setex(f"task:{task_id}", 3600, json.dumps({"status": "processing", "message": "正如火如荼地绘制中..."}))
         try:
             headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
             # 使用较长的超时时间 (10分钟)，确保长耗时任务不被中断
@@ -217,9 +219,18 @@ def process_video_generation(app, user_id, internal_task_id, payload, api_key, c
                 raise Exception(f"未获取到远程任务 ID: {submit_result}")
 
             # 2. 轮询状态
+            redis_client.setex(f"task:{internal_task_id}", 3600, json.dumps({"status": "processing", "message": "视频生成中，请耐心等待..."}))
+            
             max_retries = 90 # 提升到 15 分钟
             video_url = None
             for i in range(max_retries):
+                # 更新进度 (伪进度或保持活跃)
+                if i % 2 == 0: # 每20秒更新一次心跳，防止被认为是死任务
+                    redis_client.setex(f"task:{internal_task_id}", 3600, json.dumps({
+                        "status": "processing", 
+                        "message": f"视频生成中 (已耗时 {i * 10} 秒)..." 
+                    }))
+                
                 time.sleep(10)
                 poll_url = Config.VIDEO_POLL_API.format(task_id=remote_task_id)
                 poll_resp = requests.get(poll_url, headers=headers, timeout=30)

static/js/main.js

@@ -373,6 +373,22 @@ function handleNewFiles(files) {
     const newFiles = Array.from(files).filter(f => f.type.startsWith('image/'));
     if (newFiles.length === 0) return;
 
+    // 检查分辨率并提示压缩
+    let largeFound = false;
+    Promise.all(newFiles.map(file => new Promise(resolve => {
+        const url = URL.createObjectURL(file);
+        const img = new Image();
+        img.onload = () => {
+            if (img.width > 2048 || img.height > 2048) largeFound = true;
+            URL.revokeObjectURL(url);
+            resolve();
+        };
+        img.onerror = resolve;
+        img.src = url;
+    }))).then(() => {
+        if (largeFound) showToast('检测到图片分辨率大于 2K，将为您自动压缩至 2K 分辨率', 'info');
+    });
+
     // 如果处于设置器模式，严格限制为 1 张
     if (isSetterActive) {
         if (newFiles.length > 0) {
@@ -505,16 +521,51 @@ document.getElementById('submitBtn').onclick = async () => {
     try {
         let image_data = [];
 
-        // 1. 将图片转换为 Base64
+        // 1. 将图片转换为 Base64 (并压缩过大图片)
         if (uploadedFiles.length > 0) {
             btnText.innerText = "正在准备图片数据...";
-            const readFileAsBase64 = (file) => new Promise((resolve, reject) => {
+            const processImageFile = (file) => new Promise((resolve, reject) => {
                 const reader = new FileReader();
-                reader.onload = () => resolve(reader.result);
+                reader.onload = (e) => {
+                    const img = new Image();
+                    img.onload = () => {
+                        const maxDim = 2048;
+                        let w = img.width;
+                        let h = img.height;
+
+                        if (w <= maxDim && h <= maxDim) {
+                            resolve(e.target.result);
+                            return;
+                        }
+
+                        if (w > h) {
+                            if (w > maxDim) {
+                                h = Math.round(h * (maxDim / w));
+                                w = maxDim;
+                            }
+                        } else {
+                            if (h > maxDim) {
+                                w = Math.round(w * (maxDim / h));
+                                h = maxDim;
+                            }
+                        }
+
+                        const canvas = document.createElement('canvas');
+                        canvas.width = w;
+                        canvas.height = h;
+                        const ctx = canvas.getContext('2d');
+                        ctx.drawImage(img, 0, 0, w, h);
+                        // 保持原格式，如果不是 png 则默认 jpeg (0.9 质量)
+                        const outType = file.type === 'image/png' ? 'image/png' : 'image/jpeg';
+                        resolve(canvas.toDataURL(outType, 0.9));
+                    };
+                    img.onerror = reject;
+                    img.src = e.target.result;
+                };
                 reader.onerror = reject;
                 reader.readAsDataURL(file);
             });
-            image_data = await Promise.all(uploadedFiles.map(f => readFileAsBase64(f)));
+            image_data = await Promise.all(uploadedFiles.map(f => processImageFile(f)));
         }
 
         // 2. 并行启动多个生成任务