import logging from datetime import datetime 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 logger = logging.getLogger("videogen") # --------------------------------------------------------------------------- # 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()} 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 = ""): """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" try: _alipay_client = DefaultAlipayClient(config) _alipay_client_app_id = app_id except Exception: logger.exception("Failed to initialize Alipay client") _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() 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] 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", "") notify_url = db_configs.get("payment_alipay_notify_url", "") gateway = db_configs.get("payment_alipay_gateway", "") 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) if client is None: return None try: from alipay.aop.api.domain.AlipayTradePrecreateModel import ( AlipayTradePrecreateModel, ) from alipay.aop.api.request.AlipayTradePrecreateRequest import ( AlipayTradePrecreateRequest, ) model = AlipayTradePrecreateModel() model.out_trade_no = order.order_no model.total_amount = f"{order.amount:.2f}" model.subject = f"充值订单 {order.order_no}" body_parts = [] if order.credits > 0: body_parts.append(f"{order.credits}积分") if body_parts: model.body = " ".join(body_parts) request = AlipayTradePrecreateRequest() request.biz_model = model if notify_url: request.notify_url = notify_url response = client.execute(request) if response.code == "10000": qr_url = response.qr_code logger.info( f"Alipay precreate success: order_no={order.order_no}, " f"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: logger.exception(f"Alipay precreate exception: order_no={order.order_no}") 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 processed: order_no={order_no}, trade_no={trade_no}")