diff --git a/app.py b/app.py index ec5bd79..b4e25ba 100644 --- a/app.py +++ b/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() diff --git a/blueprints/payment.py b/blueprints/payment.py index e0317a3..8d06c8d 100644 --- a/blueprints/payment.py +++ b/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/', 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 \ No newline at end of file diff --git a/config.py b/config.py index 1421ee6..f5696c4 100644 --- a/config.py +++ b/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 # 是否使用沙箱环境 # 开发模式配置 diff --git a/services/alipay_service.py b/services/alipay_service.py index 1bf5e91..9229eab 100644 --- a/services/alipay_service.py +++ b/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 diff --git a/services/generation_service.py b/services/generation_service.py index 9a5380f..da5a347 100644 --- a/services/generation_service.py +++ b/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) # 视频目前默认为积分模式 diff --git a/services/task_service.py b/services/task_service.py index a216a66..0714eb7 100644 --- a/services/task_service.py +++ b/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) diff --git a/static/js/main.js b/static/js/main.js index ceb62ed..a9707cb 100644 --- a/static/js/main.js +++ b/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. 并行启动多个生成任务