diff --git a/video-gen-api/app/api/v1/payments.py b/video-gen-api/app/api/v1/payments.py index 122f5109..8939d2a0 100644 --- a/video-gen-api/app/api/v1/payments.py +++ b/video-gen-api/app/api/v1/payments.py @@ -16,9 +16,6 @@ from app.services.payment import ( verify_wechat_callback, verify_alipay_callback, process_payment_success_by_order_no, - _get_payment_configs, - _is_mock_mode, - _check_and_expire_order, ) router = APIRouter(prefix="/payments", tags=["payments"]) @@ -45,6 +42,7 @@ async def recharge( raise HTTPException(status_code=400, detail="不支持的支付方式") # Check if the selected payment method is enabled in admin config + from app.services.payment import _get_payment_configs, _is_mock_mode configs = await _get_payment_configs(db) if not _is_mock_mode(configs): enabled_key = f"payment_{req.method}_enabled" @@ -120,6 +118,7 @@ async def list_orders( db: AsyncSession = Depends(get_db), ): # Auto-expire stale pending orders before returning + from app.services.payment import _check_and_expire_order result = await db.execute( select(PaymentOrder) .where(PaymentOrder.user_id == current_user.id) diff --git a/video-gen-api/app/services/payment.py b/video-gen-api/app/services/payment.py index 869f5be7..c622fa22 100644 --- a/video-gen-api/app/services/payment.py +++ b/video-gen-api/app/services/payment.py @@ -2,8 +2,6 @@ import logging import os import certifi import ssl -import sys -import time as _time from datetime import datetime, timedelta # 尝试禁用 SSL 验证（用于解决证书问题） @@ -23,22 +21,10 @@ 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 -__all__ = [ - "create_recharge_order", - "verify_alipay_callback", - "verify_wechat_callback", - "process_payment_success_by_order_no", - "get_order", - "get_user_orders", - "expire_all_pending_orders", - "_get_payment_configs", - "_is_mock_mode", - "_check_and_expire_order", -] - # --------------------------------------------------------------------------- # 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) @@ -69,67 +55,20 @@ class DailyFileHandler(logging.FileHandler): self._file_handler.close() self.baseFilename = self._make_path() self._file_handler = logging.FileHandler( - self.baseFilename, mode=self.mode, encoding=self.encoding + self.baseFilename, mode="a", encoding=self.encoding ) self._file_handler.setFormatter(self.formatter) - # Delegate to underlying file handler - if self._file_handler: - self._file_handler.emit(record) - else: - super().emit(record) + self._current_date = date_str + self.stream = self._file_handler.stream + super().emit(record) -_daily_handler = DailyFileHandler(_log_dir) -_formatter = logging.Formatter( - "%(asctime)s [%(levelname)s] %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", -) -_daily_handler.setFormatter(_formatter) -logger.addHandler(_daily_handler) - -# --------------------------------------------------------------------------- -# Monkey patch alipay-sdk-python's WebUtils to fix bytes/str TypeError bug -# --------------------------------------------------------------------------- -# 这个问题是官方 SDK 的一个已知 bug：WebUtils.py 中错误地将 bytes 和 str 拼接 -_patched = False - - -def _patch_alipay_sdk(): - """Monkey patch alipay.aop.api.util.WebUtils to fix the TypeError bug""" - global _patched - if _patched: - return True - try: - from alipay.aop.api.util import WebUtils - if hasattr(WebUtils, 'do_post'): - original_do_post = WebUtils.do_post - - def patched_do_post(*args, **kwargs): - try: - return original_do_post(*args, **kwargs) - except TypeError as e: - error_str = str(e) - if 'bytes' in error_str and 'str' in error_str: - logger.warning( - "Alipay SDK WebUtils TypeError bug detected! " - "Returning empty string to avoid crash." - ) - return "" - raise - - WebUtils.do_post = patched_do_post - _patched = True - logger.info("Successfully patched alipay WebUtils.do_post") - return True - except ImportError: - pass # SDK 还没有导入 - except Exception as e: - logger.warning(f"Failed to patch alipay SDK: {e}") - return False - - -# 立即尝试 patch -_patch_alipay_sdk() +_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 @@ -148,10 +87,6 @@ async def _get_payment_configs(db: AsyncSession) -> dict[str, str]: return {c.key: c.value for c in result.scalars().all()} -def _is_mock_mode(configs: dict[str, str]) -> bool: - return configs.get("payment_mock", "false").lower() == "true" - - 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. @@ -188,16 +123,68 @@ async def expire_all_pending_orders(db: AsyncSession) -> int: f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}" ) if orders: - await db.commit() + await db.flush() return len(orders) -# --------------------------------------------------------------------------- -# Alipay client cache -# --------------------------------------------------------------------------- +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 +_alipay_web_utils_patched = False + + +def _patch_alipay_web_utils(): + """Monkey-patch alipay SDK WebUtils.do_post to fix Python 3 bytes/str TypeError. + + The SDK's do_post raises:: + + TypeError: can only concatenate str (not "bytes") to str + + when the HTTP response is non-2xx, because ``response.read()`` returns + bytes but is used directly in a str concatenation inside the SDK. + """ + global _alipay_web_utils_patched + if _alipay_web_utils_patched: + return + + import alipay.aop.api.util.WebUtils as _web_utils + + _original_do_post = _web_utils.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 TypeError as e: + err_str = str(e) + if "bytes" not in err_str and "str" not in err_str: + raise + + # SDK bug: response.read() returned bytes but was used in str concat. + # The original HTTP status is lost due to the TypeError; we raise a + # descriptive RuntimeError so the caller can handle it gracefully. + try: + from alipay.aop.api.util.WebUtils import THREAD_LOCAL + uuid = THREAD_LOCAL.uuid + except Exception: + uuid = "???" + raise RuntimeError( + f"[{uuid}] Alipay HTTP request failed (non-2xx response). " + f"The SDK raised a bytes/str TypeError. " + f"URL: {url}" + ) from e + + _web_utils.do_post = _patched_do_post + _alipay_web_utils_patched = True def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway: str = ""): @@ -217,6 +204,9 @@ def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway: ) return None + # Fix SDK's Python 3 bytes/str bug in WebUtils.do_post (once per process) + _patch_alipay_web_utils() + config = AlipayClientConfig() config.server_url = gateway or "https://openapi.alipay.com/gateway.do" config.app_id = app_id @@ -257,17 +247,32 @@ async def create_recharge_order( credits: float, price: float, label: str, - bonus_credits: float = 0, - method: str = "alipay", + bonus_credits: float = 0.0, + method: str = "wechat", ) -> PaymentOrder: - """Create a new pending payment order and call the payment gateway. - If mock mode is enabled, auto-approves. - Returns the PaymentOrder with qr_code (or None if mock). - """ - configs = await _get_payment_configs(db) - is_mock = _is_mock_mode(configs) - total_credits = credits + bonus_credits + """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, @@ -275,74 +280,60 @@ async def create_recharge_order( amount=price, credits=total_credits, payment_method=method, - status="pending" if not is_mock else "paid", - qr_url=None, + 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={is_mock}" + f"amount={price} credits={total_credits} method={method} mock={mock_mode}" ) - if is_mock: - # Mock mode: instantly credit user - await _process_payment_success(db, order, "mock_transaction_id") - await db.commit() - return order + 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("支付宝预下单失败，请检查配置或稍后重试") - # Real payment - if method == "alipay": - qr_url = _create_alipay_order(order, configs) - if qr_url: - order.qr_url = qr_url - await db.flush() - elif method == "wechat": - _create_wechat_order(order, configs) - # Wechat would get a qr_url too, but stubbed for now - - await db.commit() return order -async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool: - """Verify Wechat Pay callback signature. - Note: This is a stub implementation. - """ - configs = await _get_payment_configs(db) - wechat_api_key = configs.get("payment_wechat_api_key", "") - - if not wechat_api_key: - logger.warning("Wechat API key not configured, skipping signature verify") - return True - - try: - # TODO: Implement proper Wechat Pay signature verification - logger.info( - f"WECHAT_CALLBACK order_no={data.get('out_trade_no')} " - f"transaction_id={data.get('transaction_id')}" - ) - return True - except Exception: - logger.exception("Error verifying Wechat callback") - return False - - # --------------------------------------------------------------------------- -# Wechat – stub for now +# WeChat (stub) # --------------------------------------------------------------------------- def _create_wechat_order(order: PaymentOrder, db_configs: dict[str, str]) -> None: - """Create a Wechat Pay order. Stub for real integration.""" + """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") + logger.warning("WeChat payment config missing in database") return logger.info( - f"Wechat order created: mch_id={mch_id}, " + f"WeChat order created: mch_id={mch_id}, " f"order_no={order.order_no}, amount={order.amount}" ) @@ -383,11 +374,6 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str AlipayTradePrecreateResponse, ) - # 确保我们已经 patch 了 SDK - if not _patched: - if _patch_alipay_sdk(): - logger.info("Successfully patched alipay SDK on demand") - # 构造业务参数 model = AlipayTradePrecreateModel() model.out_trade_no = order.order_no @@ -416,19 +402,7 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str logger.warning(f"Failed to set notify_url: {e}") # 执行API调用 - try: - response_content = client.execute(request) - except TypeError as e: - error_str = str(e) - if 'bytes' in error_str and 'str' in error_str: - # 这是那个已知的 bug！尝试自己修复或者使用备选方案 - logger.error( - f"Alipay SDK bytes/str TypeError bug hit: order_no={order.order_no}" - ) - # 暂时返回 None，让前端提示失败 - return None - raise # 其他 TypeError 正常抛出 - + response_content = client.execute(request) if not response_content: logger.error(f"Alipay precreate failed: empty response, order_no={order.order_no}") return None @@ -448,13 +422,16 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str ) return None - except Exception as e: - # 处理 SDK 内部的 bytes/str 错误 - if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)): - logger.error( - f"Alipay SDK TypeError (bytes/str issue): order_no={order.order_no}, " - f"error={str(e)}" - ) + except RuntimeError as e: + # The patched WebUtils raises RuntimeError on non-2xx HTTP responses + # (the original SDK would have raised a confusing TypeError). This is + # expected — the Alipay gateway rejected the request for some reason. + logger.warning( + f"Alipay precreate HTTP error: order_no={order.order_no}, " + f"detail={str(e)}" + ) + return None + except Exception: logger.exception(f"Alipay precreate exception: order_no={order.order_no}") return None @@ -466,99 +443,128 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool: """Verify Alipay payment callback (async notify) signature. - Note: This is a simplified implementation. In production, you should - verify using the SDK's signature verification or by checking against - Alipay's public key. + + Reads the Alipay public key from the database and uses the SDK's + built-in RSA2 verification. """ - configs = await _get_payment_configs(db) - alipay_public_key = configs.get("payment_alipay_public_key", "") - - if not alipay_public_key: - logger.warning("Alipay public key not configured, skipping signature verify") + 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: - # This is a simplified check – in production, use SDK verification - # For alipay-sdk-python, you'd typically use the DefaultAlipayClient verify - from alipay.aop.api.util.SignatureUtils import SignatureUtils - - # Remove sign/sign_type from data to verify - verify_data = data.copy() - sign = verify_data.pop("sign", None) - sign_type = verify_data.pop("sign_type", None) - + sign = data.get("sign") if not sign: - logger.warning("No sign field in Alipay callback") + logger.warning("Alipay callback missing 'sign' field") return False - - # For now, just check that the callback has our order and trade status - # In production, implement proper RSA verification - logger.info( - f"ALIPAY_CALLBACK order_no={data.get('out_trade_no')} " - f"trade_no={data.get('trade_no')} status={data.get('trade_status')}" + + # 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()) ) - return True + + 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.warning("alipay-sdk-python not available, skipping signature verify") + logger.error("alipay-sdk-python not installed, skipping signature verification") return True except Exception: - logger.exception("Error verifying Alipay callback") + logger.exception("Alipay callback verification error") return False -async def _process_payment_success(db: AsyncSession, order: PaymentOrder, transaction_id: str): - """Internal: actually update order, add credits, etc. - Caller must ensure we are in a transaction. - """ - order.status = "paid" - order.transaction_id = transaction_id - await db.flush() - - await add_credits(db, order.user_id, order.credits, "recharge", order.id) - - logger.info( - f"PAYMENT_SUCCESS order_no={order.order_no} user={order.user_id} " - f"amount={order.amount} credits={order.credits} txn={transaction_id}" - ) +# --------------------------------------------------------------------------- +# WeChat callback verification (stub) +# --------------------------------------------------------------------------- -async def process_payment_success_by_order_no( - db: AsyncSession, - order_no: str, - transaction_id: str, -) -> PaymentOrder | None: - """Mark order as paid, grant credits, etc., by order number. - Used by payment callback endpoints. Transaction managed by caller. - """ +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.order_no == order_no) + select(PaymentOrder).where(PaymentOrder.id == order_id).limit(1) ) order = result.scalar_one_or_none() - if order is None: - logger.warning(f"PAYMENT_SUCCESS order not found: order_no={order_no}") - return None + if not order or order.status != "pending": + return - if order.status == "paid": - logger.info(f"PAYMENT_SUCCESS already processed: order_no={order_no}") - return order - - await _process_payment_success(db, order, transaction_id) - await db.commit() - return order - - -async def get_order(db: AsyncSession, order_no: str) -> PaymentOrder | None: - result = await db.execute( - select(PaymentOrder).where(PaymentOrder.order_no == order_no) + order.status = "paid" + order.paid_at = datetime.now() + await add_credits( + db, + order.user_id, + order.credits, + f"充值成功({order.credits}积分)", + related_id=order.id, ) - return result.scalar_one_or_none() + await db.flush() -async def get_user_orders(db: AsyncSession, user_id: str) -> list[PaymentOrder]: +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.user_id == user_id) - .order_by(PaymentOrder.created_at.desc()) + select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1) ) - return list(result.scalars().all()) + 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}" + )