chore: initialize tuhui repository

This commit is contained in:
Codex
2026-03-08 19:28:32 +08:00
commit ee10c46aae
189 changed files with 17754 additions and 0 deletions

View File

@@ -0,0 +1,10 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""易收米支付SDK"""
from .pay import create_payment
from .query import query_order
from .notify import PaymentNotify
__all__ = ['create_payment', 'query_order', 'PaymentNotify']

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1,69 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import hashlib
import aiohttp
class YsmPayApi:
"""易收米支付API工具类"""
@staticmethod
async def http_post(url, data):
"""
发送异步HTTP POST请求
Args:
url: 请求地址
data: 请求数据JSON字符串
Returns:
str: 响应内容
"""
headers = {
'Content-Type': 'application/json; charset=UTF-8',
'Accept': 'application/json',
'User-Agent': '*/*',
'Authorization': 'WECHATPAY2-SHA256-RSA2048 '
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, data=data, headers=headers, ssl=False, timeout=30) as response:
return await response.text()
except Exception as e:
print(f"Error: {str(e)}")
return None
@staticmethod
def hash_sign(data, secret):
"""
生成签名
Args:
data: 要签名的数据字典
secret: 密钥
Returns:
str: 签名字符串
"""
# 按键名升序排序
sorted_data = dict(sorted(data.items()))
# 构造签名字符串
sign_str = ""
for key, value in sorted_data.items():
# 跳过hash字段和空值
if key == 'hash' or value is None or value == '':
continue
# 添加分隔符
if sign_str:
sign_str += '&'
# 拼接键值对
sign_str += f"{key}={value}"
# 添加密钥并计算SHA256哈希
sign_str += secret
return hashlib.sha256(sign_str.encode('utf-8')).hexdigest()

View File

@@ -0,0 +1,100 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
支付成功异步回调处理模块
当用户支付成功后,支付平台会把订单支付信息异步请求到回调接口
"""
import json
import asyncio
from .api import YsmPayApi
class PaymentNotify:
"""支付通知处理类"""
def __init__(self, appid, appsecret):
"""
初始化
Args:
appid: 支付通道ID
appsecret: 密钥
"""
self.appid = appid
self.appsecret = appsecret
async def process(self, request_data):
"""
处理支付通知(异步)
Args:
request_data: 请求数据(字符串或字典)
Returns:
tuple: (是否成功, 消息)
"""
# 解析请求数据
if isinstance(request_data, str):
try:
data = json.loads(request_data)
except:
return False, "数据格式错误"
else:
data = request_data
# 验证签名
calc_hash = YsmPayApi.hash_sign(data, self.appsecret)
if data.get('hash') != calc_hash:
return False, "验签失败,签名错误"
# 获取商户订单号
mch_orderid = data.get('mch_orderid')
# 处理支付结果
if data.get('state') == 'SUCCESS':
# 在这里处理订单业务逻辑
# 注意:平台可能会多次调用本接口,请确保订单不会被重复处理
# 示例:订单处理逻辑,这里可以使用异步数据库操作
"""
order = await find_order_by_id(mch_orderid)
if order and order.status != 'paid':
await update_order_status(mch_orderid, 'paid')
# 其他业务逻辑...
"""
return True, "success"
else:
# 处理未支付的情况
return False, "订单未支付"
# FastAPI Web应用示例
"""
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post("/notify")
async def payment_notify(request: Request):
# 配置信息
appid = '20********' # 支付通道ID
appsecret = 'e605ac7******************4af5164' # AppSecret
# 创建通知处理器
notify_handler = PaymentNotify(appid, appsecret)
# 获取请求数据
request_data = await request.body()
# 处理通知
success, message = await notify_handler.process(request_data)
if success:
return Response(content=message)
else:
return JSONResponse(content={"error": message}, status_code=400)
# 启动命令: uvicorn notify:app --reload
"""

100
backend/app/ysm_sdk/pay.py Normal file
View File

@@ -0,0 +1,100 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
发起支付模块
"""
import json
import time
import random
import string
import asyncio
from .api import YsmPayApi
def generate_nonce_str(length=10):
"""生成随机字符串"""
chars = string.ascii_letters + string.digits
return ''.join(random.choice(chars) for _ in range(length))
async def create_payment(appid, appsecret, order_id, description, amount,
notify_url, nopay_url, callback_url, pay_type=1):
"""
创建支付请求(异步)
Args:
appid: 支付通道ID
appsecret: 密钥
order_id: 商户订单号
description: 订单描述
amount: 支付金额(单位:分)
notify_url: 异步回调地址
nopay_url: 未支付跳转地址
callback_url: 支付成功跳转地址
pay_type: 支付类型默认为1微信内
Returns:
str: 支付链接
"""
# 构造支付请求数据
data = {
'appid': appid, # 支付通道ID
'mch_orderid': order_id, # 商户网站订单号
'description': description, # 订单标题
'total': amount, # 订单金额(分)
'notify_url': notify_url, # 异步通知地址
'nopay_url': nopay_url, # 未支付跳转地址
'callback_url': callback_url, # 支付成功跳转地址
'time': int(time.time()), # 时间戳
'nonce_str': f"abc{int(time.time())}", # 随机字符串
'payType': pay_type, # 支付类型
}
# 生成签名
data['sign'] = YsmPayApi.hash_sign(data, appsecret)
# 发起支付请求
url = 'https://www.yishoumi.cn/u/payment' # 支付网关
response = await YsmPayApi.http_post(url, json.dumps(data))
# 解析响应
result = json.loads(response) if response else None
# 返回支付链接
if result and 'url' in result:
return result['url']
else:
# 返回错误信息以便调用者处理
if result and 'msg' in result:
raise Exception(f"支付创建失败: {result['msg']} (错误代码: {result.get('code', 'unknown')})")
else:
raise Exception("支付创建失败: 无响应或响应格式错误")
async def main():
# 示例使用
appid = 'YSMcd16b45d' # 支付通道ID
appsecret = '899850e778e8d2b53e4c4a4e88695688' # AppSecret
# 生成订单号
order_id = f"123321{int(time.time())}"
# 创建支付
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description="快充数据线",
amount=100, # 1元
notify_url="http://abc.com/notify.php",
nopay_url="http://abc.com/notify.php",
callback_url="http://abc.com/notify.php",
pay_type=1 # 微信内支付
)
if pay_url:
print(f"支付链接: {pay_url}")
else:
print("创建支付失败")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -0,0 +1,74 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
查询订单模块
"""
import json
import asyncio
from .api import YsmPayApi
async def query_order(appid, mch_orderid=None, ysm_orderid=None):
"""
查询订单状态(异步)
Args:
appid: 支付通道ID
mch_orderid: 商户订单号与ysm_orderid二选一
ysm_orderid: 易收米订单号与mch_orderid二选一
Returns:
dict: 订单信息字典,包含状态等信息
SUCCESS支付成功
REFUND转入退款
NOTPAY未支付
CLOSED已关闭
"""
# 检查参数
if not mch_orderid and not ysm_orderid:
raise ValueError("商户订单号和易收米订单号必须提供一个")
# 构造请求数据
data = {'appid': appid}
if mch_orderid:
data['mch_orderid'] = mch_orderid
elif ysm_orderid:
data['ysm_orderid'] = ysm_orderid
# 请求URL
url = 'https://www.yishoumi.cn/u/query' # 订单查询网关
try:
# 发起请求
response = await YsmPayApi.http_post(url, json.dumps(data))
# 解析响应
result = json.loads(response) if response else None
if not result:
raise Exception('服务器错误:' + str(response), 500)
return result
except Exception as e:
print(f"错误码:{getattr(e, 'code', 0)}, 错误信息:{str(e)}")
return None
async def main():
# 示例使用
appid = '20********' # 支付通道ID
# 查询订单
order_info = await query_order(
appid=appid,
mch_orderid='20230510505610', # 商户订单号
)
if order_info:
print(f"订单信息: {json.dumps(order_info, ensure_ascii=False, indent=2)}")
else:
print("查询订单失败")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -0,0 +1,157 @@
# Ysm-SDK-python
易收米支付Python SDK提供支付相关API的Python实现。
## 安装依赖
```bash
pip install -r requirements.txt
```
或者手动安装:
```bash
pip install aiohttp fastapi uvicorn jinja2 python-multipart
```
## 文件说明
- `api.py` - 基础API工具类提供HTTP请求和签名方法
- `pay.py` - 发起支付模块
- `query.py` - 订单查询模块
- `notify.py` - 支付回调处理模块
- `demo.py` - 命令行演示脚本
- `web_demo.py` - Web服务演示
- `requirements.txt` - 依赖包列表
## 使用示例
### 发起支付
```python
from pay import create_payment
# 配置信息
appid = '20********' # 支付通道ID
appsecret = 'e605ac7******************4af5164' # AppSecret
# 创建支付
pay_url = create_payment(
appid=appid,
appsecret=appsecret,
order_id="123321123321",
description="快充数据线",
amount=100, # 1元
notify_url="http://example.com/notify",
nopay_url="http://example.com/cancel",
callback_url="http://example.com/success",
pay_type=1 # 微信内支付
)
print(f"支付链接: {pay_url}")
```
### 查询订单
```python
from query import query_order
# 配置信息
appid = '20********' # 支付通道ID
# 查询订单
order_info = query_order(
appid=appid,
mch_orderid='202305105056', # 商户订单号
)
print(f"订单信息: {order_info}")
```
### 处理支付回调
```python
from flask import Flask, request
from notify import PaymentNotify
app = Flask(__name__)
@app.route('/notify', methods=['POST'])
def payment_notify():
# 配置信息
appid = '20********' # 支付通道ID
appsecret = 'e605ac7******************4af5164' # AppSecret
# 创建通知处理器
notify_handler = PaymentNotify(appid, appsecret)
# 获取请求数据
request_data = request.get_data()
# 处理通知
success, message = notify_handler.process(request_data)
if success:
# 处理成功
return message
else:
# 处理失败
return {"error": message}, 400
if __name__ == '__main__':
app.run(debug=True)
```
## 支付类型说明
- `pay_type=1`: 微信内支付
- `pay_type=2`: 微信扫码支付
- `pay_type=3`: 微信H5支付(非原生)
- `pay_type=11`: 支付宝H5支付
## 演示程序
### 1. 命令行演示
运行完整的SDK功能演示
```bash
python demo.py
```
这个演示会展示:
- 创建支付订单
- 查询订单状态
- 处理支付回调
- 不同支付类型说明
### 2. Web服务演示
启动Web演示服务
```bash
python web_demo.py
```
然后访问 `http://localhost:8000` 查看:
- 支付表单页面
- 订单查询界面
- 支付成功/取消页面
- 支付回调处理接口
### 配置说明
在使用前请修改以下配置:
1.`demo.py` 中修改:
```python
self.appid = 'your_appid' # 替换为你的AppID
self.appsecret = 'your_appsecret' # 替换为你的AppSecret
```
2. 在 `web_demo.py` 中修改:
```python
APPID = 'your_appid'
APPSECRET = 'your_appsecret'
NOTIFY_URL = "http://your-domain.com/notify" # 替换为你的回调地址
```

View File

@@ -0,0 +1,210 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
易收米支付SDK使用演示
这个demo展示了如何使用易收米支付SDK的各个功能
1. 创建支付订单
2. 查询订单状态
3. 处理支付回调
运行前请确保安装依赖:
pip install aiohttp fastapi uvicorn
使用方法:
python demo.py
"""
import json
import time
import asyncio
from pay import create_payment
from query import query_order
from notify import PaymentNotify
class YsmPayDemo:
"""易收米支付演示类"""
def __init__(self):
# 配置信息 - 请替换为你的真实配置
self.appid = 'YSMcd16b45d' # 支付通道ID
self.appsecret = '899850e778e8d2b53e4c4a4e88695688' # AppSecret
# 回调地址配置
self.notify_url = "http://your-domain.com/notify"
self.nopay_url = "http://your-domain.com/cancel"
self.callback_url = "http://your-domain.com/success"
async def demo_create_payment(self):
"""演示创建支付订单"""
print("\n=== 创建支付订单演示 ===")
# 生成唯一订单号
order_id = f"demo_{int(time.time())}"
try:
# 创建支付
pay_url = await create_payment(
appid=self.appid,
appsecret=self.appsecret,
order_id=order_id,
description="演示商品 - 快充数据线",
amount=100, # 1元 = 100分
notify_url=self.notify_url,
nopay_url=self.nopay_url,
callback_url=self.callback_url,
pay_type=1 # 微信内支付
)
if pay_url:
print(f"✅ 支付订单创建成功")
print(f"📋 订单号: {order_id}")
print(f"💰 金额: 1.00元")
print(f"🔗 支付链接: {pay_url}")
return order_id, pay_url
else:
print("❌ 支付订单创建失败")
return None, None
except Exception as e:
print(f"❌ 创建支付时出错: {str(e)}")
return None, None
async def demo_query_order(self, order_id):
"""演示查询订单状态"""
print(f"\n=== 查询订单状态演示 ===")
print(f"🔍 查询订单: {order_id}")
try:
# 查询订单
order_info = await query_order(
appid=self.appid,
mch_orderid=order_id
)
if order_info:
print("✅ 订单查询成功")
print(f"📊 订单信息:")
print(json.dumps(order_info, ensure_ascii=False, indent=2))
# 解析订单状态
state = order_info.get('state', 'UNKNOWN')
status_map = {
'SUCCESS': '✅ 支付成功',
'REFUND': '🔄 转入退款',
'NOTPAY': '⏳ 未支付',
'CLOSED': '❌ 已关闭',
}
# 检查是否是错误响应
if 'code' in order_info and order_info.get('code') == 'ORDER_NOT_EXIST':
print(f"💡 订单状态: ⚠️ 订单不存在(演示订单,正常现象)")
else:
print(f"💡 订单状态: {status_map.get(state, f'未知状态({state})')}")
return order_info
else:
print("❌ 订单查询失败")
return None
except Exception as e:
print(f"❌ 查询订单时出错: {str(e)}")
return None
async def demo_payment_notify(self):
"""演示支付回调处理"""
print(f"\n=== 支付回调处理演示 ===")
# 模拟支付成功的回调数据
mock_notify_data = {
"appid": self.appid,
"mch_orderid": "demo_1234567890",
"ysm_orderid": "YSM20240101123456",
"total": 100,
"state": "SUCCESS",
"time": int(time.time()),
"nonce_str": "abc123456"
}
# 添加签名
from api import YsmPayApi
mock_notify_data['hash'] = YsmPayApi.hash_sign(mock_notify_data, self.appsecret)
print(f"📨 模拟回调数据:")
print(json.dumps(mock_notify_data, ensure_ascii=False, indent=2))
# 创建通知处理器
notify_handler = PaymentNotify(self.appid, self.appsecret)
# 处理回调
try:
# 使用异步调用
success, message = await notify_handler.process(mock_notify_data)
if success:
print(f"✅ 回调处理成功: {message}")
else:
print(f"❌ 回调处理失败: {message}")
return success, message
except Exception as e:
print(f"❌ 处理回调时出错: {str(e)}")
return False, str(e)
def demo_payment_types(self):
"""演示不同支付类型"""
print(f"\n=== 支付类型说明 ===")
payment_types = {
1: "微信内支付 - 适用于在微信内打开的页面",
2: "微信扫码支付 - 生成二维码供用户扫码支付",
3: "微信H5支付 - 适用于手机浏览器(非微信内)",
11: "支付宝H5支付 - 适用于手机浏览器跳转支付宝"
}
for pay_type, description in payment_types.items():
print(f"💳 pay_type={pay_type}: {description}")
async def run_full_demo(self):
"""运行完整演示"""
print("🚀 易收米支付SDK演示开始")
print("=" * 50)
# 显示配置信息
print(f"📋 当前配置:")
print(f" AppID: {self.appid}")
print(f" AppSecret: {self.appsecret[:8]}***{self.appsecret[-8:]}")
# 1. 演示支付类型
self.demo_payment_types()
# 2. 创建支付订单
order_id, pay_url = await self.demo_create_payment()
if order_id:
# 3. 查询订单状态
await self.demo_query_order(order_id)
# 4. 演示回调处理
await self.demo_payment_notify()
print("\n" + "=" * 50)
print("✅ 演示完成")
# 提示用户
if pay_url:
print(f"\n💡 提示:")
print(f" 1. 可以访问支付链接进行测试: {pay_url}")
print(f" 2. 支付成功后可以再次查询订单状态")
print(f" 3. 实际使用时请替换为真实的回调地址")
async def main():
"""主函数"""
demo = YsmPayDemo()
await demo.run_full_demo()
if __name__ == "__main__":
# 运行演示
asyncio.run(main())

View File

@@ -0,0 +1,134 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
易收米支付商户信息说明
解释商户全称的显示机制和可控制的参数
"""
def explain_merchant_info():
"""解释商户信息显示机制"""
print("=" * 60)
print("易收米支付 - 商户信息显示机制说明")
print("=" * 60)
print("\n商户信息组成:")
print("1. 商户全称 - 由支付平台后台配置API无法修改")
print("2. 商品描述 - 可通过API的description参数控制")
print("3. 订单金额 - 可通过API的amount参数控制")
print("4. 订单号 - 可通过API的order_id参数控制")
print("\n无法通过API隐藏的信息:")
print("- 商户全称(需要在支付平台后台修改)")
print("- 支付平台Logo和名称")
print("- 基本的支付安全信息")
print("\n可以通过API控制的信息:")
print("- 商品描述description参数")
print("- 订单金额amount参数")
print("- 订单号格式order_id参数")
print("- 回调地址notify_url等参数")
print("\n商户全称修改方法:")
print("1. 登录易收米商户后台")
print("2. 进入商户信息设置")
print("3. 修改商户全称/商户简称")
print("4. 提交审核可能需要1-3个工作日")
print("5. 审核通过后生效")
print("\n当前可优化的设置:")
# 展示不同的description设置效果
descriptions = [
("支付", "最简洁"),
("订单支付", "通用描述"),
("服务费", "服务类"),
("充值", "充值类"),
("购买", "购买类"),
("费用", "费用类"),
("", "空描述(不推荐)")
]
print("\n可选的商品描述设置:")
for desc, note in descriptions:
print(f" description='{desc}' - {note}")
print("\n注意事项:")
print("- 商户全称是监管要求,必须显示真实商户信息")
print("- 不能完全隐藏商户信息,这违反支付规范")
print("- 只能通过合规方式优化显示内容")
print("- 建议设置简洁明了的商户名称")
def show_current_merchant_info():
"""显示当前使用的商户信息"""
print("\n" + "=" * 60)
print("当前商户配置信息")
print("=" * 60)
print("\n当前商户设置:")
print("AppID: YSMcd16b45d")
print("商户全称: 由易收米平台后台配置")
print("支付方式: 微信扫码支付")
print("\n可控制的参数示例:")
examples = [
{
"场景": "最简洁支付",
"description": "支付",
"amount": "1分钱",
"显示效果": "商户全称 + 支付 + 0.01元"
},
{
"场景": "服务支付",
"description": "服务费",
"amount": "1分钱",
"显示效果": "商户全称 + 服务费 + 0.01元"
},
{
"场景": "充值支付",
"description": "账户充值",
"amount": "1分钱",
"显示效果": "商户全称 + 账户充值 + 0.01元"
}
]
for example in examples:
print(f"\n{example['场景']}:")
print(f" 参数: description='{example['description']}'")
print(f" 效果: {example['显示效果']}")
def alternative_solutions():
"""提供替代解决方案"""
print("\n" + "=" * 60)
print("替代解决方案")
print("=" * 60)
print("\n🎯 如果需要更简洁的支付体验:")
print("1. 申请新的商户号,使用简短的商户名称")
print("2. 使用个人收款码(但功能有限)")
print("3. 集成其他支付平台如官方微信支付API")
print("4. 使用第三方聚合支付(商户名称可能更简洁)")
print("\n📱 微信官方支付API对比:")
print("- 微信官方API: 可以设置更详细的商户信息")
print("- 易收米等第三方: 商户信息由平台统一管理")
print("- 个人收款码: 显示个人姓名但无法API调用")
print("\n💼 商业建议:")
print("- 如果是正式商业用途,建议使用符合规范的商户全称")
print("- 如果是测试用途,当前配置已经足够简洁")
print("- 关注用户体验的同时要遵守支付规范")
if __name__ == "__main__":
explain_merchant_info()
show_current_merchant_info()
alternative_solutions()
print("\n" + "=" * 60)
print("总结")
print("=" * 60)
print("商户全称无法通过API完全隐藏这是支付行业的规范要求。")
print("但可以通过优化商品描述等方式让支付页面更简洁。")
print("如需修改商户全称,需要在支付平台后台操作。")
print("=" * 60)

View File

@@ -0,0 +1,249 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
隐私支付二维码生成器
生成微信扫码支付二维码,隐藏商户和商品信息
"""
import asyncio
import time
import qrcode
from pay import create_payment
async def create_private_payment():
"""创建隐私支付二维码"""
print("微信扫码支付 - 隐私模式")
print("=" * 40)
# 配置信息
appid = 'YSMcd16b45d'
appsecret = '899850e778e8d2b53e4c4a4e88695688'
# 生成订单号
order_id = f"pay_{int(time.time())}"
print(f"订单号: {order_id}")
print(f"金额: 0.01元")
print()
print("正在生成支付二维码...")
try:
# 创建扫码支付订单 - 使用简化的描述
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description="支付", # 简化商品描述
amount=1, # 1分钱
notify_url="http://localhost/notify", # 简化回调地址
nopay_url="http://localhost/cancel",
callback_url="http://localhost/success",
pay_type=2 # 微信扫码支付
)
if pay_url:
print("支付二维码生成成功!")
print()
# 生成二维码
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=8, # 稍微小一点的二维码
border=2,
)
qr.add_data(pay_url)
qr.make(fit=True)
# 创建二维码图片
qr_img = qr.make_image(fill_color="black", back_color="white")
# 保存二维码
qr_filename = f"private_qr_{order_id}.png"
qr_img.save(qr_filename)
print(f"二维码已生成: {qr_filename}")
print()
print("使用方法:")
print("1. 用微信扫描生成的二维码")
print("2. 完成0.01元支付")
print("3. 支付页面将显示最小化的商户信息")
print()
print("隐私设置:")
print("- 商品描述: 仅显示'支付'")
print("- 商户信息: 使用平台默认设置")
print("- 订单金额: 0.01元")
return order_id, pay_url, qr_filename
else:
print("支付订单创建失败")
return None, None, None
except Exception as e:
print(f"创建支付时出错: {str(e)}")
return None, None, None
async def create_custom_payment():
"""创建自定义支付二维码"""
print("\n" + "=" * 40)
print("自定义支付设置")
print("=" * 40)
# 让用户自定义设置
description_options = {
"1": "支付",
"2": "订单支付",
"3": "在线支付",
"4": "服务费用",
"5": "充值"
}
print("选择商品描述:")
for key, value in description_options.items():
print(f"{key}. {value}")
# 由于是演示,我们使用默认选项
choice = "1" # 默认选择"支付"
description = description_options.get(choice, "支付")
# 配置信息
appid = 'YSMcd16b45d'
appsecret = '899850e778e8d2b53e4c4a4e88695688'
# 生成订单号
order_id = f"custom_{int(time.time())}"
print(f"\n订单设置:")
print(f"订单号: {order_id}")
print(f"描述: {description}")
print(f"金额: 0.01元")
print("\n正在生成支付二维码...")
try:
# 创建支付订单
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description=description,
amount=1,
notify_url="http://localhost/notify",
nopay_url="http://localhost/cancel",
callback_url="http://localhost/success",
pay_type=2
)
if pay_url:
# 生成二维码
qr = qrcode.QRCode(version=1, box_size=8, border=2)
qr.add_data(pay_url)
qr.make(fit=True)
qr_img = qr.make_image(fill_color="black", back_color="white")
# 保存二维码
qr_filename = f"custom_qr_{order_id}.png"
qr_img.save(qr_filename)
print(f"自定义支付二维码已生成: {qr_filename}")
return order_id, pay_url, qr_filename
else:
print("支付订单创建失败")
return None, None, None
except Exception as e:
print(f"创建支付时出错: {str(e)}")
return None, None, None
async def batch_create_payments():
"""批量创建不同类型的支付二维码"""
print("批量生成支付二维码")
print("=" * 40)
payment_types = [
("支付", "simple"),
("订单", "order"),
("充值", "recharge"),
("服务", "service")
]
appid = 'YSMcd16b45d'
appsecret = '899850e778e8d2b53e4c4a4e88695688'
created_qrs = []
for desc, type_name in payment_types:
order_id = f"{type_name}_{int(time.time())}"
try:
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description=desc,
amount=1,
notify_url="http://localhost/notify",
nopay_url="http://localhost/cancel",
callback_url="http://localhost/success",
pay_type=2
)
if pay_url:
# 生成二维码
qr = qrcode.QRCode(version=1, box_size=6, border=2)
qr.add_data(pay_url)
qr.make(fit=True)
qr_img = qr.make_image(fill_color="black", back_color="white")
qr_filename = f"{type_name}_qr_{order_id}.png"
qr_img.save(qr_filename)
created_qrs.append({
'order_id': order_id,
'description': desc,
'filename': qr_filename,
'url': pay_url
})
print(f"{desc} - {qr_filename}")
except Exception as e:
print(f"✗ 创建{desc}支付失败: {str(e)}")
print(f"\n成功生成 {len(created_qrs)} 个支付二维码")
return created_qrs
async def main():
"""主函数"""
print("微信支付二维码生成器 - 隐私版")
print("支持隐藏商户信息和自定义商品描述")
print()
# 选择模式
print("选择生成模式:")
print("1. 隐私模式 - 最小化信息显示")
print("2. 自定义模式 - 自定义商品描述")
print("3. 批量模式 - 生成多种类型二维码")
# 默认使用隐私模式进行演示
mode = "1"
if mode == "1":
await create_private_payment()
elif mode == "2":
await create_custom_payment()
elif mode == "3":
await batch_create_payments()
print("\n" + "=" * 40)
print("提示:")
print("- 所有二维码都已保存到当前目录")
print("- 使用微信扫描任意二维码进行支付测试")
print("- 支付金额均为0.01元")
print("- 商户信息已最小化显示")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -0,0 +1,153 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
易收米支付二维码演示
生成微信扫码支付二维码,真实支付场景演示
"""
import asyncio
import time
import qrcode
from io import BytesIO
from PIL import Image
from pay import create_payment
async def create_qr_payment():
"""创建二维码支付"""
print("易收米微信扫码支付演示")
print("=" * 50)
# 配置信息
appid = 'YSMcd16b45d'
appsecret = '899850e778e8d2b53e4c4a4e88695688'
# 生成订单号
order_id = f"qr_{int(time.time())}"
print(f"订单号: {order_id}")
print(f"商品: 测试商品 - 数据线")
print(f"金额: 0.01元")
print(f"支付方式: 微信扫码支付")
print()
print("正在生成支付二维码...")
try:
# 创建扫码支付订单 - 使用 pay_type=2
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description="测试商品 - 数据线",
amount=1, # 1分钱
notify_url="http://your-domain.com/notify",
nopay_url="http://your-domain.com/cancel",
callback_url="http://your-domain.com/success",
pay_type=2 # 微信扫码支付
)
if pay_url:
print("支付订单创建成功!")
print(f"支付链接: {pay_url}")
print()
# 生成二维码
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(pay_url)
qr.make(fit=True)
# 创建二维码图片
qr_img = qr.make_image(fill_color="black", back_color="white")
# 保存二维码
qr_filename = f"payment_qr_{order_id}.png"
qr_img.save(qr_filename)
print(f"二维码已生成: {qr_filename}")
print()
print("使用方法:")
print("1. 打开微信扫一扫")
print(f"2. 扫描生成的二维码文件: {qr_filename}")
print("3. 完成0.01元支付")
print("4. 支付完成后可以查询订单状态")
print()
print("=" * 50)
print("二维码图片已保存到当前目录")
print("请使用微信扫描二维码文件进行支付")
print("=" * 50)
return order_id, pay_url, qr_filename
else:
print("支付订单创建失败")
return None, None, None
except Exception as e:
print(f"创建支付时出错: {str(e)}")
return None, None, None
async def check_payment_status(order_id):
"""检查支付状态"""
if not order_id:
return
print(f"\n检查订单状态: {order_id}")
from query import query_order
try:
# 查询订单状态
order_info = await query_order(
appid='YSMcd16b45d',
mch_orderid=order_id
)
if order_info:
state = order_info.get('state', 'UNKNOWN')
if state == 'SUCCESS':
print("支付状态: 支付成功! 🎉")
elif state == 'NOTPAY':
print("支付状态: 等待支付...")
elif 'code' in order_info and order_info.get('code') == 'ORDER_NOT_EXIST':
print("支付状态: 订单查询中...")
else:
print(f"支付状态: {state}")
print(f"详细信息: {order_info}")
else:
print("查询失败")
except Exception as e:
print(f"查询订单时出错: {str(e)}")
async def main():
"""主函数"""
# 生成支付二维码
order_id, pay_url, qr_file = await create_qr_payment()
if order_id:
# 等待用户扫码支付
input("\n按回车键查询支付状态...")
# 检查支付状态
await check_payment_status(order_id)
# 可以持续查询
while True:
check_again = input("\n是否再次查询订单状态? (y/n): ")
if check_again.lower() == 'y':
await check_payment_status(order_id)
else:
break
if __name__ == "__main__":
print("请确保已安装依赖: pip install qrcode[pil]")
print()
asyncio.run(main())

View File

@@ -0,0 +1,60 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
易收米支付SDK快速测试
简单的支付功能测试脚本
"""
import asyncio
import time
from pay import create_payment
async def quick_payment_test():
"""快速支付测试"""
print("易收米支付快速测试")
print("=" * 40)
# 配置信息 - 请替换为你的真实配置
appid = 'YSMcd16b45d'
appsecret = '899850e778e8d2b53e4c4a4e88695688'
# 生成测试订单号
order_id = f"test_{int(time.time())}"
print(f"创建测试订单: {order_id}")
print(f"金额: 0.01元")
print(f"支付类型: 微信内支付")
try:
# 创建支付订单
pay_url = await create_payment(
appid=appid,
appsecret=appsecret,
order_id=order_id,
description="测试商品",
amount=1, # 1分钱
notify_url="http://your-domain.com/notify",
nopay_url="http://your-domain.com/cancel",
callback_url="http://your-domain.com/success",
pay_type=1
)
if pay_url:
print("支付订单创建成功!")
print(f"支付链接: {pay_url}")
print()
print("使用说明:")
print(" 1. 复制上方链接到微信中打开")
print(" 2. 完成支付测试")
print(" 3. 可以使用query.py查询订单状态")
else:
print("支付订单创建失败")
except Exception as e:
print(f"创建支付时出错: {str(e)}")
if __name__ == "__main__":
asyncio.run(quick_payment_test())

View File

@@ -0,0 +1,5 @@
aiohttp>=3.8.0
fastapi>=0.68.0
uvicorn>=0.15.0
jinja2>=2.11.0
python-multipart>=0.0.5

View File

@@ -0,0 +1,444 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
易收米支付SDK Web服务演示
这个演示展示了如何在Web应用中集成易收米支付
1. 创建支付页面
2. 处理支付回调
3. 查询订单状态
运行前请安装依赖:
pip install fastapi uvicorn jinja2 python-multipart
运行方式:
python web_demo.py
然后访问: http://localhost:8000
"""
import json
import time
import asyncio
from fastapi import FastAPI, Request, Form, Response
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from pay import create_payment
from query import query_order
from notify import PaymentNotify
# 创建FastAPI应用
app = FastAPI(title="易收米支付演示", description="易收米支付SDK Web演示")
# 配置信息 - 请替换为你的真实配置
APPID = 'YSMcd16b45d'
APPSECRET = '899850e778e8d2b53e4c4a4e88695688'
# 回调地址 - 实际使用时应该是你的域名
NOTIFY_URL = "http://localhost:8000/notify"
NOPAY_URL = "http://localhost:8000/cancel"
CALLBACK_URL = "http://localhost:8000/success"
# 模板引擎
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
"""首页 - 支付表单"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>易收米支付演示</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; }
.form-group { margin: 15px 0; }
label { display: block; margin-bottom: 5px; font-weight: bold; }
input, select, textarea { width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 4px; }
button { background: #007bff; color: white; padding: 10px 20px; border: none; border-radius: 4px; cursor: pointer; }
button:hover { background: #0056b3; }
.result { margin: 20px 0; padding: 15px; background: #f8f9fa; border-radius: 4px; }
.success { border-left: 4px solid #28a745; }
.error { border-left: 4px solid #dc3545; }
</style>
</head>
<body>
<h1>🚀 易收米支付演示</h1>
<h2>创建支付订单</h2>
<form action="/create_payment" method="post">
<div class="form-group">
<label>商品名称:</label>
<input type="text" name="description" value="演示商品 - 快充数据线" required>
</div>
<div class="form-group">
<label>支付金额(元):</label>
<input type="number" name="amount" value="1.00" step="0.01" min="0.01" required>
</div>
<div class="form-group">
<label>支付类型:</label>
<select name="pay_type">
<option value="1">微信内支付</option>
<option value="2">微信扫码支付</option>
<option value="3">微信H5支付</option>
<option value="11">支付宝H5支付</option>
</select>
</div>
<button type="submit">💳 创建支付</button>
</form>
<h2>查询订单状态</h2>
<form action="/query_order" method="post">
<div class="form-group">
<label>订单号:</label>
<input type="text" name="order_id" placeholder="输入订单号" required>
</div>
<button type="submit">🔍 查询订单</button>
</form>
<h2>📚 接口说明</h2>
<ul>
<li><strong>支付回调:</strong> POST /notify</li>
<li><strong>支付成功:</strong> GET /success</li>
<li><strong>支付取消:</strong> GET /cancel</li>
<li><strong>订单列表:</strong> GET /orders</li>
</ul>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.post("/create_payment")
async def create_payment_endpoint(
description: str = Form(...),
amount: float = Form(...),
pay_type: int = Form(...)
):
"""创建支付订单"""
try:
# 生成订单号
order_id = f"demo_{int(time.time())}"
# 转换金额为分
amount_cents = int(amount * 100)
# 创建支付
pay_url = await create_payment(
appid=APPID,
appsecret=APPSECRET,
order_id=order_id,
description=description,
amount=amount_cents,
notify_url=NOTIFY_URL,
nopay_url=NOPAY_URL,
callback_url=CALLBACK_URL,
pay_type=pay_type
)
if pay_url:
# 保存订单信息(实际应用中应该保存到数据库)
order_info = {
'order_id': order_id,
'description': description,
'amount': amount,
'pay_type': pay_type,
'pay_url': pay_url,
'status': 'created',
'created_at': time.time()
}
result_html = f"""
<!DOCTYPE html>
<html>
<head>
<title>支付订单创建成功</title>
<meta charset="utf-8">
<style>
body {{ font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; }}
.success {{ background: #d4edda; border: 1px solid #c3e6cb; color: #155724; padding: 15px; border-radius: 4px; }}
.info {{ background: #f8f9fa; border: 1px solid #dee2e6; padding: 15px; border-radius: 4px; margin: 15px 0; }}
.button {{ display: inline-block; background: #007bff; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; margin: 5px; }}
.button:hover {{ background: #0056b3; }}
</style>
</head>
<body>
<h1>✅ 支付订单创建成功</h1>
<div class="success">
支付订单已成功创建!请点击下方链接进行支付。
</div>
<div class="info">
<strong>订单信息:</strong><br>
订单号: {order_id}<br>
商品: {description}<br>
金额: ¥{amount}<br>
支付类型: {pay_type}
</div>
<a href="{pay_url}" class="button" target="_blank">🚀 立即支付</a>
<a href="/" class="button">🏠 返回首页</a>
<a href="/query_order_page?order_id={order_id}" class="button">🔍 查询订单</a>
</body>
</html>
"""
return HTMLResponse(content=result_html)
else:
return JSONResponse(
content={"error": "创建支付失败"},
status_code=400
)
except Exception as e:
return JSONResponse(
content={"error": f"创建支付时出错: {str(e)}"},
status_code=500
)
@app.post("/query_order")
async def query_order_endpoint(order_id: str = Form(...)):
"""查询订单状态"""
try:
# 查询订单
order_info = await query_order(
appid=APPID,
mch_orderid=order_id
)
if order_info:
status_map = {
'SUCCESS': '✅ 支付成功',
'REFUND': '🔄 转入退款',
'NOTPAY': '⏳ 未支付',
'CLOSED': '❌ 已关闭',
}
state = order_info.get('state', 'UNKNOWN')
status_text = status_map.get(state, f'未知状态({state})')
result_html = f"""
<!DOCTYPE html>
<html>
<head>
<title>订单查询结果</title>
<meta charset="utf-8">
<style>
body {{ font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; }}
.info {{ background: #f8f9fa; border: 1px solid #dee2e6; padding: 15px; border-radius: 4px; margin: 15px 0; }}
.success {{ color: #155724; }}
.pending {{ color: #856404; }}
.failed {{ color: #721c24; }}
pre {{ background: #f8f9fa; padding: 10px; border-radius: 4px; overflow-x: auto; }}
.button {{ display: inline-block; background: #007bff; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; margin: 5px; }}
</style>
</head>
<body>
<h1>🔍 订单查询结果</h1>
<div class="info">
<strong>订单号:</strong> {order_id}<br>
<strong>状态:</strong> <span class="{'success' if state == 'SUCCESS' else 'pending' if state == 'NOTPAY' else 'failed'}">{status_text}</span>
</div>
<h3>📋 详细信息:</h3>
<pre>{json.dumps(order_info, ensure_ascii=False, indent=2)}</pre>
<a href="/" class="button">🏠 返回首页</a>
</body>
</html>
"""
return HTMLResponse(content=result_html)
else:
return JSONResponse(
content={"error": "订单不存在或查询失败"},
status_code=404
)
except Exception as e:
return JSONResponse(
content={"error": f"查询订单时出错: {str(e)}"},
status_code=500
)
@app.post("/notify")
async def payment_notify(request: Request):
"""支付成功回调处理"""
try:
# 获取请求数据
request_data = await request.body()
# 创建通知处理器
notify_handler = PaymentNotify(APPID, APPSECRET)
# 处理通知
success, message = await notify_handler.process(request_data)
if success:
# 这里应该更新数据库中的订单状态
print(f"💰 支付成功回调处理: {message}")
return Response(content=message)
else:
print(f"❌ 支付回调处理失败: {message}")
return JSONResponse(content={"error": message}, status_code=400)
except Exception as e:
print(f"❌ 处理支付回调时出错: {str(e)}")
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.get("/success")
async def payment_success(request: Request):
"""支付成功页面"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>支付成功</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; text-align: center; }
.success { background: #d4edda; border: 1px solid #c3e6cb; color: #155724; padding: 30px; border-radius: 4px; }
.button { display: inline-block; background: #007bff; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; margin: 10px; }
</style>
</head>
<body>
<div class="success">
<h1>🎉 支付成功!</h1>
<p>您的订单已支付成功,感谢您的购买!</p>
</div>
<a href="/" class="button">🏠 返回首页</a>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.get("/cancel")
async def payment_cancel(request: Request):
"""支付取消页面"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>支付取消</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; text-align: center; }
.warning { background: #fff3cd; border: 1px solid #ffeaa7; color: #856404; padding: 30px; border-radius: 4px; }
.button { display: inline-block; background: #007bff; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; margin: 10px; }
</style>
</head>
<body>
<div class="warning">
<h1>⚠️ 支付已取消</h1>
<p>您取消了支付,如有需要可以重新发起支付。</p>
</div>
<a href="/" class="button">🏠 返回首页</a>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.get("/query_order_page")
async def query_order_page(request: Request, order_id: str = None):
"""订单查询页面"""
if order_id:
# 直接查询订单
try:
order_info = await query_order(appid=APPID, mch_orderid=order_id)
if order_info:
status_map = {
'SUCCESS': '✅ 支付成功',
'REFUND': '🔄 转入退款',
'NOTPAY': '⏳ 未支付',
'CLOSED': '❌ 已关闭',
}
state = order_info.get('state', 'UNKNOWN')
status_text = status_map.get(state, f'未知状态({state})')
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>订单详情</title>
<meta charset="utf-8">
<style>
body {{ font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; }}
.info {{ background: #f8f9fa; border: 1px solid #dee2e6; padding: 15px; border-radius: 4px; margin: 15px 0; }}
pre {{ background: #f8f9fa; padding: 10px; border-radius: 4px; }}
.button {{ display: inline-block; background: #007bff; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; margin: 5px; }}
</style>
</head>
<body>
<h1>📋 订单详情</h1>
<div class="info">
<strong>订单号:</strong> {order_id}<br>
<strong>状态:</strong> {status_text}
</div>
<pre>{json.dumps(order_info, ensure_ascii=False, indent=2)}</pre>
<a href="/" class="button">🏠 返回首页</a>
</body>
</html>
"""
return HTMLResponse(content=html_content)
except:
pass
# 显示查询表单
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>订单查询</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; }
.form-group { margin: 15px 0; }
label { display: block; margin-bottom: 5px; font-weight: bold; }
input { width: 100%; padding: 8px; border: 1px solid #ddd; border-radius: 4px; }
button { background: #007bff; color: white; padding: 10px 20px; border: none; border-radius: 4px; cursor: pointer; }
</style>
</head>
<body>
<h1>🔍 订单查询</h1>
<form action="/query_order" method="post">
<div class="form-group">
<label>订单号:</label>
<input type="text" name="order_id" required>
</div>
<button type="submit">查询订单</button>
</form>
<a href="/">🏠 返回首页</a>
</body>
</html>
"""
return HTMLResponse(content=html_content)
if __name__ == "__main__":
import uvicorn
print("🚀 启动易收米支付演示服务...")
print("📱 访问地址: http://localhost:8000")
print("📋 配置信息:")
print(f" AppID: {APPID}")
print(f" AppSecret: {APPSECRET[:8]}***{APPSECRET[-8:]}")
print("=" * 50)
uvicorn.run(app, host="0.0.0.0", port=8000)