Maas-Model-platform/backend/scripts/test_alipay_sandbox.py

"""
支付宝沙箱环境测试脚本

测试 alipay.trade.page.pay 接口
沙箱应用: 2088721092817692/9021000159623614
"""
import os
import sys
import base64
import json
import uuid
from datetime import datetime
from decimal import Decimal
from urllib.parse import urlencode, quote_plus

# 添加项目根目录到路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.backends import default_backend


# ==================== 沙箱配置 ====================
# 沙箱网关
SANDBOX_GATEWAY = "https://openapi-sandbox.dl.alipaydev.com/gateway.do"

# 沙箱应用 APP_ID（你提供的）
SANDBOX_APP_ID = "9021000122696995"

# 沙箱商户 PID
SANDBOX_SELLER_ID = "2088721004566557"

# 沙箱私钥（需要从支付宝开放平台沙箱获取）
# 请将你的沙箱私钥粘贴到这里（单行格式，不含头尾）
SANDBOX_PRIVATE_KEY = """
MIIEogIBAAKCAQEAtbYqPvCXlvLf0wHzMojixChBJwgIHd69lD2LN8N6RF+h/Zpmj1wwiuuoggJf7fHJ15xmaais716RnNOt8eJnISU9w5r9JI3SRE3AB7BmNyYDqav3ot+eQ8z7j5dOPHaC/+5x46nsnGLnBbVtlL4cqQzIrIuwEMyXJvtQSlR2v1Zkg5ZVh0jOnsLkDWE+vNCYc5RyO6v0XwrXEACb/CvtW0BTKqYHBdeSkwTUNioh2iFPr74IW8prpm4ib2sA2qRD7AvF9ihJLQkU73CavruucLJswK5CFanxQy9jhZI2OsveZT41BPAr1CeMpaU4SWvNtLIJh3o2l46A+RIJiJ77uwIDAQABAoIBAHfQlsiLgZjxqm4K0h4XLlcjJ9qhrCBt91kgv4RUa3FcEYcT4N5Ublmbi6+1+yT9EhcONGUk6GqpU6Ax33cnztKHKNOqhZItxWBuV9l6Edv5P98H2jBV/Jg+N/11SgwKJNS2l/gC2lY/zI7yE0BJnsJFAKjtLnnVPTFh8o4gZXBBWtdr7U1Gl/ENvesI7xEhSGlyXlJIMHeIe03jOgu0Z/Sy7TQePul2xD7iyqVpVpXWLc1LGIts+IW7CouuELv+QNe+EXJCBODIb9TCYN2EUMx8+poo7hYjqJPpoA4VS1zI0Z9b7HFNlZkuAbID5kxn/5sO36VINvgzGFGlyE8ckRECgYEA9v1KUUyWEVFMt/UOiuIy96A2f5OLzXcvQbHJrRiXyC9loW3RdXeCLK732jRKfGxTrq1VSmT/Y9hc+JAC7XYsmw+/YRVFLx0BIVouFIsOy7l6wCpblWja8e7BpoEE5/mJSDpqq+yMBp7PV/7O2yib85xabqkZ4yyQQKJPGXpfYsMCgYEAvFc5niYwfvJ3+S2mnhCJA9HXANFIlG7SZuolR5RpZIYCMnH+cLPRSpUkENTzcFRs9HpQfVQbbWZckH4KJWC3rsvrLwUoa13iYH9MPfamUFlbTq7K3LCW9mwkTYh2ejGeiepEhyOTaKNsY9TaQpAAGqBemmZUjBLxNxM4P3Ofg6kCgYAw+1hyuRKFyq7BSKsipetfqnlEYbl2/Oz1RVHurxLi02t/US0Z86JwRB7JOlePR1htKtjgURlI7s65SK4b87Sy57OXiEVZK5Jez+iVkGJVyqnqVDwnbE0Hb8cdwzZ63sT3+wXOpLG8WmBaXiEd23baICbbDVQh9mOBPe8b6WXjSwKBgBlZkCrGLx7XGejTCSsbRKXb95lnkeJQrEcn2s6wniLmmqMZjsqymUf1nP4a+40x/9xEHlNQH8Tw0yKrpEf1paOEVOpmQlN5NoAQgj2Q1j/YVx7XDfz4U8llMtmwtWmrBhDAFGswEaYy4OLrAlmVMj5jOXiEr3qPDiKAlsCLmnvJAoGAFYIAbeDpfqslnBWY2a9N5stOKS31E7XU5jM+DqoVEITjihGzpXaYvtZIq3w87lSxU9qeqzV8LMbCcB4sGckxZUBo07KisfF8xEpnMKBzlX48IZhTaUXbT3I31XB//edUKjNVSU37+3zh2LToQyMWg86wtcYVnOhrPXbzw7HvVQ8=
"""


def format_pem_key(key_content: str, key_type: str) -> str:
    """格式化 PEM 密钥"""
    key_content = key_content.strip()
    
    # 移除可能的头尾
    for prefix in ["-----BEGIN RSA PRIVATE KEY-----", "-----BEGIN PRIVATE KEY-----"]:
        key_content = key_content.replace(prefix, "")
    for suffix in ["-----END RSA PRIVATE KEY-----", "-----END PRIVATE KEY-----"]:
        key_content = key_content.replace(suffix, "")
    
    key_content = key_content.replace("\n", "").replace("\r", "").replace(" ", "")
    
    # 每64字符换行
    lines = [key_content[i:i+64] for i in range(0, len(key_content), 64)]
    formatted = "\n".join(lines)
    
    return f"-----BEGIN {key_type}-----\n{formatted}\n-----END {key_type}-----"


def load_private_key(key_content: str):
    """加载私钥"""
    # 尝试 PKCS8 格式
    pem_content = format_pem_key(key_content, "PRIVATE KEY")
    try:
        return serialization.load_pem_private_key(
            pem_content.encode(),
            password=None,
            backend=default_backend()
        )
    except Exception:
        pass
    
    # 尝试 RSA PRIVATE KEY 格式
    pem_content = format_pem_key(key_content, "RSA PRIVATE KEY")
    try:
        return serialization.load_pem_private_key(
            pem_content.encode(),
            password=None,
            backend=default_backend()
        )
    except Exception as e:
        print(f"加载私钥失败: {e}")
        return None


def build_sign_content(params: dict, exclude: list = None) -> str:
    """构建待签名字符串"""
    exclude = exclude or []
    filtered = {k: v for k, v in params.items() if v and k not in exclude}
    sorted_keys = sorted(filtered.keys())
    pairs = [f"{k}={filtered[k]}" for k in sorted_keys]
    return "&".join(pairs)


def sign_rsa2(params: dict, private_key) -> str:
    """RSA2 签名"""
    sign_content = build_sign_content(params)
    signature = private_key.sign(
        sign_content.encode("utf-8"),
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    return base64.b64encode(signature).decode("utf-8")


def generate_payment_url(
    out_trade_no: str,
    total_amount: str,
    subject: str,
    private_key
) -> str:
    """
    生成支付宝支付页面 URL
    
    调用接口: alipay.trade.page.pay
    """
    # 业务参数
    biz_content = {
        "out_trade_no": out_trade_no,
        "total_amount": total_amount,
        "subject": subject,
        "product_code": "FAST_INSTANT_TRADE_PAY"
    }
    
    # 公共参数
    params = {
        "app_id": SANDBOX_APP_ID,
        "method": "alipay.trade.page.pay",
        "format": "JSON",
        "charset": "utf-8",
        "sign_type": "RSA2",
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "version": "1.0",
        "biz_content": json.dumps(biz_content, ensure_ascii=False, separators=(",", ":"))
    }
    
    # 签名
    params["sign"] = sign_rsa2(params, private_key)
    
    # 构建 URL
    query_string = urlencode(params, quote_via=quote_plus)
    return f"{SANDBOX_GATEWAY}?{query_string}"


def main():
    print("=" * 60)
    print("支付宝沙箱环境测试 - alipay.trade.page.pay")
    print("=" * 60)
    
    # 检查私钥配置
    if "请在这里粘贴" in SANDBOX_PRIVATE_KEY:
        print("\n⚠️  请先配置沙箱私钥！")
        print("\n获取方式：")
        print("1. 登录支付宝开放平台: https://open.alipay.com/")
        print("2. 进入沙箱环境")
        print("3. 获取沙箱应用的私钥")
        print("4. 将私钥粘贴到本脚本的 SANDBOX_PRIVATE_KEY 变量中")
        print("\n或者，你可以使用现有的 .env 配置（如果已配置沙箱密钥）")
        
        # 尝试使用 .env 中的配置
        from dotenv import load_dotenv
        load_dotenv()
        
        env_private_key = os.getenv("ALIPAY_PRIVATE_KEY", "")
        env_app_id = os.getenv("ALIPAY_APP_ID", "")
        
        if env_private_key and env_app_id:
            print(f"\n检测到 .env 配置，APP_ID: {env_app_id}")
            use_env = input("是否使用 .env 中的配置进行测试？(y/n): ").strip().lower()
            if use_env == 'y':
                private_key = load_private_key(env_private_key)
                if private_key:
                    run_test(private_key, env_app_id)
                    return
        return
    
    # 加载私钥
    private_key = load_private_key(SANDBOX_PRIVATE_KEY)
    if not private_key:
        print("❌ 私钥加载失败，请检查私钥格式")
        return
    
    run_test(private_key, SANDBOX_APP_ID)


def run_test(private_key, app_id: str):
    """执行测试"""
    global SANDBOX_APP_ID
    SANDBOX_APP_ID = app_id
    
    # 生成唯一订单号
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    unique_id = uuid.uuid4().hex[:8].upper()
    out_trade_no = f"TEST{timestamp}{unique_id}"
    
    print(f"\n📋 测试参数:")
    print(f"   APP_ID: {SANDBOX_APP_ID}")
    print(f"   订单号: {out_trade_no}")
    print(f"   金额: 0.01 元")
    print(f"   商品: 测试商品")
    
    # 生成支付链接
    try:
        payment_url = generate_payment_url(
            out_trade_no=out_trade_no,
            total_amount="0.01",
            subject="测试商品",
            private_key=private_key
        )
        
        print(f"\n✅ 支付链接生成成功！")
        print(f"\n🔗 支付链接:")
        print("-" * 60)
        print(payment_url)
        print("-" * 60)
        
        print(f"\n📝 入参示例:")
        biz_content = {
            "out_trade_no": out_trade_no,
            "total_amount": 0.01,
            "subject": "测试商品",
            "product_code": "FAST_INSTANT_TRADE_PAY"
        }
        print(json.dumps({"biz_content": biz_content}, indent=2, ensure_ascii=False))
        
        print(f"\n💡 下一步:")
        print("   1. 复制上面的支付链接")
        print("   2. 在浏览器中打开")
        print("   3. 使用沙箱买家账号登录支付")
        print("   4. 沙箱买家账号可在支付宝开放平台沙箱环境中查看")
        
    except Exception as e:
        print(f"\n❌ 生成支付链接失败: {e}")
        import traceback
        traceback.print_exc()


if __name__ == "__main__":
    main()