import logging import os import certifi import ssl from datetime import datetime, timedelta # 尝试禁用 SSL 验证（用于解决证书问题） try: _create_unverified_https_context = ssl._create_unverified_context except AttributeError: pass else: ssl._create_default_https_context = _create_unverified_https_context # 猴子补丁修复 alipay-sdk-python 的 ResponseException 错误 # 修复 WebUtils.py 中 bytes 和 str 拼接的问题 def fix_alipay_web_utils(): try: import sys from alipay.aop.api.util import WebUtils # 保存原始的 do_post 函数 original_do_post = WebUtils.do_post def patched_do_post(url, query_string, headers, params, charset, timeout): try: return original_do_post(url, query_string, headers, params, charset, timeout) except Exception as e: # 如果是 ResponseException，尝试修复错误信息 error_msg = str(e) if "invalid http status" in error_msg or "ResponseException" in error_msg: # 重新实现一个更安全的版本 import urllib.request import urllib.error import urllib.parse import ssl import json import socket from alipay.aop.api.exception.ResponseException import ResponseException logger.warning("Alipay SDK ResponseException detected, using fallback request") # 构造请求 if query_string: full_url = url + "?" + query_string else: full_url = url # 准备 headers req_headers = {} if headers: for key, value in headers.items(): req_headers[key] = value req_headers["Content-type"] = "application/x-www-form-urlencoded;charset=" + charset req_headers["Connection"] = "Keep-Alive" req_headers["Cache-Control"] = "no-cache" req_headers["User-Agent"] = "Mozilla/4.0 (compatible; MSIE 6.0; Windows XP; SV1)" # 准备 post data post_data = "" if params: post_params = [] for key, value in params.items(): post_params.append( "%s=%s" % (key, urllib.parse.quote(str(value), encoding=charset)) ) post_data = "&".join(post_params) logger.debug(f"Request URL: {full_url}") logger.debug(f"Request data: {post_data[:200]}...") # 发送请求 try: req = urllib.request.Request(full_url, data=post_data.encode(charset) if post_data else None, headers=req_headers) # 创建不验证证书的 context ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE with urllib.request.urlopen(req, timeout=timeout, context=ctx) as response: response_body = response.read().decode(charset) logger.debug(f"Response: {response_body[:500]}...") return response_body except urllib.error.HTTPError as e: try: error_body = e.read().decode(charset) except: error_body = str(e) logger.error(f"Alipay HTTP error: {e.code}, body: {error_body}") raise ResponseException(str(e.code) + "," + error_body) except Exception as e2: logger.exception(f"Alipay request failed") raise else: raise # 应用补丁 WebUtils.do_post = patched_do_post logger.info("Alipay WebUtils monkey patch applied successfully") except ImportError: # alipay 模块还没安装，跳过 pass except Exception: logger.exception("Failed to apply Alipay WebUtils monkey patch") # 应用补丁 fix_alipay_web_utils() from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.models.payment_order import PaymentOrder from app.models.system_config import SystemConfig from app.services.credits import add_credits from app.utils.id_gen import generate_id, generate_order_no # --------------------------------------------------------------------------- # Payment logger → log/payment/YYYY-MM-DD.log (one file per day, no cleanup) # --------------------------------------------------------------------------- import time as _time logger = logging.getLogger("payment") logger.setLevel(logging.INFO) _log_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "log", "payment") os.makedirs(_log_dir, exist_ok=True) class DailyFileHandler(logging.FileHandler): """Write to a file named by date, e.g. log/payment/2026-06-10.log""" def __init__(self, directory, encoding="utf-8"): self._directory = directory self._current_date = "" self._file_handler = None super().__init__(self._make_path(), mode="a", encoding=encoding, delay=False) def _make_path(self): date_str = _time.strftime("%Y-%m-%d") self._current_date = date_str return os.path.join(self._directory, f"{date_str}.log") def emit(self, record): date_str = _time.strftime("%Y-%m-%d") if date_str != self._current_date: # Day rolled over — switch to a new file if self._file_handler: self._file_handler.close() self.baseFilename = self._make_path() self._file_handler = logging.FileHandler( self.baseFilename, mode="a", encoding=self.encoding ) self._file_handler.setFormatter(self.formatter) self._current_date = date_str self.stream = self._file_handler.stream super().emit(record) _handler = DailyFileHandler(_log_dir) _handler.setFormatter(logging.Formatter( "[%(asctime)s] %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S" )) if not logger.handlers: logger.addHandler(_handler) # Orders pending payment for longer than this are auto-cancelled ORDER_EXPIRE_MINUTES = 5 # --------------------------------------------------------------------------- # Config helpers – read from system_configs table (admin panel) # --------------------------------------------------------------------------- async def _get_payment_configs(db: AsyncSession) -> dict[str, str]: """Read all payment_* configs from the database, return as a dict.""" result = await db.execute( select(SystemConfig).where(SystemConfig.key.like("payment_%")) ) return {c.key: c.value for c in result.scalars().all()} async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool: """If a pending order has passed its expiry, mark it cancelled. Returns True if the order was expired. """ if order.status != "pending": return False expiry = order.created_at + timedelta(minutes=ORDER_EXPIRE_MINUTES) if datetime.now(order.created_at.tzinfo) >= expiry: order.status = "cancelled" await db.flush() logger.info( f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} " f"amount={order.amount} created_at={order.created_at.isoformat()}" ) return True return False async def expire_all_pending_orders(db: AsyncSession) -> int: """Background task: mark all expired pending orders as cancelled. Returns the number of orders expired. """ threshold = datetime.now() - timedelta(minutes=ORDER_EXPIRE_MINUTES) result = await db.execute( select(PaymentOrder).where( PaymentOrder.status == "pending", PaymentOrder.created_at <= threshold, ) ) orders = result.scalars().all() for o in orders: o.status = "cancelled" logger.info( f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}" ) if orders: await db.flush() return len(orders) def _is_mock_mode(db_configs: dict[str, str]) -> bool: """Check if payment mock mode is enabled (from DB or env).""" db_val = db_configs.get("payment_mock", "") if db_val: return db_val.lower() in ("true", "1", "yes") return settings.PAYMENT_MOCK # --------------------------------------------------------------------------- # Alipay client (lazy singleton, recreated when config changes) # --------------------------------------------------------------------------- _alipay_client = None _alipay_client_app_id = None def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway: str = "", notify_url: str = ""): """Get or create an Alipay client. Recreated if app_id changes.""" global _alipay_client, _alipay_client_app_id if _alipay_client is not None and _alipay_client_app_id == app_id: return _alipay_client try: from alipay.aop.api.AlipayClientConfig import AlipayClientConfig from alipay.aop.api.DefaultAlipayClient import DefaultAlipayClient except ImportError: logger.error( "alipay-sdk-python is not installed. " "Install it with: pip install alipay-sdk-python" ) return None config = AlipayClientConfig() config.server_url = gateway or "https://openapi.alipay.com/gateway.do" config.app_id = app_id config.app_private_key = private_key config.alipay_public_key = public_key config.sign_type = "RSA2" config.charset = "utf-8" config.notify_url = notify_url config.notify_type = "json" # 先尝试使用 certifi 证书 config.ca_certificates = certifi.where() try: _alipay_client = DefaultAlipayClient(config, logger) _alipay_client_app_id = app_id except Exception: logger.warning("Failed to initialize Alipay client with SSL verification, trying without verification...") # 如果初始化失败，尝试不验证 SSL 证书（通过不设置 ca_certificates） try: config.ca_certificates = None # 清空证书路径，跳过验证 _alipay_client = DefaultAlipayClient(config, logger) _alipay_client_app_id = app_id logger.warning("Alipay client initialized without SSL verification") except Exception: logger.exception("Failed to initialize Alipay client even without SSL verification") _alipay_client = None _alipay_client_app_id = None return _alipay_client # --------------------------------------------------------------------------- # Create recharge order # --------------------------------------------------------------------------- async def create_recharge_order( db: AsyncSession, user_id: str, credits: float, price: float, label: str, bonus_credits: float = 0.0, method: str = "wechat", ) -> PaymentOrder: """Create a payment order. Reads payment config from the database (admin panel). Returns the order; for Alipay the ``qr_url`` attribute will be populated with the scan-to-pay URL. """ # Read config from database first db_configs = await _get_payment_configs(db) mock_mode = _is_mock_mode(db_configs) # In real mode, validate that the payment method is enabled and configured if not mock_mode: enabled_key = f"payment_{method}_enabled" if db_configs.get(enabled_key, "").lower() != "true": raise ValueError("该支付方式未启用，请联系管理员") if method == "alipay": if not db_configs.get("payment_alipay_app_id") or not db_configs.get("payment_alipay_private_key"): raise ValueError("支付宝支付未完成配置，请联系管理员") elif method == "wechat": if not db_configs.get("payment_wechat_mch_id") or not db_configs.get("payment_wechat_api_key"): raise ValueError("微信支付未完成配置，请联系管理员") total_credits = credits + bonus_credits order = PaymentOrder( id=generate_id(), user_id=user_id, order_no=generate_order_no(), amount=price, credits=total_credits, payment_method=method, status="pending", ) db.add(order) await db.flush() logger.info( f"ORDER_CREATED order_no={order.order_no} user={user_id} " f"amount={price} credits={total_credits} method={method} mock={mock_mode}" ) if mock_mode: # Mock: immediately complete payment order.status = "paid" order.paid_at = datetime.now() desc = f"充值{label}({total_credits}积分)" if bonus_credits > 0: desc += f"(含赠送{bonus_credits}积分)" await add_credits( db, user_id, total_credits, desc, related_id=order.id, ) await db.flush() else: # Real payment: delegate to WeChat or Alipay if method == "wechat": _create_wechat_order(order, db_configs) elif method == "alipay": qr_url = _create_alipay_order(order, db_configs) if qr_url: # Attach QR URL to the order instance (transient, not persisted) order.qr_url = qr_url # type: ignore[attr-defined] else: # Precreate failed — do not leave a pending order that can never be paid raise ValueError("支付宝预下单失败，请检查配置或稍后重试") return order # --------------------------------------------------------------------------- # WeChat (stub) # --------------------------------------------------------------------------- def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> None: """Create a WeChat Pay order. Stub for real integration.""" mch_id = db_configs.get("payment_wechat_mch_id", "") api_key = db_configs.get("payment_wechat_api_key", "") if not mch_id or not api_key: logger.warning("WeChat payment config missing in database") return logger.info( f"WeChat order created: mch_id={mch_id}, " f"order_no={order.order_no}, amount={order.amount}" ) # --------------------------------------------------------------------------- # Alipay – trade.precreate (当面付 预下单) # --------------------------------------------------------------------------- def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str | None: """Call Alipay ``trade.precreate`` to obtain a QR code URL. Reads all Alipay config from the database (admin panel). Returns the ``qr_code`` URL on success, or ``None`` on failure. """ app_id = db_configs.get("payment_alipay_app_id", "") private_key = db_configs.get("payment_alipay_private_key", "") public_key = db_configs.get("payment_alipay_public_key", "") gateway = db_configs.get("payment_alipay_gateway", "") notify_url = db_configs.get("payment_alipay_notify_url", "") if not app_id or not private_key: logger.warning("Alipay config missing in database (app_id / private_key)") return None client = _get_alipay_client(app_id, private_key, public_key, gateway, notify_url) if client is None: return None try: from alipay.aop.api.domain.AlipayTradePrecreateModel import ( AlipayTradePrecreateModel, ) from alipay.aop.api.request.AlipayTradePrecreateRequest import ( AlipayTradePrecreateRequest, ) from alipay.aop.api.response.AlipayTradePrecreateResponse import ( AlipayTradePrecreateResponse, ) # 构造业务参数 model = AlipayTradePrecreateModel() model.out_trade_no = order.order_no model.total_amount = f"{order.amount:.2f}" model.subject = f"充值订单 {order.order_no}" model.product_code = "QR_CODE_OFFLINE" body_parts = [] if order.credits > 0: body_parts.append(f"{order.credits}积分") if body_parts: model.body = " ".join(body_parts) # 构造请求 request = AlipayTradePrecreateRequest(biz_model=model) # 执行API调用 logger.info(f"Calling Alipay trade.precreate for order: {order.order_no}") response_content = client.execute(request) if not response_content: logger.error(f"Alipay precreate failed: empty response, order_no={order.order_no}") return None logger.info(f"Alipay response received: {response_content[:500]}...") # 解析响应结果 response = AlipayTradePrecreateResponse() response.parse_response_content(response_content) if response.is_success(): qr_url = response.qr_code logger.info(f"Alipay precreate success, qr_url: {qr_url}") return qr_url else: logger.error( f"Alipay precreate failed: code={response.code}, " f"msg={response.msg}, sub_code={response.sub_code}, " f"sub_msg={response.sub_msg}, order_no={order.order_no}" ) return None except Exception as e: # 特殊处理 bytes 和 str 拼接的错误 error_str = str(e) if "can only concatenate str (not \"bytes\") to str" in error_str: logger.error(f"Alipay SDK bytes/str concat error, order_no={order.order_no}") # 尝试直接用 urllib 发送请求作为备选方案 return _fallback_alipay_request(order, db_configs) else: logger.exception(f"Alipay precreate exception: order_no={order.order_no}") return None def _fallback_alipay_request(order: PaymentOrder, db_configs: dict[str, str]) -> str | None: """ 备选方案：直接使用 urllib 发送支付宝请求，绕过 SDK 的 WebUtils """ try: import urllib.request import urllib.error import urllib.parse import ssl from alipay.aop.api.util.Signature import sign_with_rsa from alipay.aop.api.response.AlipayTradePrecreateResponse import ( AlipayTradePrecreateResponse, ) logger.info("Using fallback Alipay request method") app_id = db_configs.get("payment_alipay_app_id", "") private_key = db_configs.get("payment_alipay_private_key", "") gateway = db_configs.get("payment_alipay_gateway", "https://openapi.alipay.com/gateway.do") charset = "utf-8" # 构造请求参数 biz_content = { "out_trade_no": order.order_no, "total_amount": f"{order.amount:.2f}", "subject": f"充值订单 {order.order_no}", "product_code": "QR_CODE_OFFLINE" } params = { "app_id": app_id, "method": "alipay.trade.precreate", "format": "json", "charset": charset, "sign_type": "RSA2", "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "version": "1.0", "biz_content": str(biz_content).replace("'", "\"") } # 排序并签名 sorted_keys = sorted(params.keys()) sign_content = "&".join([f"{k}={params[k]}" for k in sorted_keys]) sign = sign_with_rsa(private_key.encode(charset), sign_content.encode(charset)) params["sign"] = sign # 构造请求 query_string = urllib.parse.urlencode(params) full_url = gateway # 创建不验证证书的 context ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # 发送请求 req = urllib.request.Request(full_url, data=query_string.encode(charset)) req.add_header("Content-type", f"application/x-www-form-urlencoded;charset={charset}") with urllib.request.urlopen(req, timeout=30, context=ctx) as response: response_body = response.read().decode(charset) logger.info(f"Fallback Alipay response: {response_body[:500]}...") # 解析响应 resp = AlipayTradePrecreateResponse() resp.parse_response_content(response_body) if resp.is_success(): return resp.qr_code else: logger.error( f"Fallback Alipay precreate failed: code={resp.code}, " f"msg={resp.msg}, sub_code={resp.sub_code}, " f"sub_msg={resp.sub_msg}" ) return None except Exception: logger.exception("Fallback Alipay request also failed") return None # --------------------------------------------------------------------------- # Alipay callback verification # --------------------------------------------------------------------------- async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool: """Verify Alipay payment callback (async notify) signature. Reads the Alipay public key from the database and uses the SDK's built-in RSA2 verification. """ db_configs = await _get_payment_configs(db) mock_mode = _is_mock_mode(db_configs) if mock_mode: return True public_key = db_configs.get("payment_alipay_public_key", "") if not public_key: logger.warning("ALIPAY_PUBLIC_KEY not found in database, cannot verify callback") return False try: sign = data.get("sign") if not sign: logger.warning("Alipay callback missing 'sign' field") return False # Build verification params (exclude sign and sign_type) verify_data = { k: v for k, v in data.items() if k not in ("sign", "sign_type") and v is not None and v != "" } from alipay.aop.api.util.Signature import verify_with_rsa sign_content = "&".join( f"{k}={v}" for k, v in sorted(verify_data.items()) ) is_valid = verify_with_rsa( public_key.encode("utf-8"), sign_content.encode("utf-8"), sign, ) if not is_valid: logger.warning("Alipay callback signature verification FAILED") return is_valid except ImportError: logger.error("alipay-sdk-python not installed, skipping signature verification") return True except Exception: logger.exception("Alipay callback verification error") return False # --------------------------------------------------------------------------- # WeChat callback verification (stub) # --------------------------------------------------------------------------- async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool: """Verify WeChat payment callback signature.""" db_configs = await _get_payment_configs(db) mock_mode = _is_mock_mode(db_configs) if mock_mode: return True logger.info("WeChat callback verification (real mode not implemented)") return True # --------------------------------------------------------------------------- # Process successful payment # --------------------------------------------------------------------------- async def process_payment_success(db: AsyncSession, order_id: str): """Process successful payment: update order and add credits.""" result = await db.execute( select(PaymentOrder).where(PaymentOrder.id == order_id).limit(1) ) order = result.scalar_one_or_none() if not order or order.status != "pending": return order.status = "paid" order.paid_at = datetime.now() await add_credits( db, order.user_id, order.credits, f"充值成功({order.credits}积分)", related_id=order.id, ) await db.flush() async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, trade_no: str = ""): """Process successful payment by order_no (used by Alipay/WeChat callbacks). Args: db: async database session order_no: the merchant order number (out_trade_no) trade_no: the Alipay trade number (trade_no), optional """ result = await db.execute( select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1) ) order = result.scalar_one_or_none() if not order or order.status != "pending": logger.info(f"Order {order_no} not found or already processed, skipping") return order.status = "paid" order.paid_at = datetime.now() if trade_no: order.trade_no = trade_no await add_credits( db, order.user_id, order.credits, f"充值成功({order.credits}积分)", related_id=order.id, ) await db.flush() logger.info( f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} " f"amount={order.amount} credits={order.credits} trade_no={trade_no}" )